Building agents with the Claude Agent SDK

 your prompt
     │
     ▼
 ┌─────────────────────────────┐
 │ Claude thinks               │◄──────────────────────────────┐
 │ "I need to read utils.py"   │                               │
 └──────────────┬──────────────┘                               │
                │ tool call(s)                                 │ tool results
                ▼                                              │
 ┌─────────────────────────────┐                               │
 │ SDK runs the tool(s)        │───────────────────────────────┘
 └─────────────────────────────┘
          ... repeat (each repeat is one TURN) ...
                │
                ▼ no more tool calls
 ┌─────────────────────────────┐
 │ final text answer           │
 │ + ResultMessage (cost, id)  │
 └─────────────────────────────┘

# Python, with uv (recommended)
uv init && uv add claude-agent-sdk

# Python, with pip
python3 -m venv .venv && source .venv/bin/activate
pip install claude-agent-sdk

# TypeScript
npm init -y && npm pkg set type=module
npm install @anthropic-ai/claude-agent-sdk zod
npm install --save-dev tsx

export ANTHROPIC_API_KEY=sk-ant-...

sample-project/
├── CLAUDE.md                 # project rules the agent will read: "use type hints", "be brief"
├── README.md
├── .claude/skills/security-check/SKILL.md   # a reusable "skill" (part 2)
├── src/app.py                # a hardcoded API key and an SQL injection bug
├── src/utils.py              # two functions that crash on edge cases
└── tests/test_utils.py       # 5 tests; 3 fail until utils.py is fixed

# src/utils.py, as shipped
def calculate_average(numbers):
    total = 0
    for num in numbers:
        total += num
    return total / len(numbers)        # crashes on an empty list


def get_user_name(user):
    return user["name"].upper()        # crashes if user is None or has no "name"

import asyncio
from claude_agent_sdk import ClaudeAgentOptions, query
from demo_util import PROJECT, show          # (1) tiny helpers: the project path, and a message printer

async def main():                             # (2) the SDK is async, so everything lives in an async function
    options = ClaudeAgentOptions(
        cwd=PROJECT,                          # (3) the folder the agent works in
        allowed_tools=["Read", "Glob", "Grep"],   # (4) these tools run without asking for permission
        max_turns=5,                          # (5) a safety cap: at most 5 tool round-trips
    )
    async for message in query(               # (6) start the loop; it yields messages as they happen
        prompt="What does this project do? Answer in two sentences.",
        options=options,
    ):
        show(message)                         # (7) print each message in a compact form

asyncio.run(main())                           # (8) run the async function

[system:init] session=de3a42e9-... model=claude-sonnet-5 cli=2.1.281 mode=default
              tools=['Task', 'Bash', 'Edit', 'Glob', 'Grep', 'NotebookEdit', 'Read', 'Skill', 'ToolSearch', 'WebFetch', 'WebSearch', 'Write', ...]
[assistant:tool_use] Bash({'command': 'ls -la ".../demo/project"', 'description': 'List project directory contents'})
[user:tool_result] total 24\ndrwxr-xr-x 4 root root 4096 Oct 16 06:10 .\n...CLAUDE....
[assistant:tool_use] Read({'file_path': '.../demo/project/README.md'})
[user:tool_result] 1	# Sample project\n2	\n3	A tiny Python project used to demonstrate the Claude Agent SDK...
[assistant:tool_use] Bash({'command': 'ls -la ".../demo/project/src"', 'description': 'List src directory'})
[user:tool_result] total 16\n...app.py\n...utils.py
[assistant:text] This project is a tiny Python 3.13 demo for the Claude Agent SDK, with `src/utils.py` providing
                 helper functions and `src/app.py` implementing a toy user lookup. The lookup intentionally
                 contains a known SQL injection bug for demonstration purposes.
[result] subtype=success turns=4 duration=9773ms cost=$0.1260 session=de3a42e9-...

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "What does this project do? Answer in two sentences.",
  options: { cwd: PROJECT, allowedTools: ["Read", "Glob", "Grep"], maxTurns: 5 },
})) {
  if (message.type === "assistant") {                 // string check instead of isinstance
    for (const block of message.message.content) {    // one level deeper than Python
      if (block.type === "text") console.log(block.text);
      else if (block.type === "tool_use") console.log(`tool: ${block.name}`);
    }
  } else if (message.type === "result" && message.subtype === "success") {
    console.log(message.result, message.total_cost_usd);
  }
}

async for message in query(
    prompt="Review src/utils.py for bugs that would cause crashes. Fix any issues you find.",
    options=ClaudeAgentOptions(
        cwd=PROJECT,
        allowed_tools=["Read", "Edit", "Glob"],  # (1) Edit is now pre-approved
        permission_mode="acceptEdits",           # (2) and file edits need no confirmation
        max_turns=10,
    ),
):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if hasattr(block, "text"):                       # (3) a TextBlock: Claude explaining itself
                print(block.text)
            elif isinstance(block, ToolUseBlock):            # (4) a tool call: show which file it touches
                print(f"Tool: {block.name} -> {block.input.get('file_path', '')}")
    elif isinstance(message, ResultMessage):
        print(f"Done: {message.subtype} (cost ${message.total_cost_usd:.4f})")

Tool: Read -> .../project/src/utils.py
Tool: Edit -> .../project/src/utils.py
Fixed two crash bugs: `calculate_average` divided by zero on an empty list, and `get_user_name`
raised `KeyError` for a missing `"name"` key — both now guard against those cases, with type hints
added per project convention.
Done: success (cost $0.0550)

--- diff against the original file ---
-def calculate_average(numbers):
+def calculate_average(numbers: list[float]) -> float:
+    if not numbers:
+        return 0.0
     total = 0
     ...
-def get_user_name(user):
-    return user["name"].upper()
+def get_user_name(user: dict) -> str:
+    return user.get("name", "").upper()

options = ClaudeAgentOptions(
    cwd="/path/to/repo",
    model="claude-opus-5-5",
    fallback_model="claude-sonnet-5-5",
    effort="high",
    allowed_tools=["Read", "Glob", "Grep"],
    max_turns=8,
    max_budget_usd=2.0,
    env={"ANTHROPIC_BASE_URL": "https://gateway.example.com", "API_TIMEOUT_MS": "120000"},
)

const options = {
  cwd: "/path/to/repo",
  model: "claude-opus-5-5",
  fallbackModel: "claude-sonnet-5-5,claude-haiku-5-5",
  effort: "high",
  allowedTools: ["Read", "Glob", "Grep"],
  maxTurns: 8,
  maxBudgetUsd: 2,
  env: { ...process.env, ANTHROPIC_BASE_URL: "https://gateway.example.com" },  // spread process.env!
};

