Python SDK

The Tier-1 Python SDK includes hand-written polling and ConnectRPC streaming workers and two generated clients, one per protocol:

• IronflowClient — REST. Event reads, KV store, config, environment CRUD, and the remaining management routes.
• IronflowRPC / AsyncIronflowRPC — ConnectRPC. Functions, runs, projections, entity streams, schemas, webhook management, agent tools, time travel, pub/sub, and four server streams.

Neither is a superset of the other. Each carries what its protocol serves. For equivalent Tier-1 capability examples, see the Go SDK and Node SDK references.

Install ironflow-py, not ironflow

pip install ironflow-py

The distribution name is ironflow-py; the import name stays ironflow. The bare name ironflow on PyPI belongs to an unrelated third-party project (a materials-science tool from the pyiron group), so pip install ironflow installs the wrong package. A virtualenv holding both has two claimants on the top-level ironflow module and the install order silently decides which wins.

The ironflow-py distribution first reached PyPI at v0.33.1; v0.33.0 was tagged but not published. The worker described here first shipped in v0.40.0. It is not in any earlier release. Releases are published from sahina/ironflow-py over PyPI Trusted Publishing.

Methods on routes the manifest annotates with schemas return a typed models.* TypedDict and take query parameters as keyword arguments. Other routes return Any; use BaseClient.request() for undeclared options. See SDK Comparison for the feature matrix.

Correlated callback logs

ctx.logger is a standard logging.LoggerAdapter. Push, polling, and streaming callback records carry five reserved fields:

Field	Type	Value
run_id	str	Full dispatch run ID
function_id	str	Dispatch function ID, not the display name
step_id	str	Allocated memoization key during a live step.run callback; "" outside it
attempt	int	Engine run attempt, not a step retry counter
trace_id	str	Current valid OTel trace ID as 32 lowercase hex characters; "" without one

Step labels work without OpenTelemetry. Nested steps restore the enclosing label; concurrent branches have separate labels. Child tasks and thread-backed callbacks stop claiming an active step when the callback exits or times out. Their inherited valid trace context remains usable. Memoized callbacks do not run or log again; code outside steps can log on every resume. Logging has no exactly-once guarantee.

Polling and streaming keep the worker’s configured logger; push keeps the ironflow.serve hierarchy. Transport diagnostics keep their existing routes and are not given callback correlation. Application handlers, filters, levels, formatters, and intentional silence remain in control, including NullHandler. The SDK merges call-level extra on Python 3.10 too. Reserved fields override collisions; other application fields, formatting arguments, and explicit exc_info survive. Handler exceptions follow normal callback error handling.

If no application handler is reachable through the logger hierarchy, callback logging uses a local stderr fallback with all five keys. It does not call basicConfig, attach handlers to application loggers, or change levels. To print fields through your own handler, choose the exported formatter:

import logging
from ironflow.worker import CorrelationFormatter

handler = logging.StreamHandler()
handler.setFormatter(CorrelationFormatter())
application_logger = logging.getLogger("my_app")
application_logger.addHandler(handler)
# Supply application_logger as Worker(..., logger=...) or StreamingWorker(..., logger=...).

For push, configure logging.getLogger("ironflow.serve") or an ancestor. A custom formatter can omit the fields from text; they remain on the record. CorrelationFormatter() also accepts ordinary diagnostic records, showing empty identity fields and attempt 0 for them.

Use an application-owned OTel Logs bridge

Install a compatible OpenTelemetry Logs SDK only if your application wants it. The bridge below is tested on Python 3.10 with opentelemetry-sdk==1.39.1, opentelemetry-api==1.39.1, and semantic conventions 0.60b1. The SDK’s tracing extra supplies only the trace API; it does not configure log export.

import logging
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler

provider = LoggerProvider()
# Add your application's chosen processors/exporters to provider.
application_logger = logging.getLogger("my_app")
application_logger.addHandler(LoggingHandler(logger_provider=provider))
# Pass application_logger to Worker or StreamingWorker; use ironflow.serve for push.

Attach the bridge once per destination, rather than at both child and root. Each callback call emits one native record; the application bridge exports it with the five attributes and intrinsic trace ID, span ID, and trace flags from the active execution context. A valid unsampled context still correlates. Ironflow does not preformat correlation into the OTel body, configure a provider or exporter, introduce network activity, or collect these records in the engine. Your application owns exporter policy, flushing, and provider shutdown. An explicit no-op bridge remains silent. Arbitrary root logs are not captured or enriched by Ironflow.

Execution tracing

Install ironflow-py[tracing] for the OpenTelemetry API integration. Configure an OpenTelemetry tracer provider and exporter in your application before starting a worker or serving push requests. Ironflow does not install or configure an exporter. Without a provider, the API uses no-op spans.

Polling jobs, streaming jobs, and push requests start a process span parented on the engine’s W3C traceparent and tracestate. Each executed step.run starts a child step.run span. Memoized steps create no span. Each resumed execution creates a new process span using its newly assigned engine context. Valid unsampled contexts propagate even when spans are not recorded. Missing or malformed parents start a new root, independently of ambient request context. Invalid or oversized trace state is discarded without losing a valid parent.

Failures set error status. Durable yields and cancellation leave status unset. Spans end when execution exits. The integration does not copy baggage, secrets, payloads, exception messages, or stack traces into spans. The optional extra depends only on opentelemetry-api; applications own SDK and exporter setup. The base install works without selecting the extra, and tracing imports are safe when the API is absent. Some transport dependencies may install the API transitively.

Pull worker

Worker polls the server for long-running jobs. StreamingWorker receives assignments over a persistent ConnectRPC HTTP/2 connection; it is useful when you want the server to deliver jobs without polling. Both workers share the same durable-step API and drain behavior. The corresponding Go worker and Node worker examples show the other Tier-1 forms.

from ironflow.worker import StreamingWorker, function

@function(id="research-agent", triggers=[{"event": "research.requested"}])
async def research(ctx):
    plan = await ctx.step.run("plan", lambda: llm.plan(ctx.event.data))
    await ctx.step.sleep("cooldown", "10m")
    reply = await ctx.step.wait_for_event(
        "approval", event="research.approved", timeout="1d"
    )
    return await ctx.step.run("write", lambda: llm.write(plan, reply))

StreamingWorker(functions=[research]).run()

Use Worker instead when the server endpoint does not support HTTP/2.

StreamingWorker refuses a job that it cannot run. The engine tracks max_concurrent_jobs, but it can send more jobs than that, for example after a reconnect. The worker refuses a job that arrives when it is full or during a drain. It sends a nack, and the engine makes the job runnable again at once. A refusal does not use a retry attempt. The engine does not send more jobs to a draining worker. For each refusal from a full worker, the engine holds back one of that worker’s slots. It frees the slot when the worker’s heartbeat shows a free slot. An exact redelivery of an active execution sequence is dropped, not refused.

Workers track active assignments by job ID and execution sequence. If recovery assigns the same job to this worker with a new sequence, both executions keep their own slot until they finish. An older completion cannot remove the newer execution from heartbeats. An exact redelivery of an active sequence is ignored. This applies to polling and ConnectRPC streaming workers; streaming cancellation by job ID signals cancellation to each locally tracked execution of that job.

@function requires an id and an async def handler. triggers is optional: with no trigger, the function runs only through step.invoke or step.invoke_async from another function. A trigger is {"event": name} or {"cron": schedule}, with an optional CEL expression filter. The decorator returns a function definition; it does not wrap the coroutine. The other options match the Go and Node function config:

@function option	Effect
name, description	Display name and description
retry	max_attempts, initial_delay, backoff_factor, max_delay, on_exhausted. Only the keys you set are sent; see Retry and exhaustion
circuit_breaker	failure_threshold, open_duration, max_open_duration. An unset or zero key keeps the server default (5, 60 seconds, 10 minutes). max_open_duration must be at least open_duration. The server raises a window shorter than the longest probe (21.3 seconds with the default push timeout) to that minimum; see Policy
throttle	limit (1 or more), period (1 second or more) and an optional key path. A soft start-rate limit: runs over it wait in the queue
timeout	Function timeout
step_timeout	Default timeout for each step.run that sets no timeout
concurrency	limit and an optional key path
debounce	period (1 second or more), optional key path, optional max_wait (at least period)
cancel_on	List of {"event": ..., "match": ...}; a matching event cancels the run
actor_key	JSON path for an actor key. The server stores and returns it. Dispatch does not use it for routing yet
secrets	Secret names the engine resolves into ctx.secrets
recording, recording_profile	Audit recording; profile is "all", "run_lifecycle", or "steps"
metadata	Custom key-value metadata
schema	Callable that validates event.data and returns the value the handler sees

