Agents & Workflows

Your first workflow

Pick a template name or ship an inline DAG. Routing, fallbacks, and budget enforcement are handled per step.

from kairosroute import KairosRoute

kr = KairosRoute(api_key="kr-your-key")

result = kr.workflows.execute(
    workflow="research",
    input={"goal": "Summarize the latest SoC announcements from April 2026"},
    budget={"max_cost_usd": 0.25, "on_exceed": "downgrade"},
)

print(result["outputs"]["summary"])
print(f"Cost: ${result['total_cost_usd']:.4f}, Latency: {result['total_latency_ms']}ms")

Stream step events

Render progress mid-flight. SSE emits step_started, step_completed, step_failed, and a final event carrying the full execution.

for event in kr.workflows.stream(
    workflow="research",
    input={"goal": "Latest AI chips"},
):
    if event["type"] == "step_completed":
        print(f"{event['step_id']}: {event['model_id']} ({event['latency_ms']}ms)")
    elif event["type"] == "final":
        print("Done:", event["result"]["outputs"])

Inline DAGs

Templates are just JSON. Ship your own:

result = kr.workflows.execute(
    workflow={
        "name": "refactor-and-test",
        "steps": [
            {"id": "plan", "prompt": "Plan a refactor for: {{goal}}", "tier": "balanced"},
            {"id": "code", "prompt": "Implement: {{steps.plan.output}}", "tier": "high",
             "capabilities": [{"kind": "code_generation", "required": True}]},
            {"id": "test", "prompt": "Write tests for: {{steps.code.output}}",
             "depends_on": ["code"], "tier": "balanced", "strategy": "speculative"},
        ],
        "outputs": {"code": "{{steps.code.output}}", "tests": "{{steps.test.output}}"},
    },
    input={"goal": "Extract auth into a middleware"},
)

Built-in templates

Ten canned DAGs cover the most common agent shapes. GET /v1/workflows returns their structure.

Tasks: multi-step state

A task is a logical agent run identified by task_id. Calls sharing a task id roll up into a single snapshot with cumulative cost, latency, call count, and an optional budget envelope. Workflows create tasks automatically; you can also create them implicitly by passing task_id to a chat completion.

# Retrieve current state of a task
snapshot = kr.tasks.get("task_abc123", include_receipts=True)
print(snapshot["task"]["status"])
print(snapshot["task"]["total_cost_usd"])
for r in snapshot["receipts"]:
    print(r["decision"]["model_id"], "->", r["execution"]["status"])

# Cancel a running task
kr.tasks.cancel("task_abc123")

Task states: pending → planning → running → completed / failed / cancelled / budget_exceeded. Paused tasks resume on the next matching call.

Budget enforcement

Attach a BudgetSpec to any workflow or task. Three policies:

Capability-aware routing

Declare what a step needs. KR picks a model that satisfies those capabilities at the lowest cost within the quality tier.

Override default capability → routing-requirement mapping per tenant via POST /v1/capabilities.

Routing receipts

Every decision KR makes is logged as a machine-readable RoutingReceipt: the candidate pool considered, why each was filtered or chosen, the selected model and fallback chain, the actual cost/latency/status, and classifier diagnostics. Pull them via tasks.get(..., include_receipts=true) or via the API reference. Receipts feed the built-in benchmarks page and the nightly model-scoring refit loop.