Flows

Define durable execution boundaries for your AI agent workflows.

A flow is the durable boundary for one agent run — the unit your platform invokes and the runner executes. It matters because the flow is what you can later replay: every model call and tool call inside it is recorded at checkpoint boundaries, so a finished run can be reproduced faithfully and rerun with one input changed. A flow is a dynamic ZenML pipeline; it runs on the same stacks, server, and dashboard as your ZenML pipelines.

Everything inside the flow is tracked at checkpoint boundaries: persisted outputs, retry, replay, resume, and wait. Your harness (Pydantic AI, LangGraph, Claude Agent SDK, raw Python) lives inside the checkpoints. Your platform sits in front of the flow's invocation API. See Harness, Runtime, Platform for the bigger picture.

The shape of a flow

The @flow boundary defines the durable execution. Checkpoints inside it are the persisted replay boundaries; the flow body is ordinary Python that orchestrates those checkpoints. Side effects in the flow body itself are not automatically durable — put anything that needs to survive a crash, be replayed, or be rolled back behind a @checkpoint.

The flow body is the orchestration layer. The checkpoints inside are the replay boundaries.

Defining a flow

Decorate your orchestration function with @flow:

from kitaru import flow

@flow
def my_agent(url: str) -> str:
    data = fetch_data(url)
    return process_data(data)

The decorated function becomes a callable wrapper object. Inside the flow body, you compose checkpoints — the units of work whose outputs are persisted.

Running a flow

Use .run() to start an execution:

handle = my_agent.run(url="https://example.com")
print(handle.exec_id)   # unique execution identifier
# ... do other work ...
result = handle.wait()   # block until finished

.run() submits the execution and immediately returns a FlowHandle. The flow runs in the background while your code continues. Call handle.wait() when you need the result: it waits for the execution to finish, then returns the persisted run output from the flow's return statement.

For a synchronous one-liner, chain .wait():

result = my_agent.run(url="https://example.com").wait()

To target a remote stack for one execution, pass stack=:

handle = my_agent.run(url="https://example.com", stack="production")

Why the boundary matters: replay

Because the flow recorded every checkpoint, you can re-execute a finished run from a checkpoint with flow.replay(...). Keep the exec_id a run returns — that's the handle into replay.

handle = my_agent.run(url="https://example.com")
result = handle.wait()
exec_id = handle.exec_id

# Faithful rerun with no change = the control/baseline.
baseline = my_agent.replay(exec_id, at="fetch_data")

# Replay again with one flow input changed (a different model).
variant = my_agent.replay(
    exec_id,
    at="fetch_data",
    flow_overrides={"model": "claude-opus-4-8"},
)

at selects the checkpoint to re-execute from. flow_overrides changes the flow's inputs for the replay run (for example model or prompt_profile). Because the baseline reproduces the original run, a diff between baseline and variant isolates your change rather than replay noise. This re-executes the real run with one input swapped — it is not re-scoring stored outputs like an eval.

See Replay and Overrides for selector rules, checkpoint-level overrides, and the CLI/MCP entry points.

Deploying and invoking flows

Use .run() when the current source process is starting a flow directly. Use .deploy() when you want to save a reusable, versioned flow entrypoint that other processes can invoke later.

# Producer side: create a reusable deployment version.
my_agent.deploy(url="https://example.com")

# Consumer side: start a new execution from the deployed default route.
handle = my_agent.invoke(url="https://example.com")

.invoke() is the remote invocation verb for deployed flows. If you do not pass version= or tag=, it invokes the reserved default tag. To pin a specific version, pass version=2; to route through a named tag, pass tag="stable".

See Deployments for auto-versioning, tag semantics, auth context, and worked producer/consumer examples.

FlowHandle

A FlowHandle is returned by .run(). It gives you access to the running execution:

Property / Method	What it does
handle.exec_id	The unique execution identifier (a string you can store or log)
handle.status	Current execution status (refreshed on each access)
handle.wait()	Block until the execution finishes, then return the persisted run output
handle.get()	Return the persisted run output immediately if finished, otherwise raise an error

handle.get() does not wait. If the execution is still running, it raises a KitaruStateError. Use handle.wait() when you want to block. For flows that explicitly return a value, both methods return the saved run output. For flows that do not return a value, inspect the persisted artifacts instead.

How errors surface

If the flow execution fails, handle.wait() raises a typed KitaruExecutionError (or a more specific subclass) with the execution ID, final status, and the failure origin attached:

import kitaru

try:
    result = my_agent.run(url="bad-input").wait()
except kitaru.KitaruExecutionError as exc:
    print(exc.exec_id, exc.status, exc.failure_origin)

Runtime options

You can configure execution behavior at the decorator level (defaults) or override per-run via .run():

from kitaru import flow

@flow(retries=2, cache=False)
def my_agent(url: str) -> str:
    ...

# Override at call time
handle = my_agent.run(url="https://example.com", retries=3, cache=True)

Option	Default	What it controls
retries	0	Number of automatic retries on failure
cache	True	Whether checkpointed outputs can be reused from previous runs. Set False to disable.
stack	None	Target execution environment for this run (overrides the active stack default)
image	None	Container image for remote execution (string, mapping, or settings object). Supports base_image, requirements, environment, apt_packages, replicate_local_python_environment, and dockerfile.

Per-run values override decorator defaults. If you don't pass an override, the decorator default applies.

For stack, the full precedence chain is:

1. .run(..., stack="...")
2. @flow(stack="...")
3. kitaru.configure(stack="...")
4. KITARU_STACK
5. [tool.kitaru].stack in pyproject.toml
6. active stack selected via kitaru stack use ...

When a higher layer supplies stack, Kitaru binds that stack only for the submission of that execution and then restores the previous active stack. That override does not permanently switch your default stack.

Run, then deploy

.run() starts an ad-hoc execution from the current process. It's the right loop for iteration and for calls made from your own code.

For production, a flow is deployed: .deploy() captures an immutable versioned snapshot that consumers invoke by flow name. Tags route traffic between versions (default is the tag your platform normally targets), so you can roll a new version out without changing the invocation surface. Auth is workspace-scoped; there are no per-deployment tokens to rotate.

# Ad-hoc run from the current process
handle = my_agent.run(url="https://example.com")

# Deploy an immutable snapshot (new version each call).
# Parameterized flows take representative deployment-time inputs;
# consumers can override them at invocation time.
my_agent.deploy(url="https://example.com")

# Consumers that have the flow imported can invoke by attribute
handle = my_agent.invoke(url="https://example.com")

# Or invoke source-free by flow name (from Python, CLI, MCP, or HTTP)
from kitaru import KitaruClient
handle = KitaruClient().deployments.invoke(
    flow="my_agent",
    inputs={"url": "https://example.com"},
)

Rules to know

• Flow functions should compose checkpoints. The flow body is the orchestration layer — heavy work belongs in checkpoints.
• Use .run() to start flows directly from source. Direct calls (my_agent(...)) are not supported and raise KitaruUsageError. Use my_agent.run(...) for source-backed executions, or .invoke(...) for deployment-backed executions.
• Retries must be non-negative. Passing a negative retries value raises a KitaruUsageError.

Next steps

• Learn how to break work into durable units with Checkpoints
• Attach structured data to your executions with Logging and Metadata