A schema can be any callable, for example Pydantic’s Model.model_validate. It can also be async. Cron events skip it. If the callable raises, or if the payload was redacted, the run fails with a non-retryable SchemaValidationError.

Handler field	Value
ctx.event	Event with id, name, version, data, UTC timestamp, idempotency_key, source, and metadata
ctx.step	Durable Step with run, sleep, sleep_until, wait_for_event, parallel, map, invoke, invoke_async, publish, and compensate
ctx.run	RunInfo with id, function_id, attempt, max_attempts, on_exhausted, and environment (the value the worker polls with). max_attempts is 0 and on_exhausted is "" when the engine is too old to send them
ctx.logger	logging.LoggerAdapter carrying run, function, active step, attempt, and trace fields
ctx.secrets	Read-only mapping of job secrets

Worker(...) option	Default
functions	Required list of decorated functions
server_url	IRONFLOW_SERVER_URL, else http://localhost:9123
api_key	IRONFLOW_API_KEY, else none
environment	IRONFLOW_ENV, else none (no header)
worker_id	Random UUID
max_concurrent_jobs	10
heartbeat_interval	30 seconds
reconnect_delay	5 seconds
checkpoint_interval	1 second; 0 disables periodic checkpoints
poll_wait	20 seconds, at most 25: how long the server may hold an idle job poll; 0 disables. StreamingWorker does not take it
drain_timeout	60 seconds
labels	Empty mapping
logger	ironflow.worker logger
upcasters	Optional UpcasterRegistry applied to function events
projections	Projections from ironflow.projection.create_projection; a worker with only projections is allowed

Duration options accept strings such as "10m", timedelta, or numbers. Bare numbers mean seconds, not milliseconds. await worker.start() does not return until the worker drains, stops, or fails. In synchronous code, worker.run() owns the event loop: the first SIGINT or SIGTERM starts a bounded drain, and a second signal stops immediately. You can also call drain() or stop() explicitly.

Step bodies must be idempotent: work executed after the last persisted checkpoint may run again after a crash. A synchronous step callback runs in a thread; a timeout or drain cannot stop that thread. Keep such callbacks bounded and make their external writes safe to repeat. A wait_for_event timeout fails the run by default. Set on_timeout="return-null" to resume with None instead. This outcome is a completed, memoized JSON null and survives restart. The default return type is Event; opting in makes it Event | None. Approval helpers retain their fail-on-timeout default.

wait_for_event returns an Event. Event.emitted_by is the authenticated emitter that the engine recorded, for example apikey:<id> or user:<id>. The engine sets it from the credential of the caller, so the event body cannot change it. It is empty when the engine reports none. An operator with runs:cancel can inject or patch a step output, so emitted_by proves the emitter only for a step that an event resolved.

The other ctx.step methods are run(name, fn, *, timeout=None), sleep(name, duration), sleep_until(name, when), wait_for_event(name, *, event, match=None, match_value=None, payload=None, timeout="7d", on_timeout="fail") and parallel(name, branches, *, concurrency=None, on_error="fail_fast"). when is a timezone-aware datetime or ISO string in the future. A parallel branch is an async def branch(step). With on_error="collect", the result list holds the exception for a failed branch. The Steps guide explains match and match_value.

ctx.step.map(name, items, fn, *, concurrency=None, on_error="fail_fast") runs fn once per item as a parallel branch. fn takes (item, step, index). It returns a list of results in item order. Pass concurrency to cap parallel branches, and on_error="collect" to get every branch’s outcome instead of stopping at the first failure.

ctx.step.invoke(function_id, input=None, *, timeout=30) calls another function and waits for its output. The target function must have no triggers; only step.invoke or step.invoke_async can reach it. timeout is a Duration in seconds by default. If the child fails, or cannot start, step.invoke raises InvokeError, a non-retryable error with function_id, child_run_id, and cause.

ctx.step.invoke_async(function_id, input=None) starts another function and returns at once. It returns InvokeAsyncResult, a value with one field, run_id, for the started run.

ctx.step.publish(topic, data=None, *, idempotency_key=None) publishes to a pub/sub topic as a durable step and returns PublishResult (event_id, sequence). A retry does not publish twice. Without idempotency_key, the step sends a default key derived from its step ID, so the server drops a repeat. It keeps the key for IRONFLOW_RUN_RETENTION_DAYS days (forever when unset or 0) and returns the original event ID. One rare gap: if the server stores the message but cannot record it, a retry after 2 minutes stores a second copy under the same event ID. idempotency_key is at most 255 bytes with no whitespace or control characters; the server rejects anything else with InvalidArgument. It does not trigger functions; use the events API for that.

ctx.step.compensate(step_name, fn) registers an undo for step_name. If the run later fails with a non-retryable error, Ironflow calls the registered undos in reverse order, after the handler stops. A failed undo does not stop the rest. A retryable error on the last attempt counts as terminal, and runs the undos, only when retry["on_exhausted"] is "fail"; with "pause" the run is held for resume and nothing is undone.

Retry and exhaustion

The SDK sends only the retry keys you set. The server fills in the rest when each run is created, from spec.engine.retry: 25 attempts (the first included), a 1 second initial delay, a backoff factor of 2.0, a 5 minute maximum delay, and on_exhausted="pause". A function registered by an earlier SDK version keeps the values stored with it until it registers again.

on_exhausted decides what happens when a retryable failure has used every attempt. "pause" holds the run, with its completed steps, until you resume it. "fail" ends the run as failed and runs its undos. A non-retryable error always fails the run at once. function raises ValueError for any other value.

Raise RetryAfterError(message, retry_after) from a handler or a step.run body to choose the delay before the next attempt. retry_after is a Duration, so a bare number is seconds. The engine waits that long instead of the computed backoff, capped at the policy’s max_delay. It is retryable, so on_exhausted applies to it as to any retryable error. Any retryable IronflowError with a positive retry_after (for example a 429 from the REST client) sends the same delay.

from ironflow.worker import RetryAfterError, function

@function(id="sync-crm", triggers=[{"event": "crm.sync"}],
          retry={"max_attempts": 10, "on_exhausted": "fail"})
async def sync_crm(ctx):
    if ctx.run.attempt == ctx.run.max_attempts:
        ctx.logger.warning("last attempt")
    raise RetryAfterError("CRM is rate limiting us", "30s")

A throttle is soft: at most limit new runs start per period (per key value when key is set). Runs over the limit wait in the queue and are never dropped. Retries and resumed runs are not throttled.

Projections and upcasters

ironflow.projection.create_projection(*, name, events, handler, initial_state=None, mode=None, partition_key=None, max_retries=3, batch_size=100) defines a projection. mode is "managed" when initial_state is given and "external" otherwise. A managed handler takes (state, event, ctx) and returns the next state; an external handler takes (event, ctx) and returns nothing. Both may be sync or async. event is a ProjectionEvent (id, name, data, seq, timestamp, source, metadata, and entity_id, entity_type and entity_version, which are None for an ordinary event). The handler’s ctx is a ProjectionContext with event (an EventInfo: id, name, seq, timestamp) and projection (a ProjectionInfo: name, version). Pass projections to Worker(projections=[...]) or StreamingWorker(projections=[...]); a worker with only projections is allowed. Register event upcasters on an ironflow.UpcasterRegistry and pass it as upcasters=. External handlers are at-least-once, and projection runners do not apply upcasters. A runner holds an exclusive lease on its projection (ADR 0120), so only one replica reads events and the other replicas wait as standbys. The lease is required: on a server older than v0.44.0, which has no lease RPCs, the runner raises ServerTooOldError (code SERVER_TOO_OLD) and stops. A poll, a stream or a state load refused with failed_precondition, such as a projection bound to a PUBSUB stream that was recreated (ADR 0164) or one whose last rebuild failed, raises ProjectionRefusedError and stops the runner the same way, because a retry cannot fix it. The worker logs the error and keeps running.

Agents

