Python SDK

Write durable agents in Python: the duraton package authors workflows, runs an async runner over connect, calls step.ai, and talks to the REST API.

duraton is the Python runner SDK. It speaks the same connect protocol as the TypeScript SDK: a runner dials Duraton over a WebSocket, registers its workflows, and the engine drives each run one replay pass at a time. Durable steps memoize across passes, so a handler resumes exactly where it left off after a crash, deploy, sleep, or human approval.

duraton is a uv project, so add it to your project with uv add and run your runner with uv run:

uv add duraton
# or, with pip: pip install duraton

The default Anthropic provider for step.ai loads its SDK lazily, so install the extra only if you call a model through it:

uv add "duraton[anthropic]"
# or, with pip: pip install "duraton[anthropic]"

Requires Python 3.11 or newer. The runner is asyncio-native.

Define a workflow

A workflow is a name, its triggers, and an async handler that receives a StepContext. Author it with define_workflow:

from duraton import define_workflow
from duraton.context import StepContext


async def fulfill(ctx: StepContext) -> object:
    ticket = ctx.event.data
    triage = await ctx.step.run("triage", lambda: triage_ticket(ticket["subject"]))
    await ctx.step.sleep("cool-off", "30s")
    await ctx.step.emit("notify", name="ticket.triaged", data=triage)
    return {"ok": True}


fulfill_wf = define_workflow(
    "ticket.created",
    fulfill,
    triggers=[{"event": "ticket.created"}],
)

Prop

Type

The handler context

Every handler receives a StepContext: the triggering event, the durable step API, and the run's metadata.

Prop

Type

Steps

Every step takes a stable id, unique within the workflow. The result is recorded under that id, and on replay a completed step returns its saved result instead of running again. Each ctx.step method is awaitable.

# run: the unit of durable work. Pass an optional explicit input as the third argument.
triage = await ctx.step.run("triage", lambda: classify(subject))
reply = await ctx.step.run("reply", lambda: draft(ticket_id), {"ticket_id": ticket_id})

# suspend without holding a worker; survives a restart.
await ctx.step.sleep("cool-off", "30s")
await ctx.step.sleep_until("follow-up", datetime(2026, 8, 1))

# suspend until an event arrives, or None when the timeout elapses first.
reply = await ctx.step.wait_for_event("await-reply", event="ticket.replied", timeout="24h")

# invoke another workflow as a linked child run and await its result.
score = await ctx.step.run_workflow("risk", name="fraud.score", app="risk", data={"ticket_id": ticket_id})

# emit an event from inside a run.
await ctx.step.emit("shipped", name="ticket.shipped", app="notifications", data={"ticket_id": ticket_id})

Prop

Type

Approvals

ctx.step.approval parks the run in needs_attention - checkpoint kept, no worker held - until a human approves or denies it.

from duraton import ApprovalRequest

decision = await ctx.step.approval("refund-gate", ApprovalRequest(
    tool="issue-refund",
    args={"priority": "high"},
    risk="high",
    summary="Refund ticket A1 in full",
))
if decision.status == "approved":
    await ctx.step.run("refund", lambda: stripe_refund(decision.args["amount"]))

Running a runner

Runner(...).run() is the blocking entrypoint: it dials Duraton, registers, serves invokes, and reconnects with backoff until cancelled.

import asyncio
from duraton import Runner


async def main() -> None:
    # url / app / api_key fall back to DURATON_URL / DURATON_APP / DURATON_API_KEY.
    await Runner([fulfill_wf]).run()


asyncio.run(main())

Inside an existing event loop you have two non-blocking shapes:

# run in the background for the duration of the block.
async with Runner([fulfill_wf], app="support-app"):
    await do_other_work()

# or take a handle you can close explicitly.
from duraton import connect

handle = connect([fulfill_wf], app="support-app")
await handle.aclose()

Prop

Type

The ping/pong liveness and reconnect knobs (ping_interval_ms, pong_timeout_ms, reconnect_initial_ms, reconnect_max_ms) are also accepted here; see connect - Liveness for what they do and their defaults.

AI steps

ctx.step.ai makes a model call a durable step: each call takes a stable id, records its result, and returns the saved result on replay instead of calling the model again. Duraton stores only the AI metadata (model, token counts, latency) as an opaque journal block - never your prompt, the response text, or your API key.

from duraton import GenerateOptions

result = await ctx.step.ai.generate("draft-reply", GenerateOptions(
    model="claude-opus-4-8",
    prompt=f"Write a one-line apology for ticket {ticket_id}.",
))
result.text            # the response
result.usage.input_tokens, result.usage.output_tokens

