Python SDK

The experimental Python SDK for Ironflow. 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.

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.

Available from v0.33.0. Published from sahina/ironflow-py over PyPI Trusted Publishing, so every release carries an attestation binding it to a public commit.

Experimental — client only

The Python SDK provides client API methods only (emit events, list runs, manage resources). Worker execution runtime (step.run, step.sleep, push/pull mode) is not implemented. Methods on routes the manifest annotates with schemas return a typed models.* TypedDict and take their query parameters as keyword arguments. A route the manifest does not annotate still returns Any and takes no query kwargs — use the BaseClient.request() escape hatch for those. See SDK Comparison for the full feature matrix.

Installation

pip install ironflow-py   # available from v0.33.0; the import name is `ironflow`

Requires Python 3.10+ and depends on connectrpc and pyqwest — ConnectRPC support is part of the default contract, not an optional extra (ADR 0062, #1781). The REST client itself still uses only urllib and json from the standard library.

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"))

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
)

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.

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 — server expects "name", "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 Value

with IronflowRPC(server_url="http://localhost:9123") 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=Value.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.

# No watch method: the KV watch endpoint is a long-lived stream and is
# deliberately not generated. Poll kv_list_buckets_keys()/kv_get_buckets_keys(),
# use rpc.pubsub.subscribe(), or use the Go or JavaScript SDK for a live watch.

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")

# No watch method: the config watch endpoint is a long-lived stream and is
# deliberately not generated. Poll config_get(), subscribe to the related
# event stream through ConnectRPC, or use the Go or JavaScript SDK for a watch.

Secrets (Server-Only)

client.secrets_list()
client.secrets_get("my-secret")
client.secrets_create(body={"name": "my-secret", "value": "s3cret", "description": "optional"})
client.secrets_update("my-secret", body={"value": "new-value"})  # PUT — full replacement
client.secrets_patch("my-secret", body={"description": "updated"})  # PATCH — partial update
client.secrets_delete("my-secret")

Server Introspection

client.health()        # GET /health
client.ready()         # GET /ready; raises IronflowError when not ready
client.capabilities()  # transports, features, version, auth_required

Entity Streams

from ironflow import IronflowRPC
from ironflow.rpc.v1 import (
    AppendEventRequest, CreateSnapshotRequest, 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"))

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.

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()

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, Policies & Tenants

# Organizations — full CRUD
client.orgs_list()
client.orgs_get("org_123")
client.orgs_create(body={"name": "my-org"})
client.orgs_patch("org_123", body={"name": "renamed"})
client.orgs_delete("org_123")

# Roles — full CRUD plus policy attach/detach
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")
client.roles_policies("role_123", body={"policy_id": "policy_456"})         # attach
client.roles_delete_policies("role_123", "policy_456")                       # detach

# Policies — full CRUD
client.policies_list()
client.policies_get("policy_456")
client.policies_create(body={
    "name": "allow-emit",
    "effect": "deny",
    "actions": "emit:*",
    "resources": "*",
})
client.policies_patch("policy_456", body={"name": "allow-all-emit"})
client.policies_delete("policy_456")

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

Error Handling

from ironflow.client import IronflowError

try:
    run = client.events_get("nonexistent")
except IronflowError as e:
    print(f"Status: {e.status_code}")  # 404
    print(f"Code: {e.code}")            # "NOT_FOUND"
    print(f"Message: {e}")              # "run not found"

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 deliberately excluded from the REST client, because they cannot be called as request/response JSON. 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	13	Trigger, TriggerSync, InvokeFunctionSync, TriggerBatch, get, list, steps, cancel, patch step, resume, pause, paused state, inject step output
rpc.events	1	Emit an event
rpc.environments	2	Get an environment and rotate its compatibility API key
rpc.streams	7	Entity streams — append, read, info, list, history, create and get snapshots
rpc.projections	16	Register, get, list, status, rebuild lifecycle, pause/resume, SQL projections, catch-up waits, plus two streams
rpc.query	1	Raw SQL against projection tables
rpc.event_schemas	6	Register, get, list, delete, 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	11	Emit, publish, consumer groups, topic listing, topic stats, plus two streams

86 methods across 13 namespaces — 82 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.

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.

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, which the server answers with Unimplemented.

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 test-python-integration additionally builds a server and runs the SDK against it. Those tests skip in the ordinary run.