ironflow.agent builds durable AI agents on the pull worker or serve(). Each LLM turn and tool call is its own durable step. @agent(id=..., tools=[...], memory=MemoryConfig(...), max_turns=20, **function_options) returns a normal function definition. The handler receives an AgentContext with tool(defn, args), tool_by_name(name, args), llm(call=..., messages=..., tools=...), approve(event, payload=..., ttl="7d"), spawn(name, function_id=..., input=..., wait=True, timeout=30), memory and turn. ctx.memory is a Memory with await get(*, bypass_cache=False) and await append(event_name, data, *, metadata=None). approve returns an ApproveResult (approved, approver, payload, reason, emitted_by). emitted_by is the authenticated emitter of the approval event, set by the engine from the credential of the caller. approver is free text from the event data, and the caller chooses it. An operator with runs:cancel can inject or patch a step output, so emitted_by proves the emitter only for a step that an event resolved. spawn returns a SpawnResult (output, run_id). llm returns an LLMCompleteResult (content, tool_calls as LLMToolCall, finish_reason, metadata). define_tool(...) builds a ToolDefinition; its idempotent option is a ToolIdempotency, "by_call" (default) or "by_args". MemoryConfig(backend=...) takes a MemoryBackend. expose_mcp returns an ExposeMcpHandle; call await handle.unregister() to remove the tools. With pip install 'ironflow-py[validate]', tool args are checked against input_schema before the handler runs, and a mismatch raises ToolValidationError. Without the extra, they are not checked. expose_mcp(name=..., callback_url=..., tools=[...]) registers tools with the engine’s MCP server; mount the dispatch callback at DISPATCH_PATH. expose_mcp(environment=...) (default IRONFLOW_ENV, else no header) scopes registration when the SDK builds the client. Agent memory uses ctx.run.environment. In push mode, serve sets it to environment, else IRONFLOW_ENV, else None; with neither set, push step.publish sends no header. The engine rejects a loopback or private-network callback_url unless it runs with --dev or IRONFLOW_AGENT_TOOLS_ALLOW_PRIVATE=true.

agent, define_tool, expose_mcp, llm, memory, spawn and approve are Ready: the contract is settled and no breaking change is planned.

Exceptions

All of these subclass IronflowError, except SignatureError.

Exception	Module	Meaning
NonRetryableError(message, code="NON_RETRYABLE")	ironflow.worker	Raise it to fail the run with no retry.
RetryAfterError(message, retry_after, code="RETRY_AFTER")	ironflow.worker	Retryable. Raise it to choose the delay before the next attempt; retry_after is a Duration. The engine caps it at the policy’s max_delay.
StepError	ironflow.worker	A step.run body raised. The cause is __cause__. It has step_id and retryable, and carries the cause’s retry_after.
StepTimeoutError	ironflow.worker	A step.run body ran past its timeout. It is retryable.
SchemaValidationError	ironflow.worker	The function’s schema rejected the event data. No retry.
WorkerAuthError	ironflow.worker	The server answered 401 or 403 to the worker.
InvokeError	ironflow.worker	A step.invoke child failed or could not start. No retry. It has function_id, child_run_id and cause.
UpcasterChainError	ironflow	An upcaster chain is incomplete, or one upcaster raised.
PreconditionFailedError, PreconditionRequiredError, PayloadTooLargeError, UnsupportedMediaTypeError	ironflow	The REST client raises them for HTTP 412, 428, 413 and 415.
IronflowRPCError	ironflow	A ConnectRPC call failed. See Errors.
SignatureError	ironflow.serve	verify_signature(body, header, key) rejected a push request. It has a code.
AgentError	ironflow.agent	Base class for agent errors. It is non-retryable and has code and details.
MaxTurnsExceededError, ToolValidationError, DuplicateToolError, ToolNotFoundError, MemoryProjectionRequiredError	ironflow.agent	AgentError subclasses.
LLMError	ironflow.agent	Base class for the errors ctx.llm classifies: LLMRefusalError, LLMInvalidJSONError and LLMMaxTokensError.

Retry signal

IronflowClient reads the server’s error body. When the body has a boolean retryable, that flag sets error.retryable, whatever the status. Without it, the status decides: 408, 429 and 5xx are retryable, except 501 (not implemented). error.retry_after is the Retry-After header when present, else the body’s retry_after in seconds. Only idempotent methods (GET, PUT, DELETE and similar) are retried. POST and PATCH are not retried by default, except the two signed-URL calls (files_buckets_signed_urls_upload and files_buckets_signed_urls_download).

Push mode

The engine POSTs each run to an HTTP endpoint you host. serve() returns an ASGI app; handle() is the framework-free core behind it; register() registers functions as push, pointing the engine at your endpoint URL.

import asyncio

from ironflow.serve import register, serve
from ironflow.worker import function

@function(id="send-receipt", triggers=[{"event": "order.paid"}])
async def send_receipt(ctx):
    await ctx.step.run("email", lambda: send(ctx.event.data))

ironflow_app = serve([send_receipt])   # uvicorn module:ironflow_app
# FastAPI: api.mount("/ironflow", ironflow_app)

# once per deploy, e.g. from a release script — not in the module uvicorn loads
asyncio.run(register([send_receipt], endpoint_url="https://api.example.com/ironflow/"))  # trailing slash: a mount redirects "/ironflow"

serve(functions, *, signing_key=None, upcasters=None, webhooks=(), server_url=None, api_key=None, environment=None) returns an ASGI app callable — mount it under FastAPI or Starlette, or run it directly with uvicorn. signing_key defaults to IRONFLOW_SIGNING_KEY; server_url, api_key, and environment default to IRONFLOW_SERVER_URL, IRONFLOW_API_KEY, and IRONFLOW_ENV (no header if unset). They apply to webhook event emission and step.publish, not to the push handler itself. serve() has no localhost fallback for server_url: without it, step.publish fails with the non-retryable code UNSUPPORTED.

async handle(functions, *, method, path, headers, body, signing_key=None, upcasters=None, webhooks=(), server_url=None, api_key=None, environment=None) -> Response is the framework-free core: bytes in, (status, headers, bytes) out. Use it to wire push serving into a framework serve() does not cover, such as a WSGI adapter or a non-ASGI runtime.

register(functions, *, endpoint_url, server_url=None, api_key=None, environment=None) -> None is an async call that registers each function as push, pointing the engine at endpoint_url. Call it once per deploy, for example from a release script. It raises IronflowError if the engine rejects a function.

Webhook(id, transform, verify=None) serves POST /webhooks/{id}. verify takes a WebhookRequest (body, headers, method, path) and raises to reject the request; transform takes the raw body and returns a WebhookEvent (name, data, idempotency_key), which serve() emits to the engine. Both may be sync or async. Pass webhooks as serve(functions, webhooks=[Webhook(...)]).

Set IRONFLOW_SIGNING_KEY to the engine’s key. With it set, the handler rejects requests without a valid X-Ironflow-Signature. Without it, the handler runs any well-formed request it receives, and a webhook with no verify emits whatever is posted to it.

verify_signature(body, header, key, *, now=None, tolerance=300.0) checks an X-Ironflow-Signature header and raises SignatureError when it is missing, malformed, outside the tolerance (in seconds) or wrong. sign(body, key, ts) builds a header value, for example in a test.

• Short functions only. The engine gives each push request at most the push timeout (10s by default), or the function’s timeout (registered as timeoutMs) if lower. After that it retries while your handler may still be running. Use the pull worker for long work.
• Push or pull, not both. Do not serve one function id from a push app and a pull Worker. The worker re-registers the function as pull and silently undoes the push registration.

Installation

pip install ironflow-py   # import name: ironflow

Requires Python 3.10+. ConnectRPC support is included in the default install. The runtime dependencies are connectrpc~=0.12.1, meaning at least 0.12.1 and below 0.13, protobuf-py>=0.3,<0.7 for the generated message types, pyqwest>=0.10,<0.12 and typing_extensions>=4.4. The REST client itself uses only urllib and json from the standard library.

Upgrading from ConnectRPC 0.11

The SDK now uses ConnectRPC 0.12.1 with regenerated protobuf and service bindings. Update any application lockfile that still pins connectrpc to 0.11.x or protobuf-py to 0.1.1. Install the SDK with its declared dependencies so the bindings and runtime stay compatible.

Existing calls through IronflowRPC, AsyncIronflowRPC, and ironflow.rpc.v1 keep the same signatures.

Quick Start

from protobuf.wkt import Struct
from ironflow import IronflowClient, IronflowRPC
from ironflow.rpc import v1

