Dex SDK for Python

Pending Channel messages

Client.get_channel_messages(flow_id, channel) returns typed ChannelMessage envelopes in FIFO order. Each envelope contains the decoded value and the UUIDv7 assigned by Dex. Client.delete_channel_message deletes only a still-pending message and raises ChannelMessageNotFoundError after consumption or another deletion.

RPC handlers can stage channel.delete(context, message_id). Declare the RPC as @rpc(is_transactional=True) when a missing message must abort its other writes. Attribute locks already select transactional execution, but Channel deletion itself does not.

Python SDK for Dex workflow engine

New user contracts

The rewrite targets Python 3.11+ and exposes strongly typed workflow contracts from dex. This phase includes definitions, attributes, channels, streams, waits, decisions, codecs, registry validation, synchronous client calls, and synchronous worker handlers. Python owns its gRPC Client and Worker transport; the shared Rust Core is used only for BlobCache.

from datetime import timedelta
from typing import Generator

import dex

counter = dex.Attribute("counter", int)
counters_by_region = dex.AttributeMap("counters-by-region", int)
progress = dex.Stream("progress", str, 10 * 1024 * 1024)

class Run(dex.Step[str]):
    def wait_for(
        self, context: dex.Context, input: str
    ) -> dex.Wait:
        return dex.Wait.until(
            dex.Timer.by_duration(timedelta(seconds=1))
        )

    def execute(
        self, context: dex.Context, input: str
    ) -> Generator[dex.StepOutput, None, dex.StepDecision]:
        yield progress.write(context, "running")
        yield dex.heartbeat({"phase": "running"})
        return dex.graceful_complete(input)

class CounterFlow(dex.Flow[str]):
    run = Run()

    def get_flow_type(self) -> str:
        return "Counter"

    def get_steps(self) -> dex.StepList[str]:
        return dex.StepList.start_step(self.run)

    def get_persistence_schema(self) -> dex.PersistenceSchema:
        return dex.PersistenceSchema.of(counter, counters_by_region, progress)

    @dex.rpc(name="Increment")
    def increment(
        self, context: dex.Context, input: int
    ) -> dex.RPCResult[int]:
        return dex.RPCResult(input + 1)

flow = CounterFlow()
registry = dex.Registry((flow,))

Registry derives codecs from declared Python types and handler annotations. Built-in primitive types and dataclasses need no codec arguments. Register an explicit codec only for a custom encoding or a type Registry cannot derive. PersistenceSchema.of(...) accepts attributes, channels, and streams together and partitions them by definition type.

Worker and AsyncWorker synchronize all registered Indexed Attributes with Dex Server before opening their listener. Existing indexes return immediately; failure or the default two-minute deadline aborts startup. An indexed AttributeMap must provide one fixed index_key.

Initial attributes retain their value types without a public wrapper class:

options = (
    dex.StartFlowOptions()
    .with_attribute(counter, 1)
    .with_attribute(counters_by_region, "us-west", 1)
)

Opt in when declaring an Attribute or AttributeMap, and select the Store in Flow configuration:

email = dex.Attribute("customer-email", str, sync_to_attribute_store=True)
config = dex.FlowConfig(attribute_store_names=["profiles", "audit"])

Stores are asynchronous latest-state projections. Every enabled Attribute write is sent to every selected Store. Deletion writes SQL NULL, and projection failures do not roll back Flow Attributes. None preserves current targets; an explicit empty list disables future synchronization while retaining protocol presence.

pip install dex-python-sdk==0.1.0

See samples for use case examples.

Requirements

• Python 3.11+
• Dex server

Concepts

Applications implement two generic interfaces from dex:

• Flow[START_INPUT] returns StepList.start_step(...), followed by optional .other_steps(...), from one get_steps() method. The StepList generic binds the Flow input to the starting Step input. Use StepList.empty() when a Flow has no Steps.
• Step[INPUT] implements execute and optionally wait_for. The default Worker accepts ordinary synchronous handlers and generator handlers. A generator yields StepOutput progress frames and returns its final Wait or StepDecision. With AsyncWorker and Registry(..., allow_async_handlers=True), Step coroutines use AsyncContext; async RPCs keep Context.

