Metadata-Version: 2.4
Name: kairo-sdk
Version: 0.4.1
Summary: Durable workflows that run inside your Python process: a WASM core and your database (SQLite or PostgreSQL). No workflow server.
License-Expression: MIT
Project-URL: Homepage, https://github.com/r-hashi01/kairo
Project-URL: Repository, https://github.com/r-hashi01/kairo
Project-URL: Documentation, https://github.com/r-hashi01/kairo/blob/main/sdk/python/README.md
Keywords: workflow,durable-execution,workflow-engine,serverless,webassembly,llm,agents
Requires-Python: >=3.12
Description-Content-Type: text/markdown
Provides-Extra: embedded
Requires-Dist: wasmtime; extra == "embedded"
Provides-Extra: graphon
Requires-Dist: graphon==0.7.0; extra == "graphon"

# 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](https://github.com/r-hashi01/kairo) · [Serverless guide](https://github.com/r-hashi01/kairo/blob/main/docs/serverless.md) · [TypeScript SDK](https://www.npmjs.com/package/kairo-sdk) · MIT

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

## Quick start

```python
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

| `effect` | Meaning | If the process dies while it runs |
|---|---|---|
| `"unprotected"` | Safe to repeat: an LLM call, a read, a computation | It runs again |
| `"real"` (the default) | Acts on the world: a payment, an email, a write | It stops for review (`blocked`). It is never re-run blindly |

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)](#actions-over-https).

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

| Raised | Means | The step |
|---|---|---|
| `RetryableError` | It did not take effect: try again | Is retried up to `max_attempts` (default 3; 1 for a real action), `backoff` apart (default `"200ms"`, doubled each time) |
| `UnknownOutcomeError` | It may have taken effect | Is never taken for success: a real step stops for review, an unprotected one is retried |

### Workflows

| In a workflow | Does |
|---|---|
| `await ctx.call(action, input)` | Runs an action once, however often the workflow runs again |
| `await asyncio.gather(ctx.call(...), ...)` | Runs calls at once |
| `await ctx.sleep(seconds)` | Waits in the database, not in memory: the process may stop meanwhile |
| `await ctx.wait_for(signal, timeout=None)` | Waits for `k.signal(id, signal, payload)` and returns the payload; past `timeout` (seconds) it raises `TimedOutError` |
| `await ctx.now()`, `await ctx.random()` | The time and a random number, the same each time the workflow runs again |
| `await ctx.workflow(name, input)` | Runs a child workflow |

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`).

```python
@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 `RuntimeError`s.

### Running

| | |
|---|---|
| `await k.run(name, input, id=..., meta=None)` | Starts workflow `id`, or resumes it, and returns its result. The id is an idempotency key. A caller that stops waiting (its task cancelled, or timed out) gets `StoppedError`; the workflow goes on |
| `await k.submit(name, input, id=..., meta=None)` | Starts workflow `id` and returns its id once the start is durable, without waiting |
| `await k.result(id)` | The result of workflow `id`, from any process, once it has finished |
| `await k.signal(id, name, payload)` | Delivers a signal to the first wait for it; a wait the workflow has not reached yet receives it when it does |
| `await k.cancel(id)` | Cancels the workflow and the calls it waits for |
| `await k.resolve(call_id, output)`, `await k.resolve_failed(call_id, message)` | Settles a call stopped for review (a real step whose outcome is unknown), or one stuck as above: it took effect with `output`, or it failed. The embedded runtime only |
| `await k.list(workflow=, status=, since=, until=, after=, limit=)` | Workflows (not child workflows or calls), oldest first, with `meta`, `created_at` and `updated_at` |

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.

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

| `kind` | When | With |
|---|---|---|
| `RUN_STARTED` (`"run.started"`) | This process made a new run | `run_id`, `plan`, `parent`, `workflow` |
| `RUN_SETTLED` (`"run.settled"`) | A run completed, failed, was cancelled, or stopped for review (`blocked`) | `status`, `error` |
| `STEP_STARTED` (`"step.started"`) | A step's handler is about to run | `action`, `step_id`, `attempt` |
| `STEP_FINISHED` (`"step.finished"`) | It returned; also when an outcome comes on the callback (`duration` None) | `status` (`STEP_OK`, `STEP_RETRYABLE`, `STEP_FAILED`, `STEP_UNKNOWN`, `STEP_WAITING`, `STEP_PENDING`), `error`, `duration` (seconds) |
| `RUN_STUCK` (`"run.stuck"`) | A tick could not take a run on (its event waits a lease period; the other runs go on) | `error` |

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.

```python
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

```python
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.

```python
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.

```python
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](https://github.com/r-hashi01/kairo#kairod-when-you-can-keep-a-process-running) gives the same model microsecond hand-offs, RPM/TPM limits across processes, and workers in any language. The workflow code is the same:

```python
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.

```python
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](https://pypi.org/project/graphon/) 0.7.0) as kairo tasks, for plans converted by `compat/dify`. Install with `pip install "kairo-sdk[graphon]"`, then:

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

## License

MIT