# Create a client
client = IronflowClient(
    server_url="http://localhost:9123",
    api_key="ifkey_...",
)

# Emit an event (triggers any listening functions)
with IronflowRPC(server_url=client.server_url, api_key=client.api_key) as rpc:
    result = rpc.events.emit(v1.TriggerRequest(event='order.placed', data=Struct.from_python({'order_id': '123', 'customer_id': '456'}), version=1))

# List runs
rpc = IronflowRPC(server_url=client.server_url, api_key=client.api_key)
runs = rpc.runs.list(v1.ListRunsRequest())

# Get a specific run
run = rpc.runs.get(v1.GetRunRequest(id="run_abc123"))

# Cancel a run
rpc.runs.cancel(v1.CancelRunRequest(id="run_abc123", reason="duplicate"))

# Permanently delete one terminal run and its steps (cannot be undone)
rpc.runs.delete(v1.DeleteRunRequest(id="run_abc123"))

# Delete terminal runs matching a filter, and read back this call's count
removed = rpc.runs.delete_many(v1.DeleteRunsRequest(function_id="process-order")).deleted

runs.delete refuses a run that is not yet terminal, and runs.delete_many needs at least one of function_id, status or until. Both require the runs:delete action, which only the admin role holds. delete_many stops at 10,000 runs per call; re-issue the same request until deleted is 0. A count below 10000 does not mean drained — a page comes back short under row-lock contention.

Client Configuration

client = IronflowClient(
    server_url="http://localhost:9123",  # Default
    api_key="ifkey_...",                  # Optional, required for authenticated endpoints
    timeout=30.0,                         # Request timeout in seconds
)

IronflowClient also takes total_timeout (a ceiling for a whole call, retries included), max_attempts, retry_methods (default ironflow.IDEMPOTENT_METHODS), initial_backoff, max_backoff, backoff_factor and on_error. on_error is a hook that takes the error and an ErrorContext (method, endpoint, status_code); a hook that raises does not hide the call’s own error. It retries idempotent HTTP methods and the two signed-URL POSTs only. Pass retry=True to client.request() to opt a write in, alongside an Idempotency-Key header.

Typed Usage

sdk/python/ironflow/models.py holds one TypedDict per response and request shape the server’s route manifest describes. Import it alongside the client:

from ironflow import IronflowClient, models

client = IronflowClient(server_url="http://localhost:9123", api_key="ifkey_...")

keys: list[models.ApiKeyResponse] = client.api_keys_list(platform=True)
created: models.CreateAPIKeyResponse = client.api_keys_create(body={"name": "ci"})

A TypedDict is a plain dict at runtime. models.ApiKeyResponse exists for mypy and your IDE — it adds no class, no __init__, and no isinstance check. Every generated return value is wrapped in typing.cast(...), which is also a runtime no-op: it changes the value’s declared type for the type checker and does nothing when the code actually runs. There is no validation. If the server sends a shape the manifest does not describe, cast does not raise — a later result["field"] raises KeyError instead, at the point of use, not at the call site.

A route can also answer 204 No Content. _http.py returns None for any empty response body, so a route annotated with RespMayBe204 gets a nullable return type: X | None, not just X. One route does this today: workers_list_jobs(), the pull-mode worker’s job poll, which answers 204 on every poll that finds no job. Dereferencing a None result raises TypeError, not KeyError — guard it before use:

batch = client.workers_list_jobs(worker_id, available=1)
if batch is None:
    # No job available. Poll again later.
    ...
else:
    for job in batch["jobs"]:
        ...

Optional keys are expressed as a base-class split (class _FooReq(TypedDict): ... then class Foo(_FooReq, total=False): ...), not typing.NotRequired. The SDK’s floor is Python 3.10; NotRequired needs 3.11+.

Individual fields can be nullable too. A str | None field requires a check before using it as a string:

from ironflow.models import ApiKeyResponse

def show_last_use(key: ApiKeyResponse) -> None:
    last_used = key.get("last_used_at")
    if last_used is not None:
        print(last_used)

Query parameters are keyword-only kwargs, typed to match the manifest’s query schema:

client.audit_list(from_="2026-01-01", limit=50)   # -> ?from=2026-01-01&limit=50
client.api_keys_list(platform=True)               # -> ?platform=true

Booleans go on the wire lowercase (true/false), matching how the Go handlers parse them. A kwarg name that collides with a Python keyword, or with a path parameter already claiming that name, gets a trailing underscore — from becomes from_ above — but the wire name (the actual query string key) is unchanged.

Any is still the return type, with no query kwargs, for any route the manifest does not annotate with schemas.

Timestamps

Every timestamp IronflowClient returns is a string, not a datetime. The client does not decode them, so you get what the server sent.

The strings are RFC 3339. The fraction of a second has no digits or up to 9, and the count can differ between values from one route. Do not compare two of them as strings.

Parsing needs care on the SDK’s Python floor. On Python 3.10, datetime.fromisoformat rejects a trailing Z and any fraction that is not 3 or 6 digits. Python 3.11 and later accept both, and drop digits past microseconds.

For typed timestamps, use IronflowRPC or AsyncIronflowRPC. Their results carry the proto Timestamp, or None when the field is unset. resolved_timestamp on the webhook verify test result is the one string among them. A subscription frame’s payload is not decoded, so RunEvent timestamps are strings, and a timestamp the frame omits reads as "" or None.

from ironflow import IronflowRPC
from ironflow.rpc import v1

with IronflowRPC(server_url="http://localhost:9123", api_key="ifkey_...") as rpc:
    run = rpc.runs.get(v1.GetRunRequest(id="run_abc123"))
    created = run.created_at.to_datetime()  # aware datetime, UTC

This only helps for what ConnectRPC serves. Many routes are REST only, including KV, config, secrets, files, capacity, workers and most tenancy routes, so their timestamps are strings in every Python client.

to_datetime() rounds to microseconds, while a Timestamp keeps nanoseconds. To send a webhook source’s updated_at back as expected_updated_at, pass the message you received, not a value rebuilt from a datetime.

API Reference

All methods follow the {group}_{action} naming pattern in snake_case.

A representative tour, not the complete index — run make sdk-health for the full generated surface.

Events

from protobuf.wkt import Struct
from ironflow.rpc import v1
from ironflow import IronflowRPC
# Emit an event — TriggerRequest takes "event", "data", optional "metadata"
with IronflowRPC(server_url=client.server_url, api_key=client.api_key) as rpc:
    rpc.events.emit(v1.TriggerRequest(event='user.created', data=Struct.from_python({'user_id': '123'})))

# List events
events = client.events_list()

Runs

from ironflow import IronflowRPC
from ironflow.rpc import v1
from protobuf.wkt import Struct

# environment="env_default_staging" adds X-Ironflow-Environment-ID to every call. It is optional,
# and the client does not read IRONFLOW_ENV.
with IronflowRPC(server_url="http://localhost:9123", environment="env_default_staging") as rpc:
    runs = rpc.runs.list(v1.ListRunsRequest())
    run = rpc.runs.get(v1.GetRunRequest(id="run_id"))
    rpc.runs.cancel(v1.CancelRunRequest(id="run_id", reason="test"))
    rpc.runs.get_steps(v1.GetRunStepsRequest(run_id="run_id"))
    rpc.audit.get_trail(v1.GetAuditTrailRequest(run_id="run_id"))
    rpc.runs.resume(v1.ResumeRunRequest(run_id="run_id"))
    rpc.runs.patch_step(v1.PatchStepRequest(
        step_id="step_xyz", output=Struct.from_python({"corrected": True}),
        reason="manual fix",
    ))

Projections

from ironflow import AsyncIronflowRPC
from ironflow.rpc import v1

async with AsyncIronflowRPC(server_url="http://localhost:9123") as rpc:
    projections = await rpc.projections.list(v1.ListProjectionsRequest())
    projection = await rpc.projections.get(v1.GetProjectionRequest(name="my-projection"))
    status = await rpc.projections.get_status(v1.GetProjectionStatusRequest(name="my-projection"))
    await rpc.projections.rebuild(v1.RebuildProjectionRequest(name="my-projection"))
    await rpc.projections.get_rebuild_job(v1.GetRebuildJobRequest(name="my-projection"))
    await rpc.projections.cancel_rebuild(v1.CancelRebuildRequest(name="my-projection"))
    await rpc.projections.pause(v1.PauseProjectionRequest(name="my-projection"))
    await rpc.projections.resume(v1.ResumeProjectionRequest(name="my-projection"))