async with ClaudeSDKClient(options=options) as client:          # (1) opens the connection; closes it at the end
    await client.query("Which function in src/app.py has a SQL injection bug? Name it only.")
    async for message in client.receive_response():            # (2) read until this turn's ResultMessage
        print_response(message)

    await client.query("Show me the one-line fix for that function, as a code snippet only.")  # (3) same session
    async for message in client.receive_response():
        print_response(message)

Claude: `find_user`
[done: success, turns=3, cost=$0.0525]

Claude: ```python
    rows = conn.execute("SELECT * FROM users WHERE name = ?", (name,)).fetchall()

🔍 Turn one took three tool round trips (find and read the file). Turn two took **one** and used no tools: "that function" was resolved from memory, and the file was already in context. The client holds the session; you did nothing.

| Method | Use |
|---|---|
| `query(prompt)` | send a message (a string, or an async generator of messages) |
| `receive_response()` | iterate messages until the next `ResultMessage` |
| `receive_messages()` | iterate forever (for UIs) |
| `interrupt()` | stop the current turn |
| `set_permission_mode(mode)`, `set_model(model)` | change settings mid-session |
| `get_context_usage()`, `get_mcp_status()`, `get_server_info()` | inspect (part 3) |
| `rewind_files(uuid)` | undo file edits (part 3) |

### Streaming input: images and queued messages

An **async generator** of message dicts is the "stream". You can `await` between items (user input, a file arriving) and attach images.

```python
async def message_generator():
    yield {"type": "user", "message": {"role": "user", "content": "Analyze this codebase for security issues"}}
    await asyncio.sleep(2)                                  # wait for anything you like
    with open("diagram.png", "rb") as f:
        image_data = base64.b64encode(f.read()).decode()
    yield {"type": "user", "message": {"role": "user", "content": [
        {"type": "text", "text": "Review this architecture diagram"},
        {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": image_data}},
    ]}}

async with ClaudeSDKClient(options) as client:
    await client.query(message_generator())
    async for message in client.receive_response():          # stops at the FIRST result; loop again for the second
        ...

await client.query("Count from 1 to 60, one number per second, using `sleep 1` in bash between numbers.")
await asyncio.sleep(8)
await client.interrupt()                                    # (1) stop it
async for message in client.receive_response():             # (2) ALWAYS drain the interrupted turn first
    if isinstance(message, ResultMessage):
        print(message.subtype, message.terminal_reason)
await client.query("Just say hello instead.")               # (3) now continue

interrupt() sent after 8.0s
  tool call: Bash({'command': 'for i in $(seq 1 60); do echo "$i"; sleep 1; do)
interrupted turn: subtype=error_during_execution terminal_reason='aborted_tools' after 8.0s
next turn: success -> 'Hello!'

>>> /compact
[compact_boundary] trigger=manual pre_tokens=26993
>>> /clear
[conversation reset] new_conversation_id=4ac7c754-...
>>> What did we talk about before? Answer in one sentence.
Claude: This session just started, so there's no prior conversation to refer back to.

try:
    async for message in query(prompt=..., options=ClaudeAgentOptions(cwd=PROJECT, allowed_tools=["Read", "Glob"], **opts)):
        if isinstance(message, ResultMessage):
            print(f"ResultMessage: subtype={message.subtype} terminal_reason={message.terminal_reason!r}")
except CLINotFoundError as e:       # (1) no claude binary at all
    print(f"CLINotFoundError: {e}")
except ResultError as e:            # (2) the run ended with an error result. Catch BEFORE ProcessError
    print(f"ResultError: subtype={e.subtype} terminal_reason={e.terminal_reason!r}: {e}")
except ProcessError as e:           # (3) the subprocess died without a result
    print(f"ProcessError: exit_code={e.exit_code}")

=== max_turns=1 ===
ResultMessage: subtype=error_max_turns is_error=True turns=2 cost=$0.0367 terminal_reason='max_turns'
  result field: None
ResultError raised after the result: subtype=error_max_turns terminal_reason='max_turns'
  message: Claude Code returned an error result: Reached maximum number of turns (1) (exit code: 1)

=== max_budget_usd=0.02 ===
ResultMessage: subtype=error_max_budget_usd is_error=True turns=1 cost=$0.0369 terminal_reason='budget_exhausted'
  result field: None
ResultError raised after the result: subtype=error_max_budget_usd terminal_reason='budget_exhausted'

=== bad cli_path ===
CLINotFoundError: Claude Code not found at: /nonexistent/claude

async for message in query(prompt="Read README.md and summarize it in one sentence.", options=options):
    if isinstance(message, StreamEvent):
        event = message.event
        if event.get("type") == "content_block_start" and event["content_block"].get("type") == "tool_use":
            sys.stdout.write(f"\n[Using {event['content_block']['name']}...]"); in_tool = True       # (1) a tool call begins
        elif event.get("type") == "content_block_delta":
            delta = event["delta"]
            if delta.get("type") == "text_delta" and not in_tool:
                sys.stdout.write(delta["text"]); sys.stdout.flush()                                  # (2) print words as they arrive
            elif delta.get("type") == "input_json_delta":
                sys.stdout.write(".")                                                                # (3) a dot per chunk of tool input
        elif event.get("type") == "content_block_stop" and in_tool:
            sys.stdout.write(" done\n"); in_tool = False

[Using Read...].............................. done
This is a tiny demo Python project for the Claude Agent SDK, containing helper functions and a toy app with a known SQL injection bug.

--- Complete (success) ---
Stream event counts: {'message_start': 2, 'content_block_start': 3, 'content_block_delta': 43, 'content_block_stop': 3, 'message_delta': 2, 'message_stop': 2}

class Finding(BaseModel):
    file: str; line: int; severity: str; description: str

class SecurityReport(BaseModel):
    findings: list[Finding]; total_count: int; overall_risk: str

async for message in query(
    prompt="Audit every file under src/ for security problems and report them.",
    options=ClaudeAgentOptions(cwd=PROJECT, allowed_tools=["Read", "Glob", "Grep"], max_turns=8,
        output_format={"type": "json_schema", "schema": SecurityReport.model_json_schema()}),   # (1) the schema
):
    if isinstance(message, ResultMessage) and message.subtype == "success" and message.structured_output:  # (2) check BOTH
        report = SecurityReport.model_validate(message.structured_output)                                   # (3) a typed object

{
  "findings": [
    {"file": "src/app.py", "line": 5, "severity": "high",
     "description": "Hardcoded live-looking API key committed directly in source code instead of loaded from environment/secret store."},
    {"file": "src/app.py", "line": 11, "severity": "high",
     "description": "SQL injection: user-controlled `name` is interpolated via f-string directly into the SQL query instead of using a parameterized query."}
  ],
  "total_count": 2,
  "overall_risk": "high"
}

INVENTORY = {"widget": 42, "gadget": 0, "sprocket": 7}

@tool(
    "get_stock",                                                   # (1) the name Claude will call
    "Look up how many units of a product are in stock",            # (2) Claude reads this to decide WHEN to use it
    {"product": Annotated[str, "Product name, lowercase"]},        # (3) simple schema: field -> type (+ description)
    annotations=ToolAnnotations(readOnlyHint=True),                # (4) "doesn't change anything" -> may run in parallel
)
async def get_stock(args):
    product = args["product"].lower().rstrip("s")
    if product not in INVENTORY:
        return {"content": [{"type": "text", "text": f"Unknown product: {product}"}], "is_error": True}   # (5) an error Claude can read
    return {"content": [{"type": "text", "text": f"{product}: {INVENTORY[product]} units"}]}             # (6) a normal result

@tool(
    "reorder", "Place a reorder for a product",
    {"type": "object",                                             # (7) full JSON Schema for enums / optional fields
     "properties": {"product": {"type": "string"},
                    "quantity": {"type": "integer", "minimum": 1, "maximum": 1000},
                    "priority": {"type": "string", "enum": ["normal", "rush"]}},
     "required": ["product", "quantity"]})                         # (8) priority is optional
async def reorder(args):
    priority = args.get("priority", "normal")
    INVENTORY[args["product"]] = INVENTORY.get(args["product"], 0) + args["quantity"]
    return {"content": [{"type": "text", "text": f"Reorder placed: {args['quantity']} x {args['product']} ({priority}). New stock: {INVENTORY[args['product']]}"}]}

inventory_server = create_sdk_mcp_server(name="inventory", version="1.0.0", tools=[get_stock, reorder])   # (9) bundle

options = ClaudeAgentOptions(
    mcp_servers={"inventory": inventory_server},   # (10) "inventory" becomes the middle of mcp__inventory__get_stock
    allowed_tools=["mcp__inventory__*"],           # (11) approve every tool on this server
    tools=[],                                      # (12) no built-in tools: this agent can ONLY use ours
)

MCP tools visible at init: ['mcp__inventory__get_stock', 'mcp__inventory__reorder']
MCP servers: [{'name': 'inventory', 'status': 'connected', 'source': 'sdk'}]
[tool call] mcp__inventory__get_stock({'product': 'widgets'})
[tool call] mcp__inventory__get_stock({'product': 'gadgets'})
[tool result] [{'type': 'text', 'text': 'widget: 42 units'}]
[tool result] [{'type': 'text', 'text': 'gadget: 0 units'}]
[tool call] mcp__inventory__get_stock({'product': 'doohickeys'})
[tool result] ERROR Unknown product: doohickey
[tool call] mcp__inventory__reorder({'product': 'gadget', 'quantity': 25, 'priority': 'rush'})
[tool result] [{'type': 'text', 'text': 'Reorder placed: 25 x gadget (rush). New stock: 25'}]

Final answer: Widgets: 42 in stock. Gadgets: were out of stock, so I placed a rush reorder for 25 (now at 25).
Doohickeys aren't a recognized product in inventory, so I couldn't check them.

Inventory after run: {'widget': 42, 'gadget': 25, 'sprocket': 7}

const convert = tool(
  "convert_temperature",
  "Convert a temperature between Celsius and Fahrenheit",
  { value: z.number().describe("The temperature value"), from: z.enum(["C", "F"]) },   // Zod shape
  async ({ value, from }) => {                                                          // typed args
    const result = from === "C" ? (value * 9) / 5 + 32 : ((value - 32) * 5) / 9;
    return { content: [{ type: "text", text: `${value}°${from} = ${result.toFixed(1)}°${from === "C" ? "F" : "C"}` }] };
  },
  { annotations: { readOnlyHint: true } },
);
const server = createSdkMcpServer({ name: "temps", version: "1.0.0", tools: [convert] });

tool_use: mcp__temps__convert_temperature({"value":100,"from":"F"})
tool_result: [{"type":"text","text":"100°F = 37.8°C"}]
tool_use: mcp__temps__convert_temperature({"value":37,"from":"C"})
tool_result: [{"type":"text","text":"37°C = 98.6°F"}]
answer: 100°F is 37.8°C, and 37°C is 98.6°F.

options = ClaudeAgentOptions(
    mcp_servers={
        # a local program, talking over stdin/stdout ("stdio")
        "filesystem": {"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/me/projects"]},
        # a remote server over HTTP, with an auth header
        "github": {"type": "http", "url": "https://api.githubcopilot.com/mcp/",
                   "headers": {"Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}"}},
        # a remote server over SSE
        "remote-api": {"type": "sse", "url": "https://api.example.com/mcp/sse"},
    },
    allowed_tools=["mcp__filesystem__read_file", "mcp__github__list_issues", "mcp__remote-api__*"],
)

tool call ─► 1 hooks ─► 2 deny rules ─► 3 ask rules ─► 4 permission mode ─► 5 allow rules ─► 6 can_use_tool
              │ deny      │ deny          │ -> callback   │ bypass: approve     │ match: approve   │ allow / deny
              ▼           ▼               ▼               │ acceptEdits: edits  ▼                  ▼
            blocked     blocked         callback          ▼ ok                run                decision

async def can_use_tool(tool_name, input_data, context):
    if tool_name == "Bash" and "rm " in input_data.get("command", ""):                 # (1) inspect the COMMAND, not just the tool
        return PermissionResultDeny(message="User policy: deletion is not allowed. Leave the file in place.")  # (2) Claude reads this
    if tool_name == "Write":
        path = input_data["file_path"]
        if "/scratch/" not in path:
            new_path = path.rsplit("/", 1)[0] + "/scratch/" + path.rsplit("/", 1)[1]
            return PermissionResultAllow(updated_input={**input_data, "file_path": new_path})   # (3) approve with a changed input
    return PermissionResultAllow(updated_input=input_data)                             # (4) approve as requested

async def noop_hook(input_data, tool_use_id, context):   # (5) required with plain query(): keeps the control channel open
    return {"continue_": True}

async def prompt_stream():                                # (6) can_use_tool needs streaming input
    yield {"type": "user", "message": {"role": "user", "content": "Create a file named notes.txt containing 'hello', then delete it with rm."}}

async for message in query(prompt=prompt_stream(), options=ClaudeAgentOptions(
        cwd=PROJECT, permission_mode="default", can_use_tool=can_use_tool,
        hooks={"PreToolUse": [HookMatcher(matcher=None, hooks=[noop_hook])]})):
    ...

[tool_result] ERROR User policy: deletion is not allowed. Leave the file in place.

[result] success: I created notes.txt with "hello", but deletion is blocked by policy, so I left the file in place instead of running `rm`.
[result] permission_denials: [{'tool_name': 'Bash', 'tool_use_id': 'toolu_012B...', 'tool_input': {'command': 'cd ".../project" && echo \'hello\' > notes.txt && cat notes.txt && rm notes.txt && ls notes.txt 2>&1', ...}}]

Decisions made by can_use_tool:
   ('deny', 'Bash', 'cd ".../project" && echo \'hello\' > notes.txt && cat notes.txt && rm notes.txt && ls notes.txt 2>&1')

{"questions": [{"question": "How should I format the output?", "header": "Format",
                "options": [{"label": "Summary", "description": "Brief overview"}, {"label": "Detailed", "description": "Full explanation"}],
                "multiSelect": false}]}

async def my_hook(input_data, tool_use_id, context) -> dict:
    # input_data: hook_event_name, session_id, cwd, and event fields (PreToolUse: tool_name, tool_input, tool_use_id)
    # tool_use_id: links PreToolUse and PostToolUse for the same call
    return {}

async def protect_env_files(input_data, tool_use_id, context):
    file_path = input_data["tool_input"].get("file_path", "")               # (1) the Write/Edit input
    if file_path.split("/")[-1] in (".env", ".env.local"):
        return {"hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "deny",                                    # (2) block it
            "permissionDecisionReason": "Policy: .env files may not be created or modified by the agent"}}
    return {}                                                                # (3) no opinion

async def audit(input_data, tool_use_id, context):
    audit_log.append((input_data["tool_name"], tool_use_id)); return {}      # (4) every successful tool call

options = ClaudeAgentOptions(
    cwd=PROJECT, permission_mode="acceptEdits", allowed_tools=["Read", "Write", "Edit", "Glob"],
    hooks={"PreToolUse": [HookMatcher(matcher="Write|Edit", hooks=[protect_env_files])],   # (5) only Write/Edit
           "PostToolUse": [HookMatcher(hooks=[audit])]})                                   # (6) every tool

[tool_result] ERROR PreToolUse:Write hook error: Policy: .env files may not be created or modified by the agent
[tool_result] File created successfully at: .../project/config.example.txt ...
Claude: I created `config.example.txt` with that line, but a security policy blocks me from creating or
modifying `.env` files — you'll need to create that one yourself.
[result] success, denials=[{'tool_name': 'Write', 'tool_use_id': 'toolu_01Bh...', 'tool_input': {'file_path': '.../project/.env', 'content': 'DATABASE_URL=sqlite:///users.db\n'}}]

Audit log (PostToolUse): [('Write', 'toolu_01FA...')]

  tool: Edit .../src/utils.py
  tool: Bash python3 -m unittest -q tests.test_utils
  [Stop hook] iteration 1: tests PASS, CHANGELOG.md missing, stop_hook_active=False     <- blocked
  tool: Bash ls -la
  tool: Write .../CHANGELOG.md
  [Stop hook] iteration 2: tests PASS, CHANGELOG.md present, stop_hook_active=True     <- allowed
[result] success after 7 turns, $0.0575

options = ClaudeAgentOptions(
    cwd=PROJECT,
    allowed_tools=["Read", "Grep", "Glob", "Agent"],          # (1) the parent needs Agent to delegate
    agents={
        "security-reviewer": AgentDefinition(
            description="Security code reviewer. Use for finding vulnerabilities in source files.",   # (2) WHEN to use it
            prompt="You are a security reviewer. Report each vulnerability as file:line and one sentence. Be brief. Never edit files.",  # (3) its system prompt
            tools=["Read", "Grep", "Glob"],                     # (4) read-only
            model="haiku",                                      # (5) cheaper model
        ),
    },
    hooks={"SubagentStart": [HookMatcher(hooks=[on_subagent_start])],
           "SubagentStop": [HookMatcher(hooks=[on_subagent_stop])]},
)
async for message in query(prompt="Use the security-reviewer agent to review src/app.py, then repeat its findings verbatim.", options=options):
    if isinstance(message, AssistantMessage):
        inside = " (inside subagent)" if message.parent_tool_use_id else ""       # (6) messages from inside carry parent_tool_use_id
        for block in message.content:
            if isinstance(block, ToolUseBlock) and block.name in ("Agent", "Task"):   # (7) the delegation call
                print(f"Subagent invoked: subagent_type={block.input.get('subagent_type')}")

  tool: Glob
Subagent invoked: subagent_type=security-reviewer model=None
  [hook SubagentStart] type=security-reviewer id=a0401eaaf13c8f1e0
  tool (inside subagent): Read
  [hook SubagentStop]  type=security-reviewer transcript=~/.claude/projects/.../subagents/agent-a0401eaaf13c8f1e0.jsonl

Final result:
 **1. SQL Injection (src/app.py:11)** ...
 **2. Hardcoded API Key (src/app.py:5)** ...
 **3. Missing Resource Cleanup on Exception (src/app.py:9-12)** ...

Per-model usage:
  claude-sonnet-5: in=6 out=599 cost=$0.0561
  claude-haiku-4-5-20251001: in=7791 out=564 cost=$0.0106

async def run(prompt, **opts):                                  # (1) one-shot helper returning (session_id, text)
    async for message in query(prompt=prompt, options=ClaudeAgentOptions(cwd=PROJECT, max_turns=4, **opts)):
        if isinstance(message, ResultMessage):
            return message.session_id, message.result

sid, _  = await run("Remember this code word: PELICAN. Reply only with 'noted'.", allowed_tools=[])
rid, t  = await run("What was the code word? One word.", resume=sid)                                  # (2) resume
fid, t  = await run("Change the code word to HERON and reply 'changed'.", resume=sid, fork_session=True) # (3) fork
_,   t  = await run("What is the code word? One word.", resume=sid)                                   # (4) original untouched?

rename_session(sid, "codeword-demo", directory=PROJECT)         # (5) on-disk housekeeping
info   = get_session_info(sid, directory=PROJECT)
recent = list_sessions(directory=PROJECT, limit=3)
msgs   = get_session_messages(sid, directory=PROJECT)

Original session: 1a2076ae-3291-4142-81df-ab350207c695 -> 'noted'
Resumed session:  1a2076ae-3291-4142-81df-ab350207c695 -> 'PELICAN'  (same id: True)
Forked session:   a0577f7e-79e3-446d-aa6b-735719e26400 -> 'changed'  (new id: True)
Original again:        -> 'PELICAN'

get_session_info: title='codeword-demo' first_prompt="Remember this code word: PELICAN. ..." cwd=.../demo/project
list_sessions (latest 3):
  1a2076ae-...  'codeword-demo'
  a0577f7e-...  "Remember this code word: PELICAN. Reply only with 'noted'."
  56b76917-...  'Reply: ok'
get_session_messages: 6 messages, roles=['user', 'assistant', 'user', 'assistant', 'user', 'assistant']

class SessionStore(Protocol):
    async def append(self, key: SessionKey, entries: list[SessionStoreEntry]) -> None: ...   # write entries, in order
    async def load(self, key: SessionKey) -> list[SessionStoreEntry] | None: ...              # read them back (None if unknown)
    # optional: list_sessions, list_session_summaries, delete, list_subkeys

store = InMemorySessionStore()                     # for tests; S3/Redis/Postgres adapters ship in the SDK repos' examples/
async for m in query(prompt="List the Python files under src/", options=ClaudeAgentOptions(session_store=store)):
    if isinstance(m, ResultMessage): session_id = m.session_id
async for m in query(prompt="Summarize what those files do", options=ClaudeAgentOptions(session_store=store, resume=session_id)):
    ...   # resumed from the store, possibly on another host

[default (minimal)]           "I'm Claude, an AI coding agent built on Anthropic's Claude Agent SDK that can help you read, write, and debug code and work with files in your project directory via a terminal-based interface."
[claude_code preset + append] "I'm Claude Code, Anthropic's CLI assistant, here to help you with software engineering tasks like coding, debugging, and running commands in your project.\n\nBANANA"
[custom string]               "Arrr, I be Claude, a scurvy code-slingin' agent here to help ye navigate files and tasks on yer machine!"

setting_sources=['project']: 'Yes. "Use type hints on all new or edited functions."'
setting_sources=[]:          "No. I don't have any project-specific instructions ... about type hints"

---
name: security-check
description: Run a security vulnerability scan over the project source
---
Analyze every file under `src/` for security vulnerabilities, including:
- SQL injection risks
- Exposed credentials
- Insecure configurations
Report each finding as `file:line - severity - one sentence`. Do not edit files.

async for message in query(
    prompt="/security-check",                                  # (1) dispatch by name
    options=ClaudeAgentOptions(cwd=PROJECT, setting_sources=["project"], skills="all",   # (2) load project skills, allow all
                               allowed_tools=["Read", "Grep", "Glob"], max_turns=8)):
    if isinstance(message, SystemMessage) and message.subtype == "init":
        print("skills loaded:", message.data.get("skills"))    # (3) confirm

skills loaded: ['security-check', 'verify', 'debug', 'code-review', 'simplify', ...]
'security-check' in slash_commands: True
  tool: Bash
  tool: Read
  tool: Read
  tool: Read
result:
 - `src/app.py:5` - High - Hardcoded live API key (`sk-live-...`) committed in source instead of loaded from environment/secrets store.
 - `src/app.py:11` - Critical - SQL query built via f-string interpolation of unsanitized `name` parameter, enabling SQL injection.

options = ClaudeAgentOptions(plugins=[{"type": "local", "path": "./my-plugin"}])   # skills appear as /my-plugin:skill-name

my-plugin/
├── .claude-plugin/plugin.json   # optional
├── skills/my-skill/SKILL.md
├── agents/specialist.md
├── hooks/hooks.json
└── .mcp.json

options = ClaudeAgentOptions(cwd=PROJECT, enable_file_checkpointing=True, permission_mode="acceptEdits",
                             extra_args={"replay-user-messages": None},    # (1) needed to receive the UUIDs
                             max_turns=4)
async with ClaudeSDKClient(options) as client:
    await client.query("Append a line '## Changelog' followed by '- demo edit' to README.md")
    async for message in client.receive_response():
        if isinstance(message, UserMessage) and message.uuid and not checkpoint_id:
            checkpoint_id = message.uuid                                  # (2) first user message = "before anything happened"
        if isinstance(message, ResultMessage):
            session_id = message.session_id

async with ClaudeSDKClient(ClaudeAgentOptions(cwd=PROJECT, enable_file_checkpointing=True, resume=session_id)) as client:
    await client.query("")                                                # (3) empty prompt just opens the connection
    async for _ in client.receive_response():
        await client.rewind_files(checkpoint_id)                          # (4) restore
        break

File changed by agent: True
Tail after edit: 'up with a known SQL injection bug\n\n## Changelog\n- demo edit\n'
Rewound to checkpoint 53cb3f96-3508-4865-91c5-454a190b25ac
File identical to original after rewind: True

if isinstance(message, AssistantMessage) and not message.parent_tool_use_id:   # (1) main loop only
    if message.message_id not in seen_ids:                                     # (2) count each API response once
        seen_ids.add(message.message_id); steps += 1
        u = message.usage or {}
        input_tokens += u.get("input_tokens", 0)
elif isinstance(message, ResultMessage):
    u = message.usage or {}                                                    # (3) totals for the main loop
    for model, mu in (message.model_usage or {}).items():                      # (4) per model, including subagents
        print(f"{model}: cost=${mu['costUSD']:.4f} contextWindow={mu['contextWindow']}")

step 1: message_id=msg_011Cg5Rkrj... input=2 cache_read=0 cache_create=24941 stop=None
step 2: message_id=msg_011Cg5Rm2X... input=2 cache_read=24941 cache_create=129 stop=None

result.usage: input=4 output=85 cache_read=24941 cache_create=25070
result.total_cost_usd: $0.1061  (client-side estimate)
result.duration_ms=3877 duration_api_ms=3655 num_turns=2
result.model_usage[claude-sonnet-5]: cost=$0.1061 contextWindow=1000000 costBasis=list

async with ClaudeSDKClient(options) as client:
    await client.query("Reply with exactly: ready")
    async for message in client.receive_response():
        if isinstance(message, SystemMessage) and message.subtype == "init":
            d = message.data            # claude_code_version, model, permissionMode, tools, slash_commands, skills, mcp_servers, plugins
    usage  = await client.get_context_usage()
    status = await client.get_mcp_status()
    await client.set_model("claude-haiku-5-5")
    await client.query("Reply with exactly: done")

init.claude_code_version: 2.1.281
init.model: claude-sonnet-5
init.tools: [..., 'mcp__claude-code-docs__query_docs_filesystem_claude_code_docs', 'mcp__claude-code-docs__search_claude_code_docs', 'mcp__claude-code-docs__submit_feedback']
init.slash_commands: ['security-check', 'verify', 'code-review', 'clear', 'compact', 'context', 'effort', 'model', 'usage', ...]
init.mcp_servers: [{'name': 'claude-code-docs', 'status': 'connected', 'source': 'dynamic'}]

get_context_usage: 25361/1000000 tokens (3.0%), model=claude-sonnet-5, autocompact=True
  categories: [('System tools', 11446), ('MCP tools (deferred)', 1721), ('System tools (deferred)', 16341), ('Memory files', 64), ('Skills', 5576), ('Messages', 8275), ('Autocompact buffer', 33000), ('Free space', 941639)]
get_mcp_status: claude-code-docs -> connected tools=3

Turn 1 model: claude-sonnet-5
Turn 2 model (after set_model): claude-haiku-5-5

received = {"traces": [], "metrics": [], "logs": []}

class Collector(BaseHTTPRequestHandler):
    """Accepts OTLP over HTTP with JSON encoding and stores the payloads."""
    def do_POST(self):
        body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        signal = self.path.rsplit("/", 1)[-1]                    # /v1/traces -> traces
        if signal in received:
            received[signal].append(json.loads(body))
        self.send_response(200); self.send_header("Content-Type", "application/json"); self.end_headers()
        self.wfile.write(b"{}")

OTEL_ENV = {
    "CLAUDE_CODE_ENABLE_TELEMETRY": "1",
    "CLAUDE_CODE_ENHANCED_TELEMETRY_BETA": "1",             # (1) traces are beta
    "OTEL_TRACES_EXPORTER": "otlp", "OTEL_METRICS_EXPORTER": "otlp", "OTEL_LOGS_EXPORTER": "otlp",
    "OTEL_EXPORTER_OTLP_PROTOCOL": "http/json",             # (2) JSON for the demo; use http/protobuf against a real collector
    "OTEL_EXPORTER_OTLP_ENDPOINT": "http://127.0.0.1:4318",
    "OTEL_SERVICE_NAME": "tutorial-agent",
    "OTEL_RESOURCE_ATTRIBUTES": "deployment.environment=demo,tenant.id=acme",
    "OTEL_METRIC_EXPORT_INTERVAL": "1000", "OTEL_LOGS_EXPORT_INTERVAL": "1000", "OTEL_TRACES_EXPORT_INTERVAL": "1000",  # (3) flush fast
    "CLAUDE_CODE_OTEL_DIAG_STDERR": "1",                    # (4) surface exporter errors on stderr
    "NO_PROXY": "127.0.0.1,localhost",                      # (5) don't send localhost traffic through a corporate proxy
}

async for message in query(
    prompt="Read README.md and reply with its first heading only.",
    options=ClaudeAgentOptions(cwd=PROJECT, allowed_tools=["Read"], max_turns=3,
                               env=OTEL_ENV, stderr=stderr_lines.append),      # Python merges env; TS needs { ...process.env, ...otelEnv }
):
    ...
await asyncio.sleep(8)                                                          # let the final export intervals fire

[result] success: '# Sample project' cost=$0.0186

received: 0 trace batches, 4 metric batches, 4 log batches

spans (name -> count): {}

metrics: ['claude_code.active_time.total', 'claude_code.cost.usage', 'claude_code.session.count', 'claude_code.token.usage']

log events (name -> count): {'managed_settings_resolved': 1, 'user_prompt': 1, 'plugin_loaded': 2, 'tool_decision': 2, 'api_request': 3, 'tool_result': 2, 'assistant_response': 1}

# ephemeral entrypoint
async for message in query(prompt=os.environ["TASK_PROMPT"], options=ClaudeAgentOptions(max_turns=20)):
    print(message)

# hybrid: hydrate from a shared store
async for message in query(prompt=user_input, options=ClaudeAgentOptions(resume=session_id, session_store=session_store)):
    ...

from claude_agent_sdk import fold_session_summary
from claude_agent_sdk.testing import run_session_store_conformance

class SqliteSessionStore:
    """Two required methods (append, load) plus the optional ones, on one SQLite file."""

    async def append(self, key, entries):
        pk, sid, sub = key["project_key"], key["session_id"], key.get("subpath") or ""
        with self.db:
            seen = {uuids already stored for this key}                            # (1) a retried batch may re-deliver entries
            seq = (max stored seq) + 1
            for e in entries:
                if e.get("uuid") in seen: continue
                self.db.execute("INSERT INTO entries VALUES (?,?,?,?,?,?)", (pk, sid, sub, seq, e.get("uuid"), json.dumps(e)))
                seq += 1
            if not sub:                                                           # (2) summary sidecar for the main transcript only
                prev = {stored summary for (pk, sid)} or None
                summary = fold_session_summary(prev, key, entries)                # (3) SDK helper: fold new entries in
                self.db.execute("INSERT OR REPLACE INTO summaries VALUES (?,?,?,?)", (pk, sid, now_ms, json.dumps(summary["data"])))

    async def load(self, key):
        rows = SELECT entry ... ORDER BY seq
        return [json.loads(r[0]) for r in rows] if rows else None                 # (4) None, not [], when unknown

    async def list_sessions(self, project_key): ...        # [{"session_id", "mtime"}], newest first
    async def list_session_summaries(self, project_key): ...
    async def delete(self, key): ...                       # main key: cascade to subkeys and the summary
    async def list_subkeys(self, key): ...

# 1. the SDK's contract tests, each on a fresh empty store
await run_session_store_conformance(lambda: SqliteSessionStore(":memory:"))

# 2. host A writes a session into the shared store (its own config dir = its own local transcripts)
store = SqliteSessionStore(db_path)
async for m in query(prompt="Remember the code word OSPREY. Reply only: noted",
                     options=ClaudeAgentOptions(cwd=PROJECT, tools=[], max_turns=2, session_store=store,
                                                env={"CLAUDE_CONFIG_DIR": host_a})):
    if isinstance(m, ResultMessage): session_id = m.session_id

# 3. host B (a different config dir, so NO local transcript) resumes from the store
async for m in query(prompt="What was the code word? One word.",
                     options=ClaudeAgentOptions(cwd=PROJECT, tools=[], max_turns=2, session_store=store,
                                                resume=session_id, env={"CLAUDE_CONFIG_DIR": host_b})):
    ...

=== conformance suite ===
conformance: PASSED

host A: session 7841b307-51d9-4f01-852f-57f0d485d90f -> 'noted'
store now holds 16 transcript entries for that session (project_key='-tmp-demo-project')
summary sidecar: {"is_sidechain": false, "created_at": 1792135295404, "cwd": "/tmp/.../demo/project", "first_prompt": "Remember the code word OSPREY. Rep
host B: resumed 7841b307-51d9-4f01-852f-57f0d485d90f -> 'OSPREY'
host B had no local transcript: []

def confine_to(root: Path):
    """PreToolUse hook: deny any file tool whose path resolves outside the tenant's working directory."""
    async def hook(input_data, tool_use_id, context):
        target = input_data["tool_input"].get("file_path") or input_data["tool_input"].get("path")
        if target and not Path(target).resolve().is_relative_to(root.resolve()):
            print(f"  [hook] DENY {input_data['tool_name']}({target})")
            return {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny",
                                           "permissionDecisionReason": f"{target} is outside the tenant workspace"}}
        return {}
    return hook

async def ask(work, config, prompt, sources, **extra):
    options = ClaudeAgentOptions(
        cwd=str(work),
        setting_sources=sources,                                      # (1) [] = no host/project settings, no CLAUDE.md
        env={"CLAUDE_CONFIG_DIR": str(config),                        # (2) per-tenant config dir: transcripts, state
             "CLAUDE_CODE_DISABLE_AUTO_MEMORY": "1"},                 # (3) auto memory would load regardless of (1)
        max_turns=3, **extra)
    async for m in query(prompt=prompt, options=options):
        if isinstance(m, ResultMessage):
            return (m.result or "").strip(), m.session_id, len(m.permission_denials)

# each tenant: work/CLAUDE.md says "The internal project codename is BLUEBIRD" (acme) / "REDWOOD" (globex)
q = "Without using any tools: what is the internal project codename? Reply with the codename only, or 'unknown'."
await ask(a_work, a_config, q, sources=[], tools=[])                  # 1. isolation
await ask(a_work, a_config, q, sources=["project"], tools=[])         # 2. opt in to the project's CLAUDE.md
probe = f"Read the file {a_work}/notes.txt and quote its contents in one line. If the read fails, say exactly why."
await ask(b_work, b_config, probe, sources=[], allowed_tools=["Read"])                            # 3. cwd only
await ask(b_work, b_config, probe, sources=[], allowed_tools=["Read"],
          hooks={"PreToolUse": [HookMatcher(matcher="Read|Edit|Write|Glob|Grep", hooks=[confine_to(b_work)])]})  # 3b. hook

=== 1. settings isolation: setting_sources=[] means CLAUDE.md is NOT injected ===
acme   -> 'unknown'
globex -> 'unknown'

=== 2. the same tenants with project settings loaded on purpose (setting_sources=['project']) ===
acme   -> 'BLUEBIRD'
globex -> 'REDWOOD'

=== 3. file isolation: does cwd alone bound what tools can reach? ===
globex, cwd only          -> I was able to read the file, but I want to flag something before quoting it: the path points to
`/tmp/ws-acme-xcw1vyki/...`, a different workspace sandbox (acme) than the one I'm running in (...globex). The file's content
is literally labeled "acme private notes" — i.e., it appears to be another tenant's private data that happens to be reachable
from this environment, not something that should cross into this session. [...]   (denials=0)

=== 3b. file isolation done right: a PreToolUse hook (or the OS sandbox) confines paths to the tenant root ===
  [hook] DENY Read(/tmp/ws-acme-xcw1vyki/work/notes.txt)
globex, with confine hook -> The read failed: the path `/tmp/ws-acme-xcw1vyki/work/notes.txt` is outside my current tenant
workspace (`/tmp/ws-globex-s6_uwjwa/work`), so the Read tool's access-control hook blocked it with the error
"is outside the tenant workspace."   (denials=1)

=== 4. transcript isolation: each tenant's sessions live under its own CLAUDE_CONFIG_DIR ===
acme: 2 transcript(s) under /tmp/ws-acme-xcw1vyki/config
globex: 4 transcript(s) under /tmp/ws-globex-s6_uwjwa/config
tenant B's config dir contains tenant A's session: False

async def one(i):
    async for m in query(prompt=f"Session {i}: read src/app.py and say in one sentence what find_user does.",
                         options=ClaudeAgentOptions(cwd=PROJECT, allowed_tools=["Read"], max_turns=3, effort="low")):
        if isinstance(m, ResultMessage):
            return i, m.subtype, round(time.time() - t0, 1), m.total_cost_usd or 0

results = await asyncio.gather(*(one(i) for i in range(N)))       # N sessions at once, one subprocess each

session 0: success in 6.5s, $0.0421
session 1: success in 7.2s, $0.0421
session 2: success in 6.3s, $0.0420

3 sessions in parallel: wall=7.2s, peak claude processes=4, peak total RSS=1074 MiB (~268 MiB per session)
planning: agents per host = (host RAM - overhead) / per-session ceiling; at ~268 MiB a 16 GiB host fits ~53 idle-sized sessions

# app.py — one worker process, many sessions; sessions are resumed from the shared store on every request
app = FastAPI()
store = SqliteSessionStore(os.environ["SESSION_DB"])              # swap for Postgres/S3 in production

@app.post("/chat/{tenant}")
async def chat(tenant: str, body: ChatIn):
    work, config = tenant_dirs(tenant)                             # separate cwd + CLAUDE_CONFIG_DIR per tenant (5.2)
    opts = ClaudeAgentOptions(
        cwd=work, setting_sources=[], tools=["Read", "Grep", "Glob"],
        env={"CLAUDE_CONFIG_DIR": config, "CLAUDE_CODE_DISABLE_AUTO_MEMORY": "1", **OTEL_ENV},   # telemetry (section 4)
        hooks={"PreToolUse": [HookMatcher(matcher="Read|Glob|Grep", hooks=[confine_to(Path(work))])]},
        session_store=store, resume=body.session_id,               # hybrid pattern (5.1)
        max_turns=20, max_budget_usd=0.50)
    async for m in query(prompt=body.message, options=opts):
        if isinstance(m, ResultMessage):
            return {"session_id": m.session_id, "reply": m.result, "cost_usd": m.total_cost_usd}

# Dockerfile (illustrative)
FROM python:3.13-slim
RUN apt-get update && apt-get install -y --no-install-recommends bubblewrap socat && rm -rf /var/lib/apt/lists/*   # sandbox deps (section 6)
RUN pip install --no-cache-dir claude-agent-sdk fastapi uvicorn
RUN useradd -m agent
USER agent
WORKDIR /srv
COPY --chown=agent app.py .
ENV CLAUDE_CODE_DISABLE_AUTO_MEMORY=1
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8080"]
# run with: --cap-drop ALL --read-only --tmpfs /tmp --pids-limit 256, and HTTPS only through your egress proxy

options = ClaudeAgentOptions(sandbox={
    "enabled": True,
    "autoAllowBashIfSandboxed": True,     # sandboxed commands need no prompt
    "failIfUnavailable": True,            # Python falls back to unsandboxed by default; TS stops by default
    "excludedCommands": ["docker *"],
    "allowUnsandboxedCommands": True,     # model may ask to leave the sandbox -> goes to can_use_tool
    "network": {"allowedDomains": ["api.github.com"], "allowLocalBinding": True},
})

stderr_lines = []
try:
    async for m in query(
        prompt="Run `echo sandbox-test` with bash and reply with its output only.",
        options=ClaudeAgentOptions(cwd=PROJECT, allowed_tools=["Bash"], max_turns=3,
                                   sandbox={"enabled": True, "autoAllowBashIfSandboxed": True, "failIfUnavailable": fail},
                                   stderr=stderr_lines.append),                      # (1) the only place sandbox warnings go
    ):
        if isinstance(m, ResultMessage):
            print(f"result: subtype={m.subtype} terminal_reason={m.terminal_reason!r} errors={m.errors}")
except ResultError as e:                                                             # (2) one-shot query() raises on an error result
    print(f"ResultError raised: {e}")

=== sandbox enabled, failIfUnavailable=True ===
ResultError raised: Claude Code returned an error result: Sandbox required but unavailable: sandbox is enabled but dependencies
are missing: bubblewrap (bwrap) not installed, socat not installed · install missing tools (e.g. apt install bubblewrap socat)
or see https://code.claude.com/docs/en/sandboxing. Set sandbox.failIfUnavailable=false to allow unsandboxed execution. (exit code: 1)
stderr lines mentioning sandbox (2):
    Error: sandbox required but unavailable: sandbox is enabled but dependencies are missing: bubblewrap (bwrap) not installed, ...
    sandbox.failIfUnavailable is set — refusing to start without a working sandbox.

=== sandbox enabled, failIfUnavailable=False (fall back unsandboxed, warn) ===
result: subtype=success terminal_reason='completed' errors=None
result text: 'sandbox-test'
stderr lines mentioning sandbox (2):
    ⚠ Sandbox disabled: sandbox is enabled but dependencies are missing: bubblewrap (bwrap) not installed, socat not installed ...
    Commands will run WITHOUT sandboxing. Network and filesystem restrictions will NOT be enforced.

# shared by p2 and p3 (examples/python/)
async def run(prompt, schema=None, **opts):
    options = ClaudeAgentOptions(cwd=PROJECT, setting_sources=[], max_turns=opts.pop("max_turns", 5),
                                 output_format={"type": "json_schema", "schema": schema} if schema else None, **opts)
    async for m in query(prompt=prompt, options=options):
        if isinstance(m, ResultMessage):
            return (m.structured_output if schema else m.result), (m.total_cost_usd or 0)

# ---------- 1. Routing: a cheap classifier picks the specialist -------------------------------
ROUTE_SCHEMA = {"type": "object", "properties": {"kind": {"type": "string", "enum": ["bug", "docs", "question"]},
                                                 "reason": {"type": "string"}}, "required": ["kind", "reason"]}
SPECIALISTS = {                                                                   # (1) one options bundle per route
    "bug":      dict(allowed_tools=["Read", "Grep", "Glob"], max_turns=8, system_prompt="You are a debugger. Locate the defect and describe the fix in two sentences."),
    "docs":     dict(allowed_tools=["Read", "Glob"],         system_prompt="You are a technical writer. Answer in two sentences."),
    "question": dict(tools=[],                               system_prompt="Answer from general knowledge in one sentence."),
}

async def routed(request):
    route, c1 = await run(f"Classify this request: {request!r}", ROUTE_SCHEMA,
                          tools=[], model="claude-haiku-5-5", effort="low")        # (2) cheap, tool-less, typed
    answer, c2 = await run(request, **SPECIALISTS[route["kind"]])                 # (3) dispatch
    print(f"[route] {request!r}\n   -> kind={route['kind']} ({route['reason']})\n   -> {answer}\n   cost=${c1 + c2:.4f}")

# ---------- 2. Prompt chaining with a programmatic gate -----------------------------------------
OUTLINE_SCHEMA = {"type": "object", "properties": {"sections": {"type": "array", "items": {"type": "string"}, "minItems": 3}},
                  "required": ["sections"]}

async def chained():
    outline, c1 = await run("Outline a README for this project in 3-5 section titles.", OUTLINE_SCHEMA, allowed_tools=["Read", "Glob"])
    assert 3 <= len(outline["sections"]) <= 5, "gate failed"                       # (4) a deterministic gate between steps
    doc, c2 = await run("Write one sentence for each of these README sections, as a markdown list:\n"
                        + json.dumps(outline["sections"]), tools=[])                 # (5) step 2 consumes step 1's output
    print(f"[chain] outline={outline['sections']}\n{doc}\n   cost=${c1 + c2:.4f}")

# ---------- 3. Parallelization: voting -----------------------------------------------------------
VOTE_SCHEMA = {"type": "object", "properties": {"safe": {"type": "boolean"}, "reason": {"type": "string"}},
               "required": ["safe", "reason"]}

async def voted():
    prompt = "Is src/app.py safe to deploy to production as-is? Answer safe=true/false with a one-sentence reason."
    results = await asyncio.gather(*(run(prompt, VOTE_SCHEMA, allowed_tools=["Read"], effort="low") for _ in range(3)))  # (6) 3 independent judges
    votes = [r[0]["safe"] for r in results]
    print(f"[vote] votes={votes} -> majority safe={sum(votes) > 1}")                 # (7) programmatic aggregation