memoflow-sdk 0.1.2

Durable workflow client and worker SDK for MemoFlow

memoflow-py

Public alpha. memoflow-sdk is the pure-Python remote client and worker for CPython 3.11-3.14. It supports authenticated TLS/mTLS transport and resumable stream reads; it does not include an embedded server and still has the bounded parity limits documented below. See the SDK strategy.

The MemoFlow Python SDK implements explicit-step workflows with checkpoint/memoization, speaking the same wire protocol as the TypeScript SDK: the ADR-0002 WorkerSession bidirectional stream plus the WorkflowService client surface.

Named explicit steps use the same identity algorithm: step keys use the same content-addressed algorithm (sha256(kind + "\n" + name)[:16] + ":" + occurrence). Cross-language resume is currently limited to a conservative JSON subset without TypeScript payload codecs or implicit HTTP call-site identities. The protocol conformance kit (packages/protocol-kit) contains golden vectors and a live explicit-step handoff scenario.

Published releases install as memoflow-sdk while the Python import remains memoflow:

python -m pip install "memoflow-sdk==0.1.2"

Inside activities, await activity_context().append_output(...) writes durable run output with per-stream ordering and a post-commit acknowledgement. Python uses the same deterministic run/stream/offset idempotency key as TypeScript, so identical retry output is safe while changed bytes at a committed offset fail loudly. This is separate from best-effort telemetry streaming.

Install (development)

python -m pip install --require-hashes -r requirements-codegen.txt
python -m pip install -e ".[dev,http]"

Protocol runtime modules and typed stubs are included in source and wheel distributions. Repository contributors regenerate every language from the root with pnpm proto:generate; package users do not need protoc. The first command installs the repository generator's exact transitive dependency set; application users installing the published package do not need that requirements file.

A workflow

from memoflow import workflow, run, activity, signal, query, Worker, WorkflowClient

@activity
class OrderActivities:
    async def chargePayment(self, order: dict) -> dict:
        ...  # normal Python — no determinism constraints

@workflow(task_queue="orders")
class OrderWorkflow:
    @run
    async def execute(self, order: dict) -> dict:
        validated = await self.ctx.activity("OrderActivities.validateOrder", [order],
                                            retry={"max_attempts": 3})
        payment = await self.ctx.activity("OrderActivities.chargePayment", [validated])
        shipment = await self.ctx.wait_for_signal("shipmentConfirmed", timeout="7d")
        return {"orderId": order["orderId"], "trackingId": shipment["trackingId"]}

Run a worker and a client:

worker = Worker(
    "localhost:7233",
    "orders",
    max_concurrent_workflows=10,
    max_concurrent_activities=50,
    ping_interval_ms=10_000,
)
await worker.run()

client = WorkflowClient("localhost:7233", namespace="default")
await client.start(workflow_id="o-1", workflow_type="OrderWorkflow",
                   task_queue="orders", input={...})
result = await client.get_result("o-1", timeout_ms=60_000)

Namespaces are always explicit: the client never silently selects a tenant. Every unary method accepts call=RpcCallOptions(timeout_s=..., max_attempts=...). Retry is opt-in, bounded to five attempts, and only follows a structured server error marked retryable; cancel and fork deliberately reject retry configuration because they are not retry-safe.

For a protected remote server, share one immutable connection policy between clients and workers:

from pathlib import Path
from memoflow import ConnectionConfig, TlsConfig, WorkflowClient, Worker

connection = ConnectionConfig(
    api_key="...",
    tls=TlsConfig(
        root_certificates=Path("ca.pem").read_bytes(),
        # Optional mTLS pair:
        private_key=Path("client-key.pem").read_bytes(),
        certificate_chain=Path("client-cert.pem").read_bytes(),
        server_name="memoflow.example.com",
    ),
    default_rpc_timeout_s=30,
    connect_timeout_s=10,
)
client = WorkflowClient("memoflow.example.com:7233", namespace="production", connection=connection)
worker = Worker("memoflow.example.com:7233", "orders", connection=connection)