client.projections_delete("my-projection")

KV Store

# Buckets
client.kv_list_buckets()
client.kv_buckets(body={"name": "my-bucket"})  # Create
client.kv_get_buckets("my-bucket")
client.kv_delete_buckets("my-bucket")

# Keys
client.kv_list_buckets_keys("my-bucket", filter="user.*")
client.kv_get_buckets_keys("my-bucket", "my-key")
client.kv_delete_buckets_keys("my-bucket", "my-key")
client.kv_delete_buckets_keys("my-bucket", "my-key", purge=True)  # hard delete

# `body` is stored as the value verbatim (json.dumps of what you pass), so pass
# the document itself — a {"value": ...} wrapper would be stored literally.
client.kv_update_buckets_keys("my-bucket", "my-key", body={"name": "Alice"})

# Reads come back base64-encoded, because the server's value is []byte.
import base64, json
entry = client.kv_get_buckets_keys("my-bucket", "my-key")
json.loads(base64.b64decode(entry["value"]))  # {'name': 'Alice'}

# Atomic create and compare-and-swap use typed conditional-header arguments:
client.kv_update_buckets_keys("my-bucket", "my-key",
                              body={"name": "Alice"}, if_none_match="*")  # create-only
client.kv_update_buckets_keys("my-bucket", "my-key",
                              body={"name": "Bob"}, if_match="3")         # CAS
# Both raise IronflowError with status_code 412 on conflict.

# Live watch: client.watch_kv(bucket, *, key=None) is an async iterator of
# KVWatchEvent (key, value, revision, operation, bucket) over WebSocket. It needs
# `pip install 'ironflow-py[watch]'`. It first yields the current entry of each
# matching key, then each change. `value` is the decoded bytes, or None for a
# delete. A missing bucket or a rejected key raises IronflowError. Leave the loop,
# or call aclose() on the iterator, to close the socket.

Command dedup

client.command_dedup(bucket_name, *, ttl_seconds=604800) returns a CommandDedup for command idempotency over KV. Reuse one instance; it creates the bucket on first use. ttl_seconds=0 means entries never expire.

dedup = client.command_dedup("order-commands")
prior = dedup.try_claim(command_id, {"status": "claimed"})
if prior is not None:
    if prior.get("status") == "done":
        return prior["result"]
    raise RuntimeError("Command is pending; reconcile its effects before retrying")

try:
    result = run_handler()  # Effects must support idempotent retries.
except Exception:
    dedup.release(command_id)  # Only the claim winner reaches here.
    raise
dedup.finalize(command_id, {"status": "done", "result": result})
return result

A prior entry can be the unfinished claim. A completion marker is application-owned; pending entries need reconciliation and idempotent effects, not a successful return. See the CQRS walkthrough for entity-stream recovery.

Never release after finalize: it deletes the result and lets the command replay.

try_claim retries the claim up to 3 times when the winner releases between your write and your read, then raises IronflowError with code == "COMMAND_DEDUP_RACE". If the bucket was deleted, a write recreates it once.

If try_claim raises on a transport error, the claim state is unknown: the write may have committed, and the helper does not retry it. To let a retry recognize its own orphaned claim, put a token that stays the same across retries in the claim and pass is_owner:

token = message_id  # the same value on every retry of this command
claim = {"token": token, "status": "claimed"}
prior = dedup.try_claim(command_id, claim, is_owner=lambda p: p.get("token") == token and p.get("status") == "claimed")

is_owner runs only on a prior entry. It must be false for a finalized result, or a finished command replays. Run one retry lineage at a time: two concurrent callers with the same token both win.

Files

Buckets, files, signed URLs and file events. The Python client has no bucket handle: each method takes the bucket name first. See the Files guide.

client.files_buckets(body={"name": "inbox", "emitEvents": True, "allowSignedUrls": True})  # Create

# Put. `body` is bytes or a seekable file object; a seekable body is retried.
info = client.files_update_buckets_objects(
    "inbox", "scans/a.pdf", pdf_bytes,
    content_type="application/pdf",
    if_none_match=None,                                    # or if_match="<etag>"
    extra_headers={"X-Ironflow-Meta-source": "scanner-1"},  # user metadata
)

# Get. A BinaryResponse streams; close it or use `with`.
with client.files_get_buckets_objects("inbox", "scans/a.pdf", if_match=info["etag"]) as resp:
    data = resp.read()                                    # or resp.iter_bytes()

page = client.files_list_buckets_objects("inbox", prefix="scans/", delimiter="/", limit=100)
client.files_get_buckets_info("inbox", "scans/a.pdf")     # metadata only
client.files_delete_buckets_objects("inbox", "scans/a.pdf")

client.files_buckets_move("inbox", body={"from": "scans/a.pdf", "to": "archive/a.pdf"})
client.files_buckets_copy("inbox", body={"from": "archive/a.pdf", "to": "b.pdf", "toBucket": "backups"})
client.files_buckets_move_prefix("inbox", body={"from": "scans/", "to": "archive/"})   # {"count": n}

up = client.files_buckets_signed_urls_upload("inbox", body={"path": "x.png", "ttlSeconds": 600})
down = client.files_buckets_signed_urls_download("inbox", body={"path": "x.png"})

Method	Route
files_buckets(body)	Create a bucket.
files_list_buckets()	List buckets.
files_get_buckets(bucket)	Get a bucket.
files_patch_buckets(bucket, body)	Update a bucket.
files_delete_buckets(bucket)	Delete an empty bucket.
files_update_buckets_objects(bucket, path, body, *, content_type, if_match=None, if_none_match=None, extra_headers=None)	Put a file.
files_get_buckets_objects(bucket, path, *, if_match=None, range=None)	Get a file as a BinaryResponse.
files_get_buckets_info(bucket, path)	A file’s metadata.
files_list_buckets_objects(bucket, *, prefix, delimiter, cursor, limit)	List files.
files_delete_buckets_objects(bucket, path)	Delete a file.
files_buckets_move, files_buckets_copy, files_buckets_move_prefix	Move, copy, directory move.
files_buckets_signed_urls_upload, files_buckets_signed_urls_download	Create a signed URL.
files_update_signed(body, *, content_type, token), files_list_signed(*, token, ...)	Use a signed URL (PUT and GET).

Add "createOnly": True to the signed upload body, and a second PUT to the URL fails with 412. The Files guide has the signed URL, file event and cluster rules.

A body that is not bytes and not seekable raises ValueError before a request is sent. There is no stream upload. A 412, 413 or 415 raises PreconditionFailedError, PayloadTooLargeError or UnsupportedMediaTypeError. All three subclass IronflowError and are importable from ironflow. A 404 is an IronflowError with status_code == 404 and code == "not_found".

Config Management

client.config_list()
client.config_get("my-config")
client.config_create("my-config", body={"key": "value"})   # POST — full replacement
client.config_patch("my-config", body={"key": "updated"})  # shallow merge
client.config_delete("my-config")

# Conditional writes. A failed condition raises IronflowError with status_code 412.
client.config_create("my-config", body={"key": "value"}, if_none_match="*")  # create-only
client.config_create("my-config", body={"key": "next"}, if_match="3")        # replace at revision 3
client.config_patch("my-config", body={"key": "again"}, if_match="4")
client.config_delete("my-config", if_match="5")

# Live watch: client.watch_config(name) is an async iterator of ConfigWatchEvent
# (name, data, revision, updated_at). It needs `pip install 'ironflow-py[watch]'`.
# It first yields the current document, if one exists, then each change. `data`
# is None after a delete.

Secrets (Server-Only)

# Every secrets method takes the environment ID as a required keyword argument
# (the `X-Ironflow-Environment-ID` header).
env = "env_default"
client.secrets_list(x_ironflow_environment_id=env)
client.secrets_get("my-secret", x_ironflow_environment_id=env)
client.secrets_create(body={"name": "my-secret", "value": "s3cret", "description": "optional"}, x_ironflow_environment_id=env)
client.secrets_update("my-secret", body={"value": "new-value"}, x_ironflow_environment_id=env)  # PUT — full replacement
client.secrets_patch("my-secret", body={"description": "updated"}, x_ironflow_environment_id=env)  # PATCH — partial update
client.secrets_delete("my-secret", x_ironflow_environment_id=env)

Server Introspection