StepOptions.wait_for_method_timeout and execute_method_timeout bound the two handler calls. Timer and channel conditions determine how long a Step waits.

wait_for_retry and execute_retry limit one logical handler execution. With StepDurability.ASYNC, local and fallback regular activities share maximum attempts, total duration, and 1-based attempt numbers. Fallback starts immediately; later regular retries continue the backoff sequence at the cumulative attempt.

The default Step durability is synchronous. A Flow configuration can select asynchronous durability, and a Step method override has highest precedence. The default retry total duration is four hours. Regular attempts default to a two-hour method timeout and one-minute heartbeat timeout; an explicit heartbeat timeout must meet the server minimum, which defaults to ten seconds. Asynchronous durability first allows at most three local attempts in seven seconds. The local phase ignores method timeouts and heartbeat frames before falling back to a regular activity.

Step progress and heartbeat recovery

A synchronous handler yields every heartbeat and Stream write. The generator return value is the only final result:

def execute(
    self, context: dex.Context, input: str
) -> Generator[dex.StepOutput, None, dex.StepDecision]:
    yield dex.heartbeat({"offset": 10})
    yield progress.write(context, "processed 10 items")
    return dex.graceful_complete(input)

An asynchronous handler keeps its normal coroutine return. Stream writes enqueue without waiting for Stream Store acknowledgement; heartbeat waits only for the Worker output queue:

async def execute(
    self, context: dex.AsyncContext, input: str
) -> dex.StepDecision:
    writer = progress.buffered_text(context)
    writer.write("started")
    await context.heartbeat({"offset": 10})
    return dex.graceful_complete(input)

The async buffered writer has a synchronous write method, so writer.write can be passed directly as an LLM delta callback. It flushes after one second, at a soft 16 KiB UTF-8 threshold, or before the final result or error. It concatenates chunks exactly and ignores empty chunks.

A synchronous generator uses the cooperative form because only yielded StepOutput values can reach gRPC:

writer = progress.buffered_text(context)
yield from writer.write(delta)
yield from writer.flush()

Its interval is checked by the next write, and the handler must explicitly flush the tail. Retry does not restore either writer's unsent buffer or deduplicate sent batches.

Call heartbeat() or await context.heartbeat() without a value to clear previous details. Passing Python None persists a present null Value. On a later regular attempt, use context.has_last_heartbeat_value() before decoding with context.get_last_heartbeat_value(ExpectedType). A Stream frame is also an implicit heartbeat, but it preserves the latest explicit heartbeat state.

A Step may write the same Stream any number of times. Dex assigns #<stepExecutionID> as each Step message's StreamMessage.source. Client writes provide their own non-empty source; duplicate sources and # are allowed and every write appends:

client.write_stream(flow_id, progress, "frontend#preview", "rendering")
message = client.read_stream(flow_id, progress)
print(message.source)

Canceling Step executions

A successful Step can cancel queued or active executions while continuing with its normal decision:

return (
    dex.go_to(RecordQuote, quote)
    .with_canceling_sibling_steps(QuoteCarrierA, QuoteCarrierB)
    .with_canceling_steps(GlobalQuoteTimeout)
)

with_canceling_steps selects every current execution of each registered Step type. with_canceling_sibling_steps selects only executions with the same Context.from_step_execution_id as the current execution. Decisions are immutable; repeated calls form a union, and Flow-wide selection wins for the same Step type. Unregistered selectors produce an invalid Step result.

Dex resolves one snapshot after the current execution succeeds. Completed, already-canceled, and absent targets are no-ops. Next Steps created by the same decision are outside the snapshot. Dex immediately applies the next or close action; late decisions, writes, retries, and recovery Steps are discarded.

