kairo-sdk for Python

Durable workflows that run inside your Python process. No workflow server, no broker: the kairo core runs here as WebAssembly, and SQLite or PostgreSQL holds the state.

Write a workflow as an ordinary async function. Every call it makes is recorded, so after a crash or a redeploy, running it again with the same id picks up where it stopped. Finished calls return their recorded results, and nothing that acts on the world runs twice.

GitHub · Serverless guide · TypeScript SDK · MIT

pip install "kairo-sdk[embedded]"     # Python 3.12+; the extra adds wasmtime

Quick start

import asyncio
from kairo_sdk import EmbeddedBackend, Kairo, SQLiteStore

async def main():
    k = Kairo(backend=await EmbeddedBackend.open(SQLiteStore("kairo.db")))

    @k.action("classify", effect="unprotected")     # safe to run again after a crash
    def classify(req, ctx):
        return {"risky": req["amount"] > 100}

    @k.action("refund-payment")                     # effect="real" (the default): never runs twice
    def refund_payment(req, ctx):
        return {"refunded": req["order"]}

    @k.workflow("refund")
    async def refund(ctx, req):
        verdict = await ctx.call("classify", req)
        if verdict["risky"]:
            await ctx.wait_for("approve")            # wait days for a human, at no cost
        await ctx.sleep(60)                          # a durable timer (seconds)
        return await ctx.call("refund-payment", req)

    await k.start()
    print(await k.run("refund", {"order": "o-42", "amount": 50}, id="refund-o-42"))
    await k.close()

asyncio.run(main())

Action handlers may be plain functions (they run on a thread) or async functions. Each takes the input and a context.

The model

Actions and their effects

A real step is dispatched only after its intent is committed to the database. Options: timeout ("30s", "5m"), destination (a rate-limit key), max_attempts and backoff (below), and url for actions over HTTP(S).

A handler that raises fails its step for good. To say otherwise, raise one of these:

Workflows

A workflow must be deterministic between its calls: read the clock and randomness through ctx.now() and ctx.random(), and do I/O in actions.

To change a workflow's function while runs of it are under way, give it a version. A run keeps the version it started with to its end, and a process drives only the runs whose version it has. New runs start with the current version; keep the old one, draining=True, until no run of it is left ("version" in get and list).

@k.workflow("refund", version="1", draining=True)  # finishes the runs that started with it
async def refund_v1(ctx, req): ...

@k.workflow("refund", version="2")  # new runs; one current version per name
async def refund_v2(ctx, req): ...

A change to an action's settings (timeout, retries, backoff) applies to its calls already under way. One exception: while a real call's attempt is out, the call does not go on under settings that are no longer real (it could run twice). The tick reports it as RUN_STUCK; restore the action, or settle the call with k.resolve, k.resolve_failed or k.cancel.

A call that fails or is cancelled raises CallError (run_id, action, status: "failed" or "cancelled", message); a cancelled one is a Cancelled too. A workflow whose function raises fails: run and result raise WorkflowError (id, message; its __cause__ is what the function raised, when it failed in this process). Both are RuntimeErrors.

Running

To stop a resident process without leaving a real step to review, await k.drain() before await k.close(): no new step starts here (workflows stop at their next call, to go on where they are driven next), and the steps already running finish and have their outcomes recorded. To bound the wait, async with asyncio.timeout(...): the steps go on, and close then stops them. close alone is a process that stops. The embedded runtime only.

A process that drives a workflow holds a lease on it. If the process stops, another process that has the workflow takes it up once the lease expires; while the lease lives, a run elsewhere waits instead of running the workflow twice.

Limits, per process: concurrency on Kairo (steps at once), and limit (steps at once) and rate (starts a minute) on an action, counted by its destination when it has one.

Logging and observing

What goes wrong in the runtime's background work (lease renewals, sweeps, timers, workflows driven in the background, callbacks) is logged to logging.getLogger("kairo_sdk"), or to Kairo(..., logger=...).

Kairo(..., observe=fn) gives fn an Observation (a dataclass) as each thing happens, to count, time and trace runs (metrics, OpenTelemetry) without kairo depending on either. It is called inside the runtime, so it must not block; one that raises is logged and the runtime goes on.

k = Kairo(backend=backend, observe=lambda o: metrics.record(o))

With kairod (HttpBackend), only STEP_STARTED and STEP_FINISHED of the steps this process's worker runs are observed (on the worker's threads): runs are kairod's.

Testing

kairo_sdk.testing.TestEnv gives your tests a kairo in memory (SQLite :memory:), in suspend mode, on a clock the test moves (from 2026-01-01T00:00:00Z, or at). Nothing waits in real time: a workflow that sleeps or waits raises Suspended, and await env.advance(seconds) fires what comes due, in order, each at its own time.