client.health()        # GET /health; HealthResponse
client.ready()         # GET /ready; ReadinessResponse. Raises IronflowError when not ready
client.capabilities()  # ServerCapabilities: transports, features, version, auth_required

Entity Streams

from ironflow import IronflowRPC
from ironflow.rpc.v1 import (
    AppendEventRequest, CreateSnapshotRequest, DeleteStreamRequest,
    GetEntityHistoryRequest, GetSnapshotRequest, GetStreamInfoRequest,
    ListStreamsRequest, ReadStreamRequest,
)
from protobuf.wkt import Struct

with IronflowRPC(server_url="http://localhost:9123", api_key="ifkey_...") as rpc:
    rpc.streams.list_streams(ListStreamsRequest())
    rpc.streams.get_info(GetStreamInfoRequest(entity_id="entity-123"))
    rpc.streams.append_event(AppendEventRequest(
        entity_id="entity-123", entity_type="order", event_name="order.updated",
        data=Struct.from_python({"status": "shipped"}),
        expected_version=4,  # -1 skips the check; 0 requires an empty stream
        idempotency_key="ship-1", version=1,
    ))
    rpc.streams.read_stream(ReadStreamRequest(entity_id="entity-123"))
    rpc.streams.get_history(GetEntityHistoryRequest(entity_id="entity-123"))
    rpc.streams.create_snapshot(CreateSnapshotRequest(
        entity_id="entity-123", entity_type="order", entity_version=10,
        state=Struct.from_python({"status": "shipped"}),
    ))
    rpc.streams.get_snapshot(GetSnapshotRequest(entity_id="entity-123"))
    # Tombstones the stream; purge=True also deletes the events below the
    # tombstone. Irreversible. Requires `streams:delete` (admin by default).
    res = rpc.streams.delete(DeleteStreamRequest(entity_id="entity-123", purge=False))
    print(res.entity_version)  # the tombstone's version

Appends to a deleted stream fail with ConnectRPC code failed_precondition. Projections do not see the $stream.deleted tombstone unless they registered that event name.

Redaction

from ironflow.rpc.v1 import RedactEventRequest, RedactRunRequest, RedactStepRequest

with IronflowRPC(server_url="http://localhost:9123", api_key="ifkey_...") as rpc:
    # The event's data. Entity-stream snapshots derived from it are dropped too.
    rpc.events.redact(RedactEventRequest(event_id="ev_01JD2KQ7X8Y9Z"))
    # The run's input, output and error, every child step error, plus attempt errors and audit payloads.
    rpc.runs.redact(RedactRunRequest(run_id="run_abc123"))
    # The step's output, original output and error, its parent run error and attempt errors.
    # Sibling errors are unchanged. step_id is the ROW id, not the name.
    rpc.runs.redact_step(RedactStepRequest(step_id="step_01JD2KQ7X8Y9Z"))

Each replaces the payload with {"$redacted": true, "sha256": ..., "redactedAt": ...} and keeps every other column, so replay ordering and optimistic concurrency are unaffected. Irreversible, idempotent, and gated on events:redact (admin by default). Error envelopes are replaced whole, including message, code, stack, details and encrypted envelopes. Run redaction leaves step outputs unchanged. Stored error placeholders survive later whole-row saves. RedactStep and RedactRun are refused with failed_precondition while the run is non-terminal. A step’s output is normally copied into the run’s output, so call both to reach every copy; projection state you derived is never rewritten. Use ironflow.is_redacted(payload) in a reducer before trusting a value. See Redacting an Event’s Payload.

API Keys

client.api_keys_list()
client.api_keys_create(body={"name": "my-key"})
client.api_keys_get("ak_123")
client.api_keys_delete("ak_123")
client.api_keys_rotate("ak_123")

Projects & Environments

client.projects_list()
client.projects_create(body={"name": "my-project"})
client.projects_update("proj_123", body={"name": "renamed"})
client.projects_delete("proj_123")

client.environments_list()
client.environments_create(body={"name": "staging", "project_id": "proj_123"})
client.environments_update("env_123", body={"name": "production"})
client.environments_delete("env_123")

Functions

from ironflow import IronflowRPC
from ironflow.rpc.v1 import InvokeFunctionRequest, ListFunctionsRequest
from protobuf.wkt import Struct

rpc = IronflowRPC()
rpc.functions.list(ListFunctionsRequest())
# Direct invocation returns run and event IDs immediately.
rpc.functions.invoke(InvokeFunctionRequest(function_id="process-order", data=Struct.from_python({"order_id": "123"})))

Event Schemas

from ironflow import IronflowRPC
from ironflow.rpc.v1 import (
    RegisterSchemaRequest, ListSchemasRequest, GetSchemaRequest,
    DeleteSchemaRequest, TestUpcastRequest,
)
from protobuf.wkt import Struct

with IronflowRPC() as rpc:
    rpc.event_schemas.register(RegisterSchemaRequest(
        event_name="order.placed", version=1,
        schema_json='{"type":"object","properties":{"orderId":{"type":"string"}}}',
    ))
    rpc.event_schemas.list(ListSchemasRequest())
    rpc.event_schemas.get(GetSchemaRequest(event_name="order.placed"))
    rpc.event_schemas.get(GetSchemaRequest(event_name="order.placed", version=1))
    rpc.event_schemas.test_upcast(TestUpcastRequest(
        event_name="order.placed", from_version=1, to_version=2,
        data=Struct.from_python({"orderId": "123"}),
    ))
    rpc.event_schemas.delete(DeleteSchemaRequest(event_name="order.placed", version=1))

Pub/Sub Topics

from ironflow import IronflowRPC
from ironflow.rpc.v1 import PublishRequest
from protobuf.wkt import Struct

with IronflowRPC() as rpc:
    result = rpc.pubsub.publish(PublishRequest(
        topic="notifications", data=Struct.from_python({"type": "ping"}),
    ))
    print(result.event_id, result.sequence)

Publishing, topic queries, subscriptions and consumer-group management use IronflowRPC. See the ConnectRPC client below.

To acknowledge an event from a MANUAL consumer group, call ack_event. Set consumer_id to the subscription_id of the event. A subscription without a consumer group always acknowledges automatically.

from ironflow.rpc.v1 import AckEventRequest, AckType, JoinConsumerGroupRequest

with IronflowRPC() as rpc:
    for event in rpc.pubsub.join_consumer_group(JoinConsumerGroupRequest(group_name="order-workers")):
        rpc.pubsub.ack_event(AckEventRequest(
            group_name="order-workers",
            consumer_id=event.subscription_id,
            event_id=event.event_id,
            ack_type=AckType.ACK,
        ))

For a MANUAL group, the group redeliver_delay_ms is the time that JetStream waits for an acknowledgment before it redelivers the event. A NAK sets its own delay in redeliver_delay_ms on the request.

SQL

Raw SQL is ConnectRPC-only — QueryService/ExecuteSQL has no REST sibling.

from ironflow import IronflowRPC
from ironflow.rpc.v1 import ExecuteSQLRequest

with IronflowRPC(server_url="http://localhost:9123", api_key="ifkey_...") as rpc:
    result = rpc.query.execute_sql(
        ExecuteSQLRequest(query="SELECT * FROM order_stats LIMIT 10")
    )

Audit

# Global audit feed
client.audit_list()

Operations

# Armed debounce timers in this environment
client.debounce_list_entries()

# Cancel one timer. Pass the raw debounce key; the client base64url-encodes it.
client.debounce_delete_entries("env_default", "process-search", "user-42")

# Capacity views need platform credentials (an ifplatform_ key). A tenant key gets 403.
client.capacity_list_queue(limit=50)
client.capacity_list_stats()

debounce_delete_entries takes the raw key. Earlier releases sent the key unchanged. A key that is not valid base64url returned 400. A key that happens to be valid base64url, such as user123, was decoded to other bytes, so the call returned 204 and cancelled nothing. If your code encoded the key itself to work around this, pass the raw key now.

Users (Server-Only)

client.users_list()
client.users_get("user_123")
client.users_create(body={"email": "alice@example.com", "password": "s3cret", "roles": ["admin"]})
client.users_patch("user_123", body={"name": "Alice"})
client.users_patch_password("user_123", body={
    "current_password": "old-s3cret",
    "new_password": "new-s3cret",
})
client.users_delete("user_123")

Organizations, Roles, Policy Set & Tenants