Plaintext defaults to loopback only; non-loopback development targets require an explicit allow_insecure=True. Credentials are excluded from representations and unstructured transport errors. Channels set bounded message sizes, keepalive and SDK user-agent identity. Use async with WorkflowClient(...) or call await client.close(); worker session channels are owned and closed by the worker.

await worker.shutdown() stops new task admission, lets active handlers finish for up to 30 seconds, then cancels any remainder and closes the session.

Inside an activity, activity_context() exposes cooperative heartbeat/cancellation, record_llm_meta(...), bounded best-effort stream(...), and acknowledged durable append_output(...). Raise RateLimitError(message, retry_after) to carry a provider Retry-After hint into the server retry decision without changing attempt or non-retryable-error policy.

Postgres-tier workers may opt into atomic database activities with transactional_step=TransactionalStepConfig(connection_string=...). Inside the activity, await activity_context().transaction(fn) commits fn's writes and the MemoFlow COMPLETED checkpoint in the same database transaction only after the session handshake proves the targets match. A mismatch refuses before user code runs unless degrade="at-least-once" is explicit. Install the lazy driver only for this feature with pip install "memoflow-sdk[postgres]"; embedded, memory, missing-contract, and stale-contract servers fail loudly during worker startup.

AI provider calls can reserve server-authoritative capacity before any external effect:

from memoflow import AiBudgetEstimate, activity_context

ctx = activity_context()
budget = await ctx.reserve_ai_budget(AiBudgetEstimate(total_tokens=8_000, model_calls=1))
# Start provider work only after admission resolves, then persist its terminal invocation.
await budget.settle(invocation.invocation_id)

No policy snapshot crosses the SDK boundary, and settlement sends only reservation/invocation identity; the server derives actual usage from its terminal ledger. AiBudgetDeniedError means the provider effect must not start.

Administrative applications can use AiBudgetClient with typed AiBudgetPolicy, AiBudgetLimits, and AiBudgetLimit values to create/version, read and list namespace, workflow and run policies. Hierarchy overrides require an explicit reason before transport. The same client exposes list_reconciliations() and reconcile() for namespace-authorized operations. Every decision requires a stable UUID, timezone-aware timestamp, reason, and lowercase SHA-256 evidence digest; release and terminal-invocation settlement are distinct decisions.

Install memoflow[http] to use the typed retention control-plane client:

from datetime import datetime, timezone
from uuid import uuid4
from memoflow import RetentionClient

async with RetentionClient("https://control.example", api_key=api_key) as retention:
    review = await retention.preview_reconciliation(
        namespace="production", scope_kind="NAMESPACE", limit=1000
    )
    # Persist and review `review` before the explicit mutation call.
    await retention.apply_reconciliation(
        namespace="production", preview=review, audit_id=str(uuid4()),
        occurred_at=datetime.now(timezone.utc), reason="approved DATA-004 review",
    )

The same client manages versioned policies and legal holds and lists deletion jobs. It reports APPLIED versus DUPLICATE and rejects empty, truncated, or cross-namespace review artifacts before transport. Reuse the exact audit ID and timestamp when retrying an uncertain mutation.

Register schema upgrades with the process-wide codecs registry (or an isolated CodecRegistry). Outputs use the same { "$mfv": version, "value": ... } wrapper as TypeScript; unregistered steps remain byte-identical, legacy version-zero values upgrade in order, and newer-than-code payloads fail loudly. Install memoflow[telemetry] to connect the SDK to your application OpenTelemetry provider. Without it, propagation and spans are strict no-ops. With it, starts inject W3C context and workers create parented workflow/activity spans containing identifiers and attempt metadata, never payloads or credentials. Exception telemetry is fail-closed: the original exception is re-raised to application code, while spans receive only a synthetic error (or canonical numeric gRPC code) and a static redacted status. Messages, causes, tracebacks, payloads, prompts, provider responses, and credentials are never copied into the exported exception event.