RPCResult.with_canceling_steps provides the Flow-wide selector for RPCs. RPCs do not support sibling selection because they have no Step execution lineage.

Soft Flow timeout

Override Flow.handle_timeout to make a positive timeout use handler policy by default. Both synchronous and async Workers support the hook:

class Orders(dex.Flow[str]):
    async def handle_timeout(self, context: dex.Context) -> dex.StepDecision:
        await notify_expiration(context)
        return dex.force_complete("expired")

options = dex.StartFlowOptions(
    timeout=timedelta(minutes=30),
    timeout_policy=dex.FlowTimeoutPolicy.HANDLER,
)

Register async hooks with allow_async_handlers=True and run them with AsyncWorker. FAIL produces FlowErrorType.FLOW_TIMEOUT and permits Flow retry; CANCEL cancels without retry. Continue-as-new preserves the deadline, while retry runs receive a fresh budget. A zero or absent timeout disables the feature.

Registry validates every Flow, Step, RPC signature, durable name, lock, and codec before Client or Worker startup. Client methods use these typed objects instead of raw Flow, Step, or RPC strings.

Waiting and map inspection

Wait.all_of and Wait.any_of may use unnamed Conditions. Every Condition in Wait.any_combination_of must have a non-empty user ID; the same Condition instance may appear in multiple combinations.

Both Client and AsyncClient provide singleton and AttributeMap-instance overloads of wait_for_attribute_equal. They target the current run and accept only string, bool, int, or float wire values. JSON objects, bytes, and null fail before transport. AttributeMap.get_map_size/get_all_instance_keys include buffered sets and deletes. The matching ChannelMap methods are RPC-only, include buffered publishes, and omit empty instances. Keys are decoded and sorted. Use force_complete_if_channels_empty(...) for conditional completion.

Client.wait_for_flow and AsyncClient.wait_for_flow return a FlowResult after hydrating every output-bearing completion. Use single_output only when the Flow contract produces exactly one output:

output = client.wait_for_flow(flow_id).single_output(OrderResult)

result = client.wait_for_flow(flow_id)
for completion in result.completions:
    if completion.step_execution_id == expected_execution_id:
        output = completion.decode(OrderResult)

completions is an immutable tuple in server collection order. Parallel branch order is not deterministic, so select by step_type or step_execution_id. No-output Flows return an empty tuple; single_output raises ValueError for zero or multiple completions. Every terminal status returns a FlowResult; inspect status, error_type, and error_message for unsuccessful completion.

SubFlows are normal, independently addressable Flows used as durable Conditions:

def wait_for(self, context: Context, input: ChargeInput) -> Wait:
    return Wait.until(SubFlow.run(self.charge_flow, input))

def execute(self, context: Context, input: ChargeInput) -> StepDecision:
    del input
    receipt = SubFlow.get_condition_results(context).single_output(Receipt)
    return graceful_complete(receipt)

SubFlow.get_flow_id(context, index=0) remains available for a running any_of loser. SubFlowOptions configures timing, timeout policy, retry, initial target Attributes, Flow config, Condition ID, and reuse. Parent completion does not cancel an unfinished SubFlow.

Errors

Client calls raise concrete DexServiceError subclasses. Existing-Flow reads (get_attribute, describe_flow, wait_for_flow, and time_travel) raise FlowNotFoundError when the Flow does not exist. Mutations, RPCs, timer/Step waits, config updates, and continue-as-new triggers raise FlowNotActiveError when no running Flow can accept the operation.

try:
    client.publish(flow_id, orders.approved, order_id)
except dex.FlowNotActiveError:
    # The Flow is missing or already closed.
    pass

Duplicate starts, worker failures, RPC lock contention, and long-poll timeouts raise FlowAlreadyStartedError, WorkerInvocationError, RpcLockConflictError, and LongPollTimeoutError. All service errors retain code, sub_status, detail, operation, flow_id, and the original gRPC exception through Python exception chaining. Worker failures also expose worker_code, worker_error_type, and worker_error_detail. Registration, serialization, and invalid handler returns use FlowDefinitionError, ValueMappingError, and InvalidStepResultError.