# Organizations — full CRUD
client.orgs_list()
client.orgs_get("org_123")
client.orgs_create_orgs(body={"name": "my-org"})  # platform credential; orgs_create always gets 403
client.orgs_patch("org_123", body={"name": "renamed"})
client.orgs_delete("org_123")

# Roles — full CRUD
client.roles_list()
client.roles_get("role_123")
client.roles_create(body={"name": "editor", "org_id": "org_123"})
client.roles_patch("role_123", body={"name": "senior-editor"})
client.roles_delete("role_123")

# Rego policy set — authorization Layer 2, deny only
current = client.policy_set_list()                       # version 0 when the org has no set
saved = client.policy_set_update(                        # a new set starts in "monitor"
    body={"modules": {"authz.rego": rego}, "mode": "monitor"},
    if_match=str(current["version"]),                       # the version you read
)
client.policy_set_validate(body={"modules": {"authz.rego": rego}})
client.policy_set_dry_run(body={
    "input": {"request": {"action": "runs:delete"}, "subject": {"roles": ["member"]}},
})
client.policy_set_list_versions()
client.policy_set_get_versions("3")                      # one version, with its files
client.policy_set_versions_rollback("3", if_match=str(saved["version"]))  # keeps the active mode

# The platform organization's set (needs a platform key). The names differ from the tenant methods.
client.platform_policy_set_list()                                # get the active set
client.platform_policy_set_update(body=..., if_match="4")        # replace the set
client.platform_policy_set_create_validate(body=...)
client.platform_policy_set_create_dry_run(body=...)
client.platform_policy_set_list_versions()
client.platform_policy_set_get("3")                              # one version, with its files
client.platform_policy_set_create_rollback("3", if_match="4")
# Break-glass: switch a tenant organization's set to monitor mode. reason is required.
client.platform_policy_set_create("org_acme", body={"reason": "enforce set denies policy routes, ticket 4412"})

# Tenants
client.tenants_list()
client.tenants_provision(body={"org_name": "Acme", "env_name": "production"})

A 412 means another admin wrote first. A 503 with the code policy_set_not_propagated is not retryable: the write committed as the version in the message. Send the same request again with if_match set to that version (a no-op that finishes propagation). Never send it again with the old version. See Policy set errors.

Error Handling

from ironflow.client import IronflowError

try:
    event = client.events_get("nonexistent")
except IronflowError as e:
    print(f"Status: {e.status_code}")  # 404
    print(f"Code: {e.code}")            # "not_found" — the lowercase Connect code
    print(f"Message: {e}")              # the server's message

IronflowError.code is the lowercase Connect code (not_found, invalid_argument, already_exists, …) on the REST client and on IronflowRPC. A domain code keeps its own name. A reply with no code gets the Connect code of its HTTP status.

Method Count

Run make sdk-health for the current coverage report. The count is not reproduced here because it changes on every regeneration, and a stale number in the docs is worse than no number.

Note that route coverage is not the same as usable coverage: WebSocket and watch/streaming endpoints (/ws, config watch, KV watch) are not generated as REST methods, because they cannot be called as request/response JSON. The hand-written watch_kv and watch_config cover KV and config watch. The /ws endpoint has no method. ConnectRPC server streams are a different mechanism and are supported — see ConnectRPC client.

ConnectRPC client

IronflowRPC and AsyncIronflowRPC cover the capabilities the REST API does not expose. Both are generated from the protobuf definitions via Buf and Connect for Python; see ADR 0062 for why the wire protocol is not hand-written.

from ironflow import IronflowRPC
from ironflow.rpc.v1 import CreateWebhookSourceRequest

with IronflowRPC(server_url="http://localhost:9123", api_key="ifkey_...") as rpc:
    source = rpc.webhooks.create_source(
        CreateWebhookSourceRequest(name="Stripe", event_prefix="stripe")
    )
    print(source.id)

Methods take one generated protobuf request object and return generated protobuf responses. Import those types from ironflow.rpc.v1 — never from ironflow._gen, which is private and free to change.

Capability namespaces

Namespace	Methods	Covers
rpc.functions	9	Register, get, list, update status, delete, version history, get-at-version, rollback, direct invoke
rpc.runs	17	Trigger, TriggerSync, InvokeFunctionSync, TriggerBatch, get, list, steps, cancel, delete, delete many, redact run, redact step, patch step, resume, pause, paused state, inject step output
rpc.events	3	Emit an event, redact an event’s payload, explain an event’s match
rpc.environments	2	Get an environment and rotate its compatibility API key
rpc.streams	8	Entity streams — append, read, info, list, history, create and get snapshots, delete
rpc.projections	18	Register, get, list, definition history and exact revisions, status, rebuild lifecycle, pause/resume, SQL projections, catch-up waits, plus two streams
rpc.query	1	Raw SQL against projection tables
rpc.event_schemas	8	Register, get, list, delete, definition history and exact revisions, dry-run upcast, check whether enforcement is on for an event name
rpc.webhooks	11	The whole management API — sources, secrets, ingest tokens, deliveries
rpc.audit	2	Per-run audit trail and auth audit trail
rpc.agent_tools	4	Register, invoke, unregister, list
rpc.time_travel	3	Run state at a timestamp, run timeline, step output at a timestamp
rpc.pubsub	12	Emit, publish, consumer groups, acknowledge a consumer-group event (AckEvent), topic listing, topic stats, plus two streams
rpc.deployments	4	Register, get, list executable deployments, and read a function’s execution routing. Request, response, error and permission reference: Executable Deployments

102 methods across 14 namespaces: 98 unary and 4 server streams.

Which RPCs each namespace exposes — and which are deliberately absent because REST already delivers them — is recorded row by row in sdk/python/rpc-capabilities.yaml.

Async

The async client mirrors the sync one method for method. Note aclose(), not close():

from ironflow import AsyncIronflowRPC
from ironflow.rpc.v1 import GetWebhookSourceRequest

async with AsyncIronflowRPC(server_url="http://localhost:9123", api_key="ifkey_...") as rpc:
    source = await rpc.webhooks.get_source(GetWebhookSourceRequest(id="whs_123"))

Streams

Four server streams are exposed. Sync methods return an Iterator, async methods an AsyncIterator — the method itself is not awaitable, so iterate it directly:

from ironflow.rpc.v1 import SubscribeRequest

with IronflowRPC(server_url="http://localhost:9123", api_key="ifkey_...") as rpc:
    for event in rpc.pubsub.subscribe(SubscribeRequest(pattern="topic:orders.*")):
        print(event.event_id)
        if done(event):
            break          # cancelling is just leaving the loop

Cancel by breaking out, or by closing the iterator explicitly — .close() on the sync one, await .aclose() on the async one. There is no separate cancellation API: Python’s generator protocol already is one.

Closing is the deterministic form, and on a long-lived event loop it is the one to prefer. Breaking out of the loop releases the connection only when the generator is finalized, which is a garbage-collection event you do not control; closing releases it at the point you close.

A stream also cannot outlive the client that created it. Creating one sends nothing — the request goes out on first iteration — so reading a stream after close() / aclose() raises IronflowRPCError rather than quietly opening an authenticated connection from a client you already shut down.

Following one entity stream

There is no entity-stream helper. Build the pattern entity:{type}.{id}.> and pass it to pubsub.subscribe:

for event in rpc.pubsub.subscribe(SubscribeRequest(pattern="entity:order.order-123.>")):
    handle(event)

Replaying recent events

Set options.replay to receive the last N events first. The stream then stays live. The server rejects a replay above 1000.

from ironflow.rpc.v1 import SubscribeOptions, SubscribeRequest

for event in rpc.pubsub.subscribe(SubscribeRequest(
    pattern="topic:orders",
    options=SubscribeOptions(replay=10),
)):
    handle(event)

A dropped connection ends a replay stream: replay is a count, not a position. To reconnect, use start_after_sequence (below).

Dropping a stale run frame

Frames can arrive out of order across nodes. RunVersionFilter.accept(event) drops strictly lower positive versions, ranking each run and each step independently by its full identity, including dots in step IDs. Ties and zero or absent versions pass without lowering the remembered version. A deletion is kept and remembers the run ID; subsequent run and step frames for that ID are dropped, including unversioned frames. Other topics pass. The filter is not thread-safe and belongs to one subscription. Its shared cache holds at most max_runs run and step identities (default 10,000); eviction forgets both ordering and deletion history. RunEvent.from_data(event.data) decodes run payloads; version 0 means unknown.

from ironflow import RunEvent, RunVersionFilter