Prop

Type

Structured output

Pass output (a JSON Schema) to constrain the model and get a parsed value on result.output. On a parse or validation failure, generate re-prompts with the error - each re-ask its own memoized step. Add validate for rules the schema can't express.

result = await ctx.step.ai.generate("triage", GenerateOptions(
    model="claude-opus-4-8",
    prompt=f"Triage: {subject}",
    output={
        "type": "object",
        "properties": {"category": {"type": "string"}, "priority": {"type": "string"}},
        "required": ["category", "priority"],
    },
    reask=2,
    validate=lambda v: None if v["priority"] in ("low", "normal", "high") else "priority out of range",
))
result.output   # the parsed, validated dict

Prop

Type

wrap, embed, loop

wrap makes a call you already write yourself durable, unchanged:

completion = await ctx.step.ai.wrap("classify", lambda: openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": subject}],
))

embed batches your embedding call, checkpointing each batch (Anthropic has no embeddings API, so you supply the call):

from duraton import EmbedOptions

res = await ctx.step.ai.embed("embed-kb", EmbedOptions(
    model="voyage-3",
    inputs=["duplicate charge policy", "annual plan refunds", "refund SLA"],
    embed=lambda batch: voyage.embed(batch),
    batch_size=2,
))
res.vectors   # one vector per input, in input ticket

loop is a durable agent loop: every turn and every tool call is its own durable step, so an agent that crashes mid-run resumes at the last committed turn. Bring your own model call in turn; the loop runs the tools it requests and feeds results into the next turn.

from duraton import LoopOptions, LoopContext, LoopTurn, LoopHandlerTool, LoopWorkflowTool

agent = await ctx.step.ai.loop("agent", LoopOptions(
    prompt=f"Resolve the ticket about: {subject}",
    max_iterations=6,
    tools={
        "search-kb": LoopHandlerTool(handler=lambda q: search_kb(q)),
        "lookup-ticket": LoopWorkflowTool(workflow="orders.lookup", app="orders"),
    },
    turn=lambda lctx, i: call_model(lctx.prompt, lctx.history, i),
))
agent.final, agent.iterations, agent.stop_reason

turn returns a LoopTurn - either tool_calls to run, or a final answer. Keep turn, validate, and stop deterministic: they run again on replay while the memoized model results do not.

Streaming (stream=True) and the inference cache are TypeScript-only for now - the Python GenerateOptions does not expose them, and Runner takes no cache. Everything else on this page - structured output, re-ask, fallback chains, wrap, embed, loop, infer, and the provider/cost seams - is at parity. See AI steps for the full behavior.

Providers

generate resolves its provider name through the built-in registry by default. Swap it with resolve_provider on the runner, or supply your own adapter:

from duraton import Runner, create_anthropic_provider, get_provider

await Runner(
    [fulfill_wf],
    resolve_provider=lambda name: get_provider(name),
).run()

The Anthropic adapter (create_anthropic_provider) loads anthropic lazily, which is why it is the optional duraton[anthropic] extra. The API key rides each call and is never stored, journaled, or sent to Duraton; omit it to fall back to ANTHROPIC_API_KEY.

REST client

duraton.client is a typed, async HTTP client for code outside a runner: trigger events, read and control runs, and read the numbers behind the console's charts. Use it as an async context manager so its connection pool is closed.

from duraton.client import AsyncDuratonClient, ListRunsOptions, SendEventInput

async with AsyncDuratonClient() as dx:
    res = await dx.events.send(SendEventInput(name="ticket.created", app="support-app", data={"ticketId": "T-421"}))
    res.run_id

    page = await dx.runs.list(ListRunsOptions(status="failed", limit=20))
    page.runs, page.next_cursor

    async for run in dx.runs.list_all(ListRunsOptions(app="support-app")):
        print(run.id, run.status)

    run = await dx.runs.get("01HXYZ...")
    stats = await dx.runs.stats()

Both url and api_key fall back to DURATON_URL / DURATON_API_KEY. The resources mirror the REST client surface: dx.events, dx.runs, dx.approvals, dx.datasets, dx.webhooks, dx.sessions, dx.ai, dx.runners, dx.workflows, dx.apps, plus dx.flow_state(), dx.health(), and dx.ready(). A non-2xx response raises DuratonAPIError carrying the status and body. Write methods (events.send, cancel, replay, score, ...) need a secret key; the GET-backed reads accept a public key.

On this page