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), and url for actions over HTTP(S).

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.

Running

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.

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