FerricStore Python SDK

Python SDK for FerricStore and FerricFlow.

Status: public alpha 0.13.1. APIs may change before 1.0, but the SDK is tested against command construction, queue/workflow handlers, leases, retries, history, indexed attributes, named values, idempotent create, worker loops, async flows, and local FerricStore integration scenarios.

Python SDK 0.13.1 requires FerricStore 0.11.4 or newer. With FerricStore 0.11.11 it negotiates compact Stream mode 34 for homogeneous auto-ID XADD batches, compact Pub/Sub mode 35 for homogeneous PUBLISH pipelines, and the pubsub_batch_v1 event codec for compatible subscriptions. Native wire protocol v1 and generic compatibility paths remain.

FerricFlow keeps each workflow or job's state and history in one durable place. It is an explicit durable state pipeline, not a hidden deterministic replay engine:

create -> claim -> handler -> transition/complete/retry/fail

Handlers should be idempotent because work can be retried after lease expiry, worker crash, or explicit retry.

Durability is the default contract. A workflow command returns success only after the state change is accepted through FerricStore's quorum path and written to disk.

First 10 minutes

1. Install

pip install ferricstore

For local development from this repo:

python3 -m venv .venv
. .venv/bin/activate
pip install -e ".[dev]"

2. Start FerricStore

Use a local FerricStore server with the FerricStore protocol listener enabled:

ferricstore start

If you are running from the FerricStore source repo, use that repo's documented server command. The SDK examples assume:

ferric://127.0.0.1:6388

The same clients can send the same commands through a FerricStore HTTP endpoint:

from ferricstore import FlowClient

client = FlowClient.from_url(
    "https://ferricstore.example.com",
    username="platform_worker",
    password="service-secret",
)
client.kv_set("key", "value")
assert client.kv_get("key") == b"value"

The path is Python SDK -> HTTP(S) endpoint -> FerricStore. The endpoint can be the in-process ferricstore-http server or a compatible gateway. Use http:// / https:// for HTTP and ferric:// / ferrics:// for a direct native connection. HTTP is a deliberate request/response transport: ordinary commands and pipelines use the same client API, while live or connection-affine sessions such as Pub/Sub, WATCH, and transactions remain native-only. The current JSON envelope accepts JSON-compatible values; byte arguments use an automatic tagged Base64 envelope when arbitrary bytes or maps must cross JSON. See Command executors for the complete boundary.

For HTTP ACL authentication, the SDK sends the username and password using standard HTTP Basic authentication. This mode requires https://. Static bearer_token authentication remains available for compatible endpoint deployments. The SDK keeps a bounded pool of HTTP/TLS connections to the endpoint, so successive commands avoid repeated TCP and TLS handshakes. For multiplexing through an HTTP/2-capable TLS gateway, install ferricstore[http2] and pass http2=True; use max_connections to cap physical sockets and max_concurrent_requests to cap in-flight commands independently. Binary-heavy callers can install ferricstore[compact] and pass compact=True to use the endpoint's MessagePack envelope without changing command methods.

Run the repository's complete HTTP-compatible integration surface through a real TLS listener and ACL boundary with:

FERRICSTORE_IMAGE=quay.io/ferricstore/ferricstore:0.11.14@sha256:f7d29befefa15bce4b3755bf786cf7620c814f13bbd336c0d9955581b323b60e \
  scripts/run_http_integration.sh

The runner creates a private CA, rejects an unauthenticated probe and a restricted user's forbidden SET, and supplies FERRICSTORE_USERNAME, FERRICSTORE_PASSWORD, and FERRICSTORE_CA_FILE to the SDK tests. Native-only sessions continue to run in scripts/run_native_integration.sh.

3. Query durable runs

Use parameterized FQL for bounded, partition-scoped reads. Cursors are opaque and must be reused with the same query and parameters.

from ferricstore import FlowClient

client = FlowClient.from_url("ferric://127.0.0.1:6388")
query = """
FROM runs
WHERE partition_key = @partition AND type = @type AND state = @state
ORDER BY updated_at_ms ASC
LIMIT 25
RETURN RECORDS
"""
params = {"partition": "partition-a", "type": "invoice", "state": "queued"}

result = client.query(query, params)
plan = client.explain(query, params)
indexes = client.query_indexes()

query_indexes() exposes each generation's bounded covering_fields and opaque format codec identities alongside build, validation, retirement, and statistics status. format.counter is None when an index has no exact-count prefix. A format change means the derived query projection must be rebuilt; authoritative Flow state is not rewritten.

For bounded request execution, pass deadline_ms as an absolute Unix timestamp in milliseconds to query, explain, or explain_analyze.

For reusable, composable queries, use the immutable query builder. It compiles to the same parameterized FQL1 and can be passed directly to query or explain:

