First-party Python SDK · 2.0 prerelease
Durable workflows in Python.¶
Define workflows and activities, run an async worker, and start durable work from Python. Begin with self-hosted Server or connect an existing managed Cloud namespace.
Python 3.10+Async-firstFully typed
Install¶
The versionless resolver reads the public quickstart contract and invokes pip with its qualified SDK identity.
How the pieces fit¶
A client asks the runtime to do durable work. A worker receives tasks and dispatches them to your workflow and activity code.
Choose who runs the runtime¶
The workflow types, activity types, and task queue stay the same. Only the endpoint, credentials, and operating boundary change.
Self-hosted Server
Run the published Server image locally, then execute the complete Python journey below.
Start locallyDurable Workflow Cloud
Connect the same program to a provisioned namespace with separate client and worker credentials.
Run your first local workflow¶
This source-free path resolves the compatibility-qualified Server image from the same public quickstart contract as the SDK installer, then runs one Python file containing an activity, workflow, worker, and client.
1. Start Server¶
Docker keeps this first run local. Resolve the Server image without copying a prerelease sequence number into the page:
export DW_SERVER_IMAGE="$(
curl -fsSL "${DURABLE_WORKFLOW_QUICKSTART_CONTRACT_URL:-https://durable-workflow.com/quickstart-execution-contract.json}" |
python -c 'import json, re, sys
contract = json.load(sys.stdin)
if contract.get("schema") != "durable-workflow.docs.v2.quickstart-execution-contract":
raise SystemExit("The public quickstart contract has an unsupported schema.")
server = contract.get("artifacts", {}).get("server", {})
version = server.get("version")
image = server.get("image")
reference = server.get("reference")
if not isinstance(version, str) or re.fullmatch(r"[0-9]+\.[0-9]+\.[0-9]+-(alpha|beta|rc)\.[0-9]+", version) is None:
raise SystemExit("The public quickstart contract has an invalid Server prerelease.")
if not isinstance(image, str) or reference != f"{image}:{version}":
raise SystemExit("The public quickstart contract has an invalid Server reference.")
print(reference)'
)"
Then bootstrap and start that qualified image:
export DURABLE_WORKFLOW_RUNTIME_URL='http://127.0.0.1:8080'
export DURABLE_WORKFLOW_RUNTIME_NAMESPACE='default'
export DURABLE_WORKFLOW_TOKEN='local-python-example-token'
docker volume create durable-workflow-python
docker run --rm -v durable-workflow-python:/app/database \
-e DW_AUTH_DRIVER=token -e DW_AUTH_TOKEN="$DURABLE_WORKFLOW_TOKEN" \
"$DW_SERVER_IMAGE" server-bootstrap
docker rm -f durable-workflow-python-server >/dev/null 2>&1 || true
docker run -d --name durable-workflow-python-server -p 8080:8080 \
-v durable-workflow-python:/app/database \
-e DW_AUTH_DRIVER=token -e DW_AUTH_TOKEN="$DURABLE_WORKFLOW_TOKEN" \
"$DW_SERVER_IMAGE"
until curl -sf http://127.0.0.1:8080/api/ready >/dev/null; do sleep 1; done
2. Save greeter.py¶
The named constants make the authoring contract visible: the decorator and start call share a workflow type, the workflow and decorator share an activity type, and the client and worker share one task queue. Values cross those boundaries with the supported Avro authoring codec.
import asyncio
import logging
import os
from uuid import uuid4
from durable_workflow import Client, Worker, activity, workflow
WORKFLOW_TYPE = "python.greeter"
ACTIVITY_TYPE = "python.greet"
TASK_QUEUE = "python-workers"
@activity.defn(name=ACTIVITY_TYPE)
def greet(name: str) -> str:
return f"Hello, {name}!"
@workflow.defn(name=WORKFLOW_TYPE)
class GreeterWorkflow:
def run(self, ctx, name):
return (yield ctx.schedule_activity(ACTIVITY_TYPE, [name]))
async def main() -> None:
logging.basicConfig(level=logging.INFO, format="%(message)s")
async with Client(
os.environ["DURABLE_WORKFLOW_RUNTIME_URL"],
token=os.getenv("DURABLE_WORKFLOW_TOKEN"),
control_token=os.getenv("DURABLE_WORKFLOW_CLIENT_TOKEN"),
worker_token=os.getenv("DURABLE_WORKFLOW_WORKER_TOKEN"),
namespace=os.environ["DURABLE_WORKFLOW_RUNTIME_NAMESPACE"],
) as client:
worker = Worker(
client,
task_queue=TASK_QUEUE,
workflows=[GreeterWorkflow],
activities=[greet],
)
handle = await client.start_workflow(
workflow_type=WORKFLOW_TYPE,
workflow_id=f"greeting-{uuid4().hex}",
task_queue=TASK_QUEUE,
input=["world"],
)
await worker.run_until(workflow_id=handle.workflow_id, timeout=30.0)
print(await handle.result(timeout=10.0))
asyncio.run(main())
3. Run it¶
Connect a managed Cloud namespace¶
Cloud provisioning returns a namespace-scoped runtime URL and namespace value.
Pass that complete runtime URL unchanged; do not invent or append an /api
suffix because the SDK adds its own routes.
Use each credential for one job¶
- Control-plane API keyCreates and administers Cloud resources and runtime credentials. It is not passed to the Python SDK runtime client.
- Runtime client tokenStarts and controls workflows in one namespace. Pass it through
DURABLE_WORKFLOW_CLIENT_TOKEN, which maps tocontrol_token=. - Runtime worker tokenRegisters, polls, heartbeats, and completes work in that namespace. Pass it through
DURABLE_WORKFLOW_WORKER_TOKEN, which maps toworker_token=.
Replace the placeholders with values returned for your namespace, then run the
same greeter.py. Keep client and worker tokens in their respective processes
when you split the example for production.
export DURABLE_WORKFLOW_RUNTIME_URL='<provisioned-runtime-url>'
export DURABLE_WORKFLOW_RUNTIME_NAMESPACE='<provisioned-runtime-namespace>'
export DURABLE_WORKFLOW_CLIENT_TOKEN='<runtime-client-token>'
export DURABLE_WORKFLOW_WORKER_TOKEN='<runtime-worker-token>'
unset DURABLE_WORKFLOW_TOKEN
python greeter.py
Continue building¶
Versioning¶
The SDK installer and Server image resolver both read the public quickstart contract. Neither command stores a release-candidate sequence number in this page. Lock the resolved package and Server image digest in your application when you need reproducible builds.