from kairo_sdk.testing import TestEnv

async with await TestEnv.open() as env:
    env.k.action("charge")(lambda o, ctx: f"charged {o['id']}")
    env.k.workflow("refund")(refund)        # sleeps a day, then waits an hour for "approve"
    await env.start()
    with self.assertRaises(Suspended):
        await env.k.run("refund", order, id="r-1")
    await env.advance(24 * 3600)
    await env.signal("r-1", "approve", "alice")
    assert await env.k.result("r-1") == "charged o-1 by alice"
    [f"{c.kind} {c.name}" for c in await env.calls("r-1")]  # call kairo.sleep, wait approve, call charge

await env.calls(id) lists the calls a workflow made, in the order it made them: kind (call, wait, workflow), name, input, output (an action's output, a wait's payload, a child workflow's result), status and error. env.now() is the clock (a datetime); other Kairo arguments go to TestEnv.open.

Storage

SQLiteStore("kairo.db")                       # the standard library's sqlite3; one file
PostgresStore(pool)                           # your psycopg_pool.AsyncConnectionPool

PostgreSQL also wakes a wait in one process when a run settles in another (LISTEN/NOTIFY). Finished runs are removed, together with everything they started, once they have been kept long enough: keep_finished on EmbeddedBackend.open, or KAIRO_KEEP_FINISHED (24h by default, 7d, forever). A workflow that is still waiting keeps its history for as long as it waits.

Serverless

With mode="suspend", an invocation advances the workflow as far as it can and returns: run raises Suspended at a timer or a signal. A scheduler brings it back.

k = Kairo(
    backend=await EmbeddedBackend.open(PostgresStore(pool)),
    mode="suspend",
    tick_secret=os.environ["CRON_SECRET"],    # "Authorization: Bearer ..." on /tick
    wake=schedule_one_off_tick,               # async (at: unix ms) -> None
)
define_workflows(k)
await k.start()
app = k.asgi_app()                            # an ASGI app: /action, /callback, /tick

• /tick: Cron or one-off schedulers call it. It takes up due timers and steps whose process died, and answers {"next": <unix ms>}.
• wake(at): Called before a call returns, with the earliest time something is due. Schedule exactly one wake-up then, instead of polling.
• tick_handler(make): The entry point for schedulers that call a function directly (e.g. AWS Lambda from EventBridge Scheduler).

Actions over HTTP(S)

An action can live elsewhere: on a GPU box, in another service, behind a URL.

k = Kairo(backend=backend, secret=os.environ["KAIRO_ACTION_SECRET"], callback_url="https://app.example.com/kairo/callback")
k.action("render", effect="unprotected", url="https://gpu.example.com/kairo/action", async_=True)(lambda i, ctx: None)

• Calls are signed (Kairo-Signature, HMAC-SHA256 with a timestamp) and carry an Idempotency-Key.
• A 200 answer is the result. A 202 means the result comes later, as a signed callback.
• 4xx fails the step. 5xx, a timeout or a broken connection make the outcome unknown.
• HTTPS is required, except to localhost or with allow_insecure=True. Trust your own CA with ca.

The serving side is the same k.asgi_app(), with the action's real handler. A handler's RetryableError is sent as a retryable failure; its UnknownOutcomeError is answered with 502 (and an async action sends no callback), which the caller takes as unknown.

With kairod

If you can run a resident process, kairod gives the same model microsecond hand-offs, RPM/TPM limits across processes, and workers in any language. The workflow code is the same:

k = Kairo("http://127.0.0.1:8420", worker="/var/lib/kairo/worker.sock")

Workers

To serve kairod's graph plans (not workflows written as code), a worker pulls tasks over kairod's worker protocol. It is the standard library only.

from kairo_sdk import Result, Worker

def handler(task, ctx):
    ctx.emit("streamed text")                 # live output
    if ctx.cancelled.is_set():                # the step was cancelled or timed out
        return Result(error="cancelled", retryable=True)
    return Result(output={"echo": task.input})

Worker("py-1", ["my.action"], handler, concurrency=8).run("/var/lib/kairo/worker.sock")

A handler always returns a Result, even when cancelled: that frees its slot. It tells a definite failure (retryable) from an unknown outcome (unknown).

Dify nodes (graphon)

kairo_sdk.graphon runs Dify's workflow nodes (graphon 0.7.0) as kairo tasks, for plans converted by compat/dify. Install with pip install "kairo-sdk[graphon]", then:

python -m kairo_sdk.serve --socket /var/lib/kairo/worker.sock --actions dify.template-transform,dify.http-request

License

MIT