keep = RunVersionFilter()
for event in rpc.pubsub.subscribe(SubscribeRequest(pattern="system.run.>")):
    if keep.accept(event):
        run = RunEvent.from_data(event.data)
        print(run.id, run.status, run.version)

Reconnecting a subscription

pubsub.subscribe is the one exposed stream a client can reposition, and it reconnects on a transport failure only when you positioned it. Set options.start_after_sequence to the last event.sequence you processed:

from ironflow.rpc.v1 import SubscribeRequest, SubscribeOptions

cursor = 0
for event in rpc.pubsub.subscribe(SubscribeRequest(
    pattern="topic:orders.*",
    options=SubscribeOptions(start_after_sequence=cursor),
)):
    handle(event)
    cursor = event.sequence

That field is both the cursor and the opt-in. Without it a dropped connection ends the stream, because there is nowhere honest to resume from: replay is a count rather than a position, and starting again from “now” silently skips whatever arrived while you were disconnected.

Reconnects use the same 3-attempt budget and backoff as unary retries, and the budget refills on every event delivered — so a long-lived subscription that reconnects normally over days does not eventually run out. timeout still bounds the whole subscription, reconnects included.

A resumed stream is at-least-once

The frame in flight when the connection dropped can arrive twice: the server sending it is not the same as you having processed it, and the protocol has no ack for a client to close that gap with. Make your handler idempotent, or dedupe on event.sequence.

start_after_sequence is rejected with InvalidArgument alongside replay (a count and a position are different starting points) or alongside consumer_group (that path is durable and server-positioned, and its position is shared by every member).

The other three streams

stream_events, wait_catchup_stream and join_consumer_group are positioned by the server, on a durable consumer keyed to the projection or group rather than to your connection. Calling them again resumes where they left off, so they take no cursor and are never auto-reconnected — a client-side position there would move a cursor other readers share.

Response size limit

read_max_bytes caps how much of a response body the client will read. It defaults to None — no cap — and applies to every call the client makes.

rpc = IronflowRPC(server_url=..., api_key=..., read_max_bytes=8 * 1024 * 1024)

Exceeding it raises IronflowRPCError with code == "resource_exhausted" and a message naming the limit. On a stream the cap applies per message, not to the subscription as a whole.

Timeouts

timeout is in seconds, and defaults to None — no deadline.

A positive timeout is never rounded down to zero: values below one millisecond resolve to a 1ms deadline rather than to no deadline at all. Values at or below zero are rejected — pass NO_TIMEOUT for “no deadline”.

rpc = IronflowRPC(server_url=..., api_key=..., timeout=10.0)   # every call
rpc.webhooks.list_sources(request, timeout=2.0)                 # this call only

Two things worth knowing:

• On a stream, the deadline bounds the whole subscription, not the gap between events. A client-wide timeout=30.0 ends a subscription after 30 seconds even while events are still arriving. Pass NO_TIMEOUT to opt one call out:

  from ironflow import NO_TIMEOUT
  for event in rpc.pubsub.subscribe(request, timeout=NO_TIMEOUT):
      ...

• IronflowClient.timeout and IronflowRPC.timeout are both in seconds but measure different things. The REST one bounds socket inactivity per attempt; this one bounds the entire call — including every retry and backoff below.

Retries

A unary call that fails with the Connect code unavailable is sent again. That code covers every transport failure the client can see — a refused connection, DNS, a socket dropped mid-response — and a server shedding load. Nothing else is retried: every other code is a decision the server would reach again identically, and deadline_exceeded has already spent the caller’s whole budget.

Defaults: 3 attempts, exponential backoff starting at 100 ms, doubling, capped at 10 s. The same schedule IronflowClient uses. Pass max_attempts=1 to switch retries off, or a larger number to widen the budget.

rpc = IronflowRPC(server_url=..., api_key=..., max_attempts=1)  # never retry

Only side-effect-free methods are retried

The protobuf definitions annotate them idempotency_level = NO_SIDE_EFFECTS, and the client reads that annotation. There is no list inside the SDK to fall out of date, and no per-call retry= override — the method’s own contract decides.

Everything else — every create, update, delete, rotate, and emit — is sent exactly once. A transport error cannot tell you whether the server committed the write before the connection dropped, so re-sending it would risk a duplicate in an append-only stream.

What can still go wrong

• A retried read may execute more than once on the server. A response lost on the way back looks exactly like a request that never arrived, so the retry re-runs the method. Harmless for a read by definition — which is precisely why the annotation, and not the error, gates the behaviour.
• A write is never retried for you. If you need to repeat one safely, repeat it yourself with an idempotency key (EmitRequest and PublishRequest both carry idempotency_key) and treat a failed write as unknown, not failed.
• timeout is not extended. It bounds the whole call, so a retry that could not finish before the deadline is not attempted at all — you get the last real error instead of a deadline error.

Errors

Failures raise IronflowRPCError, a subclass of IronflowError, so except IronflowError still catches everything this SDK raises. Errors raised part-way through a stream are translated too.

from ironflow import IronflowRPCError

try:
    rpc.webhooks.get_source(GetWebhookSourceRequest(id="nope"))
except IronflowRPCError as err:
    print(err.code)      # "not_found" — the Connect code, as a string
    print(err.details)   # structured details the server attached, if any

err.retryable is always False and err.status_code always 0. Those fields are inherited from IronflowError and do not mean here what they mean on an HTTP error. Connect codes are not HTTP statuses, and retryable means “there is nothing left for this client to do” — by the time you hold the error, either the method’s contract barred a retry or the attempt budget ran out. It is deliberately not a “safe to repeat” flag: that property belongs to the method, not the error, and surfacing it here would invite if err.retryable: call_again() on a write. Read err.code and decide for yourself.

Both RPC clients also take on_error=, the hook IronflowClient takes. It gets the error and an ErrorContext: method is the RPC method name, endpoint is /{Service}/{Method} and status_code is None. A hook that raises does not hide the call’s own error. AsyncIronflowRPC awaits an async hook.

Not exposed, on purpose

• Capabilities still delivered by the REST client. Each capability has one Python facade.
• Worker and projection-runner transports, which an SDK drives on your behalf.
• PubSubService/SubscribeBidirectional. Python has no client for it yet. Join a consumer group with join_consumer_group and ack with ack_event.

Testing

ironflow.testing.TestClient runs a function’s handler in-process, with no engine. mock_step(name, fn) replaces a step.run body, mock_invoke(function_id, fn) supplies step.invoke and step.invoke_async, and send_event(name, data) queues an event for step.wait_for_event. Sleeps return at once. An unqueued wait with on_timeout="return-null" immediately returns None; a default wait still fails for a missing fixture. send_event(name, None) queues a real event whose data is None. An unmocked step.run executes its real body, and step.invoke needs a mock. await emit(name, data) runs the first function with a trigger for the event and returns a TestRun with status, output, error, steps (a list of TestStepRecord: name, type, output, error), compensations_ran and step_output(name). ctx.step is a TestStep, with the same methods as ironflow.worker.Step.

import asyncio
from ironflow.testing import TestClient

tc = TestClient([process_order])
tc.mock_step("validate", lambda: {"valid": True})

run = asyncio.run(tc.emit("order.placed", {"order_id": "123"}))
assert run.status == "completed"

See the Testing Workflows guide for waiting, invoke and saga patterns.

Code Generation

This SDK is auto-generated from the Ironflow server’s route manifest using cmd/sdk-gen. To regenerate after server changes:

make sdk-manifest    # Extract endpoint manifest from running server
make sdk-gen-python  # Regenerate the REST client from that manifest
make proto-python    # Regenerate the ConnectRPC client from the protobufs
make test-python     # Run the SDK gate

Two pipelines, and only the first reads the manifest. The ConnectRPC half is generated by Buf from api/proto/ironflow/v1 and is regen-diffed by make proto-python-verify, which runs in CI — so a .proto change that nobody regenerated fails there rather than shipping.

make proto-python creates an isolated generator environment from the gen extra in sdk/python/pyproject.toml. It pins connectrpc and protoc-gen-connectrpc to 0.12.1, and protobuf-py and protoc-gen-py to 0.3.0. Keep each runtime/plugin pair aligned and regenerate when changing these pins. Applications only need the runtime dependencies; the generator plugins are not required to install or use the SDK.

make test-python-integration additionally builds a server and runs the SDK against it. Those tests skip in the ordinary run.