from ferricstore import FlowFields, FlowQuery, flow_param

queued_invoices = (
    FlowQuery.runs()
    .where(
        FlowFields.partition_key.eq(flow_param("partition")),
        FlowFields.type.eq("invoice"),
        FlowFields.state.eq("queued"),
    )
    .order_by(FlowFields.updated_at_ms.desc())
    .limit(25)
    .return_records(FlowFields.run_id, FlowFields.state, FlowFields.updated_at_ms)
    .bind(partition="partition-a")
)

result = client.query(queued_invoices)

4. Create a durable queue item

from ferricstore import FlowClient, QueueClient

client = QueueClient.from_url("ferric://127.0.0.1:6388")
emails = client.queue(type="email")

emails.enqueue("email-1", payload=b"welcome:user-1", idempotent=True)

Use attributes for small indexed metadata you want to filter/count later:

emails.enqueue(
    "email-2",
    payload=b"welcome:user-2",
    partition_key="account-a",
    attributes={"account": "acme", "campaign": "summer"},
    idempotent=True,
)

flow = FlowClient.from_url("ferric://127.0.0.1:6388")
records = flow.list(
    "email",
    partition_key="account-a",
    attributes={"account": "acme"},
)
stats = flow.stats(
    "email",
    partition_key="account-a",
    attributes={"account": "acme"},
)

Attributes are not payload bytes. Use named values/value refs for large data.

FIFO Flow state policy is opt-in per state:

from ferricstore import FlowStatePolicy

flow.install_policy("email", states={"queued": FlowStatePolicy.fifo()})
emails.enqueue("email-3", payload=b"welcome", partition_key="account-a:email")

FIFO states require a partition_key; priority is for parallel states.

Policy writes are deep patches by default and return a typed snapshot with a monotonic generation. Use compare-and-swap when coordinating policy writers:

from ferricstore import StalePolicyGenerationError

snapshot = flow.policy_get("email")
try:
    snapshot = flow.install_policy(
        "email",
        expected_generation=snapshot.generation,
        states={"queued": FlowStatePolicy.fifo()},
    )
except StalePolicyGenerationError:
    snapshot = flow.policy_get("email")

Workflow install_policy() calls default to full replacement because the workflow definition is the source of truth. Pass replace=False to request a patch explicitly.

4. Run a queue worker

from ferricstore import QueueClient

client = QueueClient.from_url("ferric://127.0.0.1:6388")
emails = client.queue(type="email")


def send_email(job):
    print(f"send {job.id}: {job.payload!r}")
    return b"sent"


emails.worker(concurrency=10, batch_size=100).run(send_email)

If the handler raises, the default worker policy is retry.

5. Create a workflow/state machine

Use workflows when one durable flow moves through named states.

from ferricstore import WorkflowClient, complete, transition

client = WorkflowClient.from_url("ferric://127.0.0.1:6388")
order = client.workflow(
    type="order",
    initial_state="created",
    partition_by=("account_id", "order_id"),
)


@order.state("created")
def created(ctx):
    charge_result = ctx.step(
        name="charge-customer:v1",
        run=lambda: charge_card(
            ctx.payload,
            idempotency_key=f"{ctx.id}:charge-customer:v1",
        ),
        to_state="charge_recorded",
    )
    return transition("receipt_pending", payload=charge_result)


@order.state("receipt_pending")
def receipt_pending(job):
    send_receipt(job.id)
    return complete(result=b"ok")


order.start(
    "order-1",
    account_id="account-a",
    order_id="order-1",
    payload=b"order payload",
    idempotent=True,
)

order.worker(states=["created", "receipt_pending"], concurrency=10, batch_size=100).run()

ctx.step(...) is the easiest durable boundary inside a workflow handler. It stores the closure result, moves the workflow, and makes the refreshed lease available to the handler automatically. Its name must remain stable across retries. External calls still need a stable provider idempotency key because a worker can stop after the provider succeeds but before FerricStore commits the result.

For custom claim loops, use the same operations directly:

job = client.advance(job, to_state="validated")
job, result = client.step(
    job,
    name="charge-customer:v1",
    run=charge_customer,
    to_state="charge_recorded",
)

Both methods infer the workflow identity, current state, lease, and fencing token from job. step_continue() remains only as a deprecated low-level migration alias.

Derived workflow partitions use collision-free fpk:<byte-length>:<value> encoding in 0.6.0. Drain flows created with the old colon-joined derived keys before upgrading. Explicit partition_key values are unchanged.

6. Store and fetch named values

Use named values when different states need different pieces of data. Values are stored as FerricFlow value refs and are only hydrated when requested.

emails.enqueue(
    "email-2",
    payload=b"small routing bytes",
    values={
        "template": b"welcome template bytes",
        "profile": b"user profile snapshot",
    },
    idempotent=True,
)