Sync vs asyncio

• Sync (default): Client and Worker use blocking gRPC and a thread-pool Worker. A progress generator cooperatively hands each yielded frame to gRPC; Stream.write must therefore be yielded. Blocking Client calls inside Step.execute are safe while other pool threads remain available.
• Asyncio: AsyncClient and AsyncWorker use grpc.aio. Use Registry(..., allow_async_handlers=True) when Steps/RPCs are coroutines. Step coroutines annotate AsyncContext, call Stream.write without await, and await context.heartbeat. Inside async execute, inject AsyncClient — do not call sync Client on the Worker event loop. Async generators are rejected.

Integration scenarios live under tests/integ. They exercise the same workflows, client operations, and assertions as the Java suite against an isolated dexcli dev environment.

Implementation status

The strongly typed contracts, registry, synchronous Client/Worker, optional AsyncClient/AsyncWorker (grpc.aio), and Rust-backed BlobCache are implemented. Python owns its gRPC transport; the native bridge is limited to the shared BlobCache. Design notes: python-sdk-async-apis.md and python-sdk-step-streaming.md.

Running Dex locally

Install and start the complete local environment with dexcli:

brew install superdurable/tap/dexcli
dexcli dev

Dex Server listens on 127.0.0.1:8801. See the CLI README for endpoints and persistence options.

How To Contribute

This project uses uv for Python versions, dependencies, virtual environments, locking, building, and publishing.

To install requirements:

uv sync --locked

Run the complete Python SDK integration suite with an isolated Dex development environment:

./run-integration-tests.sh

Measure integration coverage

Run the same integration suite with Python source coverage:

./run-integration-tests.sh --coverage

Only the integration scenarios contribute execution data, and only production Python modules under dex are measured. Generated protobuf modules under dex/dexpb are excluded. The terminal report lists uncovered line ranges. The browser report starts at coverage/html/index.html; coverage/coverage.xml and coverage/lcov.info are also generated.

CI uploads LCOV to Codecov with GitHub OIDC under the sdk-python-integration flag and retains the full report as the sdk-python-integration-coverage Actions artifact.

Update IDL

Edit protos/dex.proto. Rename catalog: docs/design/idl-renames.md.

Generate stubs from IDL

make -C ../protos proto-python

Checked-in Python stubs land in dex/dexpb/.

Linting

Validate that every dex.__all__ class, function, constant, public method, argument, return value, dataclass field, enum value, and public instance attribute has a Google-style docstring:

uv run --frozen python scripts/check_public_docs.py

The checker resolves definitions from the public package export table, so private helpers and generated protobuf modules are excluded. Use help(dex.Client) or IDE hover information to read the same documentation. To run all other linting for this project:

uv run --frozen pre-commit run --show-diff-on-failure --color=always --all-files

Code of Conduct

This project is governed by the Contributor Covenant v 1.4.1. (Review the Code of Conduct and remove this sentence before publishing your project.)

Publishing to PyPI

1. Optionally run Publish Python SDK to PyPI via workflow_dispatch with a version and publish=false to validate all distributions without uploading.
2. Create a GitHub Release with tag sdk-python/vX.Y.Z (for example sdk-python/v0.1.0). CI stamps that version into pyproject.toml for the build (same idea as the TypeScript SDK release), then builds and smoke-tests Linux x86_64/ARM64, macOS x86_64/ARM64, and Windows x86_64 wheels, verifies the source distribution, and publishes with PYPI_TOKEN.
3. After publishing, bump the committed pyproject.toml / docs install line when you want the repo tip to reflect the released version.

A manual run publishes only from main, and only when publish is explicitly selected. The dispatch version input is stamped the same way as a release tag.

See CONTRIBUTING.md for monorepo tag conventions.

License

Super Durable Source License 1.0, with legacy portions under their original terms as described in LEGACY_NOTICES.md.