The full port of the canonical order-workflow example lives in examples/order_workflow/.

The context API

method	semantics
ctx.activity(type, args, retry=..., ...)	checkpointed activity with at-least-once execution
ctx.llm(type, args, ...)	documentary sugar over activity for LLM calls
ctx.sleep("5m")	durable server-side timer
ctx.wait_for_signal(name, timeout="7d")	durable wait; timeout raises catchable SignalTimeoutError
ctx.approval(name, timeout=...)	human gate — sugar over wait_for_signal("approval:{name}")
ctx.side_effect(fn)	compute once, memoize forever

Durable steps park on never-settling futures and register commands on the context; the worker drains the event loop and suspends once with everything the slice scheduled — so asyncio.gather(...) fan-out dispatches every branch concurrently (the collect-then-suspend model, mirroring the TS SDK's execute.ts).

Non-determinism is LOUD: a step-key mismatch on a memoization hit raises NonDeterminismError, and the server parks the run as resumable BLOCKED_NONDETERMINISTIC (surfaced to clients as WorkflowBlockedError — see docs/DEPLOY-PLAYBOOK.md).

Testing

python -m ruff check .
python -m mypy
python -m pytest tests
# The wire-level acceptance gate — the protocol kit driven by this SDK:
MEMOFLOW_KIT_DRIVER="python $(pwd)/kit_driver.py" \
  pnpm --filter @memoflowio/memoflow-protocol-kit run test:kit

Implicit HTTP steps — what is and isn't intercepted

A bare HTTP call inside a workflow body is a durable step — same checkpoint schema, same memoization, same NonDeterminism detection as explicit steps (using the same checkpoint semantics as explicit steps). Step identity is the CALL SITE (module:lineno + occurrence), never the URL. Interception is scoped to workflow execution slices via contextvars: activity bodies and plain scripts are byte-identical pass-through.

transport	intercepted?	durability of the record
httpx.AsyncClient	yes	eager: the record is DurableAck'd before your code sees the response
httpx.Client (sync)	yes	batched with the slice's suspension (a sync client inside an async workflow also blocks the event loop — prefer AsyncClient)
aiohttp, requests, urllib	no — pass-through	not recorded; wrap in an activity if you need durability

Mutating implicit calls (non-GET/HEAD/OPTIONS) carry an auto-injected Idempotency-Key derived from the step's own identity (N2 — same algorithm as the TS SDK; a user-supplied header always wins), so upstreams can deduplicate the at-least-once window.

Activities can renew their heartbeat lease and cooperatively observe workflow cancellation through the task-local context:

from memoflow import activity_context

async def long_running_activity() -> None:
    ctx = activity_context()
    while more_work():
        await do_one_chunk()
        if ctx is not None:
            await ctx.heartbeat()
            ctx.throw_if_cancelled()

Opt-outs always exist: send the header x-memoflow-raw: 1 on a request to bypass recording (the header is stripped before the wire), or simply run the code outside a workflow slice (interception activates only inside slices).

Scope honesty

• Remote API-key, TLS/mTLS, bounded deadlines and explicit insecure-loopback configuration are supported. Python does not ship an embedded/native runtime.
• Query, durable output resume, schedules, debounce, approvals, history reveal and server-backed restart-from-step forks participate in the checked TypeScript/Python/Go client matrix.
• Worker workflow/activity concurrency is enforced per session; set either maximum to 0 to disable that capability. Set ping_interval_ms=0 to disable periodic workflow-task lease renewal; explicit activity heartbeats are still sent by activity_context().heartbeat().
• No ctx.transaction(); the retained same-Postgres v1 source is deliberately unadvertised until a fenced v2 server contract exists.
• No payload-codec registry: steps recorded with a registered TS codec are not cross-readable from Python; unregistered steps (the default) round-trip byte-compatibly.
• No debounced starts from Python yet.