emails.worker(claim_values=["template"]).run(send_email)

Fetch one or many values directly when needed:

profile = client.value_get(owner_flow_id="email-2", name="profile")
values = client.value_mget(
    owner_flow_id="email-2",
    names=["template", "profile"],
)

Use ValueConfig or value_max_bytes in production to cap large value reads.

7. Inspect history

record = emails.get("email-1")
history = emails.history("email-1")

print(record)
for event in history:
    print(event)

History is for debugging and audit. Handlers should use claimed job data and requested values, not history replay.

8. Common errors

9. Persist LangGraph and LangChain threads

Install the optional integration and compile a graph with a FerricStore-backed checkpointer:

pip install "ferricstore[langgraph]"

from ferricstore import FlowClient
from ferricstore.langgraph import FerricStoreSaver

client = FlowClient.from_url("ferric://127.0.0.1:6388")
checkpointer = FerricStoreSaver(client)
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "conversation-123"}}
result = graph.invoke({"messages": messages}, config)

LangChain's modern create_agent API accepts the same checkpointer plus a FerricStoreStore for memory shared across conversation threads. Install ferricstore[langchain], then pass both adapters when creating the agent:

from ferricstore.langgraph import FerricStoreSaver, FerricStoreStore

agent = create_agent(
    model=model,
    tools=tools,
    checkpointer=FerricStoreSaver(client),
    store=FerricStoreStore(client),
)

Reusing a thread_id restores one conversation; an explicit store namespace can be read from different thread IDs.

The graph definition remains application code. FerricStore persists thread checkpoints, metadata, pending writes, and interrupt/resume state. Use AsyncFerricStoreSaver with AsyncFlowClient for native asynchronous I/O. Use LangGraphFlow or AsyncLangGraphFlow inside a FerricFlow state handler when the graph also needs an outer business state machine, durable retries, leases, schedules, signals, or governance. The bridge maps each Flow identity to a collision-safe LangGraph thread and translates graph completion or interrupts into normal Flow outcomes. See LangGraph and LangChain persistence for the complete contract, LangChain example, and the distinction between thread checkpoints and cross-thread memory.

What you use

• QueueClient / AsyncQueueClient for durable queues.
• WorkflowClient / AsyncWorkflowClient for explicit durable state machines.
• FlowClient / AsyncFlowClient for advanced command-level control.
• FerricStoreSaver / AsyncFerricStoreSaver for LangGraph checkpoints and FerricStoreStore / AsyncFerricStoreStore for cross-thread agent memory.
• ScheduleRecord, ScheduleFireResult, ScheduleFireDueResult, EffectResult, ApprovalResult, CircuitBreakerStatus, BudgetResult, and GovernanceOverview for typed admin/governance responses with dict fallback.
• RetryPolicy, WorkerConfig, ValueConfig, and ExceptionPolicy for runtime defaults.
• RawCodec by default, JsonCodec when you want JSON payloads.
• client.command(...) as the FerricStore low-level command escape hatch.

Async quickstart

import asyncio

from ferricstore import AsyncQueueClient


async def main():
    client = AsyncQueueClient.from_url("ferric://127.0.0.1:6388")
    emails = client.queue(type="email")

    async def handler(job):
        await send_email_async(job.payload)

    await emails.worker(concurrency=100, batch_size=500).run(handler)


asyncio.run(main())

Production shape

Use one process/service to create work and a separate long-lived worker service to claim and complete work.

web/serverless producer -> FerricStore -> worker service

Before production, configure timeouts, lease duration, backpressure behavior, graceful shutdown, and value hydration caps. The ferric:// transport defaults to one multiplexed connection with 8 request lanes; only raise connection or lane counts after profiling shows client-side saturation.

Docs

• Documentation index
• Quickstart
• SDK guide
• Configuration
• Production readiness
• Data in workflows
• Worker runtime
• Async APIs
• LangGraph and LangChain persistence
• Use cases
• Testing
• Troubleshooting

Examples

• examples/order_workflow.py: two-state workflow.
• examples/queue_worker.py: queue producer and worker.
• examples/async_queue_worker.py: async queue producer and worker.
• examples/state_machine_workflow.py: explicit workflow runner.
• examples/langgraph_checkpoint.py: LangGraph state persisted in FerricStore.
• examples/langgraph_flow.py: run a checkpointed LangGraph from FerricFlow.
• examples/protocol_commands.py: FerricStore command helpers.
• examples/protocol_kv_benchmark.py: protocol SET/GET benchmark.
• examples/protocol_dbos_benchmark.py: protocol DBOS-style queued workflow benchmark.
• examples/dbos_style_benchmark.py: DBOS-style throughput benchmark.

Contributing

See CONTRIBUTING.md, CODE_OF_CONDUCT.md, SECURITY.md, and RELEASE.md.