Metadata-Version: 2.4
Name: memoflow-sdk
Version: 0.1.1
Summary: Durable workflow client and worker SDK for MemoFlow
License-Expression: MIT
Project-URL: Homepage, https://memoflow.io
Project-URL: Documentation, https://github.com/evigasoft/memoflow/tree/main/docs
Project-URL: Repository, https://github.com/evigasoft/memoflow
Project-URL: Changelog, https://github.com/evigasoft/memoflow/blob/main/packages/memoflow-py/CHANGELOG.md
Project-URL: Issues, https://github.com/evigasoft/memoflow/issues
Keywords: durable-execution,workflows,agents,llm
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: grpcio<2,>=1.81.1
Requires-Dist: protobuf<7,>=6.33.5
Provides-Extra: dev
Requires-Dist: build<2,>=1.3; extra == "dev"
Requires-Dist: grpcio-tools==1.81.1; extra == "dev"
Requires-Dist: mypy<3,>=1.19; extra == "dev"
Requires-Dist: mypy-protobuf==5.1.0; extra == "dev"
Requires-Dist: opentelemetry-sdk<2,>=1.28; extra == "dev"
Requires-Dist: packaging<27,>=24; extra == "dev"
Requires-Dist: pytest<10,>=9; extra == "dev"
Requires-Dist: ruff<1,>=0.14; extra == "dev"
Requires-Dist: types-protobuf==6.32.1.20260221; extra == "dev"
Provides-Extra: http
Requires-Dist: httpx>=0.27; extra == "http"
Provides-Extra: postgres
Requires-Dist: psycopg[binary]<4,>=3.2; extra == "postgres"
Provides-Extra: telemetry
Requires-Dist: opentelemetry-api<2,>=1.28; extra == "telemetry"

# memoflow-py

> **Experimental.** The Python SDK has a bounded authenticated TLS/mTLS connection layer and
> resumable stream reads, but still lacks full transaction, codec and embedded parity.
> See [the SDK strategy](../../docs/SDK-STRATEGY.md).

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](../protocol-kit/))
contains golden vectors and a live explicit-step handoff scenario.

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

```bash
python -m pip install memoflow-sdk
```

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)

```bash
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

```python
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:

```python
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:

```python
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:

```python
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:

```python
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/`](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

```bash
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:

```python
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

- No API-key, TLS or mTLS channel support.
- Snapshot queries are supported; query inputs are not. Durable cron schedules support upsert,
  list, pause/resume through upsert, and idempotent deletion. Server-backed restart-from-step forks
  are supported for terminal runs. Resumable output reads and debounce clients are not yet implemented.
- 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 narrow same-Postgres atomic checkpoint path is TS-only today).
- 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.
