# README.md

# kaboo-workflows

**YAML-driven multi-agent orchestration with AG-UI and CopilotKit support, built on [strands-agents](https://github.com/strands-agents/harness-sdk)**

[![Python 3.12+](https://img.shields.io/badge/python-3.12+-blue.svg)](https://www.python.org/)
[![License](https://img.shields.io/badge/license-Apache--2.0-blue.svg)](LICENSE)
[![Docs](https://img.shields.io/badge/docs-gl--pgege.github.io-blue.svg)](https://gl-pgege.github.io/kaboo-workflows/)

**[Documentation](https://gl-pgege.github.io/kaboo-workflows/)** ·
**[Configuration guide](docs/configuration/)** ·
**[Workflow guides](docs/workflows/)** ·
**[Examples](examples/)** ·
**[Live demo](https://github.com/gl-pgege/kaboo-workflows-demo)**

> Extended with native AG-UI protocol support for CopilotKit frontends and AgentCore deployment. Now maintained at [gl-pgege/kaboo-workflows](https://github.com/gl-pgege/kaboo-workflows) (originally forked from strands-compose — see [Attribution](#attribution)).

---

## Quick Start

### 1. Install

```bash
pip install kaboo-workflows[openai]
# or with uv
uv add kaboo-workflows[openai]
```

Extras: `openai`, `ollama`, `gemini`, `agentcore-memory`

### 2. Create a config

```yaml
# config.yaml
vars:
  OPENROUTER_API_KEY: ${OPENROUTER_API_KEY}

models:
  default:
    provider: openai
    model_id: anthropic/claude-sonnet-4
    params:
      client_args:
        base_url: https://openrouter.ai/api/v1
        api_key: ${OPENROUTER_API_KEY}

agents:
  assistant:
    model: default
    system_prompt: "You are a helpful assistant."
    tools:
      - ./tools/calculator.py

entry: assistant
```

### 3. Create a tool

```{.python notest}
# tools/calculator.py
from strands.tools.decorator import tool

@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression and return the result."""
    import ast
    result = eval(compile(ast.parse(expression, mode='eval'), '<expr>', 'eval'))
    return f"Result: {result}"
```

### 4. Start the server

```bash
OPENROUTER_API_KEY=sk-... uv run kaboo-serve config.yaml
```

That's it. Your agent is now serving AG-UI SSE on `http://localhost:8080/invocations`.

### 5. Test it

```bash
# Health check
curl http://localhost:8080/ping

# Send a message
curl -s -N -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{
    "thread_id": "thread-1",
    "run_id": "run-1",
    "messages": [{"id": "msg-1", "role": "user", "content": "What is 15 * 23?"}],
    "tools": [],
    "context": [],
    "state": {},
    "forwarded_props": {}
  }'
```

You'll see AG-UI events stream back:

```
RUN_STARTED → TEXT_MESSAGE_START → TEXT_MESSAGE_CONTENT →
TOOL_CALL_START → TOOL_CALL_ARGS → TOOL_CALL_END → TOOL_CALL_RESULT →
TEXT_MESSAGE_CONTENT → TEXT_MESSAGE_END → RUN_FINISHED
```

---

## How It Works

kaboo-workflows does three things:

1. **YAML → Agents**: Your config defines models, agents, tools, hooks, MCP servers, and orchestrations. `load()` resolves everything into live strands objects.
2. **Agents → AG-UI SSE**: `kaboo-serve` wraps the resolved agents with [ag-ui-strands](https://github.com/ag-ui-protocol/ag-ui/tree/main/integrations/aws-strands/python) and serves them as AG-UI Server-Sent Events.
3. **AG-UI → CopilotKit**: Any CopilotKit frontend connects directly. Generative UI, shared state, and human-in-the-loop work out of the box.

```
YAML config → load() → strands.Agent → StrandsAgent (AG-UI) → FastAPI SSE
                                                                    ↑
Browser → CopilotKit Runtime ──────────────────────────────────────┘
```

### AgentCore Deployment

kaboo-workflows is fully compatible with [Amazon Bedrock AgentCore](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-agui.html). Deploy with the AG-UI protocol flag:

```bash
agentcore configure -e my_server.py --protocol AGUI
agentcore deploy
```

AgentCore handles auth, session isolation, and scaling. Your `kaboo-serve` server serves the same `/invocations` (AG-UI SSE) and `/ping` endpoints that AgentCore expects.

---

## Model Providers

Swap providers by changing `models.default` in your YAML:

```yaml
# OpenRouter (any model via OpenAI-compatible API)
default:
  provider: openai
  model_id: anthropic/claude-sonnet-4
  params:
    client_args:
      base_url: https://openrouter.ai/api/v1
      api_key: ${OPENROUTER_API_KEY}

# AWS Bedrock
default:
  provider: bedrock
  model_id: us.anthropic.claude-sonnet-4-6-v1:0

# Local Ollama
default:
  provider: ollama
  model_id: llama3

# Direct OpenAI
default:
  provider: openai
  model_id: gpt-4o
```

---

## Multi-Agent Orchestration

Three orchestration modes, arbitrarily nestable:

### Delegate — agent as a tool

```yaml
orchestrations:
  team:
    mode: delegate
    entry_name: coordinator
    connections:
      - agent: researcher
        description: "Research the topic."
      - agent: writer
        description: "Write the report."
entry: team
```

### Swarm — autonomous handoffs

```yaml
orchestrations:
  review:
    mode: swarm
    entry_name: drafter
    agents: [drafter, reviewer, tech_lead]
    max_handoffs: 10
entry: review
```

### Graph — deterministic DAG

```yaml
orchestrations:
  pipeline:
    mode: graph
    entry_name: writer
    edges:
      - from: writer
        to: reviewer
      - from: reviewer
        to: publisher
entry: pipeline
```

---

## CLI Reference

| Command | What it does |
|---------|-------------|
| `uv run kaboo-serve config.yaml` | Start AG-UI SSE server (port 8080) |
| `uv run kaboo-serve config.yaml --port 9000` | Custom port |
| `uv run kaboo-workflows check config.yaml` | Validate config (no side-effects) |
| `uv run kaboo-workflows load config.yaml` | Full load + MCP health check |
| `uv run kaboo-workflows eval dataset.yaml` | Run a golden dataset and score it; exits non-zero on any failure |

`eval` runs every dataset item through the same wire path as production and
scores the outcomes with deterministic checks, budgets and LLM-as-judge. See
[evaluation](docs/evaluation.md).

---

## Using as a Library

```{.python notest}
from kaboo_workflows import load
from kaboo_workflows.adapters import create_agui_app

# Option 1: Programmatic agent use
resolved = load("config.yaml")
result = resolved.entry("Hello!")

# Option 2: Create a FastAPI app for custom middleware
app = create_agui_app("config.yaml")
```

`create_agui_app` lives in `kaboo_workflows.adapters` (it is intentionally not
re-exported at the top level, to keep the top-level surface small).

---

## Public API

The top-level `kaboo_workflows` package exports a curated surface; the full,
auto-generated reference for every public module lives on the
[documentation site](https://gl-pgege.github.io/kaboo-workflows/api-reference/).

```{.python notest}
from kaboo_workflows import (
    load, load_config, load_session, resolve_infra,   # config pipeline
    make_event_queue, EventQueue,                       # streaming
    StreamEvent, EventType,                             # event protocol
    AppConfig, ConfigInput, ResolvedConfig, ResolvedInfra,
    OrchestrationBuilder,
    node_as_tool, node_as_async_tool, serialize_multiagent_result,
    create_mcp_client, create_mcp_server, MCPLifecycle,
    EventPublisher, MaxToolCallsGuard, StopGuard, ToolNameSanitizer,
    AnsiRenderer, cli_errors,
)
```

Public subpackages (import directly for the rest of the surface):

| Import path | Highlights |
|-------------|-----------|
| `kaboo_workflows.adapters` | `create_agui_app` — the primary serving entrypoint |
| `kaboo_workflows.config` | `load`, `load_config`, `load_session`, schema models (`AgentDef`, `AppConfig`, …) |
| `kaboo_workflows.hooks` | `EventPublisher`, `ContextHook`, `HistoryHook`, `InterruptHook`, guards, `ToolNameSanitizer` |
| `kaboo_workflows.mcp` | `MCPClient`, `MCPServer`, `MCPLifecycle`, transports |
| `kaboo_workflows.tools` | `ask_user`, tool loaders, `node_as_tool` |
| `kaboo_workflows.converters` | `StreamConverter`, `OpenAIStreamConverter`, `RawStreamConverter` |
| `kaboo_workflows.renderers` | `AnsiRenderer` |
| `kaboo_workflows.types` | `EventType`, `StreamEvent`, `SessionManifest` family |

A completeness test (`tests/contract/test_public_api.py`) guarantees every public
symbol has a docstring and an autodoc page, so this surface can never drift out of
sync with the docs.

---

## Examples

```bash
# Serve any example
OPENROUTER_API_KEY=... uv run kaboo-serve examples/step1/config.yaml

# Or run as a REPL
OPENROUTER_API_KEY=... uv run python examples/step1/main.py
```

See [examples/](examples/) for the full list.

### Live demo

[kaboo-workflows-demo](https://github.com/gl-pgege/kaboo-workflows-demo) is a
runnable, end-to-end reference: this library serves a YAML multi-agent pipeline as
AG-UI SSE, behind a CopilotKit runtime
([kaboo-runtime](https://github.com/gl-pgege/kaboo-runtime)) and a React UI
([kaboo-react](https://github.com/gl-pgege/kaboo-react)). See
[the kaboo stack](https://gl-pgege.github.io/kaboo-docs/) for the whole picture.

---

## Developer Setup

```bash
git clone https://github.com/gl-pgege/kaboo-workflows.git
cd kaboo-workflows
uv run just install      # uv sync --all-groups --all-extras, plus git hooks

uv run just check        # lint + type check + security scan
uv run just test         # pytest with coverage (>=70% gate)
uv run just format       # auto-format
uv run just docs-llms    # regenerate llms.txt after changing docs
```

---

## Attribution

Originally forked from [strands-compose/sdk-python](https://github.com/strands-compose/sdk-python) (Apache 2.0), original work by [Michal Galuszka](https://github.com/strands-compose). Now maintained as [kaboo-workflows](https://github.com/gl-pgege/kaboo-workflows).

# docs/api-reference.md

# API reference

A curated map of the public surface, grouped by import path. Every module below
has a full, auto-generated reference (rendered from docstrings by mkdocstrings) —
follow the links for signatures, parameters, and source.

The public surface is guarded by `tests/contract/test_public_api.py`: every
symbol in a public `__all__` must resolve, carry a docstring, and appear on one
of these pages.

## Top level — [`kaboo_workflows`](api/kaboo_workflows.md)

The curated top-level exports, for the common case:

- **Config pipeline**: `load`, `load_config`, `load_session`, `resolve_infra`,
  `AppConfig`, `ConfigInput`, `ResolvedConfig`, `ResolvedInfra`
- **Per-run configs**: `parse_config_sources`, `validate_raw_config`,
  `load_session_config`, `resolve_run_clients` — the two-stage pipeline for a
  service whose runs submit their own workflow
  ([Chapter 17](configuration/Chapter_17.md#when-the-config-itself-is-per-request))
- **Orchestration**: `OrchestrationBuilder`
- **Streaming**: `make_event_queue`, `EventQueue`, `StreamEvent`, `EventType`
- **Tools**: `node_as_tool`, `node_as_async_tool`, `serialize_multiagent_result`
- **MCP**: `create_mcp_client`, `create_mcp_server`, `MCPLifecycle`
- **Hooks**: `EventPublisher`, `MaxToolCallsGuard`, `StopGuard`, `ToolNameSanitizer`
- **Auth**: `Principal`, `get_auth_context`, `set_auth_context`, `build_auth`,
  `RelayTokenAuth`, `M2MClientCredentialsAuth`, `StaticTokenAuth`
- **Forwarded props**: `get_forwarded_props`, `set_forwarded_props`
- **Run context**: `get_thread_id`, `get_run_id`, `get_turn_id` — which
  conversation, request and logical turn a tool or hook is running in
  ([recipe](recipes/tools-with-run-context.md))
- **Telemetry**: `init_telemetry`, `telemetry_enabled`, `current_trace_id`
- **Rendering / CLI**: `AnsiRenderer`, `cli_errors`
- **Errors**: `ConfigurationError`, `SchemaValidationError`,
  `UnresolvedReferenceError`, `CircularDependencyError`,
  `ImportResolutionError` (see `kaboo_workflows.exceptions`)

## Serving — [`kaboo_workflows.adapters`](api/adapters.md)

- `create_agui_app` — turn a config into a FastAPI AG-UI / CopilotKit app. This is
  the primary serving entrypoint and lives here (not re-exported top level).
  `session_config_key=` makes each run bring its own config;
  `allowed_mcp_hosts=` bounds what a submitted config may connect to.

## Auth — [`kaboo_workflows.auth`](api/auth.md)

The outbound MCP auth strategies an `auth:` block resolves to
([Chapter 9](configuration/Chapter_09.md#outbound-auth-the-auth-field)):
`RelayTokenAuth`, `M2MClientCredentialsAuth`, `StaticTokenAuth`,
plus `build_auth` to construct one from a name and params, and
`apply_auth_to_transport_options` to wire one into a transport.

Inbound identity is separate and lives at top level: `Principal`,
`get_auth_context`, `set_auth_context`. A `relay` strategy reads the current
`Principal`, which is what an AG-UI `auth=` verifier sets.

## Configuration — [`kaboo_workflows.config`](api/config.md)

Loaders, interpolation, infra resolution, and the full YAML schema models
(`AgentDef`, `ModelDef`, `OrchestrationDef`, `DelegateOrchestrationDef`,
`SwarmOrchestrationDef`, `GraphOrchestrationDef`, `MCPServerDef`, `MCPClientDef`,
`HookDef`, `SessionManagerDef`, …).

## Hooks — [`kaboo_workflows.hooks`](api/hooks.md)

`EventPublisher`, `HistoryHook`, `InterruptHook`, `MaxToolCallsGuard`,
`MultiAgentStopGuard`, `SessionStateHook`, `StopGuard`, `ToolNameSanitizer`,
`restore_session_state`, `session_state_snapshot`, `stop_guard_from_event`.

## MCP — [`kaboo_workflows.mcp`](api/mcp.md)

`MCPClient`, `MCPServer`, `MCPLifecycle`, `create_mcp_client`,
`create_mcp_server`, and the `sse` / `stdio` / `streamable_http` transports.

## Tools — [`kaboo_workflows.tools`](api/tools.md)

`ask_user`, the `load_tool*` discovery helpers, `node_as_tool` /
`node_as_async_tool`, `resolve_tool_spec(s)`, `serialize_multiagent_result`.

## Converters — [`kaboo_workflows.converters`](api/converters.md)

`StreamConverter`, `OpenAIStreamConverter`, `RawStreamConverter`.

## Renderers — [`kaboo_workflows.renderers`](api/renderers.md)

`AnsiRenderer`.

## Testing — [`kaboo_workflows.testing`](api/testing.md)

`ScriptedModel` (a model that replays a per-agent script), `Pipeline` and
`run_pipeline` (drive a config through the AG-UI path in-process), and
`TurnResult` (the events and activity a client would see). See
[Testing your workflows](testing.md).

## Types — [`kaboo_workflows.types`](api/types.md)

`EventType`, `StreamEvent`, and the `SessionManifest` family
(`AgentDescriptor`, `OrchestrationDescriptor`, `EntryDescriptor`,
`ModelDescriptor`, `NodeRef`, `EdgeRef`).

# docs/api/adapters.md

# kaboo_workflows.adapters

The serving layer: turn a resolved config into an AG-UI / CopilotKit app.

::: kaboo_workflows.adapters
    options:
      show_root_heading: true
      members_order: source

# docs/api/auth.md

# kaboo_workflows.auth

Outbound MCP auth strategies: relay, on-behalf-of, machine-to-machine and static
tokens. These are what an `auth:` block on an MCP client resolves to — see
[Chapter 9](../configuration/Chapter_09.md#outbound-auth-the-auth-field) for the
YAML form.

::: kaboo_workflows.auth
    options:
      show_root_heading: true
      members_order: source

# docs/api/config.md

# kaboo_workflows.config

Config schema, loaders, interpolation, and infrastructure resolution.

::: kaboo_workflows.config
    options:
      show_root_heading: true
      members_order: source

# docs/api/converters.md

# kaboo_workflows.converters

Stream converters that reshape agent events for external consumers.

::: kaboo_workflows.converters
    options:
      show_root_heading: true
      members_order: source

# docs/api/hooks.md

# kaboo_workflows.hooks

Lifecycle hooks: event publishing, guards, sanitizers, and interrupts.

::: kaboo_workflows.hooks
    options:
      show_root_heading: true
      members_order: source

# docs/api/kaboo_workflows.md

# kaboo_workflows

The top-level package. See [API reference overview](../api-reference.md) for the
curated map of the public surface grouped by import path.

::: kaboo_workflows
    options:
      show_root_heading: true
      members_order: source

# docs/api/mcp.md

# kaboo_workflows.mcp

Model Context Protocol client/server lifecycle and transports.

::: kaboo_workflows.mcp
    options:
      show_root_heading: true
      members_order: source

# docs/api/renderers.md

# kaboo_workflows.renderers

Terminal renderers for streamed agent activity.

::: kaboo_workflows.renderers
    options:
      show_root_heading: true
      members_order: source

# docs/api/testing.md

# kaboo_workflows.testing

Scripted models and an in-process pipeline for testing your own workflows. See
[Testing your workflows](../testing.md) for the guide.

::: kaboo_workflows.testing
    options:
      show_root_heading: true
      members_order: source

# docs/api/tools.md

# kaboo_workflows.tools

Tool discovery, loading, and node-as-tool adapters.

::: kaboo_workflows.tools
    options:
      show_root_heading: true
      members_order: source

# docs/api/types.md

# kaboo_workflows.types

Shared event protocol and session manifest models.

::: kaboo_workflows.types
    options:
      show_root_heading: true
      members_order: source

# docs/best-practices.md

# Best practices

Rules of thumb from running kaboo-workflows in production, each with the reason
behind it. The [recipes](recipes/index.md) show them in full.

## Configuration

- **Keep models and MCP clients apart from workflows, and refer to them by
  name.** The same workflow then runs against real models in production and
  scripted ones in tests, and a workflow stored in a database never carries a
  URL or a credential ([Testing](testing.md#testing-your-own-tools),
  [per-run workflows](recipes/per-run-workflows.md)).
- **Secrets live in the base config; escape `${` in everything else.** A
  submitted workflow is interpolated against the server's environment, so a
  user-authored `${…}` would resolve there too. `$${` is a literal `${`.
- **Grant tools by exact name, a few at a time.** `tool_filters.allowed` is an
  exact-match list. Narrow clients per job and a ceiling of about 20 tools per
  agent keep prompts small and tool choices good.
- **Validate on save, not on first run.** `load_session_config` checks a
  workflow exactly as a run would and names what is missing; add your own
  rules after it and refuse any unfilled placeholder.
- **Configure only what the model supports.** Adaptive thinking, prompt caching
  and media blocks are model features; a config written for one model can fail
  at the provider on another.
- **Use module paths in workflows submitted per run.** Relative paths resolve
  from a config file's directory; a submitted workflow has none, so they
  resolve from the service's working directory.

## Prompts and skills

- **Short prompts, reusable skills.** Inline only the rules that must always
  hold; list the rest so agents load them when a task calls for it
  ([Skills](recipes/skills.md)).
- **Tell agents to reload what compression removed.** Long runs compress old
  messages, and an agent will happily carry on from a skill it can no longer
  see.
- **Keep the start of a prompt stable.** Prompt caching keys on the prefix;
  append changing lines, such as today's date, at the end.
- **Write titles, labels and schema descriptions for their reader.** Stream
  titles and tool labels are read by users; schema field descriptions are read
  by the model. Neither should look like an identifier.

## Reliability

- **Put guard hooks on every agent.** Repair transcripts before each model
  call, refuse a call that has failed twice, stop cleanly when a sign-in
  expires, and cap tool calls ([Guard hooks](recipes/guard-hooks.md)).
- **Respect Bedrock's conversation rules.** Every tool call answered in the next
  message, no two messages in a row from the same role, no empty text blocks.
  A violation stops the run without an error; `ScriptedModel` enforces the
  pairing rule so tests catch it first.
- **One voice asks the user.** Grant `ask_user` to the agent that talks to the
  user, ask once with every open question, and treat a cancelled request as an
  answer ([Ask the user](recipes/asking-the-user.md)).
- **On capped hosts, set a response budget.** A long turn then pauses and
  continues in a fresh response instead of being cut off; on AgentCore also
  report busy while a run is open
  ([Serve on AgentCore](recipes/production-serving.md)).

## State and identity

- **Conversation state belongs to the client.** History and pending interrupts
  travel on the AG-UI state channel, so any replica can serve any turn and a
  restart loses nothing. Do not keep conversation state in process memory.
- **Bound anything process-local, and key it deliberately.** Thread id for the
  conversation, run id for the request, turn id for a side effect that must
  happen once across an interrupt and its resume
  ([Tools that need run context](recipes/tools-with-run-context.md)).
- **Identity comes only from the verified principal.** Forwarded props are
  client input; validate them like any other.
- **Relay a user's token only on clients built per run.** Per-run clients are
  what keep one user's token out of another's run
  ([Call MCP tools as the signed-in user](recipes/user-identity-mcp.md)).

## Cost

- **Cache the system prompt, the tool list and the conversation.** Three of
  Bedrock's four cache points pay for themselves on the first tool loop.
- **Think where it pays.** Thinking on the agent that plans and judges, off for
  the ones executing clear instructions.
- **Do not resend what the agent no longer needs.** Clear old tool results
  within a run and mask them in history between runs
  ([Cost and context control](recipes/cost-and-context.md)).

## Testing

- **Script a test for every workflow you ship.** A scripted run exercises the
  real tools, hooks, prompts and orchestration with no model, network or
  server ([Testing your workflows](testing.md)).
- **Test what runs, not what is stored.** For composed workflows, feed the
  composer's output to the pipeline and assert on the prompt the model
  received.
- **Prove restarts on a fresh pipeline.** Resuming on a new `Pipeline` with the
  persisted state is the only test that a pause survives a deploy.
- **Keep model evaluations few and deliberate.** Scripted tests prove
  behaviour; [evaluations](evaluation.md) prove answer quality, at the price of
  a real model call per item.

## Observability

- **Turn on telemetry** and send traces to an OTLP backend such as Langfuse
  ([Evaluation](evaluation.md#enable-telemetry)).
- **Log every model call's latency and retries with its run id.** It answers
  "why was that slow?" when tracing is off.
- **Watch token usage per run** after every prompt or config change; caching in
  particular breaks silently.

# docs/concepts.md

# Concepts

A mental model for how kaboo-workflows turns a YAML file into a running,
streaming multi-agent system.

## `load()` returns plain strands objects

kaboo-workflows is a **resolver**, not a runtime wrapper. `load(config)` reads
your YAML, resolves every reference, and returns live
[strands-agents](https://github.com/strands-agents/harness-sdk) objects —
`strands.Agent`, `Swarm`, `Graph` — with nothing wrapping them. Whatever you can
do with a hand-built strands object, you can do with the one kaboo hands back.

```mermaid
flowchart LR
    YAML["config.yaml"] -->|load| Resolved["ResolvedConfig"]
    Resolved --> Entry["resolved.entry<br/>(strands Agent / Swarm / Graph)"]
    Entry -->|"kaboo-serve"| SSE["AG-UI SSE endpoint"]
    SSE -->|"Server-Sent Events"| CK["CopilotKit frontend"]
```

## The pieces

A config wires together a handful of resolvable sections (each has its own
chapter in the [Configuration reference](configuration/README.md)):

| Section | What it is |
|---------|-----------|
| `models` | Named model providers (`bedrock`, `openai`, `ollama`, `gemini`). |
| `agents` | Named agents: a model, a system prompt, tools, hooks and MCP clients. |
| `mcp_servers` | MCP servers the library starts and stops for you. |
| `mcp_clients` | Connections to MCP servers, which agents attach by name. |
| `orchestrations` | How agents compose into a multi-agent system. |
| `attachments` | How references (files + custom entities cited via `@`) reach agents. |
| `session_manager` | Where conversation state persists between runs. |
| `entry` | The root agent or orchestration a run starts from. **The only required key.** |

Tools and hooks are not top-level sections — they are declared per agent, as
`agents.<name>.tools` and `agents.<name>.hooks`. Tools are Python `@tool`
functions loaded from files or modules, plus built-ins like `ask_user`; hooks are
lifecycle callbacks for event publishing, interrupts and guards. The remaining
root keys are settings rather than wiring: `history`, `telemetry`, `runtime`,
`log_level` and `version`, plus `vars` for interpolation.

## Orchestration shapes

Agents compose through orchestrations, which nest arbitrarily:

- **delegate** — an entry agent calls other agents as tools (see
  [deep nesting](workflows/deep-nesting.md)).
- **swarm** — agents hand off to one another until one produces the answer.
- **graph** — an explicit DAG of agents with edges and an entry point.

Those three are the whole list — `mode:` accepts nothing else. Two patterns
people look for as modes are properties of a graph rather than shapes of their
own:

- **parallel** — a graph node with several outgoing edges fans out, and the
  branches run concurrently (see [parallel](workflows/parallel.md)).
- **nested** — any orchestration can be a node inside another (see
  [swarm + graph](workflows/swarm-and-graph.md)).

## Serving: the AG-UI event stream

`kaboo-serve` (or `create_agui_app` in `kaboo_workflows.adapters`) wraps the
resolved entry with [ag-ui-strands](https://github.com/ag-ui-protocol/ag-ui) and
exposes it as an AG-UI SSE endpoint. A single run emits a well-formed event
stream:

```text
RUN_STARTED → TEXT_MESSAGE_* → TOOL_CALL_* → ACTIVITY_SNAPSHOT → ACTIVITY_DELTA* → RUN_FINISHED
```

`ACTIVITY_SNAPSHOT` events carry the hierarchical activity tree that
[kaboo-react](https://github.com/gl-pgege/kaboo-react) renders, while
[kaboo-runtime](https://github.com/gl-pgege/kaboo-runtime) can persist and
replay the whole stream. Human-in-the-loop pauses surface as interrupts on
`RUN_FINISHED` and resume cleanly — see
[human-in-the-loop](workflows/human-in-the-loop.md).

Each response carries one snapshot and then `ACTIVITY_DELTA` events holding a
JSON Patch against it, so the tree crosses the wire once rather than in full on
every change. The run's closing snapshot restates it whole. Pass
`activity_deltas=False` to `create_agui_app` to send snapshots throughout.

The tree keeps one copy of everything it describes: a group's `tools` own its
tool calls and its `timeline` refers to them by `toolUseId`, and the timeline's
text entries are the only copy of its streamed prose.

### Runs larger than one response

Some hosts cap how much a single HTTP response may carry — AWS AgentCore closes
the socket at 100 MB, which reaches the browser as a dropped connection rather
than an answer. Pass `response_budget=<bytes>` to `create_agui_app` and a run
that fills its response stops at the next tool call instead: it pauses there,
the response ends with a proper `RUN_FINISHED`, and the client resumes into a
response with a fresh budget.

The pause is an ordinary interrupt, which is what makes it cheap. The pending
tool call and the agent's state are persisted and restored by the machinery
that already serves approvals, so the tool runs exactly once and nothing is
recomputed. It is marked `reason: "continuation"`, and clients resolve it
immediately rather than prompting anyone; kaboo-react does this for you. The
turn id is unchanged across the resume, so the two responses render as one
turn. The default of `0` means no cap and no pausing.

## References & attachments

Anything a user cites from the frontend with `@` — an uploaded file or a pointer
to a custom entity (a table, a dashboard) — is a **reference**. The AG-UI layer
gives inline media to the entry agent automatically; the `attachments:` config
extends that to any agent in a pipeline via a lightweight **manifest**, an
on-demand **tool** (`fetch_attachment` / `list_references`), and opt-in
**inline** `ContentBlock`s for vision/doc-capable models. File attachments are
resolved by the built-in tool; custom object kinds are resolved by your own MCP
tool. See [Attachments & Multimodal](configuration/Chapter_19.md).

Attachment URL fetches funnel through a pluggable fetcher: configure
`attachments.base_url` / `authorization` for files behind your API's auth, or
register a `ReferenceFetcher` in code for multi-store routing.

## The host side channel: `forwardedProps`

The AG-UI protocol carries a free-form `forwardedProps` object on every run —
context the *host backend* sends the runtime that is not part of the
conversation and never reaches the model as text: run-scoped credentials,
tenant ids, per-run agent configuration. kaboo binds it per request; your
tools and hooks read it back with `get_forwarded_props()`, and every agent
also receives a copy in its state under `forwarded_props`.

Two built-in features consume it:

- **Authorized attachment fetching** — `attachments.authorization: forwarded_props:<key>`
  reads a run-scoped bearer token for own-origin fetches.
- **Runtime-submitted workflow configs** — with
  `create_agui_app(session_config_key="workflow_config")`, the run's own YAML
  arrives under that key and is merged over the service's base config. This is how
  one process serves many host-defined agent types: each run gets its own agents,
  orchestration, entry and MCP client sessions, and editing an agent type takes
  effect on the next turn without a restart. See
  [Chapter 13](configuration/Chapter_13.md#the-second-merge-mode-session-overlays).

  A submitted config supersedes `forwardedProps.agent_config`, which applied only
  `system_prompt` and `model_id` behind `runtime.allow_invocation_overrides`. That
  flag is **deprecated**: an overlay expresses both, plus the structure they could
  not.

Everything on this channel is only as trustworthy as whoever set it, so a host
should stamp these server-side rather than accept them from a browser — see
[server-side props](https://gl-pgege.github.io/kaboo-runtime/server-side-props/)
in kaboo-runtime for where that goes on a Node host.

## Statelessness: what a run keeps and what it carries

A run keeps nothing. Everything a conversation accumulates arrives with the turn
on the AG-UI **state channel** — `kaboo_history` for sub-agent transcripts,
`kaboo_session` for a pending human-in-the-loop gate — and leaves in the outgoing
`STATE_SNAPSHOT` the host persists.

That is what makes the rest safe: a service can rebuild its agents every run,
scale to a second replica, or restart mid-approval, and behave identically,
because the objects being rebuilt hold behaviour rather than memory.

## Where to go next

- **[Getting started](getting-started.md)** if you haven't run it yet.
- **[Configuration reference](configuration/README.md)** for the exhaustive
  option list.
- **[Troubleshooting](troubleshooting.md)** when something doesn't resolve.

# docs/configuration/Chapter_01.md

# Chapter 1: The Basics — Your First Config

[← Back to Table of Contents](README.md)

---

A kaboo-workflows config is a YAML file with a handful of top-level sections. The only truly **required** fields are `agents` (at least one) and `entry` (which agent or orchestration to call).

Here is the absolute minimum:

```yaml
agents:
  assistant:
    system_prompt: "You are a helpful assistant."

entry: assistant
```

That's it. One agent, one entry point. Load it in Python:

```{.python notest}
from kaboo_workflows import load

resolved = load("config.yaml")
result = resolved.entry("Hello!")
print(result)
```

`resolved.entry` is a plain `strands.Agent` — nothing wrapped, nothing magic. You can call it, inspect it, pass it around — it's the real deal.

## Root-Level Sections

Here is the full list of top-level keys you can put in a config file:

| Key | Type | Required | Description |
|-----|------|----------|-------------|
| `version` | string | No | Schema version. Only `"1"` is supported. Defaults to `"1"`. |
| `vars` | dict | No | Variable definitions for `${VAR}` interpolation. |
| `models` | dict | No | Named LLM model definitions. |
| `agents` | dict | **Yes** (at least one somewhere) | Named agent definitions. |
| `orchestrations` | dict | No | Named multi-agent orchestration definitions. |
| `mcp_servers` | dict | No | Named MCP server definitions (managed lifecycle). |
| `mcp_clients` | dict | No | Named MCP client connections. |
| `session_manager` | dict | No | Global session manager (inherited by all agents). |
| `entry` | string | **Yes** | Name of the agent or orchestration to use as the entry point. |
| `history` | bool | No | Default for per-agent conversation history when an agent omits `history:`. Default: `false`. See [Chapter 7](Chapter_07.md). |
| `attachments` | dict | No | How agents receive files and references. Default: reference mode with the fetch tool on. See [Chapter 19](Chapter_19.md). |
| `telemetry` | dict | No | OpenTelemetry export. Off unless `enabled: true`. See [Chapter 15](Chapter_15.md). |
| `runtime` | dict | No | Runtime behaviour switches (`allow_invocation_overrides`, `persist_session_state`). |
| `log_level` | string | No | Logging level for kaboo_workflows. Default: `"WARNING"`. |

Two keys behave differently from the rest. `vars` is consumed during
interpolation and removed before validation, so it never reaches the config
object. Top-level keys beginning `x-` are stripped the same way, which is what
makes them usable as anchor scratch pads ([Chapter 4](Chapter_04.md)).

Sections marked as **dict** are name-keyed dictionaries — you pick the name, and it becomes the identifier:

```yaml
models:
  my_fast_model:    # <-- you chose this name
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-4-6-v1:0
```

The `x-` prefix is reserved for scratch-pad keys (YAML anchors) — they are **stripped** before validation and never reach the schema. More on that in [Chapter 4](Chapter_04.md).

## What About `vars`?

`vars` is special. It's consumed during interpolation and **removed** before schema validation. You will never see it in the final `AppConfig` object. It exists only to feed `${VAR}` references. Covered in [Chapter 3](Chapter_03.md).

> **Tips & Tricks**
>
> - You can omit `models` entirely. If an agent doesn't specify a `model`, strands uses its default (Bedrock with the default model). Handy for quick prototyping.
> - `entry` must reference something defined in either `agents` or `orchestrations`. If it doesn't, you'll get a clear error at load time.
> - Names follow the pattern `[a-zA-Z0-9_-]` and are limited to 64 characters. Spaces and special characters are auto-sanitized to underscores.

---

[Next: Chapter 2 — Models →](Chapter_02.md)

# docs/configuration/Chapter_02.md

# Chapter 2: Models — Choosing Your LLM

[← Back to Table of Contents](README.md) | [← Previous: The Basics](Chapter_01.md)

---

The `models` section defines named LLM configurations. Each model has a `provider`, a `model_id`, and optional `params`.

```yaml
models:
  fast:
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-4-6-v1:0

  creative:
    provider: openai
    model_id: gpt-4o
    params:
      temperature: 0.9

  local:
    provider: ollama
    model_id: llama3.2
    params:
      ctx_size: 8192
```

## Built-in Providers

| Provider | Package Required | Example `model_id` |
|----------|-----------------|---------------------|
| `bedrock` | *(included)* | `us.anthropic.claude-sonnet-4-6-v1:0` |
| `openai` | `pip install kaboo-workflows[openai]` | `gpt-4o` |
| `ollama` | `pip install kaboo-workflows[ollama]` | `llama3.2` |
| `gemini` | `pip install kaboo-workflows[gemini]` | `gemini-2.0-flash` |

## How Agents Reference Models

Agents reference models by **name** — the key you defined in the `models` section:

```yaml
models:
  smart:
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-4-6-v1:0

agents:
  analyst:
    model: smart     # <-- references the "smart" model above
    system_prompt: "You analyze data."
```

## Inline Models

Don't want to name a model? Define it inline directly on the agent:

```yaml
agents:
  analyst:
    model:
      provider: bedrock
      model_id: us.anthropic.claude-sonnet-4-6-v1:0
    system_prompt: "You analyze data."
```

This is handy when only one agent uses a specific model — no need to pollute the `models` section. But if two agents share the same model config, **use a named model** to avoid duplication.

An inline model is also the simplest choice in a config submitted per run ([Chapter 13](Chapter_13.md)): the agents in an overlay replace the base's, but a named model has to exist somewhere, either in the base's `models` or in the overlay's own. Inline needs neither.

## Custom Model Providers

If the built-in four providers aren't enough, you can point `provider` to a custom `Model` subclass:

```yaml
models:
  my_custom:
    provider: my_package.models:CustomModel
    model_id: my-model-v1
    params:
      api_key: ${API_KEY}
```

The class must be a subclass of `strands.models.Model`. The `model_id` and `params` are passed to its constructor.

## The `params` Dict

`params` is a pass-through dictionary — whatever you put in it gets forwarded as `**kwargs` to the model constructor. This means you can set any provider-specific parameter:

```yaml
models:
  tuned:
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-4-6-v1:0
    params:
      max_tokens: 4096
      temperature: 0.7
      top_p: 0.9
```

### OpenAI client defaults: `timeout` and `max_retries`

For the `openai` provider, `params.client_args` is forwarded to the OpenAI SDK
client. kaboo fills two keys when you don't set them: `timeout: 180` (seconds)
and `max_retries: 2`. Without a timeout, a stalled connection to the provider
(or a router like OpenRouter) hangs the run forever with no error; with these
defaults it fails visibly and retries. Override either key to tune:

```yaml
models:
  default:
    provider: openai
    model_id: anthropic/claude-sonnet-4-6
    params:
      client_args:
        api_key: ${OPENROUTER_API_KEY}
        base_url: https://openrouter.ai/api/v1
        timeout: 300        # long-running document analysis
        max_retries: 1
```

### Bedrock client settings: `client_config`

For the `bedrock` provider, `params.client_config` sets the timeouts and retry
policy of the boto client the model reaches Bedrock through. kaboo turns it into
a botocore `Config` and passes it to strands' `BedrockModel` as
`boto_client_config`. Only these keys are accepted; a misspelt one fails when the
config loads rather than being ignored.

| Key | Value | Meaning |
|-----|-------|---------|
| `read_timeout` | seconds, > 0 | How long to wait for data on the response stream. Unset, it stays at strands' Bedrock default of 120 (botocore's own is 60). |
| `connect_timeout` | seconds, > 0 | How long to wait for a connection. Unset, botocore's 60. |
| `retries.max_attempts` | integer, ≥ 0 | Retries after the first attempt. |
| `retries.mode` | `legacy` \| `standard` \| `adaptive` | botocore's retry mode. |

```yaml
models:
  claude:
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-4-6-v1:0
    params:
      client_config:
        read_timeout: 840        # long generations on a slow stream
        connect_timeout: 10
        retries: {max_attempts: 4, mode: adaptive}
```

## Thinking

Claude models on Bedrock can think before they answer. `thinking:` says whether
they do: on a model, as the default for every agent that uses it, and on an
agent, as an override of its model's.

```yaml
models:
  sonnet:
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-5
    thinking: {effort: high, display: summarized}

agents:
  planner:
    model: sonnet              # thinks, as its model does
  worker:
    model: sonnet
    thinking: false            # answers without thinking
  checker:
    model: sonnet
    thinking: {effort: low}    # thinks less, with the model's display
```

| Value | Sent to Bedrock |
|-------|-----------------|
| `false` (or `off`) | `thinking: {type: disabled}` |
| `true` | `thinking: {type: adaptive}` |
| `{effort, display}` | `thinking: {type: adaptive, display}` and `output_config: {effort}`, each only when set |
| unset | nothing, so the model's own default applies |

- `effort` is `low`, `medium`, `high`, `xhigh` or `max`.
- `display: summarized` returns a readable summary of the thinking, which streams as
  `REASONING` events; `omitted` returns none of it. Thinking is billed the same
  either way.

An agent's `false` wins over its model's setting, a mapping replaces only the
fields it sets, and `true` takes the model's settings. The fields reach strands'
`BedrockModel` as `additional_request_fields`, so a model takes `thinking:` or
`params.additional_request_fields.thinking`, not both. Other providers reject
`thinking:` when the config loads.

Agents whose setting differs from their named model's run on a variant of it, one
per setting, built for the first such agent and reused by every agent and session
after it.

Which settings a model accepts is the provider's business: some Claude models
think by default, some reject `type: disabled`, and older ones have no adaptive
thinking. Check the model's documentation.

> **Tips & Tricks**
>
> - When combined with `vars`, you can swap models at runtime: `MODEL=gpt-4o python main.py`. See [Chapter 3](Chapter_03.md).
> - If you omit `model` on an agent entirely, strands picks its built-in default (Bedrock). This is fine for quick tests but explicit is better for production.
> - The `params` dict preserves types from YAML — integers stay integers, floats stay floats. This matters for parameters like `max_tokens` that must be an int.

---

[Next: Chapter 3 — Variables →](Chapter_03.md)

# docs/configuration/Chapter_03.md

# Chapter 3: Variables — Environment-Driven Config

[← Back to Table of Contents](README.md) | [← Previous: Models](Chapter_02.md)

---

kaboo-workflows supports Docker Compose-style `${VAR}` interpolation. Define variables in the `vars` block, reference them anywhere with `${VAR}`, and optionally provide defaults with `${VAR:-fallback}`.

```yaml
vars:
  MODEL: ${MODEL:-us.anthropic.claude-sonnet-4-6-v1:0}
  TONE:  ${TONE:-friendly}
  MAX_TOKENS: ${MAX_TOKENS:-1024}

models:
  default:
    provider: bedrock
    model_id: ${MODEL}
    params:
      max_tokens: ${MAX_TOKENS}

agents:
  assistant:
    model: default
    system_prompt: "You are a ${TONE} assistant."

entry: assistant
```

## Lookup Order

When kaboo-workflows sees `${SOMETHING}`, it resolves it in this order:

1. **`vars` block** — your YAML-defined variables
2. **Environment variables** — `os.environ`
3. **Default value** — the part after `:-`
4. **Error** — if none of the above, loading fails with a clear message

So `${MODEL:-gpt-4o}` means: "use the `MODEL` var if defined, then check the environment, then fall back to `gpt-4o`."

## Override at Runtime

```bash
# Linux/macOS
TONE=formal MODEL=gpt-4o python main.py

# Windows PowerShell
$env:TONE="formal"; $env:MODEL="gpt-4o"; python main.py
```

## Variable Chaining

Variables can reference other variables:

```yaml
vars:
  BASE_MODEL: us.anthropic.claude-sonnet-4-6-v1:0
  MODEL: ${BASE_MODEL}
```

This works because kaboo-workflows resolves `vars` in two sequential passes — the first pass resolves against environment variables, the second pass resolves cross-references between vars. Circular references (`A: ${B}`, `B: ${A}`) are caught and raise a clear error.

## Type Preservation

Here's a subtle but powerful feature: when the **entire** value is a single `${VAR}` reference (not embedded in a larger string), the original type is preserved.

```yaml
vars:
  MAX_TOKENS: 1024    # This is an integer in YAML

models:
  default:
    provider: bedrock
    model_id: some-model
    params:
      max_tokens: ${MAX_TOKENS}   # Resolves to integer 1024, not string "1024"
```

But if you embed it in a string, it becomes a string:

```yaml
system_prompt: "Use max ${MAX_TOKENS} tokens"  # "Use max 1024 tokens" (string)
```

This is important for parameters that expect a specific type (like `max_tokens` needing an int).

## Variables Without Defaults

If you reference a variable that doesn't exist and has no default, loading fails immediately:

```yaml
vars:
  MODEL: ${REQUIRED_MODEL}  # No :- default!
```

```
ValueError: Variable '${REQUIRED_MODEL}' is not set in 'vars:' or environment,
and no default was provided.
Use ${REQUIRED_MODEL:-fallback} to set a fallback value.
```

This is intentional — it forces explicit configuration for deployment-critical values.

### Set-but-empty is not unset

The check is whether the name **exists**, not whether it has a useful value. An
environment variable set to the empty string exists, so its default is *not*
applied and the empty string is substituted:

```bash
GATEWAY_URL= python app.py      # ${GATEWAY_URL:-http://localhost:8090} -> ""
```

That is why a variable you want to be mandatory is better left out of `.env`
entirely than written with a blank value — blank passes the check and fails
later, somewhere less obvious.

## Escaping: Passing `${...}` Through Untouched

Some downstream consumer may have its own `${...}` syntax that must survive
interpolation — a prompt template filled in later, for instance. Write `$${` to
emit a literal `${`:

```yaml
agents:
  assistant:
    system_prompt: "Substitute $${user_name} yourself, later."
```

That loads as `Substitute ${user_name} yourself, later.` A `$$` not followed by
`{` is left exactly as written, so ordinary shell or currency text needs no
special handling.

## What Interpolation Applies To

Interpolation runs **after** the YAML is parsed, walking every string in the
tree — keys' values, list items, nested dicts alike. Two consequences are worth
knowing:

- A value that is *exactly* `${VAR}` keeps its native type, so `${MAX_TOKENS}`
  resolving to `1024` stays an integer rather than becoming `"1024"`.
- A variable embedded in a larger string is coerced to text and concatenated,
  as you would expect.

Lookup order is `vars` first, then the environment, then the `:-` default.

## Per-Source Interpolation

When you use [multi-file configs](Chapter_13.md), each file's `vars` block is interpolated independently before merging. File A's vars don't leak into File B's interpolation.

> **Tips & Tricks**
>
> - Use variables for anything that changes between environments: model IDs, API endpoints, log levels, session directories.
> - The `${VAR:-default}` pattern is your best friend for making configs self-contained — they work out of the box but can be customized via environment.
> - `vars` is removed after interpolation — it never reaches schema validation. So you can put anything in there, even nested dicts and lists (though string/number values are most common).
> - Want to see what resolved? Use `load_config()` instead of `load()` — it returns the validated `AppConfig` without starting anything.

---

[Next: Chapter 4 — YAML Anchors →](Chapter_04.md)

# docs/configuration/Chapter_04.md

# Chapter 4: YAML Anchors — DRY Config Blocks

[← Back to Table of Contents](README.md) | [← Previous: Variables](Chapter_03.md)

---

YAML has a built-in reuse mechanism: **anchors** (`&name`) and **aliases** (`*name`). kaboo-workflows embraces this for eliminating copy-paste across your config.

## The `x-` Scratch Pad

Any top-level key starting with `x-` is treated as a scratch pad — it's stripped before schema validation. Use it to define reusable blocks:

```yaml
# Define reusable blocks
x-base_prompt: &base_prompt |
  You are a helpful assistant.
  Always be concise and clear.

x-safety_hooks: &safety_hooks
  - type: kaboo_workflows.hooks:MaxToolCallsGuard
    params: { max_calls: 15 }
  - type: kaboo_workflows.hooks:ToolNameSanitizer

x-model_params: &model_params
  max_tokens: 2048
  temperature: 0.7

# Use them
models:
  default:
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-4-6-v1:0
    params: *model_params        # Reuse the model params

agents:
  researcher:
    model: default
    system_prompt: *base_prompt   # Reuse the prompt
    hooks: *safety_hooks          # Reuse the hooks

  writer:
    model: default
    system_prompt: *base_prompt   # Same prompt, no copy-paste
    hooks: *safety_hooks          # Same hooks, no copy-paste

entry: researcher
```

## How Anchors Work

1. **Define** with `&name`: `x-my_block: &my_block { key: value }`
2. **Reference** with `*name`: `field: *my_block`

An alias is a **reference, not a copy** — every alias site and the anchor itself are the same
object in the parsed document, so a tool that mutates one sees the change everywhere. This
rarely matters for config you only read, but it is why you should not treat an alias as a
template you can tweak afterwards. The `x-` prefix is a kaboo-workflows convention — YAML anchors work on any key, but `x-` keys are cleaned up so they don't trigger "unknown field" errors.

## Combining Anchors with Variables

Anchors and variables work together beautifully:

```yaml
vars:
  TONE: ${TONE:-professional}

x-base_prompt: &base_prompt |
  You are a ${TONE} assistant.
  Keep responses clear and structured.

agents:
  assistant:
    system_prompt: *base_prompt  # Gets "${TONE}" which is then interpolated
```

The order is: YAML parsing (anchors resolved) → anchor stripping (`x-*` removed) → variable interpolation (`${VAR}` replaced). So the alias `*base_prompt` is expanded first, and then `${TONE}` within it is interpolated.

## Anchors for Type-Preserving Reuse

Unlike variables (which can become strings when embedded), anchors always preserve the original YAML structure. An anchor on a dict gives you a dict, an anchor on a list gives you a list, an integer stays an integer:

```yaml
x-model_params: &params
  max_tokens: 2048    # integer
  temperature: 0.7    # float

models:
  fast:
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-4-6-v1:0
    params: *params   # max_tokens is still int 2048, not string "2048"
```

## Anchors You Can Define Anywhere

You don't *have* to use `x-` keys. Anchors can be defined on any value:

```yaml
models:
  default: &default_model        # Anchor on an entire model definition
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-4-6-v1:0

  backup: *default_model          # Exact copy of 'default'
```

The `x-` prefix is simply cleaner for "scratch pad" blocks that don't belong to any real config section.

> **Tips & Tricks**
>
> - Use `x-` blocks at the top of your file to define your project's "design system" — shared prompts, hook lists, model params.
> - Pair anchors with variables for maximum flexibility: anchors handle structure reuse, variables handle value swapping.
> - YAML anchors are resolved by the YAML parser itself — kaboo-workflows doesn't even see them. This means they work exactly as documented in the YAML spec.
> - You can use `{ key: value }` inline syntax for short dicts in anchor definitions — great for concise params: `params: { max_calls: 15 }`.

---

[Next: Chapter 5 — Tools →](Chapter_05.md)

# docs/configuration/Chapter_05.md

# Chapter 5: Tools — Giving Agents Superpowers

[← Back to Table of Contents](README.md) | [← Previous: YAML Anchors](Chapter_04.md)

---

The `tools` field on an agent is a list of spec strings that tell kaboo-workflows where to find Python tool functions.

```yaml
agents:
  analyst:
    model: default
    tools:
      - ./tools.py                          # All @tool functions from this file
      - ./tools.py:count_words              # One specific function
      - ./utils/                            # All @tool functions from all .py files in dir
      - my_package.tools                    # All @tool functions from an installed module
      - my_package.tools:special_function   # One specific function from a module
      - strands_tools.http_request          # A tool from strands' built-in tools
    system_prompt: "You analyze text using your tools."
```

## Spec Formats

| Format | What It Loads |
|--------|---------------|
| `./file.py` | All `@tool`-decorated functions from the file |
| `./file.py:func_name` | One specific function (auto-wrapped with `@tool` if needed) |
| `./dir/` | All `@tool` functions from all `.py` files in directory (skips `_`-prefixed files) |
| `module.path` | All `@tool` functions from an installed Python module |
| `module.path:func_name` | One specific function from a module |

## Writing Tool Functions

Tool functions must be decorated with `@tool` from strands:

```python
# tools.py
from strands.tools.decorator import tool

@tool
def count_words(text: str) -> int:
    """Count the number of words in the given text."""
    return len(text.split())

@tool
def reverse_text(text: str) -> str:
    """Reverse the given text."""
    return text[::-1]
```

The decorator registers the function's name, docstring (used as the tool description for the LLM), and parameter schema (derived from type hints). Functions **without** `@tool` are silently ignored when scanning a file or module.

## Single Function Lookups and Auto-Wrapping

When you use the colon syntax to load a specific function (`./file.py:my_func`), kaboo-workflows does something helpful: if the function isn't decorated with `@tool`, it auto-wraps it for you (and logs a warning). This is safe because the intent is unambiguous — you explicitly named the function:

```yaml
tools:
  - ./helpers.py:calculate_tax   # Works even without @tool decorator
```

You'll see a warning in the logs:

```
WARNING | tool=<calculate_tax> | not decorated with @tool, wrapping automatically
```

For file/module-wide scanning (without `:`), only `@tool`-decorated functions are picked up. This prevents accidentally exposing internal helper functions.

## Path Resolution

Filesystem paths are resolved **relative to the config file**, not the working directory. This is critical — it means your config works regardless of where you run the Python script from:

```
project/
├── config.yaml          # tools: [./tools/analysis.py]
├── tools/
│   └── analysis.py      # Resolved relative to config.yaml's directory
└── main.py              # Can be run from anywhere
```

Module-based specs (`module.path:func`) use the standard Python import system — the module must be importable.

## Directory Scanning

The directory spec (`./dir/`) loads all `.py` files **directly inside** the directory, skipping any file whose name starts with `_`. It is not recursive — files in subdirectories are ignored, so nest deliberately:

```
tools/
├── _helpers.py     # Skipped (underscore prefix)
├── __init__.py     # Skipped (underscore prefix)
├── analysis.py     # Loaded — all @tool functions extracted
└── formatting.py   # Loaded — all @tool functions extracted
```

> **Tips & Tricks**
>
> - Organize tools in a directory when you have many of them. One file per domain: `tools/math.py`, `tools/text.py`, `tools/database.py`.
> - The `strands_tools` package has built-in tools like `http_request`, `file_read`, `shell` — use them with `strands_tools.http_request`.
> - Tools are resolved per agent, but that is not the same as isolation: `module.path:func` specs go through the normal import system, so two agents referencing one module share the same function object. Keep tools stateless rather than relying on a copy you do not get.
> - Tool function docstrings are sent to the LLM as the tool description. Write good docstrings — they directly affect how well the model uses your tools.
> - Type hints on tool parameters become the JSON schema the LLM sees. Use `str`, `int`, `float`, `bool`, `list[str]`, etc. The more specific your types, the better the LLM calls your tools.

---

[Next: Chapter 6 — Hooks →](Chapter_06.md)

# docs/configuration/Chapter_06.md

# Chapter 6: Hooks — Middleware for Agents

[← Back to Table of Contents](README.md) | [← Previous: Tools](Chapter_05.md)

---

Hooks are lifecycle callbacks that fire at specific points during agent execution — before/after invocations, before/after tool calls, before/after model calls. They're perfect for guardrails, logging, metrics, and custom behavior.

```yaml
agents:
  assistant:
    model: default
    hooks:
      - type: kaboo_workflows.hooks:MaxToolCallsGuard
        params:
          max_calls: 10
      - type: kaboo_workflows.hooks:ToolNameSanitizer
      - type: ./my_hooks.py:AuditLogger
        params:
          log_file: ./audit.log
    system_prompt: "You are a helpful assistant."
```

## Hook Specification Formats

Hooks can be specified in two ways:

**Inline object** — with `type` and optional `params`:

```yaml
hooks:
  - type: kaboo_workflows.hooks:MaxToolCallsGuard
    params:
      max_calls: 10
```

**String shorthand** — just the import path (no params):

```yaml
hooks:
  - kaboo_workflows.hooks:ToolNameSanitizer
```

Both the `type` field and the string shorthand accept:
- `module.path:ClassName` — for installed packages
- `./file.py:ClassName` — for local files (relative to config file)

## Built-in Hooks

kaboo-workflows exports eleven hook classes, but most are wired up for you when you enable
the feature they serve — `HistoryHook`, `InterruptHook`, `ReferenceHook`, `SessionStateHook`,
`ForwardedPropsHook`, `EventPublisher` and `MultiAgentStopGuard` all appear because of some
other config key, not because you listed them. These four are the ones you add by hand:

### `MaxToolCallsGuard`

Limits how many tool calls an agent can make in a single invocation. Two-phase behavior:

`max_calls` defaults to **25**. Two-phase behaviour:

1. **First violation** — injects a system message telling the LLM to stop and write a final answer.
2. **Second violation** — if the LLM ignores the warning and calls another tool, the loop is terminated.

```yaml
hooks:
  - type: kaboo_workflows.hooks:MaxToolCallsGuard
    params:
      max_calls: 15
```

### `ToolNameSanitizer`

Some models inject extra tokens into tool names (e.g., `search<|python_tag|>` instead of `search`). This hook strips those artifacts so strands can find the tool in the registry.

```yaml
hooks:
  - type: kaboo_workflows.hooks:ToolNameSanitizer
```

No params needed — just add it.

### `MCPCallMetaHook`

Attaches per-call metadata (MCP `params._meta`) to every MCP tool call the agent makes. The metadata is computed on `BeforeToolCallEvent` — in the **caller's request context**, where per-request contextvars (the inbound `Principal`, application-bound run state) are visible — and rides the JSON-RPC message itself.

Use this when a long-lived shared MCP client must carry a *per-request* credential or correlation id: header-based auth strategies (see the `auth:` field in [Chapter 9](Chapter_09.md)) resolve on the transport's writer task, whose context was snapshotted once at client start, so they cannot see per-request state on a shared client. `_meta` can, because it is bound at call time.

```yaml
agents:
  assistant:
    mcp: [platform]
    hooks:
      - type: kaboo_workflows.hooks:MCPCallMetaHook
        params:
          provider: ./mcp/auth.py:provide_call_meta   # (tool_use) -> dict | None
```

The provider is a callable `(tool_use) -> dict | None` (or an import spec resolving to one). Its result is merged into `_meta`; unless the provider set one, `toolCallId` is stamped with the strands `toolUseId`, giving the server a stable idempotency/correlation key per tool invocation. Omit `provider` to stamp only `toolCallId`. Non-MCP tools are untouched.

### `StopGuard`

A cooperative stop mechanism — set a flag on the guard and the agent stops cleanly at the next opportunity. Useful for external cancellation (e.g., user disconnects from a web socket).

`StopGuard` needs a Python callable for `stop_check`, so it's usually wired from Python rather than pure YAML:

```python
from kaboo_workflows.hooks import stop_guard_from_event

guard, stop = stop_guard_from_event()

# add `guard` to an agent's hooks, then later:
stop.set()
```

## Writing Custom Hooks

A hook is any class that subclasses `strands.hooks.HookProvider` and implements `register_hooks()`:

```python
# my_hooks.py
from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import AfterToolCallEvent, AfterInvocationEvent

class ToolCounter(HookProvider):
    """Counts tool calls and prints a summary after each invocation."""

    def __init__(self, verbose: bool = False):
        self._count = 0
        self._verbose = verbose

    def register_hooks(self, registry: HookRegistry, **kwargs):
        registry.add_callback(AfterToolCallEvent, self._on_tool)
        registry.add_callback(AfterInvocationEvent, self._on_done)

    def _on_tool(self, event: AfterToolCallEvent):
        self._count += 1
        if self._verbose:
            print(f"  Tool #{self._count}: {event.tool_use.get('name')}")

    def _on_done(self, event: AfterInvocationEvent):
        print(f"Agent used {self._count} tools this turn.")
        self._count = 0
```

Use it in YAML:

```yaml
hooks:
  - type: ./my_hooks.py:ToolCounter
    params:
      verbose: true
```

The `params` dict is spread as `**kwargs` to your class constructor.

## Hook Execution Order

Hooks fire in the order they're listed. First hook's callbacks run before the second hook's for the same event. This matters when hooks interact — for example, `ToolNameSanitizer` should run before hooks that inspect tool names.

## Available Hook Events

These are the strands lifecycle events you can listen to:

| Event | When It Fires |
|-------|---------------|
| `BeforeInvocationEvent` | Before the agent starts processing |
| `AfterInvocationEvent` | After the agent finishes |
| `BeforeModelCallEvent` | Before each LLM API call |
| `AfterModelCallEvent` | After each LLM API call |
| `BeforeToolCallEvent` | Before each tool execution |
| `AfterToolCallEvent` | After each tool execution |
| `BeforeNodeCallEvent` | Before a graph/swarm node executes |
| `AfterNodeCallEvent` | After a graph/swarm node executes |
| `BeforeMultiAgentInvocationEvent` | Before a multi-agent orchestration starts |
| `AfterMultiAgentInvocationEvent` | After a multi-agent orchestration completes |

## Hooks on Orchestrations

Orchestrations also support hooks — applied at the orchestration level, not the individual agent level:

```yaml
orchestrations:
  pipeline:
    mode: graph
    entry_name: writer
    hooks:
      - type: kaboo_workflows.hooks:MaxToolCallsGuard
        params: { max_calls: 30 }
    edges:
      - from: writer
        to: reviewer
```

> **Tips & Tricks**
>
> - Each agent gets **fresh hook instances**. Two agents with the same hook config get independent instances — no shared state between them.
> - `params` preserves YAML types. `{ max_calls: 15 }` passes `max_calls` as an integer, `{ log_file: ./out.log }` as a string.
> - Hooks are the right place for cross-cutting concerns: rate limiting, audit logging, cost tracking, safety guardrails.
> - Combine `MaxToolCallsGuard` and `ToolNameSanitizer` as a baseline for any agent that uses tools — they handle the most common edge cases.

---

[Next: Chapter 7 — Session Persistence →](Chapter_07.md)

# docs/configuration/Chapter_07.md

# Chapter 7: Session Persistence — Memory That Survives Restarts

[← Back to Table of Contents](README.md) | [← Previous: Hooks](Chapter_06.md)

---

By default, agents are stateless — each `load()` call starts fresh. The `session_manager` section enables persistent conversation history.

Pending human-in-the-loop interrupts are a separate concern and are handled for you — see [Pending interrupts survive a restart without a store](#pending-interrupts-survive-a-restart-without-a-store) at the end of this chapter.

## Global Session Manager

Define a session manager at the root level and **every agent** inherits it:

```yaml
session_manager:
  provider: file
  params:
    storage_dir: ./.sessions
    session_id: my-session-001

agents:
  assistant:
    model: default
    system_prompt: "You remember everything."

entry: assistant
```

## Built-in Providers

| Provider | Backend | Required Package |
|----------|---------|------------------|
| `file` | Local filesystem | *(included)* |
| `s3` | Amazon S3 bucket | *(included, needs AWS creds)* |
| `agentcore` | Bedrock AgentCore Memory | `pip install kaboo-workflows[agentcore-memory]` |

### File Provider

```yaml
session_manager:
  provider: file
  params:
    storage_dir: ./.sessions
    session_id: my-session
```

Sessions are stored as files in `storage_dir`. Delete the directory to start fresh.

### S3 Provider

```yaml
session_manager:
  provider: s3
  params:
    bucket_name: my-agent-sessions
    session_id: prod-session-001
```

Requires AWS credentials in the environment.

### AgentCore Provider

The `agentcore` provider requires a unique `actor_id` per agent and **cannot** be set globally — set it per-agent instead:

```yaml
agents:
  assistant:
    model: default
    system_prompt: "You are helpful."
    session_manager:
      provider: agentcore
      params:
        actor_id: assistant
        memory_id: my-memory-store
```

## Per-Agent Session Manager

Any agent can override the global session manager with its own:

```yaml
session_manager:
  provider: file
  params:
    storage_dir: ./.sessions

agents:
  persistent_agent:
    model: default
    system_prompt: "I remember."
    # Inherits the global file session manager

  stateless_agent:
    model: default
    system_prompt: "I forget."
    session_manager: ~             # <-- Explicit opt-out with YAML null (~)
```

Setting `session_manager: ~` (YAML null) on an agent **explicitly opts it out** of the global default. This is important — without this, it would inherit the global one.

## Session ID Resolution

When no `session_id` is provided, kaboo-workflows generates a random UUID — meaning each run gets a fresh session. The resolution order is:

1. **Runtime override** — via `load_session(..., session_id="abc")`
2. **`params.session_id`** — from YAML config
3. **Random UUID** — fresh session per run

## Custom Session Manager

For anything beyond the built-in providers, point `type` to your own class:

```yaml
session_manager:
  type: my_package.sessions:RedisSessionManager
  params:
    host: localhost
    port: 6379
```

The class must be a subclass of `strands.session.SessionManager`. When `type` is set, `provider` is ignored.

## Swarm and Graph Agents, and Sessions

**Important limitation**: an agent that is a node in a **Swarm or a Graph** orchestration **cannot** have a session manager. This is a strands-agents limitation, and it applies to both shapes — a delegate orchestration is unaffected. The check runs at load time, so a global session manager plus a swarm or graph agent fails before the first run rather than during it:

```
ConfigurationError: Agent 'drafter' is in a swarm or graph orchestration and cannot
have a session manager (source: global 'session_manager:' in config).
Strands does not yet support session persistence for Swarm or Graph node agents.
Fix: Add 'session_manager: ~' to agent 'drafter' to opt out of the global default.
```

The fix: add `session_manager: ~` to each swarm or graph node agent to opt out.

> **Tips & Tricks**
>
> - For development, `file` provider with a fixed `session_id` is great — restart your script and the agent remembers your conversation.
> - For server/API deployments, use `load_session()` with a per-request `session_id`. kaboo-workflows
>   computes a single `effective_session_id` from your value and threads it to every agent and
>   orchestration, so all agents in one request share the same session folder. See
>   [the multi-tenant pattern](#the-multi-tenant-server-pattern) below.
> - Delete the `.sessions/` directory to "factory reset" your agent's memory.

## The Multi-Tenant Server Pattern

For web servers where each HTTP request needs its own session:

```{.python notest}
from kaboo_workflows import load_config, resolve_infra, load_session

# Once at startup
app_config = load_config("config.yaml")
infra = resolve_infra(app_config)
infra.mcp_lifecycle.start()

# Per request
def handle_request(user_session_id: str, message: str):
    resolved = load_session(app_config, infra, session_id=user_session_id)
    return resolved.entry(message)
```

MCP servers are shared across sessions (started once), but agents and their conversation state are created fresh per session.

## Pending Interrupts Survive a Restart Without a Store

A session manager persists the *conversation*. An interrupt is different: when an agent pauses on `ask_user` or a gated tool, what has to survive is the open gate, and it lives in the agent object's own interrupt state. That object is per-process, so before this existed a restart between the question and the answer stranded the approval — the user clicked approve and got "No agent session found for resume".

There is nothing to configure. Serving through `create_agui_app`, the pending interrupt travels on the AG-UI **state channel** under `kaboo_session`, the same channel that carries `kaboo_history`:

- On the way out, it is written into the `STATE_SNAPSHOT` your host already persists.
- On the way in, it is read from `RunAgentInput.state` and restored onto the agent that runs the resume — including one that has never seen the conversation, which is how a resume works after a restart, on a second replica, or when the session is rebuilt per run.
- A new message instead of a resume supersedes the gate, on a fresh process as on a warm one. Nothing is restored, the paused call is closed, and the cleared state goes out, so a host replaying an older snapshot cannot bring back a gate already answered.

Hosts on [kaboo-runtime](https://github.com/gl-pgege/kaboo-runtime) get this end to end for free, because the runtime persists state snapshots per thread and replays them into the next run.

Turn it off in the one case where it would be wrong:

```yaml
runtime:
  persist_session_state: false
```

The gate arrives from the client, so trusting it means trusting the client. That is correct when the AG-UI endpoint is called by your own server (the supported topology — a browser talks to your API, your API talks to kaboo), and wrong if you expose `/invocations` straight to a browser, where a user could hand you back a gate you never issued.

Two details worth knowing:

- A warm agent's own state wins over the incoming copy, so a stale snapshot cannot resurrect a gate that was already answered.
- Swarm and Graph entries are not covered: strands does not yet support session persistence for orchestration node agents, so a multi-agent entry still relies on the process staying up. A plain-agent entry — including one with delegates — is fully covered.

---

[Next: Chapter 8 — Conversation Managers →](Chapter_08.md)

# docs/configuration/Chapter_08.md

# Chapter 8: Conversation Managers — Controlling Context Windows

[← Back to Table of Contents](README.md) | [← Previous: Session Persistence](Chapter_07.md)

---

Conversation managers control how the conversation history is managed — truncation, summarization, or no management at all. This is different from session persistence (which stores history) — conversation managers decide **what's in the context window** when the LLM is called.

```yaml
agents:
  assistant:
    model: default
    conversation_manager:
      type: strands.agent:SlidingWindowConversationManager
      params:
        window_size: 40
        should_truncate_results: false
    system_prompt: "You are a helpful assistant."
```

## Built-in Conversation Managers

These come from strands-agents directly:

| Class | What It Does |
|-------|-------------|
| `strands.agent:SlidingWindowConversationManager` | Keeps the last N messages in context |
| `strands.agent:SummarizingConversationManager` | Summarizes older messages to save context |
| `strands.agent:NullConversationManager` | No management — keeps entire history |

## Using Them

```yaml
# Fixed-window approach
conversation_manager:
  type: strands.agent:SlidingWindowConversationManager
  params:
    window_size: 20

# Summarization approach
conversation_manager:
  type: strands.agent:SummarizingConversationManager
  params:
    summary_ratio: 0.3
    preserve_recent_messages: 10

# No management (pass everything)
conversation_manager:
  type: strands.agent:NullConversationManager
```

## Custom Conversation Managers

Write your own by subclassing `strands.agent.conversation_manager.ConversationManager`:

```yaml
conversation_manager:
  type: ./managers.py:MySmartManager
  params:
    strategy: semantic_relevance
```

## What the Model Is Sent: `context:`

A conversation manager decides what an agent keeps. `context:` shapes only the
copy of that conversation each model call is sent: the agent's own messages, the
chat a client stores and the history written back for it stay exactly as they
were.

### Clearing old tool results

An agent working through many tool calls resends every earlier result on every
model turn. `clear_tool_results` replaces the older ones once the input grows:

```yaml
agents:
  analyst:
    mcp: [warehouse]
    context:
      clear_tool_results:
        trigger: 60000            # estimated input tokens before anything is cleared
        keep_last: 3              # results of the 3 most recent turns always stay
        min_chars: 1500           # shorter results always stay
        keep_tools: [warehouse___get_schema]
```

| Key | Default | Meaning |
|-----|---------|---------|
| `trigger` | *(required)* | Estimated input tokens (system prompt plus messages, four characters a token) above which results are cleared before a model call. The `structuredContent` strands keeps beside an MCP tool's result isn't counted, since no model is sent it. |
| `keep_last` | `3` | How many of the most recent turns keep their results; at least 1. A turn is every result of one model response, so a batch of parallel calls stays or goes together, and the newest batch, which the model has not read yet, is always sent whole. |
| `min_chars` | `1500` | Results shorter than this many characters always stay. |
| `keep_tools` | `[]` | Tools whose results always stay. |

A cleared result keeps its call and reads:

```text
Result cleared (warehouse___run_query, queryExecutionId=8f3c, tableId=sales): call it again to see it.
```

It names the tool and up to ten `key=value` pairs found in the result's JSON for
keys ending in `id`, `Id` or `ID`, so the model can call again for the data it
needs. A cleared result also loses its `structuredContent`. A delegate's
result, another agent's report, is never cleared: it cannot be fetched again
without rerunning that agent. The rewrite depends on nothing but
the messages and leaves a cleared result as it is, so the result reads the same
on every later call and a cached prompt prefix survives.

### Calls to tools an agent does not have

Every agent, with or without `context:`, turns a `toolUse` for a tool it does not
have into a text note on the copy its model is sent:

```text
Earlier step (analyze_sql, a tool you don't have): asked {"input": "revenue in 2024"}; returned Revenue in 2024 was $42M.
```

Such calls reach an agent through a transcript it did not write: the chat, when
the workflow changed between turns, or a `history.group` shared with an agent
whose tools differ. A model shown a call it cannot make tends to make it anyway.
A delegate (a name without `___`) keeps up to 2,000 characters of input and its
whole result; an MCP tool (`<target>___<tool>`) keeps 500 characters of input
and a 500-character excerpt of its result, with its full length noted. Calls to
tools the agent has are left alone, and so is every call the agent makes itself
during the invocation: only the messages it held at its first model call are
rewritten. A tool name it gets wrong therefore reaches its model as the call
and its error, which is what lets it correct the name.

> **Tips & Tricks**
>
> - Conversation managers are per-agent. Different agents in the same config can use different strategies.
> - For orchestration coordinators that do a lot of delegating, `SlidingWindowConversationManager` with a generous `window_size` prevents context overflow from long delegate tool results.
> - Unlike session managers, there's no "global" conversation manager — set it on each agent that needs one.

---

[Next: Chapter 9 — MCP →](Chapter_09.md)

# docs/configuration/Chapter_09.md

# Chapter 9: MCP — External Tool Servers

[← Back to Table of Contents](README.md) | [← Previous: Conversation Managers](Chapter_08.md)

---

The Model Context Protocol (MCP) lets agents connect to external tool servers. kaboo-workflows supports three connection modes and manages the full server lifecycle.

## Architecture

```
mcp_servers:  → Define managed local servers (kaboo-workflows starts/stops them)
mcp_clients:  → Define connections to servers (local, remote, or subprocess)
agents:
  my_agent:
    mcp: [client_name]  → Attach MCP clients as tool providers
```

## Mode 1: Managed Local Server

You define a server, kaboo-workflows starts it in a background thread before creating agents, and stops it on shutdown:

```yaml
mcp_servers:
  calculator:
    type: ./server.py:create
    params:
      port: 9001

mcp_clients:
  calc:
    server: calculator                # References the server above
    params:
      prefix: calc                    # Tools become calc_add, calc_multiply, etc.

agents:
  assistant:
    mcp: [calc]
    system_prompt: "Use calc tools for math."

entry: assistant
```

The `type` field points to a factory function that returns an `MCPServer` instance:

```{.python notest}
# server.py
from mcp.server.fastmcp import FastMCP
from kaboo_workflows.mcp import MCPServer

class CalculatorServer(MCPServer):
    def _register_tools(self, mcp: FastMCP) -> None:
        @mcp.tool()
        def add(a: float, b: float) -> float:
            """Add two numbers."""
            return a + b

        @mcp.tool()
        def multiply(a: float, b: float) -> float:
            """Multiply two numbers."""
            return a * b

def create(name: str, port: int = 9001) -> CalculatorServer:
    return CalculatorServer(name=name, port=port)
```

The factory receives `name` (from the YAML key) plus everything in `params`.

## Mode 2: Remote URL

Connect to an existing MCP server over HTTP — no server management needed:

```yaml
mcp_clients:
  aws_docs:
    url: https://knowledge-mcp.global.api.aws
    transport: streamable-http
    params:
      prefix: aws
      startup_timeout: 30
```

## Mode 3: Stdio Subprocess

Spawn a local process that speaks MCP over stdin/stdout:

```yaml
mcp_clients:
  filesystem:
    command: ["npx", "-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
    params:
      prefix: fs
```

## The `transport` Field

Transport auto-detection usually works, but you can override it:

| Transport | When to Use |
|-----------|-------------|
| `streamable-http` | Default for URLs and managed servers. Modern MCP transport. |
| `sse` | Older Server-Sent Events transport. Auto-detected if URL ends in `/sse`. |
| `stdio` | Set automatically for `command:` mode. Not valid for managed servers. |

## Client `params`

The `params` dict on an MCP client is forwarded to strands' `MCPClient` constructor:

| Param | Type | What It Does |
|-------|------|-------------|
| `prefix` | string | Prefix all tool names from this server (e.g., `calc_add`) |
| `startup_timeout` | number | Seconds to wait for the server to respond |
| `tool_filters` | mapping | Restrict which of the server's tools the agent sees |

`tool_filters` is forwarded to strands rather than implemented here, so its
matching rules are strands'. In practice that means **exact names, not globs**:

```yaml
mcp_clients:
  platform_read:
    url: ${GATEWAY_URL}
    params:
      tool_filters:
        allowed:
          - platform___get_record
          - platform___list_records
```

Filtering one client down to a read-only subset, and pointing a second client at
the same URL with the full set, is how one server becomes two capability levels.
Give each to a different agent — an agent holding both registers the overlapping
tools twice under one name.

## Client `transport_options`

Transport-specific options forwarded to the transport factory:

```yaml
mcp_clients:
  authenticated_server:
    url: https://internal.example.com/mcp
    transport_options:
      headers:
        Authorization: "Bearer ${API_TOKEN}"
```

Available options vary by transport:

- **stdio**: `env`, `cwd`, `encoding`, `encoding_error_handler`
- **sse**: `headers`, `timeout` (default 5), `sse_read_timeout` (default 300), `auth`, `httpx_client_factory`
- **streamable-http**: `headers`, `http_client`, `terminate_on_close`, `timeout`

### Timeouts on `streamable-http`

`timeout` takes either a number of seconds applied to every phase, or a dict
naming httpx phases individually. **Phases you do not name keep httpx's 5s
default**, which is the trap: a tool that streams for minutes needs `read`
raised explicitly.

```yaml
mcp_clients:
  slow_queries:
    url: ${GATEWAY_URL}
    transport: streamable-http
    transport_options:
      terminate_on_close: false
      timeout: { read: 840 }        # connect/write/pool stay at 5s
```

Declare `transport_options` **once per client**. YAML mappings are last-wins, so
a second `transport_options:` further down the same client silently replaces the
first — including a carefully raised `read` timeout — with no warning from the
parser.

When you pass your own `http_client`, both `headers` and `timeout` are ignored;
configure them on the client you supply.

## Outbound Auth — the `auth:` Field

`transport_options.headers` is a fixed dict, decided when the config is parsed.
That is enough for a static API key and useless for anything that has to be
resolved per call — a caller's own token, or one that expires. `auth:` is the
declarative form for those.

```yaml
mcp_clients:
  platform:
    url: ${GATEWAY_URL}
    transport: streamable-http
    auth:
      type: relay
      params:
        header: Authorization
```

`auth: relay` is shorthand for the same strategy with no params, which lands in
the same place because `Authorization` is the default header. `auth:` is not
valid alongside `command:` — a stdio subprocess has no HTTP request to attach a
header to.

### The three strategies

| `type` | Whose identity | Where the token comes from | Safe on a shared client |
|--------|----------------|----------------------------|-------------------------|
| `relay` | The inbound caller | The current request's principal, unchanged | **No** — see below |
| `m2m` | The workflow service itself | OAuth2 client-credentials grant, cached | Yes |
| `static` | Whoever the token belongs to | A fixed value you supply | Yes |

`relay` resolves identity from the ambient request context, which strands
snapshots when a client **starts**. A client started at boot has no caller to
relay, so it only works when clients are resolved per run — that
is, when the app is serving runs that submit their own config
(`session_config_key=`, [Chapter 17](Chapter_17.md)). `m2m` and `static` have no
such constraint because they do not depend on who is calling.

There is deliberately no on-behalf-of strategy. Token exchange belongs to
whatever sits between the agent and the service — a gateway can exchange
against an authorization server that knows what the downstream audience should
be, where this library would have to be told, per deployment, which of several
audiences each tool call needed. An `obo` strategy existed through 0.19.0 and
was removed in 0.20.0; `build_auth("obo", ...)` now raises.

### Common params

Every strategy accepts these:

| Param | Default | Purpose |
|-------|---------|---------|
| `header` | `Authorization` | Header name, or a **list** of names |
| `scheme` | `Bearer` | Prefix before the token; `""` sends the raw value |

`header` accepting a list (0.19.0) exists for gateways that consume the header
they authenticate on. A managed gateway that validates the caller on
`Authorization` and then replaces it with its own outbound credential leaves the
target seeing nothing — so the token has to be sent twice, under two names:

```yaml
    auth:
      type: relay
      params:
        header:
          - Authorization                 # the gateway authenticates on this
          - x-kaboo-run-token             # the target reads this one
```

The same value is written to every name listed. An empty list is rejected.

### Per-strategy params

**`relay`** — forward the caller's token untouched.

| Param | Default | Purpose |
|-------|---------|---------|
| `token` | principal's token | Explicit override, for a client bound to a captured token |
| `context_headers` | none | Headers to stamp with the current run's context, each mapped to `run_id`, `thread_id` or `forwarded_props.<key>` |

`context_headers` (0.24.0) tells the services behind a gateway which run a call
belongs to, and anything the client forwarded with it, such as the user's time
zone. A header whose value the run lacks is left off, and an unknown source is
rejected when the client is built:

```yaml
    auth:
      type: relay
      params:
        header: Authorization
        context_headers:
          x-run-id: run_id
          x-timezone: forwarded_props.timezone
```

**`static`** — a fixed token on every request.

| Param | Default | Purpose |
|-------|---------|---------|
| `token` | required | The value to send |

**`m2m`** — the service's own machine identity, cached until shortly before expiry.

| Param | Default | Purpose |
|-------|---------|---------|
| `token_url` | required | OAuth2 token endpoint |
| `client_id` / `client_secret` | required | Client credentials |
| `scope` | none | Space-delimited scopes |
| `audience` | none | Audience parameter (e.g. Auth0) |
| `extra` | `{}` | Extra form fields on the token request |

### How it reaches the wire

For `streamable-http`, the strategy is attached to a dedicated `httpx.AsyncClient`
built for that MCP client; for `sse` it becomes `transport_options.auth`. Either
way the token is resolved **per request**, not once at parse time. If the
strategy resolves to no token — `relay` with no principal in context, most often
— the header is simply omitted and the request goes out unauthenticated, so a
misconfigured relay looks like a 401 from the target rather than an error here.

## Lifecycle Management

kaboo-workflows handles the startup ordering automatically:

1. Start all MCP **servers** (in parallel)
2. Wait for all servers to be **ready** (TCP port check with configurable timeout)
3. Create agents (which auto-start MCP **clients**)

On shutdown (via context manager or `.stop()`):

1. Stop all **clients** first
2. Then stop all **servers**

Always use the MCP lifecycle context manager:

```{.python notest}
resolved = load("config.yaml")

with resolved.mcp_lifecycle:
    result = resolved.entry("Hello!")
```

Or for async contexts:

```{.python notest}
async with resolved.mcp_lifecycle:
    result = await resolved.entry.invoke_async("Hello!")
```

### How Long a Client Session Lives

A **server** is a process, so it belongs to the process that started it. A **client** is a session, and its lifetime is a choice:

- Serving one fixed config (`create_agui_app("config.yaml")`), clients are opened once and held for the process, so every run reuses them.
- Serving runs that submit their own config (`session_config_key=`, see [Chapter 17](Chapter_17.md)), each run resolves its own clients and they are closed when its stream ends.

Per-run clients cost a handshake per client per turn, which is small beside a model call, and they buy two things worth more than that. `MCPClientError: the client session is not running` stops being possible rather than being retried, because a session that cannot outlive its run cannot be found dead at the start of the next one. And `relay` auth becomes reachable, because strands captures the ambient identity when a client *starts* — a client started at boot has no caller to relay.

Servers are still shared either way, so a `server:` client declared by a run binds to the process's already-running server.

## MCPClientDef Validation

Exactly **one** of `server`, `url`, or `command` must be set on each client. Setting zero or more than one raises a validation error:

```
MCPClientDef requires exactly one of 'server', 'url', or 'command'; got none.
```

## Combining Multiple MCP Sources

A single agent can use tools from multiple MCP clients:

```yaml
agents:
  super_agent:
    mcp:
      - calc_client
      - aws_knowledge
      - filesystem
    system_prompt: "You have math, AWS docs, and filesystem access."
```

> **Tips & Tricks**
>
> - The `prefix` parameter is your friend. It namespaces tools to avoid collisions: `calc_add` vs `aws_add`.
> - For development, managed servers (Mode 1) are the most convenient — everything starts and stops with your script.
> - For production, prefer remote URLs (Mode 2) — deploy MCP servers independently and connect agents to them.
> - Server transport defaults to `streamable-http`. You can also use `sse` for older MCP servers.
> - MCP servers support `server_params` which are forwarded to FastMCP constructor — useful for `stateless_http`, `json_response`, etc.

---

[Next: Chapter 10 — Orchestrations →](Chapter_10.md)

# docs/configuration/Chapter_10.md

# Chapter 10: Orchestrations — Multi-Agent Systems

[← Back to Table of Contents](README.md) | [← Previous: MCP](Chapter_09.md)

---

Orchestrations wire multiple agents into collaborative systems. Define them under the `orchestrations` section — each one has a `mode` and references agents by name.

**Key rule**: agents and orchestrations share a single namespace. You can't have an agent named `team` and an orchestration named `team`.

## Mode: Delegate

A coordinator agent calls sub-agents as tool functions. Best for hub-and-spoke patterns where one agent directs others:

```yaml
agents:
  researcher:
    model: default
    system_prompt: "You research topics thoroughly."

  writer:
    model: default
    system_prompt: "You write polished articles."

  coordinator:
    model: default
    system_prompt: |
      For every request:
        1. Call researcher to gather facts.
        2. Pass the facts to writer for the final article.
      Delegate all work — don't write content yourself.

orchestrations:
  team:
    mode: delegate
    entry_name: coordinator
    connections:
      - agent: researcher
        description: "Research a topic and return structured facts."
      - agent: writer
        description: "Write a polished article from research material."

entry: team
```

**How it works**: kaboo-workflows **forks** a new agent from the `entry_name` agent's blueprint (model, system_prompt, hooks, tools) and adds delegate tools for each connection. The original `coordinator` agent is **never mutated**. Each connection becomes an async tool that the coordinator can call.

**Fields**:

| Field | Type | Required | Description |
|-------|------|----------|-------------|
| `mode` | `"delegate"` | Yes | |
| `entry_name` | string | Yes | The agent whose blueprint is forked as coordinator |
| `connections` | list | Yes | Sub-agents to wire as tools |
| `connections[].agent` | string | Yes | Name of the target agent or orchestration |
| `connections[].description` | string | Yes | Tool description the LLM sees |
| `session_manager` | dict | No | Override session manager for the forked agent |
| `hooks` | list | No | Additional hooks for the forked agent |
| `agent_kwargs` | dict | No | Override agent kwargs (merged with entry agent's kwargs) |

## Mode: Swarm

Peer agents hand off control to each other autonomously. No central coordinator — agents decide when to pass the baton:

```yaml
agents:
  drafter:
    model: default
    system_prompt: |
      Write initial code. When done, hand off to reviewer.

  reviewer:
    model: default
    system_prompt: |
      Review code. If issues found, hand back to drafter.
      If good, hand off to tech_lead.

  tech_lead:
    model: default
    system_prompt: "Make final approval decision."

orchestrations:
  review_team:
    mode: swarm
    agents: [drafter, reviewer, tech_lead]
    entry_name: drafter
    max_handoffs: 10

entry: review_team
```

**How it works**: strands' Swarm injects a `handoff_to_agent` tool into every agent in the list. Agents call this tool to transfer control. Execution continues until one agent decides to stop or `max_handoffs` is reached.

**Fields**:

| Field | Type | Required | Default | Description |
|-------|------|----------|---------|-------------|
| `mode` | `"swarm"` | Yes | | |
| `agents` | list[str] | Yes | | Agent names participating in the swarm |
| `entry_name` | string | Yes | | Which agent starts first |
| `chat_output` | string | No | | Member agent whose streamed text becomes the assistant reply when the swarm is an AG-UI entry point. Omitted, the adapter emits the final active node's text once the run completes. Must name one of `agents`. |
| `max_handoffs` | int | No | 20 | Maximum handoffs before termination |
| `max_iterations` | int | No | 20 | Maximum iterations |
| `execution_timeout` | float | No | 900.0 | Total execution timeout (seconds) |
| `node_timeout` | float | No | 300.0 | Per-agent timeout (seconds) |
| `session_manager` | dict | No | | Swarm-level session manager |
| `hooks` | list | No | | Swarm-level hooks |

**Limitation**: All swarm nodes must be plain agents — no nested orchestrations. Node agents cannot have session managers, which applies to graph nodes too (see [Chapter 7](Chapter_07.md#swarm-and-graph-agents-and-sessions)).

## Mode: Graph

Deterministic DAG pipeline with explicit edges. Agents execute in dependency order — independent nodes can run in parallel:

```yaml
agents:
  planner:
    model: default
    system_prompt: "Create a content outline."

  writer:
    model: default
    system_prompt: "Write content following the outline."

  editor:
    model: default
    system_prompt: "Edit for clarity and correctness."

orchestrations:
  pipeline:
    mode: graph
    entry_name: planner
    edges:
      - from: planner
        to: writer
      - from: writer
        to: editor

entry: pipeline
```

**How it works**: kaboo-workflows feeds the edges to strands' `GraphBuilder`, which constructs a topological execution plan. The `entry_name` must be a node with no incoming edges (the pipeline start).

**Fields**:

| Field | Type | Required | Default | Description |
|-------|------|----------|---------|-------------|
| `mode` | `"graph"` | Yes | | |
| `entry_name` | string | Yes | | Starting node (must have no incoming edges) |
| `chat_output` | string | No | | Node whose streamed text becomes the assistant reply when the graph is an AG-UI entry point. Must name a node in the graph. |
| `edges` | list | Yes | | Edge definitions |
| `edges[].from` | string | Yes | | Source node name |
| `edges[].to` | string | Yes | | Target node name |
| `edges[].condition` | string | No | | Python callable for conditional routing |
| `max_node_executions` | int | No | | Max times any node can execute |
| `execution_timeout` | float | No | | Total pipeline timeout (seconds) |
| `node_timeout` | float | No | | Per-node timeout (seconds) |
| `reset_on_revisit` | bool | No | false | Reset agent state when a node is revisited |
| `session_manager` | dict | No | | Graph-level session manager |
| `hooks` | list | No | | Graph-level hooks |

> **Tips & Tricks**
>
> - **Delegate** is best for "boss and workers" patterns — the coordinator has full control over when and how to call sub-agents.
> - **Swarm** is best for peer collaboration — agents negotiate among themselves. Great for review/revision cycles.
> - **Graph** is best for fixed pipelines — when you know the exact processing order. Parallel execution of independent nodes is automatic.
> - The `description` field on delegate connections is what the coordinator LLM sees as the tool description — make it clear and actionable.
> - In swarm mode, guide handoff behavior through system prompts — tell each agent *when* and *to whom* they should hand off.

---

[Next: Chapter 11 — Graph Conditions →](Chapter_11.md)

# docs/configuration/Chapter_11.md

# Chapter 11: Graph Conditions — Dynamic Routing

[← Back to Table of Contents](README.md) | [← Previous: Orchestrations](Chapter_10.md)

---

Graph edges can have conditions — Python functions that decide at runtime whether an edge should fire. This unlocks feedback loops and branching pipelines.

```yaml
orchestrations:
  pipeline:
    mode: graph
    entry_name: writer
    reset_on_revisit: true
    max_node_executions: 6
    edges:
      - from: writer
        to: reviewer
      - from: reviewer
        to: writer
        condition: ./conditions.py:needs_revision
      - from: reviewer
        to: publisher
        condition: ./conditions.py:is_approved
```

## Writing Condition Functions

Condition functions receive the graph execution context and return `True` or `False`:

```python
# conditions.py

def needs_revision(context: dict) -> bool:
    """Route back to writer if reviewer says REVISE."""
    last_output = str(context.get("last_output", ""))
    return "REVISE" in last_output.upper()

def is_approved(context: dict) -> bool:
    """Route to publisher if reviewer approves."""
    last_output = str(context.get("last_output", ""))
    return "APPROVED" in last_output.upper()
```

The condition spec format is the same as tools and hooks: `./file.py:function_name` or `module.path:function_name`.

## Loops and Revisits

When an edge condition creates a loop (reviewer → writer → reviewer), you need two settings:

- **`reset_on_revisit: true`** — Resets the agent's conversation state when it's visited again. Without this, the agent accumulates context from all previous visits.
- **`max_node_executions: N`** — Safety cap on how many times any node can execute. Prevents infinite loops.

## How Conditions Are Evaluated

For a given node, all outgoing edges are checked. Edges without conditions always fire. Edges with conditions fire only if the function returns `True`. If no outgoing edge fires, the pipeline stops.

> **Tips & Tricks**
>
> - Make your reviewer agent's output deterministic by instructing it to start with keywords like "REVISE:" or "APPROVED:" — this makes condition functions simple and reliable.
> - Always set `max_node_executions` when you have loops — it's your safety net against infinite cycles.
> - `reset_on_revisit` is usually what you want for revision loops — the writer should get fresh context each time, not accumulate all previous attempts.
> - Condition functions must be callable. If you accidentally point to a non-callable (like a string or class), kaboo-workflows will raise a clear error.

---

[Next: Chapter 12 — Nested Orchestrations →](Chapter_12.md)

# docs/configuration/Chapter_12.md

# Chapter 12: Nested Orchestrations — Composing Systems

[← Back to Table of Contents](README.md) | [← Previous: Graph Conditions](Chapter_11.md)

---

Named orchestrations can reference each other. A swarm can be plugged into a delegate as a tool. A delegate can be a node in a graph. Compose them however you want.

```yaml
agents:
  researcher:
    model: default
    system_prompt: "Research topics thoroughly."

  reviewer:
    model: default
    system_prompt: "Review and improve content."

  qa_bot:
    model: default
    system_prompt: "Run quality checks on content."

  coordinator:
    model: default
    system_prompt: |
      1. Send work to content_team.
      2. Pass results to qa_bot.
      3. Return the final output.

orchestrations:
  # Inner orchestration — built first
  content_team:
    mode: swarm
    agents: [researcher, reviewer]
    entry_name: researcher
    max_handoffs: 10

  # Outer orchestration — references inner by name
  manager:
    mode: delegate
    entry_name: coordinator
    connections:
      - agent: content_team          # This is an orchestration, not a plain agent!
        description: "Content production team."
      - agent: qa_bot
        description: "Quality assurance check."

entry: manager
```

## How It Works

1. kaboo-workflows collects all orchestration dependencies.
2. It performs a **topological sort** — inner orchestrations are built before outer ones.
3. Built orchestrations become nodes in the node pool, available for outer orchestrations to reference.
4. For delegate mode, inner orchestrations are wrapped as async tools (just like regular agents).

## Circular Dependencies

Orchestrations that reference each other in a cycle are caught at load time:

```yaml
orchestrations:
  a:
    mode: delegate
    entry_name: some_agent
    connections:
      - agent: b
        description: "..."
  b:
    mode: delegate
    entry_name: other_agent
    connections:
      - agent: a          # Circular!
        description: "..."
```

```
CircularDependencyError: Circular dependency between orchestrations: ['a', 'b'].
Orchestrations cannot reference each other in a cycle.
```

## Nesting Depth

There's no hard limit on nesting depth. A delegate can reference a graph that contains a delegate that references a swarm. As long as there are no cycles, it builds.

## Graph Nodes as Orchestrations

Graph edges can reference orchestrations, not just agents. This means a graph node can be an entire swarm:

```yaml
orchestrations:
  writing_team:
    mode: delegate
    entry_name: writer
    connections:
      - agent: researcher
        description: "Research first."

  pipeline:
    mode: graph
    entry_name: writing_team        # Starts with the delegate orchestration
    edges:
      - from: writing_team
        to: editor
      - from: editor
        to: publisher
```

> **Tips & Tricks**
>
> - Draw your system topology on paper first, then translate it to YAML. Each box is an agent or orchestration, each arrow is a connection/edge.
> - Name your orchestrations descriptively — `content_team`, `review_pipeline`, `analysis_graph` — because error messages reference these names.
> - Use delegate mode as the "outer shell" for most complex systems — it gives the coordinator explicit control over when to call sub-systems.

---

[Next: Chapter 13 — Multi-File Configs →](Chapter_13.md)

# docs/configuration/Chapter_13.md

# Chapter 13: Multi-File Configs — Splitting and Merging

[← Back to Table of Contents](README.md) | [← Previous: Nested Orchestrations](Chapter_12.md)

---

Large configs can be split across multiple files. Pass a list to `load()` and they're merged:

```{.python notest}
from kaboo_workflows import load

resolved = load(["base.yaml", "agents.yaml", "mcp.yaml"])
```

## Merge Rules

**Collection sections** (dicts) are merged across files:

- `models` — merged
- `agents` — merged
- `orchestrations` — merged
- `mcp_servers` — merged
- `mcp_clients` — merged

**Singleton fields** use last-wins semantics:

Everything that is not one of those five collections is a singleton, so the rule
is simpler than a list: `entry`, `session_manager`, `history`, `attachments`,
`runtime`, `telemetry`, `log_level` and `version` all take the last file's
value.

## Duplicate Detection

If two files define the same name in the same collection section, loading fails:

```yaml
# file_a.yaml
agents:
  helper:
    system_prompt: "I help."

# file_b.yaml
agents:
  helper:                  # Duplicate!
    system_prompt: "I also help."
```

```
ValueError: Duplicate names in 'agents' across config sources: ['helper']
```

## Per-File Variable Interpolation

Each file's `vars` block is interpolated independently *before* merging. This means File A's vars don't affect File B:

```yaml
# base.yaml
vars:
  MODEL: us.anthropic.claude-sonnet-4-6-v1:0
models:
  default:
    provider: bedrock
    model_id: ${MODEL}         # Resolves from base.yaml's vars

# agents.yaml
vars:
  TONE: friendly               # This is agents.yaml's own vars
agents:
  assistant:
    model: default
    system_prompt: "You are ${TONE}."
entry: assistant
```

## Typical Split Patterns

**Infrastructure + Application**:
```
base.yaml     — vars, models, mcp_servers, mcp_clients, session_manager
agents.yaml   — agents, orchestrations, entry
```

**Environment Layering**:
```
base.yaml        — shared models, shared agents
production.yaml  — production model IDs, production entry
staging.yaml     — staging model IDs, staging entry
```

**Team-Based**:
```
models.yaml       — all model definitions
research.yaml     — researcher agents + orchestrations
content.yaml      — writer/editor agents + orchestrations
main.yaml         — coordinator agent, top-level orchestration, entry
```

## Neither File Needs to Be Complete

Individual files don't need to be valid on their own. `base.yaml` can define models without agents or entry. `agents.yaml` can reference models it doesn't define. The merged result must be valid — individual files don't.

## The Second Merge Mode: Session Overlays

Everything above describes **composition**: several files that together are one config, which is why a duplicate name is an error rather than an override.

A server whose runs bring their own workflow needs the opposite — **substitution**. That is `load_session_config(base_raw, overlay)`, and it merges by different rules:

| Section | Multi-file merge | Session overlay |
|---------|------------------|-----------------|
| `agents`, `orchestrations` | merged; duplicate name is an error | **replaced** wholesale by the overlay |
| `models`, `mcp_clients` | merged; duplicate name is an error | **unioned**, and the overlay wins a name clash |
| `entry`, `runtime`, `attachments`, `telemetry`, `history`, `log_level`, `session_manager` | last-wins | last-wins from the overlay |
| `mcp_servers` | merged | **rejected** in an overlay |

Only `agents` and `orchestrations` are genuinely replaced. `entry` is last-wins
like any other singleton, which usually looks the same — an overlay that brings
its own agents names its own entry — but differs in one case worth knowing: an
overlay that omits `entry` inherits the base's, and the base's entry agent has
just been replaced out of existence. That fails at load rather than
falling back silently:

```
SchemaValidationError: Invalid config at '': Value error, entry 'base_agent'
is not defined under agents: or orchestrations:.
```

Replace rather than merge is what makes a submitted config a whole workflow instead of an addition to someone else's. Union on `models` and `mcp_clients` is what makes the base useful: it holds the shared infrastructure every workflow draws on by name, and an overlay may still add its own.

```{.python notest}
from kaboo_workflows import load_session_config, parse_config_sources

base_raw = parse_config_sources("config.yaml")   # once, at boot

config = load_session_config(base_raw, submitted_yaml)  # per run
```

`base_raw` is never mutated, so one parse safely serves concurrent runs. The overlay is interpolated as its own source, so a `${OPENROUTER_KEY}` in a config authored in your database resolves from the serving process's environment — the secret stays where it belongs.

Two things an overlay may not do, because a submitted config is data from somewhere else even when its author is trusted:

- `mcp_servers:` is rejected outright. Starting a server means running a command.
- `mcp_clients:` with `command:` is rejected, and a `url:` is checked against `allowed_mcp_hosts` when you pass one. Unlike a Python import path, a URL needs no code on your machine to send your agent's context somewhere else.

Superset enforcement comes free from validation you already have: an overlay agent naming a client nobody defined fails with `Agent 'x' references MCP client 'y' which is not defined. Available: [...]`.

See [Chapter 17](Chapter_17.md) for how this fits the loading pipeline, `create_agui_app(session_config_key=...)` for the serving side, and [`examples/21_runtime_configs`](https://github.com/gl-pgege/kaboo-workflows/tree/main/examples/21_runtime_configs) for a base and two submitted workflows you can run.

> **Tips & Tricks**
>
> - Use multi-file configs when your single file exceeds ~200 lines. It makes diffs cleaner and team collaboration easier.
> - The `entry` field should typically go in the "application" file, not the "infrastructure" file — it's the most likely to change between use cases.
> - File paths for tools/hooks/servers are resolved relative to the file they appear in. If `agents.yaml` says `tools: [./tools.py]`, it looks for `tools.py` next to `agents.yaml`.

---

[Next: Chapter 14 — Agent Factories →](Chapter_14.md)

# docs/configuration/Chapter_14.md

# Chapter 14: Agent Factories — Custom Agent Construction

[← Back to Table of Contents](README.md) | [← Previous: Multi-File Configs](Chapter_13.md)

---

The `type` field on an agent lets you replace the standard `strands.Agent()` constructor with your own factory function. The `agent_kwargs` dict passes extra parameters to it.

```yaml
agents:
  assistant:
    type: ./factory.py:create_agent
    model: default
    system_prompt: "You are a helpful assistant."
    agent_kwargs:
      greeting: "Ahoy, captain!"
      personality: pirate
```

## Writing a Factory

kaboo-workflows calls your factory with all standard agent parameters plus `agent_kwargs`. Precisely, and always, these:

`name`, `agent_id` (both the YAML key), `model`, `system_prompt`, `description`,
`tools`, `hooks`, `conversation_manager`, `session_manager` — then everything in
`agent_kwargs` spread on top.

So a factory needs `**kwargs` to absorb the ones it does not care about, and
should name only the parameters it actually uses:

```python
# factory.py
from strands import Agent

def create_agent(
    *,
    name: str,
    greeting: str = "Hello!",
    personality: str = "friendly",
    **kwargs,
) -> Agent:
    """Custom factory that injects personality into the system prompt."""
    system_prompt = kwargs.pop("system_prompt", "") or ""
    enhanced = f"{system_prompt}\nPersonality: {personality}. Greet with: {greeting}"

    return Agent(
        name=name,
        system_prompt=enhanced,
        **kwargs,
    )
```

**Important**: `strands.Agent.__init__` does NOT accept `**kwargs` — it has explicit parameters only. Your factory **must** consume any custom keys from `agent_kwargs` before forwarding the rest to `Agent()`.

## The `agent_kwargs` Dict

`agent_kwargs` accepts any key-value pairs that get spread into your factory call. It can also pass valid `Agent()` parameters directly:

```yaml
agents:
  assistant:
    model: default
    system_prompt: "You are helpful."
    agent_kwargs:
      record_direct_tool_call: true
      trace_attributes:
        team: content
        environment: production
```

When `type` is **not** set (standard agent construction), `agent_kwargs` is passed directly to `strands.Agent()`. Any invalid key will raise `TypeError` at construction time.

## Delegate `agent_kwargs` Override

Delegate orchestrations can also specify `agent_kwargs`, which get **merged** over the entry agent's kwargs (orchestration values win on conflict):

```yaml
agents:
  coordinator:
    model: default
    system_prompt: "Coordinate work."
    agent_kwargs:
      record_direct_tool_call: false

orchestrations:
  team:
    mode: delegate
    entry_name: coordinator
    agent_kwargs:
      record_direct_tool_call: true     # Overrides the agent's value
    connections:
      - agent: worker
        description: "Do the work."
```

> **Tips & Tricks**
>
> - Agent factories are great for custom `Agent` subclasses — your factory can return `MySpecialAgent(...)` which extends strands' `Agent`.
> - The factory must return a `strands.Agent` instance — kaboo-workflows checks this and raises `TypeError` if it doesn't.
> - Use factory paths in the same format as hooks and tools: `./local/file.py:function_name` or `my_package.factory:create_agent`.

---

[Next: Chapter 15 — Event Streaming →](Chapter_15.md)

# docs/configuration/Chapter_15.md

# Chapter 15: Event Streaming — Real-Time Observability

[← Back to Table of Contents](README.md) | [← Previous: Agent Factories](Chapter_14.md)

---

When you have nested orchestrations running, you need visibility into what's happening. `wire_event_queue()` attaches event publishers to every agent and orchestrator and funnels all events into a single async queue.

```{.python notest}
import asyncio
from kaboo_workflows import AnsiRenderer, load

async def main():
    resolved = load("config.yaml")
    queue = resolved.wire_event_queue()

    async def invoke():
        try:
            await resolved.entry.invoke_async("Analyze LLM trends.")
        finally:
            await queue.close()

    asyncio.create_task(invoke())

    renderer = AnsiRenderer()
    while (event := await queue.get()) is not None:
        renderer.render(event)
    renderer.flush()

asyncio.run(main())
```

## Event Types

Every event is a `StreamEvent` dataclass with four fields: `type`, `agent_name`, `timestamp`, and `data`.

### Session lifecycle events

These two events bracket every invocation. They are produced by the queue layer, not by individual agents.

| Event Type | Description | `data` payload |
|------------|-------------|----------------|
| `SESSION_START` | First event on the queue — emitted before any agent activity | `{"session_id": "<id or null>", "manifest": {SessionManifest}}` — agents, orchestrations, entry point, model info, session manager locations |
| `SESSION_END` | Last typed event before the stream closes | `{"session_id": "<id or null>"}` |

The `SESSION_START` payload wraps the full wired topology snapshot together with the effective session id. Use the `manifest` key to restore conversation history, render an architecture diagram, or audit which models are in use — before any agent has run.

### Per-agent events

| Event Type | Description |
|------------|-------------|
| `AGENT_START` | Agent begins processing |
| `TOKEN` | Individual token streamed from LLM |
| `REASONING` | Reasoning/thinking content from LLM |
| `TOOL_START` | Tool execution begins |
| `TOOL_END` | Tool execution completes |
| `INTERRUPT` | Agent pauses for human input |
| `AGENT_COMPLETE` | Agent finishes — `data` carries `usage` metrics, `text` (final output string), and `message` (raw message dict) |
| `ERROR` | Model or execution error — `data` carries `text` and `exception_type` |

`AGENT_COMPLETE` is not guaranteed on every finish. When a run stops because an
agent raised an interrupt, `INTERRUPT` is emitted and the publisher returns
without a completion event, so a consumer that waits for `AGENT_COMPLETE` before
releasing a turn will hang on any human-in-the-loop pause. Treat `INTERRUPT` as
an equally valid end of turn.

### Stream group events

| Event Type | Description |
|------------|-------------|
| `STREAM_GROUP_START` | An agent in a `stream.group` begins; `data` carries `parent_group`, `tool_call_id` and the `task` it was handed |
| `STREAM_GROUP_END` | That group finishes |

Once an agent is in a group, **every** event it emits also carries
`stream_group` and `stream_title` in `data`, so a consumer can route events to
the right activity card without tracking the start event. A grouped agent that
sets no `title` gets one derived from its name — `data_analyst` becomes
`Data Analyst`.

### Multi-agent events

| Event Type | Description |
|------------|-------------|
| `NODE_START` | Graph/swarm node begins |
| `NODE_STOP` | Graph/swarm node completes |
| `HANDOFF` | Swarm agent hands off to another |
| `MULTIAGENT_START` | Multi-agent orchestration begins |
| `MULTIAGENT_COMPLETE` | Multi-agent orchestration completes |

## AnsiRenderer

The built-in `AnsiRenderer` prints colored terminal output — agent names, tool calls, reasoning traces, tokens — all streaming live. Perfect for development and debugging.

## Custom Event Consumers

Events are `StreamEvent` dataclasses with `.asdict()` for serialization:

```{.python notest}
while (event := await queue.get()) is not None:
    data = event.asdict()
    # Send to websocket, log to file, push to metrics system...
```

A typical consumer pattern that handles the session lifecycle:

```{.python notest}
while (event := await queue.get()) is not None:
    if event.type == "session_start":
        session_id = event.data.get("session_id")
        manifest = event.data["manifest"]  # full topology snapshot
        entry = manifest["entry"]          # {"name": "...", "kind": "agent|orchestration"}
    elif event.type == "session_end":
        session_id = event.data.get("session_id")
    else:
        # per-agent or multi-agent event
        process(event)
```

## What Is Configured Where

The **queue** is a Python concern: you create it and call `wire_event_queue()`,
and there is no YAML for it. The hooks that fills in (`EventPublisher`) listen to
the same lifecycle events as your YAML-defined hooks, and the two coexist.

How an agent's output is **labelled** in that stream, though, is YAML:

```yaml
agents:
  researcher:
    model: default
    stream:
      group: research      # groups this agent's events with others in `research`
      title: Researching    # human-readable label for the group
```

`stream:` takes `group` and `title`, both optional. Agents sharing a `group`
bracket their combined output with `STREAM_GROUP_START` and `STREAM_GROUP_END`
events, which is how a UI renders several agents' work as one collapsible
activity rather than an interleaved mess.

> **Tips & Tricks**
>
> - Call `wire_event_queue()` only **once** per `ResolvedConfig` — it mutates agents and orchestrators by adding hooks. Calling it twice would double-attach publishers.
> - Call `queue.flush()` between requests to clear stale events from a previous invocation. This also resets the `SESSION_START` / `SESSION_END` guards so the next cycle can re-emit them.
> - The queue has a max size of 10,000. If your agent generates more events than the consumer processes, events are dropped with a warning.
> - `SESSION_START` is emitted synchronously by `wire_event_queue()` before any agent runs. `SESSION_END` is emitted by `queue.close()` — always call it in a `finally` block.

---

[Next: Chapter 16 — Name Sanitization →](Chapter_16.md)

# docs/configuration/Chapter_16.md

# Chapter 16: Name Sanitization — How Names Are Handled

[← Back to Table of Contents](README.md) | [← Previous: Event Streaming](Chapter_15.md)

---

Names in config (agent names, model names, MCP client/server names, orchestration names) follow strict rules:

## Valid Names

- Characters: `[a-zA-Z0-9_-]`
- Length: 1–64 characters
- Examples: `researcher`, `fast-model`, `content_writer_v2`

## Automatic Sanitization

If you use characters outside the valid set (spaces, dots, special characters), kaboo-workflows sanitizes them:

- Invalid characters → underscores
- Consecutive underscores → single underscore
- Leading/trailing underscores → stripped
- Names over 64 characters → truncated

```yaml
agents:
  "My Cool Agent!":      # Sanitized to: My_Cool_Agent
    system_prompt: "Hi."
```

A warning is logged when sanitization happens:

```
WARNING | section=<agents>, original=<My Cool Agent!>, sanitized=<My_Cool_Agent> | sanitized collection key
```

## Reference Updates

When a name is sanitized, **all references** to it throughout the config are updated automatically — `entry`, `model` references, `mcp` lists, orchestration agents, connections, and edges.

## Namespace Collisions

Agents and orchestrations share a single namespace. You cannot have:

```yaml
agents:
  team:
    system_prompt: "I'm an agent."

orchestrations:
  team:                    # ERROR: collides with agent 'team'
    mode: swarm
    agents: [team]
    entry_name: team
```

```
ValueError: Name collision between agents and orchestrations: ['team'].
Names must be unique within each section.
```

Models, MCP servers, and MCP clients each have their own independent namespaces — a model and an agent can share a name (though it's confusing and not recommended).

> **Tips & Tricks**
>
> - Stick to `snake_case` for names. It's valid, readable, and never needs sanitization.
> - Avoid naming an orchestration and an agent the same thing — even accidentally similar names can confuse you when debugging.

---

[Next: Chapter 17 — The Loading Pipeline →](Chapter_17.md)

# docs/configuration/Chapter_17.md

# Chapter 17: The Loading Pipeline — What Happens Under the Hood

[← Back to Table of Contents](README.md) | [← Previous: Name Sanitization](Chapter_16.md)

---

When you call `load("config.yaml")`, here's exactly what happens:

## Step 1: Parse Sources

Each config source (file path or raw YAML string) is parsed with `yaml.safe_load()`. `Path` objects are always treated as files. Strings are files if the path exists on disk, otherwise parsed as inline YAML — with two refinements worth knowing: a **multi-line** string is never probed as a path, and a single-line string that looks like a path (contains a separator, or ends `.yaml`/`.yml`) but does not exist raises `FileNotFoundError` rather than being parsed as YAML and failing confusingly later.

## Step 2: Strip Anchors and Interpolate Variables

For each source independently:
1. Extract and remove the `vars` block
2. Strip `x-*` keys (YAML anchor scratch pads)
3. Interpolate `${VAR}` references using `vars` + environment

## Step 3: Rewrite Relative Paths

All filesystem-based specs (`./file.py:func`, `./tools/`) are rewritten to absolute paths anchored to the config file's directory. This ensures the config works regardless of the working directory.

## Step 4: Sanitize Collection Keys

Names in all collection sections are sanitized to `[a-zA-Z0-9_-]`. Cross-references are updated automatically.

## Step 5: Merge (If Multi-File)

Collection sections are combined, duplicate names detected, singleton fields use last-wins.

## Step 6: Normalize, Then Validate Against Schema

`normalize()` runs first and checks `version`: only `"1"` exists, and anything else is rejected here rather than producing a confusing field error further down.

The normalized dict is then validated against the Pydantic models. Invalid fields, missing required values, wrong types — all caught here. Only the **first** Pydantic error is surfaced, reshaped as `SchemaValidationError`:

```
Invalid config at 'agents.analyst.model': Input should be a valid string
```

## Step 7: Validate References

Cross-references are checked, raising `UnresolvedReferenceError` — which, unlike the schema error above, lists the names that *are* available:
- Agent `model` references → must exist in `models`
- Agent `mcp` references → must exist in `mcp_clients`
- MCP client `server` references → must exist in `mcp_servers`
- Orchestration agent references → must exist in `agents` or `orchestrations`

## Step 8: Resolve Infrastructure

Models, MCP servers, MCP clients, and session managers are created as Python objects. Nothing is started yet.

`resolve_infra()` also enforces one rule that is easier to catch here than later: a **global** `session_manager` with `provider: agentcore` is rejected, because that provider needs an `actor_id` unique to each agent.

## Step 9: Start MCP Lifecycle

MCP servers are started in background threads. The pipeline waits for all servers to be ready (TCP port check). This happens **before** agent creation because `Agent.__init__` auto-starts MCP clients which need running servers.

## Step 10: Create Agents

Each agent definition is resolved: model looked up, tools loaded, hooks instantiated, MCP clients attached, session manager wired. Each agent is a fresh `strands.Agent` instance.

Agents that are nodes of a **swarm or graph** are checked here and rejected if they carry a session manager, global or per-agent, because strands cannot persist those (see [Chapter 7](Chapter_07.md#swarm-and-graph-agents-and-sessions)).

## Step 11: Wire Orchestrations

Orchestrations are topologically sorted and built in dependency order. Inner orchestrations first, outer orchestrations reference the already-built inner ones.

## Step 12: Return ResolvedConfig

The final `ResolvedConfig` has:
- `agents` — dict of all agents by name
- `orchestrators` — dict of all built orchestrations by name
- `entry` — the entry point (Agent, Swarm, or Graph)
- `mcp_lifecycle` — for managing shutdown
- `app_config` — the validated `AppConfig` the rest was built from

## Advanced Topic: `load()` vs `load_config()` + `resolve_infra()` + `load_session()`

Most users only need:

```{.python notest}
from kaboo_workflows import load

resolved = load("config.yaml")
```

That one call runs the whole pipeline:

1. Parse YAML, then **per source**: pop `vars`, strip `x-*` anchors, interpolate, rewrite relative paths
2. Sanitize names, per source
3. Merge sources
4. Normalize, validate schema, validate references
5. Apply `log_level`
6. Initialise telemetry — process-wide, and the first call wins, so a later `load()` cannot change it
7. Resolve infrastructure
8. Start MCP lifecycle
9. Create agents and orchestrations

Sanitizing and interpolating happen **before** the merge, per source, which is why one file's `vars` never leak into another's interpolation.

But kaboo-workflows also exposes the lower-level split because **config parsing** and **session creation** are not always the same thing.

### What counts as "config"?

`load_config()` returns a validated `AppConfig` — just structured data.

At this point, nothing is started and no live strands objects exist yet:

- no `Agent` instances
- no orchestration objects
- no started MCP servers
- no connected MCP clients

This step is useful when you want to parse and validate once at process startup, fail fast on bad YAML, and keep the validated config around.

### What counts as "infrastructure"?

`resolve_infra(app_config)` turns the validated config into the shared runtime pieces:

- resolved model objects
- resolved MCP server objects
- resolved MCP client objects
- a cold `mcp_lifecycle`

> **Note:** `resolve_infra()` does **not** build session manager instances. Session managers are
> created per agent and per orchestration at session time (inside `load_session()`), once a real
> session ID is known. This avoids creating orphan filesystem folders before a session actually starts.

Important nuance: **resolved** does not mean **started**.

After `resolve_infra()`:

- MCP servers exist as Python objects, but are not running yet
- MCP clients exist as Python objects, but are not connected yet
- agents still do not exist
- orchestrations still do not exist

You then start the shared MCP runtime explicitly:

```{.python notest}
from kaboo_workflows.config import load_config, resolve_infra

app_config = load_config("config.yaml")
infra = resolve_infra(app_config)
infra.mcp_lifecycle.start()
```

### What `load_session()` does

`load_session(app_config, infra, session_id=...)` is the final step. It uses the already-started shared infrastructure to create a **fresh** `ResolvedConfig` for one session:

- fresh agents
- fresh orchestrations
- fresh entry point
- the same shared MCP lifecycle

This is the key distinction:

- `resolve_infra()` gives you **shared process-level infrastructure**
- `load_session()` gives you **session-level agent graph built on top of that infrastructure**

### Why this split matters for multi-tenant deployments

In a multi-tenant server, you usually do **not** want to re-parse YAML, re-resolve models, or restart MCP servers on every request. Those are process-level concerns.

Instead, you want:

- one validated config shared by the process
- one resolved infrastructure shared by the process
- one started MCP lifecycle shared by the process
- one fresh set of agents per tenant/session/request

Typical pattern:

```{.python notest}
from kaboo_workflows.config import load_config, load_session, resolve_infra

# Once at process startup
app_config = load_config("config.yaml")
infra = resolve_infra(app_config)
infra.mcp_lifecycle.start()

# Per request / websocket / tenant session
resolved = load_session(app_config, infra, session_id="tenant-123")
result = resolved.entry("Hello!")
```

This avoids paying the startup cost repeatedly while still keeping per-session agent state isolated.

### Session manager nuance

Session manager instances are **not** built during `resolve_infra()`. Instead, `load_session()`
computes a single `effective_session_id` and threads it down to every agent and orchestration leaf:

1. If you pass `session_id="my-id"` to `load_session()`, that value is used as-is.
2. If you do **not** pass a `session_id` but the config declares a global `session_manager:`,
   kaboo-workflows looks for a `session_id` in `session_manager.params`. If found, that value is
   used; otherwise a fresh `uuid.uuid4()` is generated once and shared by all agents in that
   call — matching the "one folder per CLI run" behaviour.
3. If neither a `session_id` is provided nor a global `session_manager:` is configured, no session
   manager instances are created at all. Each individual agent or orchestration with a per-agent
   `session_manager:` will generate its own ID as usual.

This design guarantees exactly one folder is created per `load_session()` call, never an orphan
folder from `resolve_infra()`.

### When the config itself is per-request

The split above assumes one config, many sessions. Push it one step further and the config becomes per-request too: the run submits it, and the process only holds the base it layers over.

That needs one more seam, because `load_config()` both parses *and* validates in one call, and here you need the raw parsed base kept around to merge each run's overlay into. So the two halves are separately available.

The base is still validated at boot, and still needs an `entry` — `entry` is required and has no default. Give it a fallback agent, as [`examples/21_runtime_configs`](https://github.com/gl-pgege/kaboo-workflows/tree/main/examples/21_runtime_configs) does, so the service is valid on its own and a run that submits no workflow still has something to answer with.

```{.python notest}
from dataclasses import replace

from kaboo_workflows import (
    load_session,
    load_session_config,
    parse_config_sources,
    resolve_infra,
    resolve_run_clients,
    validate_raw_config,
)

# Once at boot: parse the base, keep it raw.
base_raw = parse_config_sources("config.yaml")
infra = resolve_infra(validate_raw_config(base_raw))
infra.mcp_lifecycle.start(pin_clients=False)   # servers only; clients are per-run

# Per run: merge, validate, resolve.
config = load_session_config(base_raw, submitted_yaml, allowed_mcp_hosts=["gateway.internal"])
clients = resolve_run_clients(config, infra)
clients.start(pin_clients=False)               # the run owns these sessions
resolved = load_session(
    config,
    replace(infra, clients=clients.clients, mcp_lifecycle=clients),
    session_id=thread_id,
)
```

Both additions matter. Without `clients.start(...)`, the run's MCP sessions are
never opened; without `mcp_lifecycle=clients`, `load_session` keeps the
**boot-time** lifecycle, so shutting the run down closes nothing it opened and
the sessions leak for the life of the process.

`create_agui_app(config_path, session_config_key="workflow_config")` does exactly this for you, reading the overlay from `forwardedProps` and closing the run's clients when its stream ends. See [Chapter 13](Chapter_13.md) for the merge rules, and [`examples/21_runtime_configs`](https://github.com/gl-pgege/kaboo-workflows/tree/main/examples/21_runtime_configs) for three runs resolving three different workflows against one base.

What is left at process level shrinks to what is genuinely process-level: the parsed base, the model objects, and any MCP server processes. Everything a conversation touches — agents, orchestrations, entry, client sessions — is built for the run and released with it.

This is only safe because conversation state does not live in those objects. History arrives each turn in `state.kaboo_history`, and a pending interrupt arrives in `state.kaboo_session` ([Chapter 7](Chapter_07.md)). Rebuilding is therefore indistinguishable from retaining, which is what lets you run a second replica, restart mid-approval, or serve a workflow edited thirty seconds ago.

### Mental model

Use this rule of thumb:

- **`load()`** = convenience API for scripts and local apps
- **`load_config()`** = validate and freeze the declarative config
- **`resolve_infra()`** = build shared runtime dependencies, but do not start them yet
- **`load_session()`** = build one session's live agents/orchestrations from shared infra
- **`parse_config_sources()` + `load_session_config()`** = the config itself varies per run

If you're building a CLI, a notebook, or a one-shot script, use `load()`.

If you're building a long-running web server with many user sessions, use `load_config()` + `resolve_infra()` once, then `load_session()` for each session.

If those sessions are different *workflows*, keep the base raw and merge an overlay per run.

---

[Next: Chapter 18 — Full Reference →](Chapter_18.md)

# docs/configuration/Chapter_18.md

# Chapter 18: Full Reference — Every Field at a Glance

[← Back to Table of Contents](README.md) | [← Previous: The Loading Pipeline](Chapter_17.md)

---

## Root Config

```yaml
version: "1"          # Optional, defaults to "1"
vars: {}              # Variable definitions (removed after interpolation)
models: {}            # Named model definitions
agents: {}            # Named agent definitions (required: at least one)
orchestrations: {}    # Named orchestration definitions
mcp_servers: {}       # Named MCP server definitions
mcp_clients: {}       # Named MCP client connections
session_manager: {}   # Global session manager
history: false        # Default for per-agent history when an agent omits history:
attachments: {}       # Global reference/attachment policy (AttachmentsDef)
runtime: {}           # Runtime behavior toggles (RuntimeDef)
telemetry: {}         # OpenTelemetry export (TelemetryDef), off by default
entry: "name"         # Required: entry point agent or orchestration
log_level: "WARNING"  # Optional: DEBUG, INFO, WARNING, ERROR

# Top-level keys beginning x- are stripped before validation, like vars,
# which is what makes them usable as YAML anchor scratch pads.
```

## ModelDef

```yaml
models:
  name:
    provider: bedrock | openai | ollama | gemini | module.path:CustomModel
    model_id: "model-identifier"
    params: {}        # Provider-specific kwargs. For bedrock, client_config:
                      # {read_timeout, connect_timeout, retries: {max_attempts, mode}}
                      # sets the boto client (Chapter 2)
    thinking: null    # bedrock only: false | true | {effort, display}, the
                      # default for every agent on this model (Chapter 2)
```

## AgentDef

```yaml
agents:
  name:
    type: null                     # Custom factory: module.path:factory_func
    agent_kwargs: {}               # Extra kwargs for Agent() or custom factory
    model: "model_name"            # String ref to models: or inline ModelDef
    thinking: null                 # false | true | {effort, display}, laid over
                                   # the model's thinking: (Chapter 2)
    system_prompt: "..."           # System prompt string
    description: "..."             # Agent description (used in orchestration tools)
    tools: []                      # List of tool spec strings
    hooks: []                      # List of HookDef objects or import path strings
    mcp: []                        # List of MCP client names
    tool_labels: {}                # Tool name -> display label mapping
    conversation_manager: null     # ConversationManagerDef
    session_manager: null          # Per-agent SessionManagerDef (overrides global);
                                   # ~ opts this agent out of the global one
    stream: null                   # {group, title} — how this agent's output is
                                   # labelled in the event stream (Chapter 15)
    interrupt: null                # InterruptDef, or true for defaults (Chapter 6):
                                   # {tools: [], ask_user: true, ttl_seconds: null}
    history: null                  # true/false, or a HistoryDef (below); falls back
                                   # to the root history: default
    context: null                  # ContextDef (below): what the model is sent
    attachments: null              # none | reference | inline | bool | {enabled, inline}
    output_schema: null            # Import path to a Pydantic model the agent must
                                   # return (module.path:Model)
```

## HistoryDef

```yaml
agents:
  name:
    history:
      enabled: true                # Remember across turns of the conversation
      group: null                  # Shared transcript bucket (default: the agent's path)
      keep: full                   # full | masked — masked writes back each finished
                                   # call without its tool results or reasoning
      summarize_over: null         # Tokens; with keep: masked, summarise the oldest
                                   # calls once the written-back memory passes this
      summarize_model: null        # A name under models:; required with summarize_over
```

A delegate orchestration's forked agent remembers exactly as its `entry_name`
agent's `history:` says. See [Multi-turn history](../workflows/multi-turn-history.md).

## ContextDef

```yaml
agents:
  name:
    context:
      clear_tool_results:          # Off when absent (Chapter 8)
        trigger: 60000             # Required: estimated input tokens
        keep_last: 3               # Most recent turns whose results always stay (>= 1)
        min_chars: 1500            # Shorter results always stay
        keep_tools: []             # Tools whose results always stay
```

## AttachmentsDef

```yaml
attachments:
  default: reference | none        # Baseline for agents without their own attachments:
  tool: true                       # Expose list_references / fetch_attachment tools
  base_url: null                   # Origin for the host's own attachment routes
  authorization: null              # forwarded_props:<key> | env:<VAR> (own-origin only)
  content_url_template: null       # e.g. /attachments/{id}/content — upgrades
                                   # attachment-kind object refs to fetchable transport
```

Per-agent override on `AgentDef.attachments` (shorthands normalize to `{enabled, inline}`):

```yaml
agents:
  vision:
    attachments: inline            # manifest + inline media (ContentBlocks)
  researcher:
    attachments: reference         # manifest only (inherits if omitted)
  writer:
    attachments: none              # excluded from references
```

## RuntimeDef

```yaml
runtime:
  persist_session_state: true        # Carry pending interrupts on the AG-UI state
                                     # channel, so an approval survives a restart
                                     # (Chapter 7). Disable only when the AG-UI
                                     # endpoint is exposed straight to a browser.
  allow_invocation_overrides: false  # DEPRECATED — apply forwardedProps.agent_config
                                     # (system_prompt / model_id) per invocation.
                                     # Submit a config instead (Chapter 13).
```

## TelemetryDef

```yaml
telemetry:
  enabled: false                   # Opt in; KABOO_TELEMETRY_ENABLED overrides
  service_name: kaboo-workflows    # OTel service name
  console: false                   # Also export spans to stdout
  sample_ratio: 1.0                # 0.0-1.0
  trace_attributes: {}             # Static attributes on every span
  otlp:
    endpoint: null                 # Falls back to KABOO_OTLP_ENDPOINT / OTEL_*
    headers: null                  # Falls back to KABOO_OTLP_HEADERS / OTEL_*
```

Telemetry is initialised once per process, first call wins, so it is not
something a per-run submitted config can change.

## HookDef

```yaml
hooks:
  # Inline object form
  - type: module.path:ClassName    # or ./file.py:ClassName
    params: {}                     # Constructor kwargs

  # String shorthand (no params)
  - module.path:ClassName
```

## SessionManagerDef

```yaml
session_manager:
  provider: file | s3 | agentcore  # Built-in provider name
  type: null                        # Custom class: module.path:ClassName (overrides provider)
  params: {}                        # Constructor kwargs (session_id, storage_dir, etc.)
```

## ConversationManagerDef

```yaml
conversation_manager:
  type: strands.agent:SlidingWindowConversationManager
  params: {}                       # Constructor kwargs (window_size, etc.)
```

## MCPServerDef

```yaml
mcp_servers:
  name:
    type: ./server.py:create       # Factory function: module.path:func or ./file.py:func
    params: {}                     # Forwarded to factory (port, host, etc.)
```

## MCPClientDef

```yaml
mcp_clients:
  name:
    # Exactly one of:
    server: "server_name"          # Reference to mcp_servers entry
    url: "https://..."             # External MCP server URL
    command: ["cmd", "arg"]        # Stdio subprocess command

    transport: null                # Override: "streamable-http" | "sse" | "stdio"
    params: {}                     # Forwarded to strands MCPClient
                                   # (prefix, startup_timeout, tool_filters)
    transport_options: {}          # Transport-specific options. Declare once per
                                   # client — a second block silently replaces it.
    tool_labels: {}                # Tool name -> display label (agent labels win)
    auth: null                     # Outbound auth; "relay" shorthand expands to the
                                   # block below. Not valid with command:.
```

### MCPClientAuthDef

```yaml
    auth:
      type: relay | m2m | static
      params:
        header: Authorization      # One name, or a list to send the token twice
        scheme: Bearer             # "" sends the raw value
        # relay:  token
        # static: token (required)
        # m2m:    token_url, client_id, client_secret, scope, audience, extra
```

`relay` resolves the caller's identity from the request context, so it only
works when clients are resolved per run. See [Chapter 9](Chapter_09.md).

## DelegateOrchestrationDef

```yaml
orchestrations:
  name:
    mode: delegate
    entry_name: "agent_name"       # Agent blueprint to fork
    connections:
      - agent: "target_name"      # Agent or orchestration name
        description: "..."         # Tool description for LLM
    session_manager: null          # Override session manager
    hooks: []                      # Additional hooks
    agent_kwargs: {}               # Override agent kwargs (merged)
```

## SwarmOrchestrationDef

```yaml
orchestrations:
  name:
    mode: swarm
    agents: [agent1, agent2]       # Participating agents
    entry_name: "agent1"           # Starting agent
    chat_output: null              # Member agent whose text becomes the reply
    max_handoffs: 20               # Max handoffs
    max_iterations: 20             # Max iterations
    execution_timeout: 900.0       # Total timeout (seconds)
    node_timeout: 300.0            # Per-agent timeout (seconds)
    session_manager: null          # Swarm-level session manager
    hooks: []                      # Swarm-level hooks
```

## GraphOrchestrationDef

```yaml
orchestrations:
  name:
    mode: graph
    entry_name: "start_node"       # Node with no incoming edges
    chat_output: null              # Node whose text becomes the reply
    edges:
      - from: "node_a"
        to: "node_b"
        condition: null            # Optional: ./file.py:func or module:func
    max_node_executions: null      # Safety cap for loops
    execution_timeout: null        # Total timeout (seconds)
    node_timeout: null             # Per-node timeout (seconds)
    reset_on_revisit: false        # Reset agent state on revisit
    session_manager: null          # Graph-level session manager
    hooks: []                      # Graph-level hooks
```

## create_agui_app

```{.python notest}
create_agui_app(
    config_path,                    # Base config. With session_config_key set, the
                                    # layer every submitted config merges over.
    endpoint="/invocations",
    ping_path="/ping",
    cors_origins=None,              # Defaults to ["*"]
    cors_allow_credentials=True,
    auth=None,                      # Inbound verifier -> Principal (raise to reject)
    session_config_key=None,        # forwardedProps key carrying this run's config.
                                    # None serves config_path alone.
    allowed_mcp_hosts=None,         # Hosts a submitted mcp_client URL may point at
)
```

`session_config_key` changes the shape of the service, not just a setting: each run gets its own agents, orchestration, entry and MCP client sessions, released when its stream ends. See [Chapter 13](Chapter_13.md) for the merge rules and [Chapter 17](Chapter_17.md) for the pipeline.

---

[Next: Chapter 19 — Attachments & Multimodal →](Chapter_19.md)

**Bonus**: [Quick Recipes →](Quick_Recipes.md)

# docs/configuration/Chapter_19.md

# Chapter 19: Attachments & Multimodal — References for Every Agent

[← Back to Table of Contents](README.md) | [← Previous: Full Reference](Chapter_18.md)

---

Your frontend can attach things to a message — an uploaded PDF, a screenshot, or a pointer to some entity like a database table or a dashboard. kaboo-workflows calls all of these **references**, and this chapter is about getting them to the agents that need them.

By default, only the **entry** agent sees an attachment (that is handled automatically by the AG-UI layer). But in a pipeline the agent that actually needs the file is often a sub-agent three hops in. The `attachments:` config makes references reach any agent you choose — as a lightweight text **manifest**, an on-demand **tool**, or fully **inline** media that a vision model can literally see.

## The two kinds of reference

Everything cited from the frontend is one of two transports:

- **Attachment** — a file/blob (`pdf`, `png/jpeg/gif/webp`, `csv/xlsx/doc/docx/html/txt/md`, video). Sent as multimodal message parts; resolvable to bytes/text/URL by the built-in `fetch_attachment` tool.
- **Object** — a pointer to a custom entity (`table`, `dashboard`, `ticket`, …). Never a blob — just `{kind, id, name, meta}`, sent in AG-UI `state.kaboo_references`, and resolved by **your own MCP tool** (there is no generic resolver in the library).

Both show up to agents the same way: as a manifest line naming the `kind`, `id`, and `name`. What the agent *does* with that id — call `fetch_attachment`, call your `query_table` MCP tool — depends on the kind.

## Three modes

Each in-scope agent runs in one of three modes:

- **`reference`** (baseline) — the agent gets a text manifest of what exists, plus (if enabled) the shared tool to fetch/resolve on demand. Cheap; never materializes bytes.
- **`inline`** (per-agent opt-in) — the agent *additionally* receives resolved media as strands `ContentBlock`s, so a vision/doc-capable model sees the file directly. Heavier — use only where a model must look at the pixels.
- **`none`** — the agent is excluded; no manifest, no tool.

The manifest says *what exists*; the tool lets the agent *act*; inline lets the model *see*.

## Config surface

Global defaults live on the root `attachments:` block; per-agent overrides on `AgentDef.attachments`:

```yaml
attachments:
  default: reference     # reference | none — baseline for in-scope agents
  tool: true             # expose list_references / fetch_attachment
  base_url: null         # origin of the host's own attachment routes (optional)
  authorization: null    # forwarded_props:<key> | env:<VAR> (own-origin only)
  content_url_template: null  # e.g. /attachments/{id}/content (optional)

agents:
  vision_analyst:
    attachments: { inline: true }   # manifest + inline media
  researcher: {}                    # inherits default: reference + tool
  writer:
    attachments: none               # excluded
```

### Shorthands

`attachments:` on an agent accepts the same shorthand style as `history:`:

| Value | Meaning |
|-------|---------|
| _(omitted / `null`)_ | inherit the global `default` |
| `none` or `false` | excluded — no references |
| `reference` | manifest only |
| `inline` or `true` | manifest **and** inline media |
| `{ enabled: true, inline: true }` | explicit mapping |

## How it flows

1. **Frontend** (kaboo-react) mints a stable id per reference. Files added from the shared **+** / **@** popover become `InputContent` parts with `kaboo_id` / `kaboo_kind` / `kaboo_name` in the part metadata; objects picked from the same popover go into `state.kaboo_references`. Both render as inline chips in the input.
2. **`kaboo_endpoint`** parses both the message content parts and `state.kaboo_references` into a request-scoped registry (`set_references`).
3. **`ReferenceHook`** (wired per in-scope agent) injects the manifest into the agent's system prompt on `BeforeInvocationEvent`, and — for `inline` agents — prepends resolved `ContentBlock`s on `BeforeModelCallEvent`.
4. The agent reads an id from the manifest and calls `fetch_attachment(id)` (files) or your MCP tool (custom kinds).

Because the registry is request-scoped and the hook is per-agent, references reach **every** in-scope agent in a pipeline — not just the entry.

## Built-in tools

When `attachments.tool: true` (the default), in-scope agents get two tools:

- **`list_references()`** — returns every reference in the current request (`kind`, `id`, `name`, `mime_type`).
- **`fetch_attachment(reference_id)`** — resolves a file attachment: decoded text for text types, or a document/image `ContentBlock` for media. URLs are fetched server-side; base64 data is decoded.

Custom object kinds are **not** covered by `fetch_attachment` — you resolve those with your own MCP tool (e.g. `query_table(id)`). The manifest hands the agent the `kind`+`id`; wiring the resolver is the documented extension point.

## Media a tool returns

A tool behind an MCP gateway answers in text: the gateway turns each HTTP
response into JSON or text, and binary responses do not pass. To let the model
*see* an image or a document a tool produced, return it as data inside the
tool's JSON:

```json
{
  "summary": "3 problems in 2 of 11 widgets",
  "attachments": [
    {"name": "Rows 1-7", "mime_type": "image/jpeg", "data": "<base64>"}
  ]
}
```

Every agent the config builds carries `ToolAttachmentsHook`. It removes
`attachments` from the result before the model reads it and appends each entry
as a media block of the same tool result, after a text block with its `name`,
so the bytes never reach the model as text. Supported types are those of
`fetch_attachment`; at most 8 entries of up to 3.75 MB each are unpacked, and
anything dropped is named in a text block. Strands carries media in a tool
result to every provider (its OpenAI formatter moves it into a following user
message). No storage or second fetch is involved: the bytes ride the gateway
response the tool already returns.

## Authorized fetching & attachments beyond the first turn

By default, URL sources are fetched server-side **without credentials** — fine
for presigned/public URLs, not for files behind your API's auth. Three knobs
change that:

```yaml
attachments:
  base_url: ${API_BASE_URL}                    # your API's origin
  authorization: forwarded_props:runToken      # or env:<VAR>
  content_url_template: /attachments/{id}/content
```

- **`base_url`** — relative reference URLs (`/attachments/…`) resolve against
  it.
- **`authorization`** — where the bearer token comes from:
  `forwarded_props:<key>` reads it from the run's AG-UI `forwardedProps` (a
  run-scoped credential your backend sends per invocation); `env:<VAR>` reads
  a static token. The token is attached **only** to URLs under `base_url` —
  presigned/public URLs on other origins keep the unauthenticated default, so
  your credential never reaches third-party hosts.
- **`content_url_template`** — entries in `state.kaboo_references` with
  `kind: "attachment"` are files, not custom objects. When they carry a
  `meta.url` — or this template can synthesize one from the id — they are
  upgraded to fetchable attachment transport, so `fetch_attachment` works on
  **every** turn, not only the first (where the file also rides the message as
  a multimodal part).

Hosts with several authenticated stores or non-HTTP retrieval can go further
and register a custom strategy in code — it receives the full `Reference` for
routing:

```python
from kaboo_workflows.tools import set_reference_fetcher

def my_fetcher(url: str, *, reference=None) -> bytes | None:
    ...  # route by reference.meta, sign requests, read from object storage

set_reference_fetcher(my_fetcher)  # once at startup, before serving
```

The registered fetcher applies everywhere bytes are resolved: the
`fetch_attachment` tool, `inline` media resolution, and the entry message's
multimodal parts (kaboo routes ag-ui-strands' internal fetch through the same
funnel).

## Inline media requirements

For `inline` to work, the agent's `model:` must be vision/doc-capable, and a few limits apply (inherited from the AG-UI → strands conversion):

- **Fetchable URLs.** URL sources are fetched server-side — unauthenticated by default (presigned/public URLs), or authorized for your own origin via `attachments.base_url` / `authorization` (see above). Have your frontend `onUpload` store the file and return a fetchable URL.
- **Format allowlist.** Images (`png/jpeg/gif/webp`) and documents (`pdf/csv/doc/docx/xls/xlsx/html/txt/md`) plus video. Unsupported or absent MIME types are skipped.
- **Audio is unsupported** and dropped.
- **Inline bloats history and event-log replay.** Prefer `reference` mode by default; keep base64 out of the transport by returning URLs from `onUpload`.

## Frontend wiring (kaboo-react)

Everything is a **reference provider** registered on `<KabooProvider references={…}>` and rendered by the `KabooReferenceInput` slot. Two transports, both shown as interactive inline chips in the input:

- **Files → `uploadProvider({ onUpload })`.** Choosing "Attach a file" (from the `+` button or the `@` popover) opens the picker, uploads via `onUpload`, and drops a file chip inline. Every file is stamped with `kaboo_id` / `kaboo_kind` / `kaboo_name` and rides the message as an attachment part.
- **Custom objects → your own `ReferenceProvider`.** Selected objects sync into `state.kaboo_references` and appear inline as `@name`.

```tsx
// references.tsx — file upload + a custom object provider (database tables)
import { uploadProvider, type ReferenceProvider } from "kaboo-react";

const tableProvider: ReferenceProvider = {
  id: "table",
  label: "Tables",
  icon: <Table2 size={16} />,
  search: (q) => TABLES.filter((t) => t.label.includes(q)),
  toReference: (item) => ({
    transport: "object",     // pointer only — never a blob
    kind: "table",           // matches your MCP resolver
    id: item.id,
    name: item.label,
  }),
};

export const referenceProviders = [
  uploadProvider({ accept: "image/*,.pdf", onUpload }),
  tableProvider,
];
```

```tsx
// App.tsx — one slot drives files + objects, + and @ share one popover
import { KabooProvider, KabooReferenceInput } from "kaboo-react";

<KabooProvider runtimeUrl="/api/copilotkit" agent={entry} references={referenceProviders}>
  <CopilotChat input={KabooReferenceInput} />
</KabooProvider>
```

`KabooReferenceInput` keeps CopilotKit's native input chrome (send button, disclaimer, theme) but replaces the plain textarea with a lightweight editor where each reference is an **interactive inline chip** — click a chip to swap it for another, or its `×` to remove it. The `+` button and the `@` key open the **same** searchable popover (an "Attach a file" action plus every searchable provider's items); it dismisses on outside-click, `Escape`, or when the `@` token is cleared. On submit the editor builds the multimodal message itself: files become attachment parts, and object references ride `state.kaboo_references` (kept in sync by `useReferences()` and cleared only once the run has started, so it never gets wiped from an in-flight send). The matching MCP resolver on the pipeline (`kind: "table"` → `research_resolve_table`) turns the object pointer into real data.

`onUpload` should store the file and return a **fetchable (presigned/public) URL**. That keeps the AG-UI transport and the persisted event log small, and is required for server-side URL fetching (there is no auth on that fetch). When omitted, kaboo-react base64-encodes the file — fine for small files/tests, heavy for large ones.

> **Tips & Tricks**
>
> - Start with `reference` everywhere; add `inline` only to the specific agent whose model must see the file.
> - The id the model reads in the manifest is the same id `fetch_attachment` / your MCP tool resolves against — frontend-minted, or a deterministic server fallback when absent.
> - Referenced ids are client-supplied and untrusted. Your resolver MCP tool must authorize access.
> - Objects (tables, dashboards) are always pointers — never send them as blobs.

---

[← Back to Table of Contents](README.md)

# docs/configuration/Quick_Recipes.md

# Quick Recipes

[← Back to Table of Contents](README.md) | [← Previous: Full Reference](Chapter_18.md)

---

Copy-paste-ready configs for common patterns.

## The Kitchen Sink

Everything in one config:

```yaml
vars:
  MODEL: ${MODEL:-us.anthropic.claude-sonnet-4-6-v1:0}
  TONE: ${TONE:-professional}

x-hooks: &safety_hooks
  - type: kaboo_workflows.hooks:MaxToolCallsGuard
    params: { max_calls: 20 }
  - type: kaboo_workflows.hooks:ToolNameSanitizer

models:
  default:
    provider: bedrock
    model_id: ${MODEL}

session_manager:
  provider: file
  params:
    storage_dir: ./.sessions
    session_id: ${SESSION_ID:-default}

agents:
  researcher:
    model: default
    hooks: *safety_hooks
    tools:
      - ./tools/research.py
    system_prompt: |
      You are a ${TONE} assistant.
      You specialize in research.
    session_manager: ~              # Opt out (used in swarm)

  reviewer:
    model: default
    hooks: *safety_hooks
    system_prompt: "You review content."
    session_manager: ~              # Opt out (used in swarm)

  qa_bot:
    model: default
    hooks: *safety_hooks
    system_prompt: "Run QA checks."

  coordinator:
    model: default
    hooks: *safety_hooks
    conversation_manager:
      type: strands.agent:SlidingWindowConversationManager
      params: { window_size: 40 }
    system_prompt: "Coordinate the team."

orchestrations:
  content_team:
    mode: swarm
    agents: [researcher, reviewer]
    entry_name: researcher
    max_handoffs: 10

  pipeline:
    mode: delegate
    entry_name: coordinator
    connections:
      - agent: content_team
        description: "Content production team."
      - agent: qa_bot
        description: "Quality assurance."

entry: pipeline
log_level: ${LOG_LEVEL:-WARNING}
```

## Minimal Single Agent

```yaml
agents:
  bot:
    system_prompt: "You answer questions."
entry: bot
```

## Two Models, One Agent

```yaml
models:
  fast:
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-4-6-v1:0
  smart:
    provider: openai
    model_id: gpt-4o

agents:
  assistant:
    model: ${WHICH_MODEL:-fast}
    system_prompt: "You are helpful."
entry: assistant
```

---

[← Back to Table of Contents](README.md)

# docs/configuration/README.md

# YAML Configuration Guide

**Everything you need to know about writing kaboo-workflows YAML configs — from zero to production.**

kaboo-workflows lets you describe entire multi-agent systems in YAML and get back live, fully wired strands objects. This guide walks you through every configuration option, from the simplest one-agent setup to nested multi-orchestration systems with MCP servers, hooks, session persistence, conditional graph pipelines, and multi-file configs.

No prior YAML expertise required. We start simple and build up.

---

## Table of Contents

1. [The Basics — Your First Config](Chapter_01.md)
2. [Models — Choosing Your LLM](Chapter_02.md)
3. [Variables — Environment-Driven Config](Chapter_03.md)
4. [YAML Anchors — DRY Config Blocks](Chapter_04.md)
5. [Tools — Giving Agents Superpowers](Chapter_05.md)
6. [Hooks — Middleware for Agents](Chapter_06.md)
7. [Session Persistence — Memory That Survives Restarts](Chapter_07.md)
8. [Conversation Managers — Controlling Context Windows](Chapter_08.md)
9. [MCP — External Tool Servers](Chapter_09.md)
10. [Orchestrations — Multi-Agent Systems](Chapter_10.md)
11. [Graph Conditions — Dynamic Routing](Chapter_11.md)
12. [Nested Orchestrations — Composing Systems](Chapter_12.md)
13. [Multi-File Configs — Splitting and Merging](Chapter_13.md)
14. [Agent Factories — Custom Agent Construction](Chapter_14.md)
15. [Event Streaming — Real-Time Observability](Chapter_15.md)
16. [Name Sanitization — How Names Are Handled](Chapter_16.md)
17. [The Loading Pipeline — What Happens Under the Hood](Chapter_17.md)
18. [Full Reference — Every Field at a Glance](Chapter_18.md)
19. [Attachments & Multimodal — References for Every Agent](Chapter_19.md)

**Bonus**: [Quick Recipes](Quick_Recipes.md) — Copy-paste-ready configs for common patterns.

---

That covers everything kaboo-workflows YAML has to offer. When in doubt, check the [examples](https://github.com/gl-pgege/kaboo-workflows/tree/main/examples) — each one is a self-contained demo of the concepts above. And remember: after `load()`, what you get back are plain strands objects. No wrappers, no subclasses. Just the real deal, fully wired and ready to go.

Happy composing! 🎼

# docs/evaluation.md

# Evaluation

kaboo-workflows treats evaluation as a first-class, opt-in feature with two
loops:

- **Online**: live runs emit OpenTelemetry traces (agent → cycle → model →
  tool spans, with tokens and cost) to any OTLP backend. User feedback from the
  UI attaches scores to the exact trace.
- **Offline**: a golden dataset of real questions runs headlessly through the
  *same* wire path as production, and pluggable scorers gate the result with a
  nonzero exit code.

Everything is off by default and zero-overhead when disabled: with telemetry
off, no OTel SDK is configured and all instrumentation is a no-op.

```mermaid
flowchart LR
    subgraph online [Online loop]
        Run[Live workflow run] -->|OTel spans| Hub[(OTLP backend, e.g. Langfuse)]
        UI[kaboo-react TurnFeedback] -->|"score {traceId, value}"| Host[Host backend]
        Host -->|create_score| Hub
    end
    subgraph offline [Offline loop]
        Golden[Golden dataset YAML/JSONL] --> Runner[kaboo-workflows eval]
        Runner -->|headless runs| Scores[Scorers]
        Scores -->|report + exit code| Gate[CLI / pytest]
        Scores -->|optional push| Hub
    end
    Hub -->|bad traces promoted| Golden
```

## Online tracing

### Enable telemetry

Add a `telemetry` section to your config:

```yaml
telemetry:
  enabled: true
  service_name: my-agent-service     # OTel service.name (default: kaboo-workflows)
  otlp:
    endpoint: http://localhost:3010/api/public/otel   # traces go to <endpoint>/v1/traces
    headers:                          # or a single "k=v,k2=v2" string
      Authorization: Basic <base64(public:secret)>
      x-langfuse-ingestion-version: "4"
  sample_ratio: 1.0                   # 0..1, trace-id ratio sampling
  console: false                      # also print spans to stdout (debugging)
  trace_attributes:                   # static attributes stamped on every span
    deployment.environment: local
```

Or via environment variables (no config change needed):

| Variable | Effect |
| --- | --- |
| `KABOO_TELEMETRY_ENABLED` | `true`/`false` — overrides `telemetry.enabled` in *both* directions (kill switch) |
| `KABOO_OTLP_ENDPOINT` or `OTEL_EXPORTER_OTLP_ENDPOINT` | OTLP base endpoint |
| `KABOO_OTLP_HEADERS` or `OTEL_EXPORTER_OTLP_HEADERS` | `k=v,k2=v2` header string |

When enabled, kaboo initializes strands' `StrandsTelemetry` (agent, cycle,
model, and tool spans following the GenAI semantic conventions) with a batch
span processor — spans stream out incrementally, nothing buffers per run.
Initialization is process-wide and idempotent; an existing global
`TracerProvider` is respected.

### Conversation attributes

Every span is stamped with the identifiers kaboo already tracks, so traces
correlate 1:1 with what the UI shows:

| Attribute | Source |
| --- | --- |
| `session.id` | AG-UI `threadId` |
| `user.id` | authenticated principal (`sub`/`username` claim), when auth is on |
| `kaboo.run.id` | AG-UI `runId` |
| `kaboo.turn.id` | kaboo's logical turn id (stable across interrupt/resume) |

### Trace id in the event stream

The active trace id rides outward so hosts can attach scores to the exact
trace:

- `AGENT_COMPLETE` events carry `trace_id`;
- activity snapshots expose `traceId` per group and `traceByRun` per run
  (`ACTIVITY_SNAPSHOT` → kaboo-react's `StreamGroup.traceId` /
  `ActivityState.traceByRun`);
- `current_trace_id()` (exported from `kaboo_workflows`) returns the ambient
  trace id, `None` when telemetry is off.

### Langfuse quickstart (reference backend)

Any OTLP backend works; [Langfuse](https://langfuse.com) is the reference
because it adds score APIs, LLM-as-judge evaluators, and annotation queues on
top of the traces. Self-hosted:

1. Run Langfuse v4 (web + worker + ClickHouse + MinIO + Redis, reusing your
   Postgres with a dedicated `langfuse` database).
2. Create an org/project and an API key pair in the UI.
3. Point telemetry at it:

```bash
KABOO_OTLP_ENDPOINT=http://localhost:3010/api/public/otel
KABOO_OTLP_HEADERS="Authorization=Basic $(echo -n pk-lf-…:sk-lf-… | base64),x-langfuse-ingestion-version=4"
KABOO_TELEMETRY_ENABLED=true
```

Traces appear under the project within seconds of a run; `session.id` groups
them per conversation.

### User feedback

kaboo-react ships a `TurnFeedback` component / `useTurnFeedback` hook that
resolves the turn's `traceId` and calls a host-provided `onFeedback` callback
— the library stores nothing. The host backend forwards the payload to the
backend's score API (for Langfuse: `create_score(trace_id=…,
name="user-feedback", value=…)`). See the kaboo-react feedback guide.

## Offline evals

Install the extras you need:

```bash
pip install 'kaboo-workflows[evals]'                 # jsonschema for json_schema scorer
pip install 'kaboo-workflows[evals,eval-deepeval]'   # + DeepEval metrics
pip install 'kaboo-workflows[evals,langfuse]'        # + push experiments to Langfuse
```

### Golden dataset format

YAML (or JSONL, one item per line):

```yaml
name: golden-v1                # dataset name (defaults to the file stem)
config: ./config.yaml          # workflow config, relative to this file
scorers:                       # applied to every item
  - type: budget
    max_cost: 2.0
    max_latency_s: 300
items:
  - id: revenue-q3
    input: "What was total revenue in Q3?"
    expect:                    # item-specific scorers
      - type: contains
        value: ["revenue", "Q3"]
      - type: tool_called
        tool: run_sql
      - type: judge
        rubric: "States the Q3 revenue figure with the currency and cites the source table."
  - id: chart
    input: "Plot monthly signups for 2025"
    expect:
      - type: trajectory
        tools: [sql_analyst, chart_visualizer]
```

Item fields: `id` (defaults to `item-<index>`), `input` (required),
`expect`, `state`, `forwarded_props`, `timeout_s` (default 600),
`metadata`.

Every item is implicitly checked by `run_succeeded` (the run finished without
an error) before any configured scorer.

### Built-in scorers

| Type | Checks | Keys |
| --- | --- | --- |
| `contains` / `not_contains` | substring(s) in the final text | `value`, `case_sensitive` |
| `regex` | pattern matches the final text | `pattern`, `flags` (`imsx`) |
| `tool_called` | a tool ran | `tool`, `agent`, `min_count` |
| `trajectory` | tools appear in order (subsequence) | `tools` |
| `budget` | cost / latency / token caps | `max_cost`, `max_latency_s`, `max_input_tokens`, `max_output_tokens`, `max_total_tokens` |
| `json_schema` | structured output (or text-as-JSON) validates | `schema` |
| `judge` | LLM-as-judge against a rubric | `rubric`, `model`, `threshold` (default 0.7), `name` |
| `deepeval` | any DeepEval metric | `metric`, `threshold`, metric kwargs |

The trajectory includes sub-agent invocations: a delegation or handoff to
`sql_analyst` counts as a step named `sql_analyst`, followed by that agent's
own tool calls — so `trajectory` and `tool_called` work across orchestration
modes.

The `judge` scorer resolves its model from the workflow config's own
`models:` section (`model: sonnet-4-5`), or takes an inline mapping
(`model: {provider: openai, model_id: …}`). Judges see the user input, the
final answer, and the tool trajectory, and return a 0–1 score with reasoning.

Custom scorers plug in via the same import-spec mechanism as model providers:

```yaml
expect:
  - type: my_pkg.scoring:GroundednessScorer   # or ./scorers.py:ClassName
    some_kwarg: 42
```

The class needs a `name` attribute and an
`async def score(item, capture) -> ScoreResult` method. The `capture`
(`RunCapture`) exposes `text`, `tools` (with per-agent attribution),
`structured_outputs`, `usage` (tokens + cost), `latency_s`, `groups`, and
`error`.

### Running

```bash
kaboo-workflows eval datasets/golden.yaml                     # all items
kaboo-workflows eval datasets/golden.yaml --item revenue-q3   # subset (repeatable)
kaboo-workflows eval datasets/golden.yaml --max-items 5
kaboo-workflows eval datasets/golden.yaml --output results.jsonl
kaboo-workflows eval datasets/golden.yaml --config overrides.yaml
kaboo-workflows eval datasets/golden.yaml --push-langfuse     # upload as a dataset experiment
```

Each item runs headlessly through the production wire path — the same agents,
orchestrations, MCP tools, and hooks your service uses — one at a time, with
per-item progress and a summary table. The exit code is nonzero when any
scorer fails, so the command gates CI as-is.

`--push-langfuse` uploads the items to a Langfuse dataset (idempotent by item
id) and records the run as an experiment with every scorer verdict attached
as a score. Requires the `langfuse` extra plus `LANGFUSE_PUBLIC_KEY`,
`LANGFUSE_SECRET_KEY`, and `LANGFUSE_BASE_URL` (or `LANGFUSE_HOST`).

### From Python / pytest

```python
from kaboo_workflows.evals import assert_eval, run_eval

async def test_golden_dataset():
    report = await run_eval("datasets/golden.yaml")
    assert_eval(report)   # raises AssertionError with the summary on failure
```

`run_eval` also accepts `config=`, `item_ids=`, `max_items=`, `output=`, and
an `on_item=` progress callback; the returned `EvalReport` exposes `ok`,
`failed`, `total_cost`, `summary()`, and `to_dict()`.

### DeepEval metrics

With the `eval-deepeval` extra, any DeepEval metric wraps into a scorer:

```yaml
expect:
  - type: deepeval
    metric: deepeval.metrics:AnswerRelevancyMetric
    threshold: 0.8
```

## Scale and toggle guarantees

- **Off by default** — no OTel SDK configured means every instrumentation
  point is a no-op; `KABOO_TELEMETRY_ENABLED=false` force-disables even a
  config that says `enabled: true`.
- **Any pipeline size** — spans export through batch processors, the eval
  runner streams items one at a time and appends each JSONL line as it
  completes; nothing buffers a whole run in memory.
- **No vendor lock-in** — the library speaks OTLP and a thin scorer protocol;
  Langfuse, Phoenix, or a bare collector are config swaps.

# docs/getting-started.md

# Getting started

This page takes you from an empty directory to a running multi-agent server in
a few minutes. Every snippet here is exercised in CI.

!!! note "Compatibility"
    kaboo-workflows requires **Python 3.12+** and is built on
    [strands-agents](https://github.com/strands-agents/harness-sdk). Model
    provider SDKs are optional extras (`openai`, `ollama`, `gemini`,
    `agentcore-memory`) — install the one you need.

## 1. Install

```bash
pip install "kaboo-workflows[openai]"
# or with uv
uv add "kaboo-workflows[openai]"
```

## 2. Write a config

The whole system is a single YAML file: models, agents, tools, and the `entry`
point. Save this as `config.yaml`:

```yaml
vars:
  OPENROUTER_API_KEY: ${OPENROUTER_API_KEY}

models:
  default:
    provider: openai
    model_id: anthropic/claude-sonnet-4
    params:
      client_args:
        base_url: https://openrouter.ai/api/v1
        api_key: ${OPENROUTER_API_KEY}

agents:
  assistant:
    model: default
    system_prompt: "You are a helpful assistant."

entry: assistant
```

`${VAR}` interpolates from the environment (see
[Chapter 1](configuration/Chapter_01.md)), so no secrets live in the file.

## 3a. Serve it (AG-UI SSE)

`kaboo-serve` resolves the config and serves your agents as AG-UI
Server-Sent Events that any [CopilotKit](https://copilotkit.ai) frontend can
consume:

```bash
OPENROUTER_API_KEY=sk-... uv run kaboo-serve config.yaml
```

The server listens on `http://localhost:8080` — `POST /invocations` for the
event stream and `GET /ping` for health.

## 3b. Or drive it from Python

`load()` hands back live, fully wired `strands` objects. There are no wrappers:
`resolved.entry` is a real `strands.Agent` (or a `Swarm`/`Graph` for
orchestrations), so you call it directly.

```{.python notest}
from kaboo_workflows import load

resolved = load("config.yaml")
agent = resolved.entry              # a plain strands.Agent
try:
    agent("What is 15 * 23?")       # streams the reply to stdout
finally:
    resolved.mcp_lifecycle.stop()   # always release MCP servers/clients
```

Wrap CLI-style entrypoints in `cli_errors()` for friendly error messages:

```{.python notest}
from kaboo_workflows import cli_errors, load

with cli_errors():
    resolved = load("config.yaml")
    resolved.entry("Hello!")
```

## Next steps

- **[Concepts](concepts.md)** — the mental model behind `load()` and serving.
- **[Configuration reference](configuration/README.md)** — every YAML option,
  chapter by chapter.
- **[Workflows](workflows/deep-nesting.md)** — worked, tested multi-agent
  patterns (delegate, swarm, graph, parallel, HITL).
- **[Examples](https://github.com/gl-pgege/kaboo-workflows/tree/main/examples)** —
  ~20 self-contained, load-tested example projects.
- **[Troubleshooting](troubleshooting.md)** — fixes for the common first-run
  errors.

# docs/index.md

# kaboo-workflows

**YAML-driven multi-agent orchestration with AG-UI and CopilotKit support, built
on [strands-agents](https://github.com/strands-agents/harness-sdk).**

Describe an entire multi-agent system in YAML — models, agents, tools, hooks, MCP
servers, and nested orchestrations — and `load()` hands back live, fully wired
strands objects. `kaboo-serve` then serves them as AG-UI Server-Sent Events that
any [CopilotKit](https://copilotkit.ai) frontend can consume.

## Start here

<div class="grid cards" markdown>

- **[Getting started](getting-started.md)** — install, write a config, and serve
  or drive it from Python in a few minutes.
- **[Concepts](concepts.md)** — the mental model: `load()` returns plain strands
  objects; orchestration shapes; the AG-UI event stream.
- **[Configuration reference](configuration/README.md)** — the complete
  19-chapter reference for every YAML option.
- **[Workflow guides](workflows/deep-nesting.md)** — worked, tested examples of
  deep nesting, swarm+graph, parallelism, HITL, history, and error handling.
- **[Recipes](recipes/index.md)** — production patterns: serving on AgentCore,
  per-run workflows, skills, user identity, guard hooks, cost control.
- **[Testing your workflows](testing.md)** — scripted models and an in-process
  pipeline to test a workflow end to end with no model or network.
- **[Best practices](best-practices.md)** — rules of thumb from production, and
  why each one holds.
- **[API reference](api-reference.md)** — the full public Python surface,
  auto-generated from docstrings.
- **[Troubleshooting](troubleshooting.md)** — fixes for the common first-run
  errors.

</div>

## The kaboo stack

kaboo-workflows is the orchestration engine. It pairs with:

- **[kaboo-runtime](https://github.com/gl-pgege/kaboo-runtime)** — a CopilotKit
  runtime plugin for persisting and replaying agent event logs (in-memory or
  Postgres `ThreadStore`s).
- **[kaboo-react](https://github.com/gl-pgege/kaboo-react)** — React components
  for rendering agent activity.
- **[kaboo-workflows-demo](https://github.com/gl-pgege/kaboo-workflows-demo)** —
  a runnable, end-to-end reference wiring all three together.

See [the kaboo stack](https://gl-pgege.github.io/kaboo-docs/) for the whole
picture.

## Everything is proven

Nothing documented here ships without a test: every example config is
load-validated, every complex workflow guide is backed by a passing end-to-end
test, executable doc snippets run in CI, and a completeness gate guarantees every
public symbol has a docstring and an API page.

# docs/recipes/asking-the-user.md

# Ask the user, and request inputs

An agent that guesses when it should ask produces confident wrong answers. The
library pauses a run for a question with `ask_user`, gates risky tools behind
an approval, and resumes the same run with the answer
([Human-in-the-loop](../workflows/human-in-the-loop.md) has the protocol).
This recipe adds the parts a product needs on top: custom requests — "attach a
file", "pick a data source" — that a UI renders as its own panel.

## The agent side

```yaml
agents:
  assistant:
    tools:
      - ./tools/requests.py:request_files
      - ./tools/requests.py:request_datasources
    interrupt:
      tools: []             # tools that need approval before running
      ask_user: true        # grant the built-in ask_user tool
      ttl_seconds: 86400    # stamp an expiry the client can show and enforce
    system_prompt: |
      When the question is ambiguous, ask once with ask_user, putting every
      open question into that one call. When the data you need is not
      attached, call request_datasources rather than asking in prose.
```

Grant `ask_user` to the agent that talks to the user, and usually not to the
analysts it delegates to: questions should come from one voice, and a
sub-agent's question pauses the whole run anyway.

## A custom request is a tool that interrupts

Any tool can pause the run. Declare it with `context=True` and call
`tool_context.interrupt`; whatever the client sends back on resume becomes the
interrupt's return value, and the tool turns it into a result the model can
read:

```python
import json
from typing import Any, Literal

from strands import tool
from strands.types.tools import ToolContext

Kind = Literal["sql", "kafka", "prometheus"]


def _answer(response: Any) -> str:
    if response is None:
        return "The user did not provide a selection."
    if isinstance(response, dict) and response.get("status") == "cancelled":
        return "The user declined the request."
    return json.dumps(response, ensure_ascii=False)


@tool(name="request_datasources", context=True)
def request_datasources(
    tool_context: ToolContext, reason: str, kinds: list[Kind] | None = None
) -> dict:
    """Pause and ask the user to select one or more data sources.

    Use this when the analysis needs a data source the run does not have.

    Args:
        reason: Why the data source is needed, shown to the user.
        kinds: The kinds of data source to offer; all when omitted.
    """
    response = tool_context.interrupt(
        name="request_datasources",
        reason={
            "type": "form",
            "kind": "datasources",
            "kinds": list(kinds) if kinds else ["sql", "kafka", "prometheus"],
            "questions": [{"question": reason, "type": "text"}],
        },
    )
    return {"status": "success", "content": [{"text": _answer(response)}]}
```

The `reason` payload is the contract with your UI:

- **`type`** is one the client already renders. kaboo-react renders
  `approval` and `form`; ours registers one `form` renderer that branches on
  **`kind`** to show the file picker or the data source picker.
- **`questions`** keeps the plain form meaningful wherever the custom panel is
  not registered, so an older client degrades to a text question rather than a
  dead end.
- **A cancelled request is an answer.** Tell the model the user declined, so it
  can continue without the input instead of asking again.

## What the user experiences

- The run pauses; the chat shows the panel; the answer resumes the *same* run,
  so the agent continues mid-thought rather than starting over.
- A paused run survives a restart or a second replica: the pending question
  travels in the AG-UI state the client sends back
  ([A gate outlives the process](../workflows/human-in-the-loop.md#a-gate-outlives-the-process-that-opened-it)).
- If the user ignores the question and types something new, the new message
  starts a fresh run. A pending question is restored only when the client
  resumes it.
- `ttl_seconds` only stamps an expiry; enforcing it is the client's or host's
  job.

## Test it

A scripted model can call your request tool, and `Pipeline.turn(resume=...)`
answers it, including on a fresh pipeline to prove the restart case
([Testing](../testing.md#interrupts-and-resume)).

# docs/recipes/attachments-and-images.md

# Images and file attachments

Users attach spreadsheets, PDFs and screenshots; tools produce images worth
looking at. [Chapter 19](../configuration/Chapter_19.md) documents every knob.
This recipe is the setup Kaboo runs, and why each setting is there.

## The setup

```yaml
attachments:
  default: reference
  tool: true
  base_url: ${API_BASE_URL:-http://localhost:3000}
  authorization: forwarded_props:runToken
  content_url_template: /api/storage/download/{id}

agents:
  orchestrator: {}                       # manifest + fetch tool
  chart_reviewer:
    attachments: {inline: true}          # also sees images directly
  formatter:
    attachments: none                    # never needs the files
```

- **`default: reference`** gives every agent a manifest of what the user
  attached and nothing heavier. A manifest line costs a few tokens; an inlined
  PDF costs thousands on every turn it stays in history.
- **`tool: true`** grants `list_references` and `fetch_attachment`, so an agent
  pulls a file's content only when its task needs it.
- **`base_url` + `authorization`** let the server fetch files from your own API
  with the caller's token, taken from the run's `forwardedProps`. The token is
  sent only to URLs under `base_url`; presigned and third-party URLs are
  fetched without it.
- **`content_url_template`** makes a file fetchable on *every* turn, not only
  the turn it was attached on — the client sends the file's id in state, and
  the template turns it into a URL.
- **`inline: true`** only on the agent that must see pixels, on a model that
  can. Everyone else works from the manifest.

## Tools that return images

A tool behind an MCP gateway can only answer in text, so return images as
base64 inside the result's JSON under `attachments`. The library strips them
out of the text and hands them to the model as image blocks of the same tool
result
([Chapter 19](../configuration/Chapter_19.md#media-a-tool-returns)):

```json
{
  "summary": "3 problems in 2 of 11 widgets",
  "attachments": [{"name": "Rows 1-7", "mime_type": "image/jpeg", "data": "<base64>"}]
}
```

Kaboo uses this to close the loop on dashboards: after building one, the
visualizer calls a verification tool that renders the dashboard in a headless
browser and returns screenshots, and the agent fixes what it can see — a
chart with no data, a title cut off — before it reports back. Agents that can
see their output catch problems no schema check will.

Keep images small and few: at most 8 per result are unpacked, each up to
3.75 MB, and every one stays in the agent's history.

## Custom kinds of reference

Users can cite things that are not files — a table, a dashboard, a ticket.
These arrive as references with a `kind` and an `id`, appear in the same
manifest, and are resolved by your own tools (`get_dashboard(id)`), not by
`fetch_attachment`. A data source cited this way is also what
[tag-activated skills](skills.md#which-skills-an-agent-gets) match on.

## Frontend

kaboo-react mints the reference ids, sends files as multimodal message parts on
the turn they are attached and as state on later turns, and renders the
attachment chips
([Chapter 19](../configuration/Chapter_19.md#frontend-wiring-kaboo-react)).

# docs/recipes/chat-ui-activity.md

# Activity a chat UI can show

A multi-agent run can take minutes. Users trust it when they can see it
working — which specialist is on the job, what it is doing right now, in
words rather than function names. Two pieces of config produce that, and
kaboo-react renders it.

## What the client receives

`create_agui_app` folds every agent's stream into **activity groups** and sends
them alongside the chat reply: a snapshot first, then JSON Patches as they
change (`activity_deltas`). Each group is one card in the UI:

- **the agent's title**, and the task it was handed;
- **its tool calls**, each with a readable label, arguments and result;
- **its text**, its token usage, and any [structured output](structured-outputs.md);
- **its children**, for agents it delegated to.

The entry agent has no card: its stream *is* the chat reply.

## Name the cards

```yaml
agents:
  orchestrator:
    model: default
  sql_analyst:
    stream:
      title: Analyzing SQL data
  sql_visualizer:
    stream:
      group: sql_visualization
      title: Creating SQL visualizations
```

Titles are what a user reads, so write them as what the agent is *doing*,
and in the user's language — "Analyzing SQL data", not "sql_analyst". An agent
with no title gets one derived from its name, which is rarely the phrase you
would choose. Agents that share a `group` render as one card, which suits a
pair that works on the same thing in turn.

## Label the tools

```yaml
mcp_clients:
  web_research:
    url: ${GATEWAY_URL}
    transport: streamable-http
    params:
      tool_filters:
        allowed: [web-research___search, web-research___read_pages]
    tool_labels:
      web-research___search: Searching the web
      web-research___read_pages: Reading web pages
```

A tool call shows its label (`toolLabel`) instead of
`web-research___read_pages`. Label every tool a user will see; an unlabelled
one shows its raw name.

## Keep it fast

- **Leave `activity_deltas` on.** Sending a full tree on every change grows
  with the run; on long multi-agent runs that is what fills a response
  ([Serve on AgentCore in production](production-serving.md#stay-under-the-response-cap)).
- **Do not route the chat reply through a group.** The entry agent's text is
  the answer and streams as ordinary AG-UI text messages.

## Test it

`TurnResult.groups` holds the cards as the UI would, so a test can assert the
tree a user would see — which agents worked, nested under whom
([Testing](../testing.md#multi-agent-runs-and-activity)).

# docs/recipes/cost-and-context.md

# Cost and context control

A data-analysis run can make dozens of model calls, each resending a long
system prompt, a tool list and every earlier tool result. Most of the cost —
and most context-window trouble — comes from that resending. Each setting
below is documented in its chapter; this recipe is the combination Kaboo runs,
in order of payoff.

## 1. Cache what every call repeats

```yaml
models:
  default:
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-5
    params:
      cache_prompt: default        # the system prompt
      cache_tools: default         # the tool list
      cache_config: {strategy: auto}   # the conversation up to the last message
      max_tokens: 32000
```

Three of Bedrock's four cache points: a tool loop re-reads its system prompt,
tools and earlier turns from the cache instead of paying for them again on
every call. Loaded history keeps a cache point of its own
([Multi-turn history](../workflows/multi-turn-history.md#loaded-memory-is-cached-on-bedrock)).
Anything that rewrites the start of the prompt on every call — a timestamp at
the top of a system prompt, say — defeats the cache; append such lines at the
end instead.

## 2. Think where it pays

```yaml
models:
  default:
    thinking: {effort: high, display: summarized}

agents:
  orchestrator:
    thinking: {effort: medium}     # routing and synthesis
  sql_analyst:
    thinking: false                # writes code against a known schema
```

Set thinking on the model as a default, then turn it down per agent
([Chapter 2](../configuration/Chapter_02.md#thinking)). Agents doing
mechanical work against clear instructions rarely gain from it; the agent that
plans and judges does. Adaptive thinking needs a model that supports it — a
config written for Sonnet 5 sent to an older model fails at the provider.

## 3. Clear old tool results

```yaml
agents:
  sql_analyst:
    context:
      clear_tool_results:
        trigger: 60000
        keep_last: 3
        keep_tools: [sql___get_schema]
```

Query results are the bulk of a long analysis, and an agent rarely needs the
ninth-from-last one verbatim. Past the trigger, older results are replaced by a
line naming the tool and the ids in the result, so the agent can fetch again
what it actually needs ([Chapter 8](../configuration/Chapter_08.md#clearing-old-tool-results)).
Keep reference results such as a schema.

## 4. Bound what carries across turns

```yaml
agents:
  orchestrator:
    conversation_manager:
      type: strands.agent:SlidingWindowConversationManager
      params: {window_size: 80}
    history:
      group: orchestrator
  sql_analyst:
    history:
      group: sql
      keep: masked
      summarize_over: 40000
      summarize_model: utility
```

A sliding window bounds what an agent keeps within a run. History bounds what
it carries into the next turn: masked history writes back each finished tool
result as a one-line pointer the agent can call again, and `summarize_over`
has a cheap model condense the oldest calls once the masked memory still grows
too large
([Multi-turn history](../workflows/multi-turn-history.md#keeping-a-masked-memory)).
Agents that share a history `group` share one memory — useful when two agents
take turns as the same voice.

## 5. Keep toolboxes small

Every tool's schema rides along on every call, and a model choosing among
forty tools chooses worse than one choosing among eight. Kaboo refuses a
workflow that gives an agent more than 20 tools, and splits each service's
tools into narrow MCP clients — an `_explore` client for read-only discovery
beside the full one — so an agent is granted exactly what its job needs
([Call MCP tools as the signed-in user](user-identity-mcp.md#relay-the-token-on-every-mcp-call)).

## 6. Put ceilings on runaway work

- `MaxToolCallsGuard` caps tool calls per invocation, so a confused agent stops
  instead of looping ([Chapter 6](../configuration/Chapter_06.md)).
- [Refusing repeated failures](guard-hooks.md#refuse-a-call-that-has-already-failed-twice)
  stops the commonest loop of all: the same broken call, again.
- A small, deterministic model for utility work — naming a conversation,
  classifying a request — costs a fraction of the analyst model:

    ```yaml
    models:
      utility:
        provider: bedrock
        model_id: us.anthropic.claude-haiku-4-5-20251001-v1:0
        params: {temperature: 0, max_tokens: 200}
    ```

## Measure it

Token usage per run arrives in the activity stream and in `TurnResult.usage_by_run`
in tests; with [telemetry](../evaluation.md#enable-telemetry) on, every model
call's tokens and latency land in your tracing backend. Measure before and
after each change — caching in particular is easy to break silently.

# docs/recipes/guard-hooks.md

# Guard hooks for every agent

Some failures have nothing to do with any particular workflow: a transcript
the provider rejects, a model retrying the same broken call, a user whose
sign-in expires mid-run. Hooks
([Chapter 6](../configuration/Chapter_06.md)) are where to catch them, once,
for every agent. These are the guards Kaboo puts on every agent of every
workflow.

## Answer every tool call before the model sees the transcript

Bedrock rejects a turn unless every tool call in an assistant message is
answered by a tool result in the very next message — and strands reacts to the
rejection by stopping the run silently. Transcripts break this rule more often
than you would think: a turn cut off mid-tools, an interrupted call replayed
from the client, a message the host persisted twice. Bedrock is just as strict
about two messages in a row from the same role, and about empty text blocks.

Repairing the transcript before every model call turns each of those into a
run that carries on:

```python
from typing import Any

from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import BeforeModelCallEvent

UNANSWERED = (
    "This call did not run: the turn that made it ended before it could. "
    "Call it again if you still need its result."
)


def _blank(block: Any) -> bool:
    return isinstance(block, dict) and set(block) == {"text"} and not str(block["text"]).strip()


def alternating(messages: list[dict]) -> list[dict]:
    merged: list[dict] = []
    for message in messages:
        content = [b for b in message.get("content") or [] if not _blank(b)]
        if not content:
            continue
        if merged and merged[-1]["role"] == message["role"]:
            merged[-1] = {**merged[-1], "content": merged[-1]["content"] + content}
        else:
            merged.append({**message, "content": content})
    for message in merged:
        if message["role"] == "user":
            results = [b for b in message["content"] if "toolResult" in b]
            message["content"] = results + [b for b in message["content"] if b not in results]
    return merged


def answered(messages: list[dict]) -> list[dict]:
    messages = alternating(messages)
    repaired: list[dict] = []
    for index, message in enumerate(messages):
        repaired.append(message)
        if message["role"] != "assistant":
            continue
        asked = [b["toolUse"]["toolUseId"] for b in message["content"] if "toolUse" in b]
        following = messages[index + 1] if index + 1 < len(messages) else None
        replies = following if following and following["role"] == "user" else None
        given = {
            b["toolResult"]["toolUseId"]
            for b in (replies or {}).get("content", [])
            if "toolResult" in b
        }
        missing = [
            {"toolResult": {"toolUseId": t, "status": "error", "content": [{"text": UNANSWERED}]}}
            for t in asked
            if t not in given
        ]
        if not missing:
            continue
        if replies is not None:
            replies["content"] = missing + replies["content"]
        else:
            repaired.append({"role": "user", "content": missing})
    return repaired


class AnswerEveryToolUse(HookProvider):
    def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
        registry.add_callback(BeforeModelCallEvent, self._answer)

    def _answer(self, event: BeforeModelCallEvent) -> None:
        event.agent.messages[:] = answered(event.agent.messages)


broken = [
    {"role": "user", "content": [{"text": "Run the report"}]},
    {"role": "assistant", "content": [{"toolUse": {"toolUseId": "t1", "name": "run", "input": {}}}]},
    {"role": "user", "content": [{"text": "Any news?"}]},
]
fixed = answered(broken)
assert fixed[2]["content"][0]["toolResult"]["toolUseId"] == "t1"
```

The inserted result is an error the model can read, so it calls the tool again
if it still needs it.

## Refuse a call that has already failed twice

A model that gets an error back sometimes sends the identical call again, and
again. Refusing the third identical attempt, with a message telling it to
change course, breaks the loop without ending the run:

```python
import json
from typing import Any

from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import AfterToolCallEvent, BeforeInvocationEvent, BeforeToolCallEvent

REPEATED_FAILURE = (
    "This exact call has already failed twice, so it was not sent again. Read the "
    "earlier errors, then change the arguments or take a different approach."
)


def failed(result: Any) -> bool:
    return isinstance(result, dict) and result.get("status") == "error"


class RefuseRepeatedFailures(HookProvider):
    def __init__(self, limit: int = 2) -> None:
        self.limit = limit
        self.failures: dict[str, int] = {}

    def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
        registry.add_callback(BeforeInvocationEvent, lambda event: self.failures.clear())
        registry.add_callback(BeforeToolCallEvent, self._refuse)
        registry.add_callback(AfterToolCallEvent, self._count)

    @staticmethod
    def _key(tool_use: Any) -> str:
        return json.dumps([tool_use.get("name"), tool_use.get("input")], sort_keys=True, default=str)

    def _refuse(self, event: BeforeToolCallEvent) -> None:
        if self.failures.get(self._key(event.tool_use), 0) >= self.limit:
            event.cancel_tool = REPEATED_FAILURE

    def _count(self, event: AfterToolCallEvent) -> None:
        key = self._key(event.tool_use)
        if failed(event.result):
            self.failures[key] = self.failures.get(key, 0) + 1
        else:
            self.failures.pop(key, None)
```

The key is the tool name *and* its input: the same tool with different
arguments is a new attempt, which is exactly what the message asks for. Ours
also counts a result whose JSON body says `"success": false` or carries an
`"error"`, since many MCP services report failure that way. The library's
`MaxToolCallsGuard` is the coarser companion: a ceiling on tool calls per
invocation, whatever they are.

## Stop when the sign-in expires

A run relaying the user's token ([Call MCP tools as the signed-in
user](user-identity-mcp.md)) can outlive it, and then every tool call fails
authorization. Check the token's expiry before each model call and each tool
call, and answer with one clear message instead:

```python
import time
from typing import Any

from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import BeforeModelCallEvent, BeforeToolCallEvent
from strands.models import Model

from kaboo_workflows import get_auth_context

SIGNED_OUT = (
    "Your sign-in expired while I was working, so I stopped here. Sign in again "
    "and send your question once more; what I built so far is kept."
)


def signed_out() -> bool:
    principal = get_auth_context()
    expiry = principal.claims.get("exp") if principal else None
    return isinstance(expiry, (int, float)) and time.time() >= expiry


class SignedOutModel(Model):
    def update_config(self, **model_config: Any) -> None:
        pass

    def get_config(self) -> Any:
        return {}

    def structured_output(self, *args: Any, **kwargs: Any) -> Any:
        raise NotImplementedError

    async def stream(self, *args: Any, **kwargs: Any) -> Any:
        yield {"messageStart": {"role": "assistant"}}
        yield {"contentBlockDelta": {"delta": {"text": SIGNED_OUT}}}
        yield {"contentBlockStop": {}}
        yield {"messageStop": {"stopReason": "end_turn"}}


class StopOnExpiredSignIn(HookProvider):
    def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
        registry.add_callback(BeforeModelCallEvent, self._answer_signed_out)
        registry.add_callback(BeforeToolCallEvent, self._refuse_tool)

    def _answer_signed_out(self, event: BeforeModelCallEvent) -> None:
        if signed_out():
            event.agent.model = SignedOutModel()

    def _refuse_tool(self, event: BeforeToolCallEvent) -> None:
        if signed_out():
            event.cancel_tool = SIGNED_OUT


assert not signed_out()
```

Swapping in a model that replies with the message makes it the agent's answer,
so it reaches the chat like any other reply. An agent with an `output_schema`
cannot answer in prose; on strands versions with `AfterToolsEvent`, ours
cancels such an agent's invocation and ends its turn after the current tools
instead.

## Two small ones

- **Today's date.** Models do not know what day it is. A
  `BeforeInvocationEvent` hook that appends "Today's date is …" in the user's
  time zone (`get_forwarded_props()["timezone"]`) to the system prompt fixes
  every "last month" question.
- **Model timing.** A `BeforeModelCallEvent` / `AfterModelCallEvent` pair that
  logs each call's latency, time to first event and retry count with the run id
  (`get_run_id()`) answers "why was that slow?" without a tracing backend.

## Put them on every agent

Either add them when composing the workflow, as Kaboo does
([Compose a workflow per run](per-run-workflows.md)), or declare them once in
the config and alias them:

```yaml
x-guards: &guards
  - my_service.hooks:AnswerEveryToolUse
  - my_service.hooks:RefuseRepeatedFailures
  - my_service.hooks:StopOnExpiredSignIn

agents:
  analyst:
    hooks: *guards
  visualizer:
    hooks: *guards
```

`x-` blocks exist only to hold anchors; the loader strips them
([Chapter 4](../configuration/Chapter_04.md)).

## Test them

Each guard has a failure it exists for, so script the model into that failure
and assert the run survived it — a broken transcript passed in as `messages`,
the same failing call scripted three times
([Testing](../testing.md#hooks-and-guards)).

# docs/recipes/index.md

# Recipes

The configuration chapters say what every field does. These recipes show how
the pieces fit together in a real service — each one is a pattern running in
production in [Kaboo](https://gl-pgege.github.io/kaboo-docs/), written so you
can lift it into your own project.

| Recipe | Reach for it when |
|--------|-------------------|
| [Serve on AgentCore in production](production-serving.md) | You deploy behind AWS AgentCore Runtime, or any host with response caps and idle reaping |
| [Compose a workflow per run](per-run-workflows.md) | One service runs many workflows — per tenant, per project, per admin-authored row |
| [Skills](skills.md) | Agents need reusable instructions, loaded only when a task calls for them |
| [Call MCP tools as the signed-in user](user-identity-mcp.md) | Tools behind a gateway must act with the caller's permissions |
| [Tools that need run context](tools-with-run-context.md) | A tool keeps state per conversation or must act once per turn |
| [Guard hooks for every agent](guard-hooks.md) | Runs should survive malformed transcripts, retry loops and expired sign-ins |
| [Ask the user, and request inputs](asking-the-user.md) | An agent needs an answer, an approval, a file or a data source mid-run |
| [Structured outputs a UI can render](structured-outputs.md) | An agent's result should arrive as typed data for a card, not prose |
| [Images and file attachments](attachments-and-images.md) | Users attach files, or tools return images |
| [Activity a chat UI can show](chat-ui-activity.md) | Users should see which agent is working on what, in words |
| [Cost and context control](cost-and-context.md) | Long, tool-heavy runs cost too much or outgrow the context window |

[Testing your workflows](../testing.md) covers proving any of these with a
scripted model, and [Best practices](../best-practices.md) collects the rules
of thumb behind them.

# docs/recipes/per-run-workflows.md

# Compose a workflow per run

One service can run a different workflow on every request: a workflow per
tenant, per project, per row an admin edits in a UI. Nothing is restarted and
nothing is read from disk at run time. The host stores workflows, composes the
final YAML for each run, and submits it with the run; the agent service layers
it over its own base config
([Chapter 17](../configuration/Chapter_17.md#when-the-config-itself-is-per-request),
`examples/21_runtime_configs/`).

## Split the config in two

**The base config** lives with the service and holds what every workflow
shares, including anything secret:

```yaml
# config.yaml — the service's own file
models:
  analyst-model:
    provider: bedrock
    model_id: ${BEDROCK_MODEL_ID:-us.anthropic.claude-sonnet-5}
    params: {region_name: "${BEDROCK_REGION:-us-east-1}"}

mcp_clients:
  sql:
    url: ${GATEWAY_URL}
    transport: streamable-http
    auth: {type: relay}
    params:
      tool_filters: {allowed: [sql___run_query, sql___get_schema]}

attachments:
  default: reference

agents:
  assistant:                       # the fallback when a run submits nothing
    model: analyst-model
    system_prompt: You are a helpful data assistant.
entry: assistant
```

**A stored workflow** holds only `agents`, `orchestrations` and `entry`, and
refers to models and MCP clients by name:

```yaml
agents:
  analyst:
    model: analyst-model
    mcp: [sql]
    system_prompt: You answer questions from the sales database.
entry: analyst
```

The overlay replaces the base's `agents`, `orchestrations` and `entry`, adds to
its `models` and `mcp_clients`, and may not declare `mcp_servers` at all — a run
may not launch a process. Stored workflows that only *name* the base's models
and clients never carry a URL or a credential.

## Compose, then submit

Composition is plain dictionary work on the host. Ours applies the user's
choices (thinking level, model override), drops agents for data sources the run
does not have, inlines [skills](skills.md), adds the
[hooks every agent needs](guard-hooks.md), stamps today's date, and fills in
IDs such as the current project:

```python
import yaml

from kaboo_workflows import load_session_config, parse_config_sources

EVERY_AGENT_HOOKS = ["my_service.hooks:AnswerEveryToolUse"]


def compose(stored: str, *, project_id: str) -> str:
    doc = yaml.safe_load(stored)
    for agent in doc.get("agents", {}).values():
        agent["hooks"] = [*agent.get("hooks", []), *EVERY_AGENT_HOOKS]
    text = yaml.safe_dump(doc, sort_keys=False)
    text = text.replace("{{project_id}}", project_id)
    return text.replace("${", "$${")


STORED = """
agents:
  analyst:
    model: analyst-model
    system_prompt: |
      You answer questions about project {{project_id}}.
      Quote templates such as ${name} literally.
entry: analyst
"""

BASE = parse_config_sources("""
models:
  analyst-model: {provider: bedrock, model_id: us.anthropic.claude-sonnet-5}
agents:
  assistant: {model: analyst-model, system_prompt: Fallback.}
entry: assistant
""")

composed = compose(STORED, project_id="p-42")
config = load_session_config(BASE, composed)

prompt = config.agents["analyst"].system_prompt
assert "project p-42" in prompt
assert "${name}" in prompt
assert config.agents["analyst"].hooks == EVERY_AGENT_HOOKS
```

The run then carries the composed text in `forwardedProps`, under the key the
service was started with (`create_agui_app(session_config_key="workflow_config")`):

```json
{
  "threadId": "t-1",
  "runId": "r-1",
  "messages": [{"id": "m-1", "role": "user", "content": "How did Q3 go?"}],
  "forwardedProps": {"workflow_config": "agents:\n  analyst: ..."}
}
```

## Escape interpolation in anything authored outside the repository

A submitted workflow is interpolated against **the service's environment**, on
purpose, so an admin-authored config can say `${API_KEY}` without the secret
living in the database. The flip side: any `${…}` in user-authored text — a
prompt that quotes template syntax, or one written to fish for
`${AWS_SECRET_ACCESS_KEY}` — resolves against the server too.

`$${` is the escape for a literal `${`. We escape every `${` in the composed
workflow, which means stored workflows never read the environment and
everything secret stays in the base config. If your workflow authors are
trusted and need environment values, escape only the fields they write freely,
such as prompts.

## Keep it safe and testable

- **Validate on save**, with the same loader a run uses
  ([production recipe](production-serving.md#validate-a-workflow-before-saving-it)),
  so a broken workflow is an error in the editor rather than a failed run.
- **Pass `allowed_mcp_hosts`** to `create_agui_app` and `load_session_config`
  if a workflow can declare its own MCP clients.
- **Refuse leftovers.** After filling placeholders, fail composition if any
  remain; a prompt that still says `{{dashboard_id}}` produces confident
  nonsense.
- **Test the composed text**, not the stored one:
  `Pipeline(base, session_config_key="workflow_config")` with the composer's
  output in `forwarded_props` runs it end to end
  ([Testing](../testing.md#runtime-composed-workflows)).

# docs/recipes/production-serving.md

# Serve on AgentCore in production

`create_agui_app` already speaks AgentCore Runtime's HTTP contract: AG-UI on
`POST /invocations`, a health check on `GET /ping`. This recipe covers what a
production deployment adds on top — the settings worth turning on, and four
behaviours of the host that the defaults do not account for.

## The app

```{.python notest}
import os
from pathlib import Path

import strands.models.bedrock as bedrock

from kaboo_workflows.adapters import create_agui_app

from my_service.auth import verify_run_token

bedrock.DEFAULT_READ_TIMEOUT = int(os.environ.get("BEDROCK_READ_TIMEOUT", "300"))

app = create_agui_app(
    Path(__file__).parent / "config.yaml",
    session_config_key="workflow_config",
    allowed_mcp_hosts=["gateway.internal.example.com"],
    auth=verify_run_token,
    ping_path=None,
    activity_deltas=True,
    response_budget=CONTINUATION_BUDGET,
)
```

- **`session_config_key`** lets each run bring its own workflow, layered over
  the base config — see [Compose a workflow per run](per-run-workflows.md).
- **`allowed_mcp_hosts`** bounds what a submitted workflow's MCP clients may
  connect to. Set it whenever workflows are authored anywhere but your
  repository: an arbitrary URL needs no code to exfiltrate data.
- **`auth`** verifies every request and binds the caller, so MCP tools can run
  as them — see [Call MCP tools as the signed-in user](user-identity-mcp.md).
- **`ping_path=None`** because the packaged health check always answers
  healthy, and AgentCore needs more than that (below).
- **`activity_deltas`** sends activity as JSON Patches after the first
  snapshot. Full snapshots on every change are what fill a response on long,
  multi-agent runs.
- **The Bedrock read timeout** defaults to strands' value, which a long
  thinking turn can exceed. Raise it for the slowest call you expect.

AgentCore Runtime expects the container on port 8080.

## Report busy while a run is open

AgentCore decides a session is idle from its `/ping` status, not from whether a
request is still on the wire. A session that only ever answers `Healthy` can be
reaped mid-run, which reaches the user as a stream that ends with no terminal
event. Answer `HealthyBusy` while any invocation is open:

```{.python notest}
class RunCounter:
    def __init__(self) -> None:
        self.active = 0

    @property
    def status(self) -> str:
        return "HealthyBusy" if self.active else "Healthy"


runs = RunCounter()


@app.get("/ping")
def ping() -> dict[str, str]:
    return {"status": runs.status}
```

Increment `runs.active` when a `POST /invocations` arrives and decrement it when
the response ends — in the ASGI middleware below, which sees both.

## Stay under the response cap

AgentCore closes the socket once a response reaches 100 MiB, with no error of
its own. The run dies without a terminal event, and CopilotKit reports
whichever tool call was open as failed — the user is told a healthy tool broke.
Two budgets keep that from happening:

1. **`response_budget`** pauses the run at the next tool boundary once the
   response has carried that many bytes, and continues the same turn in a fresh
   response. The user loses nothing. Clients resolve the pause on their own; it
   is an interrupt marked `reason: "continuation"`
   ([Human-in-the-loop](../workflows/human-in-the-loop.md)).
2. **A hard backstop** for a response that fills without ever reaching a tool
   call: wrap the ASGI `send` so it forwards whole SSE frames, and once the
   budget is spent, close the stream with a real `RUN_FINISHED` and a short
   message saying the answer was cut off. A truncated answer with a reason
   beats a failure with none.

Give the pause room to fire first:

```{.python notest}
AGENTCORE_RESPONSE_CAP = 104_857_600
HARD_BUDGET = AGENTCORE_RESPONSE_CAP - 10 * 1024 * 1024
CONTINUATION_BUDGET = HARD_BUDGET * 3 // 4
```

The margin under the cap leaves room for the closing events and for
chunked-transfer framing, which counts against the cap but is not yours to
measure. When cutting the stream, split only on the blank line between SSE
frames — cutting an arbitrary chunk leaves the client half an event.

## Non-streaming calls over `/invocations`

AgentCore Runtime forwards only `POST /invocations`. Anything else your service
answers — validating a workflow before an admin saves it, running a workflow
once for a background job — has to ride that path too. Dispatch on a field in
the body, in a **pure ASGI** middleware, and replay the body downstream for
ordinary AG-UI runs:

```{.python notest}
class NonStreamingOverInvocations:
    def __init__(self, app, runs: RunCounter) -> None:
        self.app = app
        self.runs = runs

    async def __call__(self, scope, receive, send) -> None:
        if scope["type"] != "http" or scope["path"] != "/invocations" or scope["method"] != "POST":
            await self.app(scope, receive, send)
            return
        self.runs.active += 1
        try:
            body = await read_body(receive)
            if b'"my_validate"' in body[:4096]:
                await answer_validation(body, scope, send)
                return
            await self.app(scope, replay(body, receive), budgeted(send, body))
        finally:
            self.runs.active -= 1


app.add_middleware(NonStreamingOverInvocations, runs=runs)
```

Starlette's `BaseHTTPMiddleware` will not do here: once it has consumed the
body, its disconnect listener receives the replayed `http.request` and breaks
the downstream SSE stream. Authenticate the non-streaming branches with the
same verifier the AG-UI path uses.

## Validate a workflow before saving it

`load_session_config` runs the same overlay, schema and reference checks a run
would, without building anything:

```{.python notest}
from kaboo_workflows import load_session_config, parse_config_sources

BASE_RAW = parse_config_sources("config.yaml")


def validate(workflow_yaml: str) -> dict:
    try:
        config = load_session_config(
            BASE_RAW, workflow_yaml, allowed_mcp_hosts=["gateway.internal.example.com"]
        )
    except Exception as exc:
        return {"valid": False, "error": str(exc)}
    return {"valid": True, "entry": config.entry, "agents": sorted(config.agents)}
```

The error names the bad reference and lists what is available, which makes it
a good message to show an admin. Add your own rules after it. Ours refuse an
agent holding more than 20 tools, a tool granted twice by overlapping MCP
clients, and a prompt that names a tool the agent is not granted.

## Run a workflow once, without streaming

For a background job, build the run the same way a streamed run is built and
answer with the final text:

```{.python notest}
from dataclasses import replace

from kaboo_workflows import (
    load_session,
    load_session_config,
    resolve_infra,
    resolve_run_clients,
    set_auth_context,
    validate_raw_config,
)

BASE_INFRA = resolve_infra(validate_raw_config(BASE_RAW))


async def run_once(workflow_yaml: str, prompt: str, principal) -> str:
    config = load_session_config(BASE_RAW, workflow_yaml, allowed_mcp_hosts=ALLOWED)
    set_auth_context(principal)
    clients = resolve_run_clients(config, BASE_INFRA)
    clients.start(pin_clients=False)
    try:
        resolved = load_session(
            config, replace(BASE_INFRA, clients=clients.clients, mcp_lifecycle=clients)
        )
        result = await resolved.entry.invoke_async(prompt)
        return str(result).strip()
    finally:
        clients.stop()
```

`set_auth_context` comes first so relayed MCP calls go out as the caller, and
the run's clients stop with the call, as they do with a streamed run.

## Related

- [Chapter 17: when the config itself is per-request](../configuration/Chapter_17.md#when-the-config-itself-is-per-request)
- [Evaluation: enable telemetry](../evaluation.md#enable-telemetry) for traces
  in Langfuse or any OTLP backend
- `examples/agentcore/` for inbound auth styles on AgentCore

# docs/recipes/skills.md

# Skills

A skill is a named piece of instructions — how to write SQL against this
warehouse, how to lay out a dashboard, how to brief a banker on a customer —
that agents use only when a task calls for it. Skills keep system prompts short
and let one well-written instruction serve every agent that needs it.

kaboo-workflows has no `skills:` field of its own. Skills are composed into the
workflow on the host, before the run is submitted — an application of
[Compose a workflow per run](per-run-workflows.md). This recipe is the pattern
Kaboo runs: a skill catalogue in a database, three ways to load a skill, and a
single tool agents call to load one.

## Three loading strategies

| Strategy | What happens | Use it for |
|----------|--------------|------------|
| `full` | Inlined into the agent's system prompt as top-priority rules | Short rules that must always apply: investigation discipline, house style |
| `catalog` | Listed by name and when-to-use; the agent loads it with a tool call | Guides only some tasks need: a tool manual, a web research protocol |
| `rag` | Listed like a catalog skill; the agent loads matching sections by query | Long references: an SDK manual, a widget reference |

Every agent's prompt ends with the skills it has *not* loaded, so an agent can
always discover the one that fits — and most skills cost nothing until used.

## Which skills an agent gets

- **Bound** in the stored workflow, with a host-side key the composer removes
  before submitting:

    ```yaml
    agents:
      sql_analyst:
        skills: [sql-guide, efficiency-discipline]
        mcp: [sql, skills]
    ```

- **Activated by tags.** A skill tagged with a data source's name or kind is
  loaded whenever the run has that data source attached — a skill tagged
  `offers-redshift` switches on for every run that attaches the connection
  named `offers-redshift`. We scope tag-activated skills to the agents that
  work with that kind of data, and the orchestrator always qualifies.
- **Gated by tools.** A skill that only makes sense with certain MCP clients
  (`requires_clients`) is only offered to agents that hold one of them.

## The composer

```python
from dataclasses import dataclass, field

CATALOG_HEADER = (
    "\n\n<available_skills>\n"
    "These skills are NOT loaded yet. Before starting any task, check whether one\n"
    "of them covers it, and call `load_skill` with its name if so. Do the work only\n"
    "after the relevant skill is loaded. If a skill you loaded earlier is no longer\n"
    "visible in your context (long runs compress old messages), load it again.\n\n"
)


@dataclass
class Skill:
    name: str
    description: str
    content: str
    strategy: str = "full"
    rules: str | None = None
    tags: list[str] = field(default_factory=list)
    requires_clients: list[str] = field(default_factory=list)


def compose_skills(doc: dict, skills: dict[str, Skill], active_tags: set[str]) -> None:
    for agent in doc.get("agents", {}).values():
        bound = agent.pop("skills", [])
        held = set(agent.get("mcp", []))

        def usable(skill: Skill) -> bool:
            return not skill.requires_clients or bool(held & set(skill.requires_clients))

        names = [name for name in dict.fromkeys(bound) if name in skills]
        names += [
            s.name
            for s in skills.values()
            if s.name not in names and set(s.tags) & active_tags and usable(s)
        ]

        full = [skills[n] for n in names if skills[n].strategy == "full"]
        if full:
            rules = "\n\n".join(f"### {s.name}\n{s.content}" for s in full)
            agent["system_prompt"] = (
                f'<rules priority="1">\n{rules}\n</rules>\n\n'
                f"<agent_instructions>\n{agent['system_prompt']}\n</agent_instructions>"
            )

        catalog = [
            s
            for s in skills.values()
            if (s.name in names and s.strategy != "full") or (s.name not in names and usable(s))
        ]
        if catalog:
            lines = "\n".join(f"- **{s.name}**: {s.rules or s.description}" for s in catalog)
            agent["system_prompt"] += CATALOG_HEADER + lines + "\n</available_skills>"


SKILLS = {
    "efficiency-discipline": Skill(
        "efficiency-discipline", "Avoid redundant work", "Never repeat a tool call."
    ),
    "sql-guide": Skill(
        "sql-guide", "SQL patterns", "...", strategy="rag", rules="Use when writing SQL"
    ),
    "nbo-analysis": Skill(
        "nbo-analysis", "Offer revenue analysis", "Quantify every claim.", tags=["offers"]
    ),
}

doc = {
    "agents": {
        "sql_analyst": {
            "system_prompt": "You analyse data.",
            "skills": ["efficiency-discipline"],
            "mcp": ["sql", "skills"],
        }
    }
}
compose_skills(doc, SKILLS, active_tags={"offers"})

prompt = doc["agents"]["sql_analyst"]["system_prompt"]
assert "skills" not in doc["agents"]["sql_analyst"]
assert "### efficiency-discipline" in prompt and "### nbo-analysis" in prompt
assert "- **sql-guide**: Use when writing SQL" in prompt
```

The wording of the two blocks matters. Inlined rules are framed as overriding
the agent's own instructions, so a house rule beats a workflow author's
shortcut. The catalog tells the agent to check *before* starting, to load
*before* doing the work, and to reload a skill that long-run compression has
pushed out of its context — without that last line, agents carry on from a
skill they can no longer see.

## The load tool

Agents load `catalog` and `rag` skills with one tool. Ours is a skills service
behind the MCP gateway, so every agent reaches it as `skills___load_skill` by
holding the `skills` client. A large reference skill answers a bare call with
its table of contents and asks for a query; a call with a query gets the
matching sections from a vector and keyword search.

A project without a skills service can start with a local tool over the same
catalogue:

```python
from strands import tool

LIBRARY = {
    "sql-guide": "## Joins\nPrefer explicit JOIN ... ON.\n\n## Windows\nUse ROW_NUMBER() for top-N.",
}


@tool
def load_skill(name: str, query: str | None = None) -> str:
    """Load a skill's instructions by name.

    Args:
        name: The skill to load, as listed under available skills.
        query: For a long reference skill, the topic to load sections about.
    """
    content = LIBRARY.get(name)
    if content is None:
        return f"No skill named {name!r}. Pass one of: {', '.join(sorted(LIBRARY))}."
    if query is None:
        return content
    sections = [s for s in content.split("\n\n") if query.lower() in s.lower()]
    return "\n\n".join(sections) or f"No section of {name!r} mentions {query!r}."


assert "ROW_NUMBER" in load_skill("sql-guide", "windows")
```

## Prove it

Feed the composer's output to a `Pipeline` with a scripted model, and assert on
the system prompt the model received
([Testing](../testing.md#asserting-what-the-model-was-sent)): the full skill is
inlined, the catalog lists the rest, and nothing named under `skills:` reached
the library.

# docs/recipes/structured-outputs.md

# Structured outputs a UI can render

Some results are better as a card than a paragraph: the dashboard an agent
built, an order's status, a list of findings with their sources. Give that
agent an `output_schema` and its result arrives as typed data alongside the
chat, with the schema's name telling the UI how to draw it.

## Declare the schema

```python
from pydantic import BaseModel, Field


class DashboardSnippet(BaseModel):
    """Dashboard produced or updated by this agent, presented as a live card."""

    dashboard_id: str | None = Field(
        default=None,
        description="ID of the dashboard created or updated in this run. Null if none was produced.",
    )
    title: str | None = Field(default=None, description="Dashboard title.")
    summary: str | None = Field(
        default=None, description="One sentence describing what the dashboard shows."
    )


assert DashboardSnippet().dashboard_id is None
```

```yaml
agents:
  visualizer:
    output_schema: ./schemas.py:DashboardSnippet
    stream: {title: "Creating visualizations"}
```

The model reads the class docstring and every field description, so write them
for the model: say what goes in each field and what to do when there is
nothing to report. Optional fields with a `None` default let an agent that
produced nothing say so, instead of inventing an ID to satisfy the schema.

## What the client receives

The agent's activity group carries two extra fields once it finishes:

| Field | Value |
|-------|-------|
| `structuredOutput` | The filled schema, as JSON |
| `outputSchemaName` | The class name, here `DashboardSnippet` |

Register one renderer per schema name in the UI; a group with no renderer for
its schema falls back to plain text. Keep structured output for the agents
whose result *is* a card — usually a specialist the orchestrator delegates to —
and let the orchestrator answer the user in prose.

## Test it

A scripted model produces the structured output by calling a tool named after
the schema class:

```python
import asyncio
import tempfile
import textwrap
from pathlib import Path

from kaboo_workflows.testing import Pipeline

project = Path(tempfile.mkdtemp())
(project / "schemas.py").write_text(
    textwrap.dedent(
        '''
        from pydantic import BaseModel, Field

        class OrderCard(BaseModel):
            """An order's status, shown as a card."""

            order_id: str | None = Field(default=None, description="The order's number.")
            status: str | None = Field(default=None, description="Where the order is now.")
        '''
    )
)
(project / "config.yaml").write_text(
    textwrap.dedent(
        """
        x-model: &scripted kaboo_workflows.testing:ScriptedModel
        agents:
          assistant:
            model:
              provider: *scripted
              model_id: scripted
              params:
                script: [{tool: tracker, input: "order 42"}]
                final_text: "Order 42 has shipped."
          tracker:
            output_schema: ./schemas.py:OrderCard
            model:
              provider: *scripted
              model_id: scripted
              params:
                script: [{tool: OrderCard, input: {order_id: "42", status: "shipped"}}]
        orchestrations:
          team:
            mode: delegate
            entry_name: assistant
            connections:
              - {agent: tracker, description: "Tracks orders"}
        entry: team
        """
    )
)


async def main() -> None:
    pipe = Pipeline(str(project / "config.yaml"))
    result = await pipe.turn("Where is order 42?")
    pipe.close()

    card = result.group("assistant.tracker")
    assert card["outputSchemaName"] == "OrderCard"
    assert card["structuredOutput"] == {"order_id": "42", "status": "shipped"}


asyncio.run(main())
```

## Watch for

- **Hooks that end a turn early** must respect the schema: an agent with an
  `output_schema` cannot answer in free text. See the structured variant of
  [`StopOnExpiredSignIn`](guard-hooks.md#stop-when-the-sign-in-expires).
- **Relative paths** in `output_schema` resolve from the config file's
  directory, like tool paths; in a workflow submitted per run, which has no
  file, they resolve from the service's working directory — prefer a module
  path (`my_service.schemas:DashboardSnippet`) there.

# docs/recipes/tools-with-run-context.md

# Tools that need run context

A tool is a plain function, but some need to know where they are running:
which conversation, which request, which user. The library binds that context
per run, and these functions read it from inside any tool or hook:

| Function | Returns | Use it for |
|----------|---------|------------|
| `get_thread_id()` | The conversation's id | State that should outlive one request, such as values an agent stores for later turns |
| `get_run_id()` | This request's id; a resume is a new request | Logs, metrics, per-request scratch space |
| `get_turn_id()` | The logical turn's id, the same across an interrupt and its resume | Keys for side effects that must happen once per turn |
| `get_forwarded_props()` | What the client forwarded with the run | Client settings such as the user's time zone |
| `get_auth_context()` | The run's `Principal`, or `None` | The caller's identity and token claims |

All of them return `None` (or an empty dict) outside a run, so a tool still
works when you call it directly in a unit test.

## Share values between agents in one run

Agents in a multi-agent run often need to hand each other something larger
than a delegation message: the findings one analyst produced, the table
another selected. A small run-scoped store does it, keyed by thread and run:

```python
import json
from collections import OrderedDict
from threading import Lock

from strands import tool

from kaboo_workflows import get_run_id, get_thread_id

_MAX_RUNS = 256
_stores: OrderedDict[str, dict[str, str]] = OrderedDict()
_lock = Lock()


def _store() -> dict[str, str]:
    key = f"{get_thread_id()}:{get_run_id()}"
    with _lock:
        store = _stores.setdefault(key, {})
        _stores.move_to_end(key)
        while len(_stores) > _MAX_RUNS:
            _stores.popitem(last=False)
        return store


@tool
def set_workflow_variable(name: str, value: str) -> dict:
    """Store a named value for other agents in this run to read.

    Args:
        name: The variable's name, such as "sql_findings".
        value: The value; serialize structured data as JSON.
    """
    _store()[name] = value
    return {"status": "ok", "name": name}


@tool
def get_workflow_variable(name: str) -> dict:
    """Read a value an earlier agent in this run stored.

    Args:
        name: The variable's name.
    """
    store = _store()
    if name not in store:
        return {"status": "not_found", "name": name, "available": sorted(store)}
    return {"status": "ok", "name": name, "value": store[name]}


set_workflow_variable("selected_table", json.dumps({"schema": "sales", "table": "orders"}))
assert get_workflow_variable("selected_table")["status"] == "ok"
```

Three details carry the weight:

- **The store is bounded.** One process serves many runs; an unbounded
  module-level dict is a slow memory leak.
- **A miss lists what is available.** The model recovers from a wrong name in
  one step instead of guessing.
- **Values are strings.** Asking the model for JSON keeps the tool schema
  simple and the contents inspectable.

A process-local store suits a single process per session, as on AgentCore
Runtime. Behind a load balancer, keep the same tool interface over Redis or a
database.

## Do a side effect once per turn

A run that pauses for approval and resumes is two requests, so the run id
changes between them. Key anything that must not happen twice on the turn id:

```python
from strands import tool

from kaboo_workflows import get_thread_id, get_turn_id

_sent: set[tuple[str | None, str | None, str]] = set()


@tool
def send_summary(recipient: str) -> str:
    """Email the conversation's summary to a recipient.

    Args:
        recipient: The email address to send to.
    """
    key = (get_thread_id(), get_turn_id(), recipient)
    if key in _sent:
        return f"Already sent to {recipient} this turn."
    _sent.add(key)
    return f"Sent to {recipient}."


assert send_summary("ops@example.com") == "Sent to ops@example.com."
assert send_summary("ops@example.com").startswith("Already sent")
```

## Read what the client forwarded, and who is asking

```python
from datetime import datetime
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from strands import tool

from kaboo_workflows import get_auth_context, get_forwarded_props


@tool
def today() -> str:
    """Return today's date in the user's time zone."""
    zone = get_forwarded_props().get("timezone")
    try:
        tz = ZoneInfo(zone) if isinstance(zone, str) and zone else None
    except (ZoneInfoNotFoundError, ValueError):
        tz = None
    return datetime.now(tz).date().isoformat()


@tool
def whoami() -> str:
    """Return the signed-in user's name."""
    principal = get_auth_context()
    return principal.claims.get("name", "unknown") if principal else "anonymous"


assert whoami() == "anonymous"
```

Treat forwarded props as user input: they come from the client, so validate
them, as `today` does with the time zone. Identity comes only from
`get_auth_context()`, which the server's verifier set
([Call MCP tools as the signed-in user](user-identity-mcp.md)).

## Wire them in

Reference tools from the config by file or module:

```yaml
agents:
  analyst:
    tools:
      - ./tools/context_kv.py:set_workflow_variable
      - ./tools/context_kv.py:get_workflow_variable
```

Relative paths resolve from the config file's directory.

# docs/recipes/user-identity-mcp.md

# Call MCP tools as the signed-in user

When an agent's tools read a database or change a dashboard, they should do so
with the permissions of the person who asked — not a service account that can
see everything. The chain has three links: verify the caller when the run
arrives, bind them to the run, and relay their token on every MCP call.

```text
browser ──token──► host ──forwardedProps.runToken──► agent service
                                                       │ auth= verifier → Principal
                                                       ▼
                                   MCP client (auth: relay) ──token──► gateway ──► service
```

## Verify the caller when the run arrives

`create_agui_app(auth=...)` takes an async function that receives each
`/invocations` request and returns a `Principal` — or raises
`HTTPException(401)` to refuse it. Ours reads the token the host stamps into
`forwardedProps.runToken` and verifies it against the identity provider's
JWKS:

```{.python notest}
import os

import jwt
from fastapi import HTTPException, Request

from kaboo_workflows import Principal

JWKS = jwt.PyJWKClient(os.environ["OIDC_JWKS_URI"], cache_keys=True)


async def verify_run_token(request: Request) -> Principal | None:
    body = await request.json()
    token = (body.get("forwardedProps") or {}).get("runToken")
    if not token:
        raise HTTPException(status_code=401, detail="Missing forwardedProps.runToken")
    try:
        key = JWKS.get_signing_key_from_jwt(token)
        claims = jwt.decode(
            token,
            key.key,
            algorithms=["RS256"],
            issuer=os.environ.get("OIDC_ISSUER"),
            options={"verify_aud": False},
        )
    except jwt.PyJWTError as exc:
        raise HTTPException(status_code=401, detail=f"Invalid token: {exc}") from exc
    return Principal(token=token, claims=claims)
```

The token travels in the body rather than a header because the host — a
CopilotKit runtime, a backend — is what calls the agent service, and it may
run work on the user's behalf with a token it obtained itself. Audience checks
belong at the services that finally receive the relayed token; which audience
a token carries depends on the gateway in between.

## Relay the token on every MCP call

`auth: relay` sends the bound principal's token with every request the MCP
client makes ([Chapter 9](../configuration/Chapter_09.md#outbound-auth-the-auth-field)).
Behind a gateway with one target per service, give each target its own client
with an exact tool allow-list, and anchor the auth block once:

```yaml
mcp_clients:
  dashboard:
    url: ${GATEWAY_URL}
    transport: streamable-http
    auth: &relay_auth
      type: relay
      params:
        header: [Authorization, x-run-token]
        context_headers:
          x-run-id: run_id
          x-timezone: forwarded_props.timezone
    params:
      tool_filters:
        allowed: [dashboard___get_dashboard, dashboard___update_widgets]
    tool_labels:
      dashboard___update_widgets: Updating the dashboard
    transport_options:
      terminate_on_close: false
      timeout: {read: 840}

  sql:
    url: ${GATEWAY_URL}
    transport: streamable-http
    auth: *relay_auth
    params:
      tool_filters:
        allowed: [sql___run_query, sql___get_schema]
```

- **Two headers** because a managed AgentCore Gateway authenticates the caller
  on `Authorization` and then replaces it with its own credential for the
  target. The second, allowlisted header is the one the service reads.
- **`context_headers`** tell the services which run a call belongs to, and
  pass along anything the client forwarded, such as the user's time zone.
- **`tool_filters.allowed`** names are exact matches, not globs. Narrow clients
  (a read-only `sql_explore` beside a full `sql`) let you grant an agent only
  what its job needs.
- **`terminate_on_close: false`** and a long read timeout suit gateways that
  hold long tool calls open.

## Only relay on clients built per run

A relayed token is per-request, so the client carrying it must be too. With
`session_config_key` set, `create_agui_app` builds a run's MCP clients for that
run and closes them when it ends — the token one run relays can never reach
another user's run. A process serving one fixed config shares its clients;
relay is only safe there when the process serves a single user, as an AgentCore
Runtime microVM does per session (`examples/agentcore/` shows both shapes).
`m2m` and `static` strategies carry no per-user identity and are safe on a
shared client anywhere.

## When there is no principal, or it expires

- A relay client with no principal in context omits the header, so the target
  answers 401. That is the first thing to check when a tool unexpectedly fails
  authorization.
- A long run can outlive the user's token. Stop it with a clear message instead
  of letting every tool fail —
  [`StopOnExpiredSignIn`](guard-hooks.md#stop-when-the-sign-in-expires).
- Background jobs that run with no user should use the service's own identity:
  have the host obtain a client-credentials token and send it the same way, so
  the agent service needs no second code path.

# docs/testing.md

# Testing your workflows

`kaboo_workflows.testing` lets you test a workflow end to end with no model
provider, no network and no server. Two pieces do the work:

- **`ScriptedModel`** stands in for the LLM. Each agent gets a script of
  steps — call this tool, hand off to that agent, answer with this text — and
  replays it the same way on every run.
- **`Pipeline`** drives a config through the same AG-UI path
  `create_agui_app` serves, in-process, and returns what a client would have
  seen as a `TurnResult`: the AG-UI events, the reply text, the state to carry
  into the next turn, and the activity groups a chat UI renders.

Because the real strands event loop runs, a scripted test exercises your
tools, hooks, prompts, orchestration wiring, interrupts and history exactly
as production does. Only the model's choices are fixed.

Use scripted tests for behaviour and wiring, and [evaluations](evaluation.md)
for answer quality against a real model. The two complement each other.

## A first test

```python
import asyncio

from kaboo_workflows.testing import Pipeline

CONFIG = """
agents:
  assistant:
    system_prompt: You answer questions about orders.
    model:
      provider: kaboo_workflows.testing:ScriptedModel
      model_id: scripted
      params:
        final_text: "Your order shipped yesterday."
entry: assistant
"""


async def main() -> None:
    pipe = Pipeline(CONFIG)
    result = await pipe.turn("Where is my order?")
    pipe.close()

    assert result.finished()
    assert result.text() == "Your order shipped yesterday."


asyncio.run(main())
```

`Pipeline` takes what `load` takes: a file path, a raw YAML string, or a list
of either. In a pytest suite with `pytest-asyncio`, write the body as an
`async def test_…` function instead of calling `asyncio.run`. Call
`pipe.close()` when the config starts MCP servers or clients;
`run_pipeline(config, message)` builds a pipeline, drives one turn and closes
it for you.

## Writing a script

A script is a list of steps, one per model turn. The step index resets when a
new user message arrives and advances on every model call within the turn, so
a turn that calls two tools and then answers is three steps. When the script
runs out, the model replies with `final_text`.

| Step | What the model does |
|------|---------------------|
| `{text: "..."}` | Answers with this text and ends the turn |
| `{tool: lookup, input: "42"}` | Calls a tool. A string input fills the tool's first parameter; a mapping is passed as the full input |
| `{tools: [{tool: a, input: ...}, {tool: b, input: ...}]}` | Calls several tools in one message, as a model making parallel calls does |
| `{handoff: reviewer, message: "..."}` | Hands off to another agent in a swarm |
| `{error: "..."}` | Raises mid-turn, to test how a failure surfaces |

Tool names match the tools strands exposes that turn, exactly first and then
by substring, so a prefixed MCP tool such as `sql___run_query` matches
`{tool: run_query}`. A delegate connection appears as a tool named after the
connected agent, so `{tool: researcher}` delegates to `researcher`.

Give every agent its own scripted model. A model shared by several agents
shares one script and one step counter, and their turns interleave.

The scripted model is as strict as Bedrock about conversation shape: a turn
whose tool results do not pair one-to-one with the previous message's tool
calls raises the same `ValidationException` Bedrock would. A malformed
transcript, such as a replayed answer to an interrupted call, fails your test
instead of failing silently in production.

## Testing your own tools

Test the workflow you ship, not a copy of it. Keep the models in their own
file and the workflow — agents, prompts, tools, orchestration — in another,
referring to models by name. Production loads the real models file; tests load
a scripted one under the same names
([multi-file configs](configuration/Chapter_13.md) merge them):

```text
my_service/
  models.yaml          # production models
  workflow.yaml        # agents, prompts, tools, orchestration
  tools.py
tests/
  models.yaml          # the same model names, scripted
  test_assistant.py
```

```yaml
# my_service/workflow.yaml — relative paths resolve from this file
agents:
  assistant:
    model: assistant-model
    system_prompt: You answer questions about orders.
    tools: [./tools.py:lookup_order]
entry: assistant
```

```yaml
# my_service/models.yaml
models:
  assistant-model:
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-5
```

```yaml
# tests/models.yaml
models:
  assistant-model:
    provider: kaboo_workflows.testing:ScriptedModel
    model_id: scripted
    params:
      script:
        - {tool: lookup_order, input: "42"}
      final_text: "Order 42 has shipped."
```

```{.python notest}
# tests/test_assistant.py
from pathlib import Path

from ag_ui.core import EventType

from kaboo_workflows.testing import Pipeline

ROOT = Path(__file__).parents[1]


async def test_assistant_looks_up_the_order() -> None:
    pipe = Pipeline([str(ROOT / "tests/models.yaml"), str(ROOT / "my_service/workflow.yaml")])
    result = await pipe.turn("Where is order 42?")
    pipe.close()

    tool_results = result.of_type(EventType.TOOL_CALL_RESULT)
    assert "shipped" in tool_results[0].content
    assert result.text() == "Order 42 has shipped."
```

The script decides *that* the tool is called and with what input; the tool
itself runs for real. Assert on its result event, or on whatever side effect it
has, such as a row written to a test database. Name a model per agent in
`workflow.yaml` (`coordinator-model`, `researcher-model`) so each can carry its
own script.

## Asserting what the model was sent

Declare the scripted model under top-level `models:` and the pipeline keeps the
instance at `pipe.infra.models[name]`. Its `calls` hold every conversation it
was sent and `system_prompts` every system prompt, which is how you test that a
prompt was composed the way you meant — for example, that a skill was inlined.

```python
import asyncio

from kaboo_workflows.testing import Pipeline

CONFIG = """
models:
  scripted:
    provider: kaboo_workflows.testing:ScriptedModel
    model_id: scripted
    params: {final_text: "Revenue rose 4%."}
agents:
  analyst:
    model: scripted
    system_prompt: |
      <rules>Cite the table every figure comes from.</rules>
      You analyse revenue.
entry: analyst
"""


async def main() -> None:
    pipe = Pipeline(CONFIG)
    await pipe.turn("How did revenue do?")
    pipe.close()

    model = pipe.infra.models["scripted"]
    assert "<rules>Cite the table" in model.system_prompts[0]
    assert model.calls[0][-1]["content"][0]["text"] == "How did revenue do?"


asyncio.run(main())
```

## Multi-agent runs and activity

`TurnResult.groups` is the activity tree a chat UI renders: one group per
agent stream, nested under the group that called it. The entry agent's own
stream is the chat reply, so it has no group.

```python
import asyncio

from kaboo_workflows.testing import run_pipeline

CONFIG = """
x-model: &scripted kaboo_workflows.testing:ScriptedModel
agents:
  coordinator:
    model:
      provider: *scripted
      model_id: scripted
      params:
        script: [{tool: researcher, input: "find last quarter's revenue"}]
        final_text: "Revenue was 42M."
  researcher:
    stream: {title: "Researching"}
    model: {provider: *scripted, model_id: scripted, params: {final_text: "42M."}}
orchestrations:
  team:
    mode: delegate
    entry_name: coordinator
    connections:
      - {agent: researcher, description: "Finds figures"}
entry: team
"""


async def main() -> None:
    result = await run_pipeline(CONFIG, "What was revenue last quarter?")

    assert result.text() == "Revenue was 42M."
    assert result.agent_names() == {"researcher"}
    assert result.group_ids() == ["coordinator.researcher"]


asyncio.run(main())
```

`children_of(group_id)` walks the tree for nested delegation, and
`group(group_id)` returns one group as the UI holds it.

## Interrupts and resume

A run that pauses on `ask_user` or a gated tool finishes with an `interrupt`
outcome. To resume it, send the next turn the way a client does: the previous
turn's `state()` and `messages()`, plus the answer in `resume`.

```python
import asyncio

from kaboo_workflows.testing import Pipeline

CONFIG = """
agents:
  assistant:
    interrupt: true
    model:
      provider: kaboo_workflows.testing:ScriptedModel
      model_id: scripted
      params:
        script:
          - {tool: ask_user, input: {question: "Delete the draft?", options: ["yes", "no"]}}
        final_text: "Deleted."
entry: assistant
"""


async def main() -> None:
    pipe = Pipeline(CONFIG)
    first = await pipe.turn("Clean up my drafts", thread_id="c1", run_id="r1")
    outcome = first.run_outcome()
    assert outcome.type == "interrupt"

    second = await pipe.turn(
        thread_id="c1",
        run_id="r2",
        state=first.state(),
        messages=first.messages(),
        resume=[
            {
                "interruptId": outcome.interrupts[0].id,
                "status": "resolved",
                "payload": {"answer": "yes"},
            }
        ],
    )
    pipe.close()

    assert second.finished()
    assert second.text() == "Deleted."


asyncio.run(main())
```

To prove a paused run survives a restart, resume on a **new** `Pipeline`
built from the same config. It shares nothing with the first, so the only
thing carrying the pause across is the state you pass — the same guarantee a
deployed service gives. Each new pipeline builds fresh scripted models, so
assert on the resumed tool result rather than on a scripted reply.

## Hooks and guards

Hooks run in a pipeline exactly as in production, so you test one by declaring
it in the config and asserting on what the run did. This scripts a coordinator
into calling its helper twice and checks the built-in tool-call guard refused
the second call:

```python
import asyncio

from ag_ui.core import EventType

from kaboo_workflows.testing import Pipeline

CONFIG = """
x-model: &scripted kaboo_workflows.testing:ScriptedModel
agents:
  coordinator:
    model:
      provider: *scripted
      model_id: scripted
      params:
        script:
          - {tool: helper, input: "first"}
          - {tool: helper, input: "second"}
    hooks:
      - type: kaboo_workflows.hooks:MaxToolCallsGuard
        params: {max_calls: 1}
  helper:
    model: {provider: *scripted, model_id: scripted, params: {final_text: "Done."}}
orchestrations:
  team:
    mode: delegate
    entry_name: coordinator
    connections:
      - {agent: helper, description: "Helps"}
entry: team
"""


async def main() -> None:
    pipe = Pipeline(CONFIG)
    result = await pipe.turn("Get help twice")
    pipe.close()

    results = result.of_type(EventType.TOOL_CALL_RESULT)
    assert any("tool call limit" in r.content for r in results)


asyncio.run(main())
```

Your own hooks work the same way: reference them by import spec
(`type: my_service.hooks:RefuseRepeatedFailures`) and script the model into
the situation the hook exists for. The
[guard hooks recipe](recipes/guard-hooks.md) shows the hooks worth having.

## Runtime-composed workflows

A service that serves a different workflow on every run
([Chapter 17](configuration/Chapter_17.md#when-the-config-itself-is-per-request))
is tested the same way. Pass the base config and `session_config_key`, then
submit each run's workflow in `forwarded_props`:

```python
import asyncio

from kaboo_workflows.testing import Pipeline

BASE = """
models:
  analyst-model:
    provider: kaboo_workflows.testing:ScriptedModel
    model_id: scripted
    params: {final_text: "Revenue rose 4%."}
agents:
  fallback:
    model: analyst-model
    system_prompt: You are a fallback assistant.
entry: fallback
"""

SUBMITTED = """
agents:
  analyst:
    model: analyst-model
    system_prompt: You analyse revenue.
entry: analyst
"""


async def main() -> None:
    pipe = Pipeline(BASE, session_config_key="workflow_config")
    result = await pipe.turn(
        "How did revenue do?", forwarded_props={"workflow_config": SUBMITTED}
    )
    pipe.close()

    assert result.text() == "Revenue rose 4%."
    assert pipe.infra.models["analyst-model"].system_prompts[0] == "You analyse revenue."


asyncio.run(main())
```

Feed it the exact YAML your composer produces, and the test covers the
composition too.

## Evaluations with real models

Scripted tests prove the workflow does what it is told. Whether a real model
tells it the right things is a quality question: run a golden dataset through
`kaboo-workflows eval` or `run_eval` from pytest and gate CI on the scorers
([Evaluation](evaluation.md#offline-evals)). Keep evaluations few and
deliberate — each item is a real model call.

# docs/troubleshooting.md

# Troubleshooting

Fixes for the errors you're most likely to hit first. If something here doesn't
cover your case, check the [Configuration reference](configuration/README.md) or
open an issue.

## Config won't load / validation errors

kaboo-workflows validates the whole config up front and raises a precise error
pointing at the offending key. Common causes:

- **Unknown agent/model reference** — an `agents.*.model` or `entry` names
  something that isn't defined. Names must match a key under `models:` /
  `agents:` exactly.
- **Missing `entry`** — every config needs an `entry` naming the root agent or
  orchestration.
- **Bad tool path** — `tools: [./tools/foo.py]` paths are resolved relative to
  the config file; a missing file or a file with no `@tool`-decorated function
  raises at load time.

Wrap your entrypoint in `cli_errors()` (from `kaboo_workflows`) to get a
formatted, human-readable message instead of a raw traceback.

## Provider / credentials errors

- **`${OPENROUTER_API_KEY}` is empty** — `${VAR}` interpolates from the
  environment. Export it before running (`OPENROUTER_API_KEY=sk-... kaboo-serve
  config.yaml`). Interpolation happens *before* the model is constructed, so a
  missing var usually surfaces as an auth error from the provider.
- **`ModuleNotFoundError` for a provider** — install the matching extra:
  `pip install "kaboo-workflows[openai]"` (also `ollama`, `gemini`,
  `agentcore-memory`).

## MCP lifecycle

- **Server not ready / client can't connect** — the lifecycle starts all
  servers before any client. If you call `load()` yourself, remember it starts
  the lifecycle for you; you are responsible for stopping it.
- **Hanging process on exit** — always stop the lifecycle in a `finally`:

```{.python notest}
from kaboo_workflows import load

resolved = load("config.yaml")
try:
    resolved.entry("hello")
finally:
    resolved.mcp_lifecycle.stop()
```

See the [MCP guide](configuration/Chapter_09.md) for details.

## Serving / CORS

- **Frontend can't reach the server** — `kaboo-serve` listens on
  `http://localhost:8080` (`POST /invocations`, `GET /ping`). Point your
  CopilotKit `runtimeUrl` (or a dev proxy) at it.
- **CORS blocked in the browser** — serve the frontend through the same origin
  or a dev proxy rather than hitting `:8080` cross-origin. The
  [kaboo-workflows-demo](https://github.com/gl-pgege/kaboo-workflows-demo) shows
  a working proxy setup.

## Nothing streams back

A malformed client transcript (a dangling or duplicate tool result) makes the
model reject the history — an event-less hang. The adapter detects imbalance and
fails fast with a `MALFORMED_HISTORY` error rather than hanging silently; check
that each assistant tool call has exactly one matching tool result.

# docs/workflows/deep-nesting.md

# Deep nested delegation (3+ levels)

A delegate **connection** can target another orchestration, not just a plain
agent. Nesting orchestrations this way gives you arbitrary depth from a single
top-level entry.

```
research_pipeline (coordinator)
  └─▶ research_team (lead)
        └─▶ field_team (field_researcher ─▶ verifier)
```

## Config

Each level is a delegate whose connection is the orchestration below it.
kaboo-workflows topologically sorts them, building `field_team` first.

```yaml
orchestrations:
  field_team:
    mode: delegate
    entry_name: field_researcher
    connections:
      - { agent: verifier, description: "Double-check the collected data" }
  research_team:
    mode: delegate
    entry_name: lead
    connections:
      - { agent: field_team, description: "Collect and verify field data" }
  research_pipeline:
    mode: delegate
    entry_name: coordinator
    connections:
      - { agent: research_team, description: "Research the topic end to end" }

entry: research_pipeline
```

Full runnable project: [`examples/15_deep_nesting`](https://github.com/gl-pgege/kaboo-workflows/tree/main/examples/15_deep_nesting).

## Run

```bash
uv run python examples/15_deep_nesting/main.py
```

## What you'll observe

- The top entry (`coordinator`) runs *as* the orchestration and owns the reply,
  wrapping the whole chain.
- The deepest leaf (`verifier`) surfaces in the activity tree nested transitively
  under `research_team` — so a user can drill from the coordinator down to the
  verifier.

## Proven by

`tests/e2e/test_complex.py::test_delegate_three_levels_of_nesting` drives the same
shape with deterministic scripted models and asserts the transitive nesting and
the coordinator's wrapping reply.

# docs/workflows/errors-and-rejection.md

# Errors & rejection

Two "unhappy path" behaviours, both handled safely.

## 1. Error isolation

If a node raises mid-run, the failure is isolated to that node: earlier nodes
stay `completed`, the failing node's activity card is marked `error`, and the run
**terminates** rather than hanging.

```
review_swarm (swarm): planner ─▶ researcher ─▶ writer
```

```yaml
orchestrations:
  review_swarm:
    mode: swarm
    agents: [planner, researcher, writer]
    entry_name: planner

entry: review_swarm
```

Full runnable project: [`examples/20_error_and_rejection`](https://github.com/gl-pgege/kaboo-workflows/tree/main/examples/20_error_and_rejection).

### Proven by

`tests/e2e/test_cross_cutting.py::test_swarm_node_error_is_isolated_and_run_terminates`
— the earlier node stays completed, the failing node is marked `error`, and the
run terminates (never stalls).

## 2. Rejection

Rejection uses the same gate mechanism as [human-in-the-loop](human-in-the-loop.md).
When the client resumes an interrupt with `{"status": "cancelled"}` (or simply
does not address it), the gated tool receives a cancellation sentinel ("The user
declined to answer.") and the agent proceeds **without fabricating an approval**.
kaboo never silently approves an unaddressed interrupt — the safe default is to
cancel.

```json
// resume payload declining the gate
[{ "interruptId": "<id>", "status": "cancelled" }]
```

### Proven by

`tests/e2e/test_complex.py::test_rejected_interrupt_finishes_cleanly` — a declined
gate resumes to a clean `RUN_FINISHED` with no error and no hang.

# docs/workflows/human-in-the-loop.md

# Human-in-the-loop

Set `interrupt: true` on an agent to grant it the built-in `ask_user` tool. When
the agent calls it, the run pauses and the question surfaces to the client; on
resume the same call returns the user's answer.

```
review_team (delegate):
  coordinator ─▶ field_agent (interrupt: true, asks the user)
```

## Config

```yaml
agents:
  field_agent:
    interrupt: true          # grants the built-in ask_user tool
    system_prompt: |
      Before finalizing anything risky, call ask_user to confirm. If several
      details are unclear, ask them together in one ask_user call.

orchestrations:
  review_team:
    mode: delegate
    entry_name: coordinator
    connections:
      - { agent: field_agent, description: "Gated field work with user confirmation" }

entry: review_team
```

Full runnable project: [`examples/18_hitl`](https://github.com/gl-pgege/kaboo-workflows/tree/main/examples/18_hitl).

## Key behaviours

- **Interrupts bubble up.** A gate fired by a nested sub-agent surfaces to the
  single top-level handler and pauses the entire run — no matter how deep it is.
- **Parallel gated tool calls.** An agent can raise several gates in one turn
  (a multi-question form, or two tool calls). Each becomes its own interrupt with
  a distinct tool-call id, and the run resumes once all are answered.
- **Positions.** HITL works for a plain entry agent, a delegate sub-agent, and
  swarm/graph nodes alike.

## Run

HITL needs a client with a resume UI — serve it and connect a CopilotKit
frontend:

```bash
OPENROUTER_API_KEY=... uv run kaboo-serve examples/18_hitl/config.yaml
```

## Tool gates and expiry

Beyond `ask_user`, `interrupt.tools` lists tool names that pause for approval
*before executing*. `interrupt.ttl_seconds` optionally stamps an `expiresAt`
ISO-8601 timestamp onto every gate interrupt (AG-UI `Interrupt.expiresAt`), so
clients can render a countdown and servers can expire unanswered approvals —
kaboo itself never auto-expires; enforcement stays with the caller.

```yaml
agents:
  field_agent:
    interrupt:
      tools: [transition_work_item, delete_work_item]
      ask_user: false
      ttl_seconds: 86400   # each gate carries expiresAt = now + 24h
```

## Resume protocol

On pause, the run finishes with an `interrupt` outcome listing the pending
interrupts (each with an `id`). The client resumes by sending, per interrupt,
either `{"status": "resolved", "payload": ...}` or `{"status": "cancelled"}`
(see [errors & rejection](errors-and-rejection.md)).

For tool gates, a resolved payload of `{"status": "approved", "tool_input":
{...}}` executes the gated call with the user's edited arguments instead of
the agent's — approving a subset of a bulk operation is one resume, not a
round-trip through the model. A payload of `{"status": "cancelled"}` rejects
the call just like a cancelled entry.

## A gate outlives the process that opened it

An approval is the one pause that can last hours, which makes it the pause most
likely to be interrupted by a deploy. The gate itself lives in the agent object's
interrupt state, so a restart between the question and the answer used to strand
it: the user clicked approve and got `No agent session found for resume`.

Served through `create_agui_app`, the pending gate now travels on the AG-UI state
channel under `kaboo_session` and is restored onto whichever agent runs the
resume — including one that has never seen the conversation. A resume therefore
works after a restart, on a second replica, and when the session is rebuilt per
run. Only a resume restores it: a new message supersedes the gate, on a fresh
process as on a warm one, so a host that replays an older snapshot cannot bring
back a gate already answered. Nothing to configure; see
[Chapter 7](../configuration/Chapter_07.md#pending-interrupts-survive-a-restart-without-a-store)
for the switch that turns it off and why you would.

The exception is a Swarm or Graph *entry*: strands does not yet persist state for
orchestration node agents, so a multi-agent entry still needs the process to stay
up between question and answer. A plain agent, with or without delegates, is
covered.

## A pause nobody answers

Not every interrupt is a question. When a host caps the size of one HTTP
response, `create_agui_app(response_budget=...)` uses the same machinery to
split a large turn across two: the run stops at the next tool call, the
response closes properly, and the client resumes at once. The interrupt is
marked `reason: "continuation"` so a client can tell it apart and resolve it
without showing anyone a prompt. Everything above still applies — the tool runs
once, the state travels on `kaboo_session`, a different replica can pick it up
— and the turn id is unchanged, so the user sees one turn. See
[concepts](../concepts.md#runs-larger-than-one-response).

## Proven by

- `tests/e2e/test_cross_cutting.py::test_a_full_response_pauses_the_run_and_the_next_one_finishes_it`
  (the budget pauses an ungated tool call and the resume completes it, once).
- `tests/e2e/test_cross_cutting.py::test_ask_user_interrupt_then_resume`
  (plain / delegate / swarm / graph positions).
- `tests/e2e/test_complex.py::test_parallel_interrupts_surface_together_and_resume`
  (two gates in one step, distinct ids, both resumed).
- `tests/e2e/test_cross_cutting.py::test_approval_survives_a_restart_of_the_service`
  (the resuming turn runs on a fresh process that never saw the question).
- `tests/e2e/test_runtime_configs.py::test_approval_survives_the_session_being_rebuilt`
  (the same, within one process, when every run builds its own session).

# docs/workflows/multi-turn-history.md

# Multi-turn history

A delegate sub-agent starts fresh every turn by default. Set `history: true` and
its transcript is seeded from `state.kaboo_history` on the way in and captured
back into the outgoing snapshot on the way out — so it can remember across turns.

```
assistant_team (delegate):
  coordinator ─▶ memo (history: true)
```

## Config

```yaml
agents:
  memo:
    history: true
    system_prompt: "Use the notes from previous turns when answering follow-ups."

orchestrations:
  assistant_team:
    mode: delegate
    entry_name: coordinator
    connections:
      - { agent: memo, description: "Records and recalls notes across turns" }

entry: assistant_team
```

Full runnable project: [`examples/19_multiturn_history`](https://github.com/gl-pgege/kaboo-workflows/tree/main/examples/19_multiturn_history).

## A delegate orchestration remembers as its entry agent

A delegate orchestration runs a fork of its `entry_name` agent under the
orchestration's own name. That fork reads and writes the transcript its entry
agent's `history:` names, so `analyze_sql` below shares `sql_analyst`'s `sql`
group, exactly as `sql_analyst` would if it were connected directly:

```yaml
agents:
  sql_analyst:
    history: {group: sql}

orchestrations:
  analyze_sql:
    mode: delegate
    entry_name: sql_analyst
    connections:
      - { agent: query_runner, description: "Runs a SQL query" }
  team:
    mode: delegate
    entry_name: coordinator
    connections:
      - { agent: analyze_sql, description: "Answers questions with SQL" }
```

A fork whose entry agent has history off gets nothing, and the entry
orchestration never does: its transcript is the chat.

## Keeping a masked memory

A remembering agent replays everything it ever did on every later turn,
including tool results its reports have long since summarised.
`keep: masked` writes back a lighter copy:

```yaml
agents:
  sql_analyst:
    history: {group: sql, keep: masked}
```

Each finished call keeps its brief, its text, its tool calls and its final
report; it loses its reasoning and the `structuredContent` copy strands keeps of
an MCP tool's result, and each tool result becomes
`Result cleared (<tool>[, <id>=<value>…]): call it again to see it.` A result
survives when it is shorter than the agent's `context.clear_tool_results.min_chars`
(1500 without one), when its tool is in that block's `keep_tools`, or when it is
a delegate's report.

- **Only the written-back copy is masked.** Within the run the agent keeps its
  whole transcript, so a second call in the same run still sees everything.
- **A call paused on an interrupt is written back whole.** It has not finished;
  the resume needs every result.
- **A memory stored in full is masked when the agent loads it**, so switching an
  agent to `masked` takes effect on its existing memory.

`keep: full`, the default, writes back exactly what was captured.

## A summary backstop

Masking still grows with every call. `summarize_over` caps it:

```yaml
models:
  cheap: {provider: bedrock, model_id: us.anthropic.claude-haiku-4-5-20251001-v1:0}

agents:
  sql_analyst:
    history:
      group: sql
      keep: masked
      summarize_over: 20000       # estimated tokens, four characters each
      summarize_model: cheap
```

When the masked memory still exceeds `summarize_over`, `summarize_model` condenses
its oldest calls into one leading pair: a user message, `Summary of your earlier
work in this conversation`, and the summary. The newest calls that fit three
quarters of the budget stay as they were, and so does a paused call. The
summariser is told to keep IDs, decisions, exact figures and open issues.

Anything a report marks to keep goes in verbatim. A block of the final report
that opens with a `To remember` heading or line runs to the next heading of the
same level (a plain line runs to the next heading of any level) and is copied
into the summary as written, through every later summary too:

```markdown
## Findings
Revenue for 2024 was $42.1M.

## To remember
- Revenue is stored in cents in `sales.amount`.
- Query `8f3c` is the one the finance team signed off.
```

The summary is written while the run goes on. The agent returns to its caller
at once, with the masked memory in place, and the run waits for the summary
only when it writes the history back: up to 20 seconds, after which it keeps
the masked memory and cancels the summary. A summary written from an older
copy, because the agent was called again in the same run, is cancelled too.

If summarising fails, or the summariser takes longer than 60 seconds, the
masked memory is written back as it is and a warning is logged.
`summarize_over` requires `keep: masked`, and `summarize_model` must name a
model under `models:`; both are checked when the config loads.

## Loaded memory is cached on Bedrock

When a remembering agent on a Bedrock model is seeded with memory, the copy its
model is sent ends the last loaded message with a `cachePoint`, so every later
model turn of the call reads the memory from Bedrock's prompt cache. If turning
foreign calls into notes leaves the cache point opening a message, it moves to
the end of the message before, since Bedrock rejects a cache point with nothing
before it. A model with its own `cache_config` is left alone, since strands then
places its cache points itself. Cache points are never written back.

`cache_config` caches the whole conversation instead: strands puts the cache
point at the end of the latest user message on every request, so each turn of
a tool loop reads everything before it from the cache. Set it as a mapping on a
Bedrock model, beside `cache_prompt` and `cache_tools`, for three cache points
of the four Bedrock allows:

```yaml
models:
  sonnet:
    provider: bedrock
    model_id: us.anthropic.claude-sonnet-4-5-20250929-v1:0
    params:
      cache_prompt: default
      cache_tools: default
      cache_config: {strategy: auto}
```

A key other than `strategy` and `ttl` fails when the config loads.

The run's usage in the activity stream (`usageByRun`) carries `cacheHitRate`,
the share of the prompt read from the cache, for any run that read or wrote it:
`cacheReadInputTokens / (inputTokens + cacheReadInputTokens + cacheWriteInputTokens)`.
Bedrock counts the three separately, so `inputTokens` alone is only the
uncached part.

## What the model is sent is not what is stored

A transcript can hold calls to tools the agent about to run does not have — a
shared group's other agents, or an earlier version of the workflow. The model is
sent those as short notes, and an agent with `context.clear_tool_results` is sent
its older tool results cleared; see
[Chapter 8](../configuration/Chapter_08.md#what-the-model-is-sent-context). Both
change only what the model is sent. The chat the client stores and the history
written back keep the original calls.

## The client owns storage

kaboo enriches the data structure; it does **not** choose your datastore. The
captured transcript comes back in the run's outgoing history snapshot, and the
client feeds it back in `state.kaboo_history` on the next turn. For turnkey
persistence (in-memory or Postgres `ThreadStore`s behind a CopilotKit runtime),
use [kaboo-runtime](https://github.com/gl-pgege/kaboo-runtime).

## The state channel carries more than history

`kaboo_history` has a sibling. A pending human-in-the-loop gate travels the same
way, under `kaboo_session`, for the same reason: the run is stateless, so anything
the conversation accumulates has to arrive with the turn. That is what lets an
approval survive a restart — see
[human-in-the-loop](human-in-the-loop.md#a-gate-outlives-the-process-that-opened-it).

Together, the two keys are the reason a server can rebuild its agents per run and
still be indistinguishable from one that kept them: a session holds behaviour,
the state channel holds memory.

## Run

```bash
OPENROUTER_API_KEY=... uv run kaboo-serve examples/19_multiturn_history/config.yaml
```

## Proven by

- `tests/e2e/test_cross_cutting.py::test_subagent_history_round_trips_and_accumulates`
  (turn 2 is seeded from turn 1 and grows).
- `tests/e2e/test_cross_cutting.py::test_history_disabled_sub_agent_does_not_persist`
  (the stateless-by-default contrast).
- `tests/e2e/test_cross_cutting.py::test_a_forked_delegate_entry_remembers_through_its_entry_agents_group`
  (a fork remembers through its entry agent's group).
- `tests/e2e/test_cross_cutting.py::test_a_masked_sub_agent_writes_back_a_paused_call_whole_and_a_finished_one_masked`
  and `tests/runtime/test_masked_history.py` (masking on write-back and on load).
- `tests/e2e/test_cross_cutting.py::test_a_memory_over_summarize_over_is_written_back_with_its_oldest_calls_summarised`
  and `tests/runtime/test_memory_summary.py` (the summary keeps ids and `To remember` blocks).
- `tests/e2e/test_model_input.py` and `tests/runtime/test_memory_cache_point.py`
  (notes, the cache point, and what the client keeps).

# docs/workflows/parallel.md

# Parallel (top-level + nested)

A graph fans out to parallel branches by giving one source node multiple outgoing
edges. Because a graph node can itself be an orchestration, you get nested
parallelism for free.

```
pipeline (graph):
  strategist ─▶ [ writer , subteam ] ─▶ editor
  subteam (graph): sub_lead ─▶ [ analyst_a , analyst_b ] ─▶ sub_end
```

## Config

```yaml
orchestrations:
  subteam:
    mode: graph
    entry_name: sub_lead
    chat_output: sub_end
    edges:
      - { from: sub_lead, to: analyst_a }
      - { from: sub_lead, to: analyst_b }
      - { from: analyst_a, to: sub_end }
      - { from: analyst_b, to: sub_end }
  pipeline:
    mode: graph
    entry_name: strategist
    chat_output: editor
    edges:
      - { from: strategist, to: writer }
      - { from: strategist, to: subteam }
      - { from: writer, to: editor }
      - { from: subteam, to: editor }

entry: pipeline
```

Full runnable project: [`examples/17_parallel`](https://github.com/gl-pgege/kaboo-workflows/tree/main/examples/17_parallel).

## Run

```bash
uv run python examples/17_parallel/main.py
```

## What you'll observe

- `writer` and `subteam` run in parallel after `strategist`.
- Inside `subteam`, `analyst_a` and `analyst_b` run in parallel before `sub_end`.
- `editor` only runs once both top-level branches finish, and owns the reply.

## Proven by

- `tests/e2e/test_complex.py::test_parallel_top_and_nested_batches_all_run`
  (all leaves run; editor merges).
- `tests/e2e/test_compositions.py::test_graph_parallel_batch_all_run`
  (top-level parallel batch).

# docs/workflows/swarm-and-graph.md

# Swarm + graph combined

A graph's parallel batch can mix orchestration **kinds**. Here one branch is an
autonomous peer swarm and the other is a delegating coordinator — both run in
parallel, then a merge node combines them.

```
pipeline (graph):
  strategist ─▶ [ research_team (swarm) , audit_team (delegate) ] ─▶ editor
```

## Config

```yaml
orchestrations:
  research_team:
    mode: swarm
    agents: [researcher_a, researcher_b]
    entry_name: researcher_a
  audit_team:
    mode: delegate
    entry_name: auditor
    connections:
      - { agent: checker, description: "Focused compliance/accuracy check" }
  pipeline:
    mode: graph
    entry_name: strategist
    chat_output: editor
    edges:
      - { from: strategist, to: research_team }
      - { from: strategist, to: audit_team }
      - { from: research_team, to: editor }
      - { from: audit_team, to: editor }

entry: pipeline
```

Full runnable project: [`examples/16_swarm_in_graph`](https://github.com/gl-pgege/kaboo-workflows/tree/main/examples/16_swarm_in_graph).

## Run

```bash
uv run python examples/16_swarm_in_graph/main.py
```

## What you'll observe

- Swarm members (`researcher_a`, `researcher_b`) nest under the swarm node.
- The delegate's `checker` nests under the audit sub-tree via its tool call.
- Only the `editor` (the `chat_output`) carries the chat reply.

## Proven by

`tests/e2e/test_compositions.py::test_kitchen_sink_mixes_swarm_and_delegate_in_one_graph`.
