Agent Blueprint

Agent Blueprint - Build Production-Grade AI Agents

Generate complete, working AI agents using battle-tested patterns studied from production-grade agentic systems. Every pattern is independently implemented and adapted for any framework, language, or purpose.

When to Use

Use this skill when:

• Building any kind of AI agent or agentic tool
• Creating a coding assistant, research agent, or task automation agent
• Designing an agent architecture with tools, permissions, and context management
• Scaffolding a new agent project from scratch
• Cloning or recreating Claude Code-like functionality
• Building multi-step automated workflows with planner + executor architecture
• Embedding an agent inside a web application via HTTP/SSE API
• Adding human-in-the-loop approval checkpoints to an autonomous agent
• Making an agent production-ready (observability, cost control, long tasks)

Do NOT use when:

• Building a simple chatbot without tools (use prompt engineering instead)
• Creating a non-agentic API wrapper (use your SDK directly)

The 5-Phase Build Process

Phase 1: Define Your Agent

Before writing any code, answer these 4 questions. Write answers to a AGENT.md spec file in your project:

1. PURPOSE: What does this agent do in one sentence?
2. TOOLS: What actions can it take? (file ops, shell, web, APIs, custom)
3. TRIGGERS: When should it act autonomously vs wait for input?
4. OUTPUT: What does "done" look like? (code changes, reports, files, etc.)

Based on your answers, choose a template:

For TypeScript/Bun: use references/typescript-agent-loop.md as your starting point and examples/typescript-research-agent.md as a runnable example. For Rust: use references/rust-agent.md.

Not sure which template? Use this quick selector:

Phase 2: Implement Core Components

Build these 5 components in order. Each is a separate module:

agent/
  core/
    agent_loop.py      # The main while-true loop (Phase 2.1)
    tool_registry.py   # Tool definitions and dispatch (Phase 2.2)
    permissions.py     # Allow/deny/ask permission system (Phase 2.3)
    context.py         # Message history and compaction (Phase 2.4)
    system_prompt.py   # Dynamic prompt assembly (Phase 2.5)

Phase 2.1: The Agent Loop

The heart of every agent. This is the while(true) pattern that drives all agentic behavior:

async def agent_loop(messages, tools, system_prompt, api_client):
    while True:
        # 1. Manage context window
        messages = compact_if_needed(messages)

        # 2. Call the LLM with streaming
        response = await api_client.stream(
            system=system_prompt,
            messages=messages,
            tools=tools.to_api_schema(),
        )

        # 3. Process response blocks
        assistant_content = []
        for block in response:
            if block.type == "text":
                assistant_content.append(block)
            elif block.type == "tool_use":
                # 4. Check permissions
                decision = permissions.check(block.name, block.input)
                if decision == "deny":
                    assistant_content.append(tool_error(block.id, "Permission denied"))
                    continue

                # 5. Execute tool
                result = await tools.execute(block.name, block.input)
                messages.append(tool_result(block.id, result))
                assistant_content.append(block)

        messages.append(assistant_message(assistant_content))

        # 6. Continue if tools were used, stop if text-only
        if not any(b.type == "tool_use" for b in assistant_content):
            break

    return messages

Critical patterns:

• Stream responses for real-time feedback
• Execute concurrency-safe tools in parallel
• Handle prompt_too_long errors by compacting context
• Track token usage and cost per turn

Load references/agent-loop.md for the complete loop with error recovery, streaming, and cost tracking.

Phase 2.2: The Tool System

Every tool implements this interface:

class Tool:
    name: str                    # Unique identifier (e.g., "bash", "read_file")
    description: str             # What it does (shown to LLM)
    input_schema: dict           # JSON Schema for parameters
    is_read_only: bool           # True if no side effects
    is_concurrency_safe: bool    # True if parallel execution is safe

    async def call(self, input: dict, context: dict) -> ToolResult:
        """Execute the tool and return result."""

    def check_permissions(self, input: dict) -> PermissionDecision:
        """Return allow/deny/ask for this specific invocation."""

    def validate_input(self, input: dict) -> ValidationResult:
        """Validate parameters before execution."""

Built-in tools to implement first:

Load references/tool-system.md for complete tool implementations with Zod/Pydantic schemas.

Phase 2.3: The Permission System

Defense-in-depth permission checking:

def check_permission(tool_name, tool_input, context):
    # Layer 1: Always-deny rules (hardcoded safety)
    if matches_deny_rule(tool_name, tool_input):
        return DENY

    # Layer 2: Tool-specific checks
    tool_check = tools[tool_name].check_permissions(tool_input)
    if tool_check in (ALLOW, DENY):
        return tool_check

    # Layer 3: Always-allow rules (user-configured patterns)
    if matches_allow_rule(tool_name, tool_input):
        return ALLOW

    # Layer 4: Safety checks for sensitive paths
    if is_sensitive_path(tool_input):
        return ASK

    # Layer 5: Ask the user
    return ASK

Permission rule format: "ToolName(pattern)" — e.g., "Bash(git *)", "Read", "Edit(*.ts)"

Load references/permission-system.md for the complete 7-layer permission pipeline.

Phase 2.4: Context Management

The #1 challenge in production agents. Implement these 3 strategies:

Strategy 1: Auto-Compact — When approaching context limits, summarize old messages:

if token_count(messages) > context_window - buffer:
    messages = await summarize_old_messages(messages, keep_recent=5)

Strategy 2: Tool Result Budget — Persist large outputs to disk, keep summaries in context:

if len(tool_result.content) > MAX_INLINE_SIZE:
    path = save_to_disk(tool_result.content)
    tool_result.content = f"[Result too large. Full output at: {path}]"

Strategy 3: Micro-Compact — Replace old tool results with [Old tool result cleared]:

for msg in messages[:-RECENT_WINDOW:]:
    if msg.type == "tool_result" and msg.age > TURNOVER_THRESHOLD:
        msg.content = "[Old tool result content cleared]"

Load references/context-management.md for the complete context lifecycle.

Phase 2.5: System Prompt Assembly

Build the system prompt as a string array, not a single blob:

def build_system_prompt(tools, context):
    parts = []

    # Static prefix (cacheable)
    parts.append(identity_section())        # "You are..."
    parts.append(tool_instructions(tools))  # Per-tool usage rules
    parts.append(output_rules())            # Formatting, verbosity

    # Dynamic boundary (changes per turn)
    parts.append(current_datetime())        # Timestamp
    parts.append(git_context(context))      # Branch, status
    parts.append(user_context(context))     # CLAUDE.md or project config

    return parts

Key pattern: Split static/dynamic content with a cache boundary. Static prefix gets cached by the API (10x cheaper). Dynamic suffix changes every turn.

Load references/system-prompts.md for the complete prompt template.

Phase 3: Add the Interaction Layer

Choose your UI:

Terminal UI component tree (React/Ink pattern):

App
  Messages          # Virtual scrollable message list
  Spinner            # "Thinking..." / "Running bash..." indicator
  PromptInput        # User input with history, autocomplete
  PermissionDialog   # "Allow Bash(git push)?" prompt
  CostDisplay        # Running token/cost counter

Phase 4: Add Advanced Features

These make a good agent into a great one:

Phase 5: Validate and Ship

Run the validation script to verify your agent has all critical components:

python scripts/validate_agent.py ./my-agent

This checks:

• Agent loop handles tool_use and text responses
• Tool registry with proper schemas
• Permission system with deny/allow/ask
• Context compaction for long conversations
• System prompt assembly
• Error handling for API failures
• Cost tracking

Templates

Ready-to-use starting points. Copy and customize:

Examples

Complete walkthroughs of real agent builds:

Scripts

Reference Files

Deep-dive documentation for each subsystem. Load when you need implementation details:

Example: Multi-Agent Orchestrator

An agent that spawns sub-agents for parallel research and synthesis.

Pattern

User Request
     │
     ▼
Orchestrator Agent (plans and delegates)
     │
     ├──→ Sub-Agent 1 (research React)
     ├──→ Sub-Agent 2 (research Vue)     ← Parallel execution
     └──→ Sub-Agent 3 (research Svelte)
     │
     ▼
Synthesize into unified report

Implementation

import asyncio
from agent.loop import agent_loop, spawn_sub_agent
from agent.tools import create_default_tool_registry
from agent.permissions import PermissionSystem
from agent.context import ContextManager
from agent.prompt import build_system_prompt
from agent.api import APIClient

async def orchestrate(task: str):
    api = APIClient()
    tools = create_default_tool_registry()

    # Step 1: Decompose the task
    decomposition_prompt = f"""Break this task into 3 independent sub-tasks.
    Return ONLY a JSON array of objects with "task" and "focus" fields.

    Task: {task}"""

    decomposition = await spawn_sub_agent(
        prompt=decomposition_prompt,
        tools=tools,
        api_client=api,
    )

    import json
    subtasks = json.loads(decomposition)

    # Step 2: Execute sub-tasks in parallel
    results = await asyncio.gather(*[
        spawn_sub_agent(
            prompt=f"Research and analyze: {st['task']}\nFocus on: {st['focus']}",
            tools=tools,
            api_client=api,
        )
        for st in subtasks
    ])

    # Step 3: Synthesize
    synthesis_prompt = f"""Synthesize these research results into a unified report.

    Original task: {task}

    Sub-task results:
    {chr(10).join(f'## {st["task"]}{chr(10)}{r}' for st, r in zip(subtasks, results))}

    Write a comprehensive report combining all findings."""

    report = await spawn_sub_agent(
        prompt=synthesis_prompt,
        tools=tools,
        api_client=api,
        system_prompt=["You are a report writer. Produce clear, structured reports."],
    )

    return report

if __name__ == "__main__":
    import sys
    task = " ".join(sys.argv[1:])
    report = asyncio.run(orchestrate(task))
    print(report)

When to Use This Pattern

• Research tasks with multiple independent angles
• Code reviews across multiple files/modules
• Documentation generation for multi-component systems
• Any task that benefits from parallel execution

Cost Considerations

Sub-agents each consume tokens independently. For 3 sub-agents with 20 turns each:

• Total turns: ~60
• Estimated cost: ~$0.30-1.00 (Sonnet)
• Time savings: ~3x faster than sequential

Example: Building a Full Claude Code Clone in Python

A complete walkthrough of building a production-grade coding agent.

What We're Building

A terminal-based coding assistant that:

• Reads, writes, and edits files
• Executes bash commands
• Searches code with glob and grep
• Manages context for long conversations
• Tracks token costs
• Asks for permission before destructive actions

Step 1: Project Setup

mkdir my-claude-clone && cd my-claude-clone
mkdir agent
pip install anthropic pydantic rich

Step 2: Copy Reference Implementations

Copy these files from the skill references:

Step 3: Create the API Client Wrapper

# agent/api.py
import anthropic
from typing import AsyncGenerator

class APIClient:
    def __init__(self, model: str = "claude-sonnet-4-6"):
        self.client = anthropic.AsyncAnthropic()
        self.model = model

    async def stream(self, system, messages, tools, stream=True):
        async with self.client.messages.stream(
            model=self.model,
            max_tokens=8192,
            system=system,
            messages=messages,
            tools=tools,
        ) as stream:
            async for event in stream:
                yield event

Step 4: Create the Entry Point

# main.py
import asyncio
import sys
from agent.api import APIClient
from agent.loop import agent_loop
from agent.tools import create_default_tool_registry
from agent.permissions import PermissionSystem
from agent.context import ContextManager
from agent.prompt import build_system_prompt
from agent.cost import CostTracker
from rich.console import Console

console = Console()

async def main():
    api = APIClient()
    tools = create_default_tool_registry()
    permissions = PermissionSystem()
    context = ContextManager(api_client=api)
    cost = CostTracker()
    system_prompt = build_system_prompt(tools)

    # Interactive mode
    console.print("[bold blue]My Claude Clone[/] — type your prompt (Ctrl+C to exit)")
    
    while True:
        try:
            prompt = console.input("[bold green]> [/]")
            if prompt.strip() in ("exit", "quit", "q"):
                break

            messages = [{"role": "user", "content": prompt}]
            
            async for event in agent_loop(
                messages=messages,
                tools=tools,
                permissions=permissions,
                context_manager=context,
                system_prompt=system_prompt,
                api_client=api,
                cost_tracker=cost,
            ):
                if event.type == "streaming_text" and event.text:
                    console.print(event.text, end="")
                elif event.type == "tool_result":
                    icon = "[green]OK[/]" if event.result.success else "[red]FAIL[/]"
                    console.print(f"\n[{event.tool_name}] {icon}")
                elif event.type == "done":
                    console.print()
                elif event.type == "cost":
                    console.print(
                        f"[dim]Tokens: {event.input_tokens + event.output_tokens} "
                        f"Cost: ${event.total_cost_usd:.4f}[/]",
                    )

        except KeyboardInterrupt:
            console.print("\n[bold]Goodbye![/]")
            break

if __name__ == "__main__":
    asyncio.run(main())

Step 5: Run It

export ANTHROPIC_API_KEY=sk-ant-...
python main.py

What You Get

My Claude Clone — type your prompt (Ctrl+C to exit)
> List all Python files and find any TODO comments

[Glob] OK
[Grep] OK

I found 3 Python files with 5 TODO comments:

1. **agent/loop.py:42** - `# TODO: add retry logic for rate limits`
2. **agent/tools.py:128** - `# TODO: support concurrent tool execution`
3. **agent/context.py:67** - `# TODO: implement micro-compact`

Tokens: 4231 Cost: $0.0127

Next Steps

• Add streaming output with Rich live display
• Implement interactive permission prompts
• Add /compact, /cost, /help slash commands
• Add session save/restore
• Add sub-agent support for parallel work

Example: TypeScript Research Agent

Build a streaming research agent in TypeScript using Bun and the Anthropic SDK.

What We're Building

A research assistant that:

• Searches the web for information (plug in Brave/Tavily/SerpAPI)
• Runs bash commands with async, non-blocking execution via Bun.spawn
• Streams response tokens to the terminal as they arrive
• Loops until the model stops calling tools

Setup

mkdir my-research-agent && cd my-research-agent
bun init -y
bun add @anthropic-ai/sdk

agent.ts

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

// --- Tool schemas (inline JSON — no zod-to-json-schema needed) ---

const tools: Anthropic.Tool[] = [
  {
    name: "bash",
    description: "Execute a shell command and return stdout/stderr.",
    input_schema: {
      type: "object" as const,
      properties: {
        command: {
          type: "string",
          description: "Shell command to execute",
        },
      },
      required: ["command"],
    },
  },
  {
    name: "web_search",
    description: "Search the web for current information.",
    input_schema: {
      type: "object" as const,
      properties: {
        query: {
          type: "string",
          description: "Search query",
        },
      },
      required: ["query"],
    },
  },
];

// --- Tool execution ---

async function executeTool(name: string, input: Record<string, string>): Promise<string> {
  switch (name) {
    case "bash": {
      try {
        const proc = Bun.spawn(["bash", "-c", input.command], {
          stdout: "pipe",
          stderr: "pipe",
        });
        const stdout = await new Response(proc.stdout).text();
        const stderr = await new Response(proc.stderr).text();
        await proc.exited;
        return stdout + (stderr ? `\nSTDERR: ${stderr}` : "");
      } catch (e: unknown) {
        return `Error: ${e instanceof Error ? e.message : String(e)}`;
      }
    }

    case "web_search": {
      // Replace this stub with a real search API.
      // Brave Search: set BRAVE_API_KEY and call https://api.search.brave.com/res/v1/web/search
      // Tavily:       set TAVILY_API_KEY and call https://api.tavily.com/search
      // SerpAPI:      set SERPAPI_KEY and call https://serpapi.com/search
      const apiKey = process.env.BRAVE_API_KEY;
      if (!apiKey) {
        return `[web_search stub] Query: "${input.query}" — set BRAVE_API_KEY to enable real search.`;
      }
      const url = `https://api.search.brave.com/res/v1/web/search?q=${encodeURIComponent(input.query)}&count=5`;
      const res = await fetch(url, {
        headers: { "X-Subscription-Token": apiKey, Accept: "application/json" },
      });
      const data = (await res.json()) as { web?: { results?: Array<{ title: string; url: string; description: string }> } };
      const results = data.web?.results ?? [];
      return results
        .map((r) => `[${r.title}](${r.url})\n${r.description}`)
        .join("\n\n") || "No results found.";
    }

    default:
      return `Unknown tool: ${name}`;
  }
}

// --- Agent loop ---

async function agent(prompt: string): Promise<void> {
  const messages: Anthropic.MessageParam[] = [
    { role: "user", content: prompt },
  ];

  while (true) {
    // Collect streamed content blocks
    const contentBlocks: Anthropic.ContentBlock[] = [];
    let currentText = "";
    let currentToolUse: Partial<Anthropic.ToolUseBlock> & { partial_json?: string } = {};

    const stream = client.messages.stream({
      model: "claude-sonnet-4-6",
      max_tokens: 4096,
      system: `You are a research assistant. Use web_search to find current information and bash for
local tasks. Provide well-sourced answers and cite URLs when available.`,
      tools,
      messages,
    });

    for await (const event of stream) {
      if (event.type === "content_block_start") {
        if (event.content_block.type === "text") {
          currentText = "";
        } else if (event.content_block.type === "tool_use") {
          currentToolUse = {
            type: "tool_use",
            id: event.content_block.id,
            name: event.content_block.name,
            partial_json: "",
          };
        }
      } else if (event.type === "content_block_delta") {
        if (event.delta.type === "text_delta") {
          process.stdout.write(event.delta.text);
          currentText += event.delta.text;
        } else if (event.delta.type === "input_json_delta") {
          currentToolUse.partial_json = (currentToolUse.partial_json ?? "") + event.delta.partial_json;
        }
      } else if (event.type === "content_block_stop") {
        if (currentText) {
          contentBlocks.push({ type: "text", text: currentText });
          currentText = "";
        } else if (currentToolUse.id) {
          contentBlocks.push({
            type: "tool_use",
            id: currentToolUse.id!,
            name: currentToolUse.name!,
            input: JSON.parse(currentToolUse.partial_json || "{}"),
          } as Anthropic.ToolUseBlock);
          currentToolUse = {};
        }
      }
    }

    messages.push({ role: "assistant", content: contentBlocks });

    // Collect tool calls
    const toolUses = contentBlocks.filter(
      (b): b is Anthropic.ToolUseBlock => b.type === "tool_use"
    );

    if (toolUses.length === 0) {
      process.stdout.write("\n");
      break;
    }

    // Execute tools and build result message
    const toolResults: Anthropic.ToolResultBlockParam[] = [];
    for (const toolUse of toolUses) {
      const result = await executeTool(toolUse.name, toolUse.input as Record<string, string>);
      process.stdout.write(`\n[${toolUse.name}] OK\n`);
      toolResults.push({
        type: "tool_result",
        tool_use_id: toolUse.id,
        content: result,
      });
    }

    messages.push({ role: "user", content: toolResults });
  }
}

// --- Entry point ---

const prompt = process.argv.slice(2).join(" ") || "What are the best practices for building AI agents in 2026?";
agent(prompt).catch(console.error);

Usage

# Run directly with Bun
bun run agent.ts "Research the latest developments in AI agent frameworks"

# With a real search API (Brave)
BRAVE_API_KEY=your-key bun run agent.ts "What is the current state of LLM context windows?"

Expected Output

I'll research AI agent frameworks for you.

[web_search] OK

Based on my research, here are the key developments in AI agent frameworks in 2026:

1. **LangGraph** remains dominant for stateful multi-agent systems...
2. **Anthropic's Agent SDK** introduced native streaming tool execution...

Sources:
- [LangGraph docs](https://langchain-ai.github.io/langgraph/)
- [Anthropic Blog](https://anthropic.com/research)

<div align="center"> <img src="media/banner.png" alt="agent-blueprint" width="100%"> </div>

agent-blueprint

Build production-grade AI agents from scratch. Complete patterns, templates, and reference implementations for any language or framework.

What This Is

A skill for AI coding agents (Claude Code, Cursor, etc.) that guides you through building complete agentic systems — tool registries, permission systems, context management, streaming APIs, human-in-the-loop workflows, and HTTP serving layers.

Two audiences:

• Workflow builders — planner + executor pipelines, HITL checkpoints, multi-agent orchestration
• Application embedders — HTTP/SSE APIs, multi-user session management, React frontend integration

Install

npx skills add OthmanAdi/agent-blueprint -g

Works with Claude Code, Cursor, and any agent that supports the skills protocol. Then just describe what you want to build — the skill activates automatically.

What's Inside

agent-blueprint/
  SKILL.md                    # Main skill — loaded by the agent
  references/                 # Deep-dive docs (load on demand)
    agent-loop.md             # Streaming agent loop (Python)
    typescript-agent-loop.md  # Streaming agent loop (TypeScript/Bun)
    tool-system.md            # Tool registry & execution pipeline
    permission-system.md      # 7-layer permission pipeline
    context-management.md     # Context window lifecycle
    system-prompts.md         # Prompt caching patterns
    mcp.md                    # MCP server integration
    memory.md                 # Session persistence & long-term memory
    rust-agent.md             # Rust implementation (tokio + reqwest)
    production-principles.md  # 85% compounding problem, Manus principles
    planner-executor.md       # Three-agent pattern (planner/executor/verifier)
    human-in-the-loop.md      # HITL approval matrix & tiered delegation
    serving.md                # FastAPI + SSE, multi-user sessions
    long-tasks.md             # File-based planning for 100+ step tasks
    observability.md          # Tracing, OpenTelemetry, cost dashboards
    cost-optimization.md      # Prompt caching, model routing, batch API
    framework-guide.md        # LangGraph vs Mastra vs Vercel AI SDK 2026
    architecture.md           # System diagram & dependency graph
  templates/
    minimal-agent/            # ~200 lines Python, 2 tools
    coding-agent/             # Full coding assistant, 9 tools
    research-agent/           # Web search + report generation
    task-agent/               # Multi-step orchestration
    workflow-agent/           # Planner + executor + HITL
    api-agent/                # FastAPI + SSE + React frontend
  examples/
    python-coding-agent.md    # Claude Code clone walkthrough
    typescript-research-agent.md
    multi-agent-orchestrator.md
  scripts/
    validate_agent.py         # Check your agent has all required components
    scaffold.py               # Generate project from template

Example Prompt

Build me a Python workflow agent that processes customer support tickets. Pull tickets from Postgres, classify them with an LLM into 5 categories, route urgent ones to Slack and low-priority to email. Ask for my approval before sending anything. Log every decision. Use the workflow-agent template with human-in-the-loop checkpoints.

The skill picks the right template, loads the relevant references, and generates a complete working project.

Language Support

License

Apache 2.0 — see LICENSE.

---

Disclaimer: This project is not affiliated with, endorsed by, or associated with Anthropic PBC. "Claude" and "Claude Code" are trademarks of Anthropic. No Anthropic source code is included in this repository. All implementations are original and independently written.

The Agent Loop

The complete streaming agent loop with error recovery, parallel tool execution, and cost tracking.

Complete Implementation

import asyncio
import json
from typing import AsyncGenerator

async def agent_loop(
    messages: list[dict],
    tools: ToolRegistry,
    permissions: PermissionSystem,
    context_manager: ContextManager,
    system_prompt: list[str],
    api_client: APIClient,
    max_turns: int = 50,
    cost_tracker: CostTracker | None = None,
) -> AsyncGenerator[Event, None]:
    turn = 0
    consecutive_errors = 0

    while turn < max_turns:
        turn += 1

        # --- Context Management Pipeline ---
        messages = context_manager.apply_budget(messages)
        messages = context_manager.snip_stale(messages)
        messages = context_manager.auto_compact_if_needed(
            messages, system_prompt, tools
        )

        # Check hard context limit
        if context_manager.is_over_limit(messages, system_prompt):
            yield ErrorEvent("Context limit exceeded. Use /compact to summarize.")
            break

        # --- Call LLM with Streaming ---
        try:
            stream = api_client.stream(
                system=system_prompt,
                messages=messages,
                tools=tools.to_api_schema(),
                stream=True,
            )

            # Collect response blocks
            text_blocks = []
            tool_blocks = []
            thinking_blocks = []
            usage = None

            async for event in stream:
                if event.type == "message_start":
                    usage = event.message.usage
                elif event.type == "content_block_start":
                    if event.content_block.type == "text":
                        yield StreamingTextEvent(start=True)
                    elif event.content_block.type == "tool_use":
                        yield ToolStartEvent(
                            tool_name=event.content_block.name,
                            tool_id=event.content_block.id,
                        )
                elif event.type == "content_block_delta":
                    if event.delta.type == "text_delta":
                        text_blocks.append(event.delta.text)
                        yield StreamingTextEvent(text=event.delta.text)
                    elif event.delta.type == "input_json_delta":
                        yield ToolInputEvent(partial_json=event.delta.partial_json)
                    elif event.delta.type == "thinking_delta":
                        thinking_blocks.append(event.delta.thinking)
                        yield ThinkingEvent(thinking=event.delta.thinking)
                elif event.type == "message_delta":
                    if event.delta.stop_reason:
                        stop_reason = event.delta.stop_reason
                    if event.usage:
                        usage = event.usage

            consecutive_errors = 0

        except PromptTooLongError:
            # Context too large - compact and retry
            messages = await context_manager.reactive_compact(
                messages, system_prompt
            )
            continue
        except APIError as e:
            consecutive_errors += 1
            if consecutive_errors >= 3:
                yield ErrorEvent(f"API error after 3 retries: {e}")
                break
            yield RetryEvent(error=str(e), attempt=consecutive_errors)
            await asyncio.sleep(2 ** consecutive_errors)
            continue

        # --- Track Costs ---
        if cost_tracker and usage:
            cost_tracker.add(usage)
            yield CostEvent(
                input_tokens=usage.input_tokens,
                output_tokens=usage.output_tokens,
                cache_read=usage.cache_read_input_tokens,
                cache_creation=usage.cache_creation_input_tokens,
                total_cost_usd=cost_tracker.total_cost(),
            )

        # --- Build Assistant Message ---
        content = []
        if thinking_blocks:
            content.append({"type": "thinking", "thinking": "".join(thinking_blocks)})
        if text_blocks:
            content.append({"type": "text", "text": "".join(text_blocks)})

        for tb in tool_blocks:
            content.append(tb)

        messages.append({"role": "assistant", "content": content})

        # --- No tools used = done ---
        tool_uses = [b for b in content if b.get("type") == "tool_use"]
        if not tool_uses:
            yield DoneEvent(stop_reason=stop_reason, turn_count=turn)
            break

        # --- Execute Tools ---
        # Separate into concurrent and sequential
        concurrent = []
        sequential = []
        for tu in tool_uses:
            tool = tools.get(tu["name"])
            if tool and tool.is_concurrency_safe:
                concurrent.append(tu)
            else:
                sequential.append(tu)

        # Run concurrent tools in parallel
        if concurrent:
            results = await asyncio.gather(*[
                execute_tool_with_permissions(
                    tu, tools, permissions, context_manager
                )
                for tu in concurrent
            ])
            for tu, result in zip(concurrent, results):
                yield ToolResultEvent(tool_name=tu["name"], result=result)
                messages.append({
                    "role": "user",
                    "content": [{
                        "type": "tool_result",
                        "tool_use_id": tu["id"],
                        "content": result.content if result.success else f"Error: {result.error}",
                        "is_error": not result.success,
                    }],
                })

        # Run sequential tools one at a time
        for tu in sequential:
            result = await execute_tool_with_permissions(
                tu, tools, permissions, context_manager
            )
            yield ToolResultEvent(tool_name=tu["name"], result=result)
            messages.append({
                "role": "user",
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": tu["id"],
                    "content": result.content if result.success else f"Error: {result.error}",
                    "is_error": not result.success,
                }],
            })

    else:
        yield MaxTurnsEvent(max_turns=max_turns)


async def execute_tool_with_permissions(
    tool_use: dict,
    tools: ToolRegistry,
    permissions: PermissionSystem,
    context_manager: ContextManager,
) -> ToolResult:
    tool = tools.get(tool_use["name"])
    if not tool:
        return ToolResult(success=False, error=f"Unknown tool: {tool_use['name']}")

    # Validate input
    validation = tool.validate_input(tool_use["input"])
    if not validation.valid:
        return ToolResult(success=False, error=validation.error)

    # Check permissions
    decision = permissions.check(tool.name, tool_use["input"])
    if decision == "deny":
        return ToolResult(success=False, error="Permission denied")

    # Execute
    try:
        result = await tool.call(tool_use["input"], context={})

        # Persist large results
        if len(str(result.content)) > context_manager.max_inline_size:
            path = context_manager.persist_result(result.content)
            result.content = f"[Result too large. Full output saved to: {path}]"

        return result
    except Exception as e:
        return ToolResult(success=False, error=str(e))

Event Types

from dataclasses import dataclass

@dataclass
class Event:
    type: str

@dataclass
class StreamingTextEvent(Event):
    type: str = "streaming_text"
    text: str = ""
    start: bool = False

@dataclass
class ToolStartEvent(Event):
    type: str = "tool_start"
    tool_name: str = ""
    tool_id: str = ""

@dataclass
class ToolInputEvent(Event):
    type: str = "tool_input"
    partial_json: str = ""

@dataclass
class ThinkingEvent(Event):
    type: str = "thinking"
    thinking: str = ""

@dataclass
class ToolResultEvent(Event):
    type: str = "tool_result"
    tool_name: str = ""
    result: "ToolResult" = None

@dataclass
class CostEvent(Event):
    type: str = "cost"
    input_tokens: int = 0
    output_tokens: int = 0
    cache_read: int = 0
    cache_creation: int = 0
    total_cost_usd: float = 0.0

@dataclass
class DoneEvent(Event):
    type: str = "done"
    stop_reason: str = ""
    turn_count: int = 0

@dataclass
class ErrorEvent(Event):
    type: str = "error"
    message: str = ""

@dataclass
class RetryEvent(Event):
    type: str = "retry"
    error: str = ""
    attempt: int = 0

@dataclass
class MaxTurnsEvent(Event):
    type: str = "max_turns"
    max_turns: int = 0

Sub-Agent Pattern

Spawn child agents for parallel work:

async def spawn_sub_agent(
    prompt: str,
    tools: ToolRegistry,
    api_client: APIClient,
    system_prompt: list[str] | None = None,
) -> str:
    sub_system = system_prompt or [
        "You are a sub-agent. Complete the task and return results.",
    ]
    messages = [{"role": "user", "content": prompt}]
    result_text = []

    async for event in agent_loop(
        messages=messages,
        tools=tools,
        permissions=PermissionSystem.auto_allow(),
        context_manager=ContextManager(),
        system_prompt=sub_system,
        api_client=api_client,
        max_turns=20,
    ):
        if isinstance(event, StreamingTextEvent) and event.text:
            result_text.append(event.text)
        if isinstance(event, DoneEvent):
            break

    return "".join(result_text)

Architecture Overview

The complete architecture of a production AI coding agent, reverse-engineered from Claude Code.

System Diagram

┌─────────────────────────────────────────────────┐
│                   Entry Point                    │
│  CLI arg parse → Auth → Init → Launch REPL      │
└──────────────────────┬──────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────┐
│              Query Engine (Session)               │
│  Owns: messages, abort, usage, file cache        │
│  submitMessage() → yields SDKMessage stream      │
└──────────────────────┬──────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────┐
│              Agent Loop (while true)              │
│                                                   │
│  ┌─────────┐  ┌──────────┐  ┌────────────────┐  │
│  │ Context  │→│ LLM API  │→│ Tool Execution │  │
│  │ Manage   │  │ Stream   │  │ Pipeline       │  │
│  └─────────┘  └──────────┘  └────────────────┘  │
│       ↑                           │               │
│       └───── tool results ────────┘               │
│                                                   │
│  Exit when: no tool_use in response               │
└──────────────────────┬──────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────┐
│              Tool Execution Pipeline              │
│                                                   │
│  Schema Validate → Hooks → Permission → Execute  │
│  → Post Hooks → Persist Large Results             │
└──────────────────────────────────────────────────┘

Core Modules

Dependency Graph

main.py
  ├── cli.py (argument parsing)
  ├── auth.py (API key management)
  ├── repl.py (interactive UI)
  │     └── query_engine.py (session orchestrator)
  │           ├── agent_loop.py (while-true loop)
  │           │     ├── api_client.py (LLM streaming)
  │           │     ├── context.py (message management)
  │           │     │     └── compact.py (summarization)
  │           │     ├── tool_executor.py (dispatch + hooks)
  │           │     │     ├── tools.py (registry)
  │           │     │     └── permissions.py (allow/deny)
  │           │     └── cost_tracker.py (usage)
  │           └── system_prompt.py (prompt assembly)
  └── headless.py (non-interactive mode)

Key Design Decisions

1. Async Generator Pattern

The agent loop is an async generator that yields events as they happen. This enables:

• Real-time streaming to any UI (terminal, web, API)
• Cancellation via abort signals
• Clean composition with sub-agents

2. Tool as Data

Every tool is a data structure (name, schema, handler), not a class hierarchy. This enables:

• Dynamic tool registration (add tools at runtime)
• MCP server integration (external tools look the same)
• Easy testing (mock any tool)

3. Prompt Cache Optimization

The system prompt is split into:

• Static prefix (never changes → gets cached by API → 10x cheaper)
• Dynamic boundary marker (signals cache break)
• Dynamic suffix (changes per turn → not cached)

4. Streaming Tool Execution

When the LLM streams multiple tool_use blocks:

• Concurrency-safe tools execute in parallel
• Non-safe tools wait for exclusivity
• Results yielded in original order

5. Defense-in-Depth Permissions

Seven independent layers check every tool call: 1. Hardcoded deny rules 2. Tool-specific permission logic 3. User-configured allow rules 4. Safety checks for sensitive paths 5. Interactive user prompts 6. Pre-tool hooks (can block/modify) 7. Post-tool hooks (can modify output)

Technology Stack Choices

For Python Agents

For TypeScript Agents

For Other Languages

The patterns are language-agnostic. Implement the same interfaces using:

• Rust: tokio + serde + clap
• Go: goroutines + struct tags + cobra
• Java: Virtual Threads + Jackson + picocli

Context Management

Strategies for managing the context window in long-running agent conversations.

The Problem

LLMs have fixed context windows. In a coding agent, conversations grow fast:

• System prompt: ~10K tokens
• Each tool call + result: ~1-5K tokens
• User messages: ~0.5-2K tokens each
• After 20 turns: ~50-100K tokens used

Without context management, the agent crashes when it exceeds the limit.

Three-Layer Strategy

Layer 1: Tool Result Budget

Persist large outputs to disk, keep references in context:

class ContextManager:
    max_inline_size = 30_000  # characters
    context_window = 200_000  # tokens
    buffer_tokens = 13_000    # safety margin
    persist_dir = ".agent-results"

    def apply_budget(self, messages: list[dict]) -> list[dict]:
        for msg in messages:
            if msg.get("role") == "user":
                for block in msg.get("content", []):
                    if block.get("type") == "tool_result":
                        content = block.get("content", "")
                        if len(content) > self.max_inline_size:
                            path = self._persist(content)
                            block["content"] = (
                                f"[Result too large ({len(content)} chars). "
                                f"Full output saved to: {path}]\n"
                                f"Preview: {content[:2000]}..."
                            )
        return messages

    def _persist(self, content: str) -> str:
        import hashlib, os
        h = hashlib.sha256(content.encode()).hexdigest()[:12]
        os.makedirs(self.persist_dir, exist_ok=True)
        path = os.path.join(self.persist_dir, f"{h}.txt")
        with open(path, "w") as f:
            f.write(content)
        return path

Layer 2: Snip/Micro-Compact

Replace old tool results with placeholders:

    def snip_stale(self, messages: list[dict], recent: int = 5) -> list[dict]:
        tool_result_count = 0
        for msg in reversed(messages):
            if msg.get("role") == "user":
                for block in msg.get("content", []):
                    if block.get("type") == "tool_result":
                        tool_result_count += 1

        results_to_snip = max(0, tool_result_count - recent)
        snipped = 0

        for msg in messages:
            if snipped >= results_to_snip:
                break
            if msg.get("role") == "user":
                for block in msg.get("content", []):
                    if block.get("type") == "tool_result" and snipped < results_to_snip:
                        original = block.get("content", "")
                        if len(original) > 200:
                            block["content"] = "[Old tool result content cleared]"
                            snipped += 1

        return messages

Layer 3: Auto-Compact (Full Summarization)

When approaching the limit, summarize the conversation:

    async def auto_compact_if_needed(
        self,
        messages: list[dict],
        system_prompt: list[str],
        tools,
    ) -> list[dict]:
        token_count = self._count_tokens(messages, system_prompt)
        threshold = self.context_window - self.buffer_tokens

        if token_count < threshold:
            return messages

        return await self._summarize(messages, system_prompt)

    async def _summarize(
        self,
        messages: list[dict],
        system_prompt: list[str],
    ) -> list[dict]:
        summary_prompt = [
            "Summarize the conversation so far. Preserve:",
            "1. Primary request and intent",
            "2. Key technical decisions made",
            "3. Files created/modified (with paths)",
            "4. Errors encountered and their fixes",
            "5. Pending tasks",
            "6. Current work in progress",
            "",
            "Be specific. Include file paths, line numbers, and code snippets.",
        ]

        # Call LLM to summarize (separate API call)
        summary = await self._call_for_summary(messages, summary_prompt)

        # Replace old messages with summary
        return [
            {
                "role": "user",
                "content": f"[Conversation compacted]\n\n{summary}",
            },
            messages[-1],  # Keep the most recent message intact
        ]

    def _count_tokens(self, messages: list[dict], system_prompt: list[str]) -> int:
        # Rough estimate: 1 token ≈ 4 chars
        total = sum(len(p) for p in system_prompt) // 4
        for msg in messages:
            content = msg.get("content", "")
            if isinstance(content, str):
                total += len(content) // 4
            elif isinstance(content, list):
                for block in content:
                    if isinstance(block, dict):
                        text = block.get("text", "") or block.get("content", "") or str(block)
                        total += len(text) // 4
        return total

Complete ContextManager

import os
import hashlib

class ContextManager:
    def __init__(
        self,
        context_window: int = 200_000,
        buffer_tokens: int = 13_000,
        max_inline_size: int = 30_000,
        persist_dir: str = ".agent-results",
        api_client = None,
    ):
        self.context_window = context_window
        self.buffer_tokens = buffer_tokens
        self.max_inline_size = max_inline_size
        self.persist_dir = persist_dir
        self.api_client = api_client

    def apply_budget(self, messages: list[dict]) -> list[dict]:
        return self._apply_budget_impl(messages)

    def snip_stale(self, messages: list[dict], recent: int = 5) -> list[dict]:
        return self._snip_stale_impl(messages, recent)

    async def auto_compact_if_needed(
        self, messages: list[dict], system_prompt: list[str], tools=None
    ) -> list[dict]:
        return await self._auto_compact_impl(messages, system_prompt)

    def is_over_limit(self, messages: list[dict], system_prompt: list[str]) -> bool:
        return self._count_tokens(messages, system_prompt) > self.context_window

    def persist_result(self, content: str) -> str:
        return self._persist(content)

Cost Optimization for Production AI Agents

How to reduce agent costs by 40–90%. Real techniques, real numbers.

---

1. The Numbers (Why This Matters)

Before you optimize anything, understand the scale of the problem:

• Unoptimized multi-agent enterprise system: $10,000–$150,000/month
• With proper optimization: 40–90% reduction is achievable
• Most teams leave 60–70% savings on the table through poor context management alone

The two biggest cost drivers in practice: 1. Context size — you pay for every input token on every turn. Old tool results that sit in context forever are pure waste. 2. Model selection — using claude-sonnet-4-6 to summarize a JSON blob that claude-haiku-4-5 could handle at one-tenth the price.

The techniques below address both. Apply them in order — the first two are highest ROI.

---

2. Technique 1: Prompt Caching (40–90% savings on system prompt costs)

Anthropic charges $0.30/MTok for cache reads versus $3.00/MTok for standard input. That's a 10x reduction on anything that hits the cache.

The rule: stable content goes first, dynamic content goes last. One changed token invalidates everything from that point forward.

def split_for_cache(system_prompt_parts: list[str]) -> list[dict]:
    """
    Split system prompt blocks into cacheable (static) and non-cacheable (dynamic).
    Static parts get cache_control, dynamic parts don't.
    """
    # Anything that doesn't change turn-to-turn is static:
    # - Agent persona and role
    # - Tool definitions
    # - Core instructions
    # - Example outputs
    #
    # Anything that changes is dynamic:
    # - Current date/time
    # - User name or project context
    # - Session-specific state
    STATIC_BLOCK_COUNT = 3  # tune this for your prompt structure

    result = []
    for i, part in enumerate(system_prompt_parts):
        block: dict = {"type": "text", "text": part}
        if i < STATIC_BLOCK_COUNT:
            block["cache_control"] = {"type": "ephemeral"}
        result.append(block)
    return result


# What NOT to do — this kills your cache hit rate:
BAD_SYSTEM_PROMPT = f"""
You are a helpful assistant. Today is {datetime.now()}.  # <-- timestamp in static section
You have access to the following tools...
"""

# What to do instead — separate static from dynamic:
STATIC_SYSTEM_PROMPT = """
You are a helpful assistant. You have access to the following tools...
[all tool definitions here]
[all core instructions here]
"""

def build_messages(user_query: str, current_date: str) -> list[dict]:
    return [
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": STATIC_SYSTEM_PROMPT,
                    "cache_control": {"type": "ephemeral"},  # this block gets cached
                },
                {
                    "type": "text",
                    "text": f"Current date: {current_date}\n\nUser query: {user_query}",
                    # no cache_control — dynamic, not cached
                },
            ],
        }
    ]

Cache rules summary: 1. Never put timestamps or session data in the cached prefix 2. Tool definitions belong in the static prefix — they almost never change 3. User context (name, current project) goes in the dynamic suffix 4. A one-token change anywhere in a cached block invalidates from that point forward

Cost calculation:

def calculate_caching_savings(
    input_tokens_per_session: int,
    sessions_per_day: int,
    cache_hit_rate: float = 0.80,
    days: int = 30,
) -> dict:
    """
    Calculate monthly savings from prompt caching.
    Prices: $3.00/MTok standard input, $0.30/MTok cache read.
    """
    STANDARD_PRICE = 3.00 / 1_000_000    # per token
    CACHE_READ_PRICE = 0.30 / 1_000_000  # per token

    total_sessions = sessions_per_day * days
    total_tokens = input_tokens_per_session * total_sessions

    cached_tokens = total_tokens * cache_hit_rate
    uncached_tokens = total_tokens * (1 - cache_hit_rate)

    cost_without_caching = total_tokens * STANDARD_PRICE
    cost_with_caching = (cached_tokens * CACHE_READ_PRICE) + (uncached_tokens * STANDARD_PRICE)

    savings_usd = cost_without_caching - cost_with_caching
    savings_pct = savings_usd / cost_without_caching

    return {
        "cost_without_caching_usd": round(cost_without_caching, 2),
        "cost_with_caching_usd": round(cost_with_caching, 2),
        "monthly_savings_usd": round(savings_usd, 2),
        "savings_percent": round(savings_pct * 100, 1),
    }


# Example: 50k token system prompt, 1000 sessions/day, 80% cache hit rate
result = calculate_caching_savings(50_000, 1_000)
# {'cost_without_caching_usd': 4500.0, 'cost_with_caching_usd': 720.0,
#  'monthly_savings_usd': 3780.0, 'savings_percent': 84.0}

---

3. Technique 2: Model Routing (40–60% savings)

Not every task needs claude-sonnet-4-6. Routing simple tasks to claude-haiku-4-5 costs roughly 10x less while producing identical results for the right task types.

Price comparison (per MTok, input/output):

• claude-haiku-4-5: ~$0.25 / $1.25
• claude-sonnet-4-6: ~$3.00 / $15.00

class ModelRouter:
    """Route tasks to the cheapest model that can handle them correctly."""

    FAST_MODEL = "claude-haiku-4-5"    # cheap, fast, great for simple tasks
    SMART_MODEL = "claude-sonnet-4-6"  # balanced, production default

    # Tasks where the fast model performs identically to the smart model
    SIMPLE_TASK_PATTERNS = [
        "summarize",
        "extract",
        "format",
        "classify",
        "list",
        "convert",
        "parse",
        "translate",
        "count",
    ]

    # Tasks that require reasoning — always use the smart model
    COMPLEX_TASK_PATTERNS = [
        "debug",
        "architect",
        "design",
        "analyze",
        "evaluate",
        "plan",
        "reason",
        "explain why",
    ]

    def select_model(self, task_description: str, context: dict | None = None) -> str:
        task_lower = task_description.lower()

        # Explicit complexity overrides
        if any(p in task_lower for p in self.COMPLEX_TASK_PATTERNS):
            return self.SMART_MODEL

        # Simple task heuristics
        if self._is_simple_task(task_lower, context):
            return self.FAST_MODEL

        return self.SMART_MODEL

    def _is_simple_task(self, task: str, context: dict | None) -> bool:
        # Pattern match
        if any(p in task for p in self.SIMPLE_TASK_PATTERNS):
            return True
        # Short structured output is simple
        if context and context.get("expected_output_format") in ("json", "list", "boolean"):
            return True
        return False


router = ModelRouter()

async def call_agent(task: str, messages: list[dict]) -> str:
    model = router.select_model(task)
    response = await client.messages.create(
        model=model,
        messages=messages,
        max_tokens=1024,
    )
    return response.content[0].text

Where to apply routing in a multi-agent system:

# Orchestrator: uses SMART_MODEL (makes decisions, plans)
# Summarizer subagent: uses FAST_MODEL (compresses tool outputs)
# Classifier subagent: uses FAST_MODEL (routes requests)
# Code executor: uses SMART_MODEL (generates correct code)
# Result formatter: uses FAST_MODEL (converts JSON to markdown)

AGENT_MODEL_MAP = {
    "orchestrator": "claude-sonnet-4-6",
    "summarizer": "claude-haiku-4-5",
    "classifier": "claude-haiku-4-5",
    "code_generator": "claude-sonnet-4-6",
    "formatter": "claude-haiku-4-5",
    "researcher": "claude-sonnet-4-6",
}

---

4. Technique 3: Context Discipline (addresses 60–70% of total costs)

This is the most important optimization. It's not glamorous. Most teams skip it. Every token in context is billed on every turn. A 10-turn session with a 100k token result sitting unmanaged in context costs 10x more than it needs to.

Three rules:

import tempfile
import os

# Rule 1: Tool result budget — never carry the full result inline
MAX_INLINE_CHARS = 30_000

def cap_tool_result(result: str, tool_name: str) -> str:
    """
    If a tool result exceeds the inline budget, write it to disk and
    return a reference with a short preview.
    """
    if len(result) <= MAX_INLINE_CHARS:
        return result

    # Write full result to temp file
    tmp = tempfile.NamedTemporaryFile(
        mode="w",
        suffix=f"_{tool_name}.txt",
        delete=False,
        prefix=".tool_output_",
    )
    tmp.write(result)
    tmp.close()

    return (
        f"[Result too large ({len(result):,} chars). Full output at: {tmp.name}]\n"
        f"Preview (first 2000 chars):\n{result[:2000]}"
    )


# Rule 2: Snip stale tool results from old turns
def snip_old_results(messages: list[dict], keep_recent_turns: int = 5) -> list[dict]:
    """
    Replace tool_result content in older turns with a placeholder.
    Keeps the most recent N turns fully intact.
    """
    tool_result_indices = [
        i for i, m in enumerate(messages)
        if m.get("role") == "tool" or (
            m.get("role") == "user" and
            isinstance(m.get("content"), list) and
            any(c.get("type") == "tool_result" for c in m.get("content", []))
        )
    ]

    # Keep the most recent `keep_recent_turns` tool results
    indices_to_clear = tool_result_indices[:-keep_recent_turns] if len(tool_result_indices) > keep_recent_turns else []

    result_messages = []
    for i, msg in enumerate(messages):
        if i in indices_to_clear:
            if isinstance(msg.get("content"), list):
                cleared_content = []
                for block in msg["content"]:
                    if block.get("type") == "tool_result":
                        cleared_content.append({**block, "content": "[Cleared — older than context window]"})
                    else:
                        cleared_content.append(block)
                result_messages.append({**msg, "content": cleared_content})
                continue
        result_messages.append(msg)
    return result_messages


# Rule 3: Compact messages when approaching the context limit
async def compact_if_needed(
    messages: list[dict],
    current_token_count: int,
    context_window: int = 200_000,
    threshold: float = 0.80,
) -> list[dict]:
    """
    Summarize older messages when token count crosses 80% of context window.
    Preserves the system prompt (index 0) and last 10 messages.
    """
    if current_token_count < context_window * threshold:
        return messages

    # Identify the slice to summarize: everything except first (system) and last 10
    if len(messages) <= 11:
        return messages

    to_summarize = messages[1:-10]
    preserved_tail = messages[-10:]

    summary_prompt = (
        "Summarize the following conversation history concisely. "
        "Preserve all decisions made, tools called, and key findings. "
        "Output only the summary, no preamble.\n\n"
        + "\n".join(f"{m['role']}: {str(m.get('content', ''))[:500]}" for m in to_summarize)
    )

    summary_response = await client.messages.create(
        model="claude-haiku-4-5",  # cheap model for summarization
        max_tokens=1024,
        messages=[{"role": "user", "content": summary_prompt}],
    )

    summary_message = {
        "role": "user",
        "content": f"[Context summary — {len(to_summarize)} messages compressed]\n{summary_response.content[0].text}",
    }

    return [messages[0], summary_message] + preserved_tail

---

5. Technique 4: Tool Set Curation

Every tool definition you include in your API call is tokens — paid for on every turn. 40 tool definitions in a typical SDK can add 8,000–15,000 tokens per turn.

# Bad: all tools loaded for every call regardless of task
def bad_agent_setup():
    registry.register_all_tools()  # 40 tools = 12,000 extra tokens per turn
    return registry.get_all()


# Good: load only what's needed for the current task type
BASE_TOOLS = ["ReadTool", "GlobTool", "GrepTool"]

TASK_TOOL_MAP = {
    "coding":   ["BashTool", "EditTool", "WriteTool"],
    "research": ["WebSearchTool", "WebFetchTool"],
    "data":     ["BashTool", "WriteTool"],
    "writing":  ["ReadTool", "WriteTool"],
}

def get_tools_for_task(task_type: str) -> list:
    tool_names = BASE_TOOLS + TASK_TOOL_MAP.get(task_type, [])
    return [registry.get_tool(name) for name in tool_names]


# Even better: classify the task first (using haiku), then load tools
async def dynamic_tool_loading(user_query: str) -> list:
    task_type = await classify_task(user_query)  # cheap haiku call
    return get_tools_for_task(task_type)


async def classify_task(query: str) -> str:
    response = await client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=10,
        messages=[{
            "role": "user",
            "content": (
                f"Classify this task as one of: coding, research, data, writing, other.\n"
                f"Reply with only the single word classification.\n\nTask: {query}"
            ),
        }],
    )
    return response.content[0].text.strip().lower()

Token savings per turn from tool curation:

• 40 tools → 5 tools: saves ~10,000 tokens per turn
• On a 20-turn session at $3/MTok: saves ~$0.60/session
• At 1,000 sessions/day: ~$18,000/month in tool overhead alone

---

6. Technique 5: Batch API for Non-Interactive Tasks

For anything that doesn't need a real-time response — scheduled reports, bulk analysis, background processing — the Batch API gives a flat 50% cost reduction.

import anthropic

client = anthropic.Anthropic()

def submit_batch_jobs(prompts: list[str]) -> str:
    """Submit a batch of requests. Returns the batch ID."""
    requests = [
        {
            "custom_id": f"job_{i}",
            "params": {
                "model": "claude-sonnet-4-6",
                "max_tokens": 1024,
                "messages": [{"role": "user", "content": prompt}],
            },
        }
        for i, prompt in enumerate(prompts)
    ]

    batch = client.beta.messages.batches.create(requests=requests)
    return batch.id


def poll_batch_results(batch_id: str) -> list[dict]:
    """Poll until complete and return results."""
    import time

    while True:
        batch = client.beta.messages.batches.retrieve(batch_id)
        if batch.processing_status == "ended":
            break
        time.sleep(60)  # batch API has no webhooks — poll every minute

    results = []
    for result in client.beta.messages.batches.results(batch_id):
        if result.result.type == "succeeded":
            results.append({
                "custom_id": result.custom_id,
                "text": result.result.message.content[0].text,
            })
    return results


# Use case: nightly document summarization
# Instead of running 500 summaries in real-time at $3/MTok:
# Submit as batch → processed within 24h → billed at $1.50/MTok

Where batch API applies: nightly reports, bulk data processing, training data generation, offline evaluation pipelines, scheduled analysis jobs.

Where it does NOT apply: anything a user is waiting on.

---

7. Monthly Cost Calculator

def estimate_monthly_cost(
    sessions_per_day: int,
    avg_turns_per_session: int,
    avg_tokens_per_turn: int,
    cache_hit_rate: float = 0.70,
    smart_model_fraction: float = 0.60,  # fraction of calls on sonnet vs haiku
    batch_fraction: float = 0.0,          # fraction run via Batch API
    model: str = "claude-sonnet-4-6",
) -> dict:
    """
    Estimate monthly cost under different optimization scenarios.
    Returns cost for: no optimization, caching only, routing added, fully optimized.

    Pricing assumptions (per MTok):
    - claude-sonnet-4-6 input: $3.00, output: $15.00, cache read: $0.30
    - claude-haiku-4-5 input:  $0.25, output:  $1.25, cache read: $0.03
    """
    SONNET_IN  = 3.00   / 1_000_000
    SONNET_OUT = 15.00  / 1_000_000
    SONNET_CACHE = 0.30 / 1_000_000
    HAIKU_IN   = 0.25   / 1_000_000
    HAIKU_OUT  = 1.25   / 1_000_000
    HAIKU_CACHE  = 0.03 / 1_000_000
    BATCH_DISCOUNT = 0.50  # 50% off for batch API

    total_sessions = sessions_per_day * 30
    turns_total = total_sessions * avg_turns_per_session

    # Assume output is ~30% of input tokens
    input_t = avg_tokens_per_turn
    output_t = int(avg_tokens_per_turn * 0.30)

    # Scenario 1: No optimization (all sonnet, no cache, no batch)
    no_opt = turns_total * (input_t * SONNET_IN + output_t * SONNET_OUT)

    # Scenario 2: Caching added
    cached_input_cost = (
        input_t * cache_hit_rate * SONNET_CACHE +
        input_t * (1 - cache_hit_rate) * SONNET_IN
    )
    with_caching = turns_total * (cached_input_cost + output_t * SONNET_OUT)

    # Scenario 3: Model routing added (split between sonnet and haiku)
    haiku_turns = turns_total * (1 - smart_model_fraction)
    sonnet_turns = turns_total * smart_model_fraction
    with_routing = (
        sonnet_turns * (
            input_t * cache_hit_rate * SONNET_CACHE +
            input_t * (1 - cache_hit_rate) * SONNET_IN +
            output_t * SONNET_OUT
        ) +
        haiku_turns * (
            input_t * cache_hit_rate * HAIKU_CACHE +
            input_t * (1 - cache_hit_rate) * HAIKU_IN +
            output_t * HAIKU_OUT
        )
    )

    # Scenario 4: Fully optimized (routing + caching + batch for eligible fraction)
    batch_turns = turns_total * batch_fraction
    interactive_turns = turns_total * (1 - batch_fraction)
    # Context discipline cuts effective input tokens by ~40%
    optimized_input_t = int(input_t * 0.60)

    fully_optimized = (
        interactive_turns * smart_model_fraction * (
            optimized_input_t * cache_hit_rate * SONNET_CACHE +
            optimized_input_t * (1 - cache_hit_rate) * SONNET_IN +
            output_t * SONNET_OUT
        ) +
        interactive_turns * (1 - smart_model_fraction) * (
            optimized_input_t * cache_hit_rate * HAIKU_CACHE +
            optimized_input_t * (1 - cache_hit_rate) * HAIKU_IN +
            output_t * HAIKU_OUT
        ) +
        batch_turns * BATCH_DISCOUNT * (
            smart_model_fraction * (optimized_input_t * SONNET_IN + output_t * SONNET_OUT) +
            (1 - smart_model_fraction) * (optimized_input_t * HAIKU_IN + output_t * HAIKU_OUT)
        )
    )

    return {
        "sessions_per_month": total_sessions,
        "without_optimization_usd": round(no_opt, 2),
        "with_caching_usd": round(with_caching, 2),
        "with_routing_usd": round(with_routing, 2),
        "fully_optimized_usd": round(fully_optimized, 2),
        "total_savings_pct": round((1 - fully_optimized / no_opt) * 100, 1) if no_opt > 0 else 0,
    }


# Example: 500 sessions/day, 15 turns each, 20k tokens/turn
result = estimate_monthly_cost(500, 15, 20_000, cache_hit_rate=0.75, smart_model_fraction=0.5, batch_fraction=0.3)
# without_optimization_usd: ~$40,500
# fully_optimized_usd:       ~$4,200
# total_savings_pct:          ~89.6%

---

Quick Reference: Optimization Priority

Do them in order. Prompt caching and context discipline together will get most teams to 60–70% savings before you touch anything else.

Framework Guide — Choosing the Right Agent Framework

Most developers reach for a framework before understanding what they're building. This guide helps you pick the right tool — or decide to build without one.

---

Quick Decision Matrix

---

When to Use agent-blueprint Patterns INSTEAD of a Framework

Frameworks solve common problems. They also add layers you may not need or want.

Use agent-blueprint patterns directly when:

• You need full control over every line of code — no hidden retry logic, no opaque prompt injection
• You're building a framework or platform — don't build on top of another framework; you'll fight it
• Your tool execution is unusual — streaming tool results, parallel fan-out, hardware APIs, custom protocols
• You're optimizing aggressively for cost/latency — framework overhead adds tokens and roundtrips
• You're learning — understanding the raw agent loop makes you a better engineer regardless of what you eventually use
• Edge or embedded deployment — framework cold starts and dependencies are unacceptable at the edge

The agent loop is not complex. It is a while loop, a tool dispatcher, and state management. Every framework is that loop plus opinions. Know the loop first.

---

2026 Framework Landscape

LangGraph

• ~30k GitHub stars. Dominant for production Python agent workflows.
• Directed graph model: nodes are functions, edges define control flow
• Checkpointing to PostgreSQL or Redis — full state persistence and replay
• interrupt_before pauses the graph at a named node for human approval
• 15B+ traces processed via LangSmith
• Choose this for: any Python agent that needs reliability, HITL, or audit trails

CrewAI

• Role-based multi-agent: Researcher, Analyst, Writer personas
• Claims 450M monthly workflows, 60% Fortune 500
• BUT: 3x more tokens than LangChain, 3x slower in benchmarks
• Best use case: rapid prototyping and demos. Migrate to LangGraph for production.

Mastra

• TypeScript-native, Y Combinator-backed ($13M seed)
• PayPal, Adobe, Docker in production
• 40+ model providers, native MCP support, supervisor pattern, LSP diagnostics
• Choose this for: any TypeScript agent not tied to a Next.js stack

Vercel AI SDK 6 (Feb 2026)

• ToolLoopAgent class, needsApproval for HITL, full MCP support
• Built-in DevTools, Fluid compute for long-running agents
• Choose this for: Next.js or full-stack TypeScript where you want streaming UI + agent in one place

OpenAI Agents SDK

• 10.3M monthly downloads, 19k GitHub stars
• Claims 100+ LLM support. Simplest path from prototype to production.
• Choose this for: teams already in the OpenAI ecosystem who want minimal setup

Anthropic Agent SDK

• 4.6k stars. Deepest MCP integration of any framework.
• Anthropic models only.
• Choose this for: Claude-committed teams who want best-in-class MCP tooling

Microsoft Agent Framework

• AutoGen is retired (maintenance mode only). Do not start new projects on AutoGen.
• Microsoft Agent Framework went GA Q1 2026. Replacement for AutoGen.
• AG2 is a community fork at ag2.ai — lives on, but fragmented ecosystem.
• Choose this for: Azure shops with enterprise governance requirements

Google ADK

• New as of late 2025. OpenTelemetry built in. Google Cloud + Gemini native.
• Still early — API surface is changing. Budget 20–30% extra time for breaking changes.
• Choose this for: Gemini-native teams on Google Cloud who can absorb churn

---

LangGraph — Setup and Core Patterns

Graph = nodes + edges + state. Every run is reproducible and resumable.

from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres import PostgresSaver
from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]

def call_model(state: AgentState) -> AgentState:
    # Call LLM, return updated messages
    ...

def call_tools(state: AgentState) -> AgentState:
    # Execute tool calls from last message
    ...

def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "call_tools"
    return "end"

# Build graph
builder = StateGraph(AgentState)
builder.add_node("call_model", call_model)
builder.add_node("call_tools", call_tools)
builder.set_entry_point("call_model")
builder.add_conditional_edges("call_model", should_continue)
builder.add_edge("call_tools", "call_model")

# Compile with persistence
graph = builder.compile(
    checkpointer=PostgresSaver.from_conn_string("postgresql://user:pass@localhost/agentdb"),
    interrupt_before=["call_tools"]  # pause for human approval before any tool call
)

# Run
config = {"configurable": {"thread_id": "session-abc-123"}}
result = graph.invoke({"messages": [user_message]}, config=config)

Key strength: interrupt_before=["call_tools"] pauses execution and persists state. Resume with the same thread_id after human review — the graph continues exactly where it stopped.

---

CrewAI — Setup and Core Patterns

Agents have roles. Tasks have expected outputs. The crew executes sequentially or in parallel.

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Research Analyst",
    goal="Find accurate, current information on the given topic",
    backstory="You are a meticulous researcher with 10 years of experience in competitive intelligence.",
    verbose=True
)

writer = Agent(
    role="Content Writer",
    goal="Write clear, structured reports from research findings",
    backstory="You turn dense research into readable executive summaries.",
    verbose=True
)

research_task = Task(
    description="Research the current state of {topic}. Find key players, recent developments, and market size.",
    agent=researcher,
    expected_output="Markdown report with sources, minimum 500 words"
)

write_task = Task(
    description="Write an executive summary based on the research provided.",
    agent=writer,
    expected_output="500-word executive summary with 3 key takeaways",
    context=[research_task]
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff(inputs={"topic": "AI agent frameworks 2026"})
print(result.raw)

Warning: CrewAI uses 3x more tokens than an equivalent LangChain implementation and runs 3x slower in benchmarks. Use it to validate business logic and role definitions. Migrate to LangGraph for production.

---

Mastra — Setup and Core Patterns

TypeScript-native. MCP built in. 40+ providers from one API.

import { Mastra, createAgent, createTool } from "@mastra/core";
import { z } from "zod";

// Define a tool
const webSearch = createTool({
  id: "web-search",
  description: "Search the web for current information",
  inputSchema: z.object({ query: z.string() }),
  outputSchema: z.object({ results: z.array(z.string()) }),
  execute: async ({ context }) => {
    // your search implementation
    return { results: [] };
  }
});

// Define an agent
const researchAgent = createAgent({
  name: "researcher",
  model: {
    provider: "ANTHROPIC",
    name: "claude-sonnet-4-6"
  },
  tools: { webSearch },
  instructions: `You are a research specialist.
  Search for current, accurate information and cite your sources.
  Always verify facts before including them in your response.`
});

// Register with Mastra
const mastra = new Mastra({
  agents: { researcher: researchAgent }
});

// Run
const result = await mastra.getAgent("researcher").generate(
  "Research the current state of LangGraph adoption in production"
);

console.log(result.text);

Supervisor pattern (multi-agent):

const supervisorAgent = createAgent({
  name: "supervisor",
  model: { provider: "ANTHROPIC", name: "claude-sonnet-4-6" },
  agents: { researcher: researchAgent, writer: writerAgent },
  instructions: "Coordinate researcher and writer agents to produce reports."
});

---

Vercel AI SDK 6 — Setup and Core Patterns

Best for Next.js. Streaming UI + agent loop coexist in one stack.

import { ToolLoopAgent } from "ai";
import { anthropic } from "@ai-sdk/anthropic";
import { tool } from "ai";
import { z } from "zod";

// Define tools
const bashTool = tool({
  description: "Execute a shell command",
  parameters: z.object({ command: z.string() }),
  needsApproval: true, // HITL — prompts user before execution
  execute: async ({ command }) => {
    // your bash execution
    return { output: "" };
  }
});

const readFileTool = tool({
  description: "Read a file from the filesystem",
  parameters: z.object({ path: z.string() }),
  execute: async ({ path }) => {
    // your file reading
    return { content: "" };
  }
});

// Create agent
const agent = new ToolLoopAgent({
  model: anthropic("claude-sonnet-4-6"),
  tools: { bash: bashTool, readFile: readFileTool },
  maxSteps: 20,
  system: "You are a helpful coding assistant."
});

// In a Next.js API route (app/api/agent/route.ts):
export async function POST(req: Request) {
  const { prompt } = await req.json();

  const { stream } = await agent.streamText({ prompt });
  return stream.toDataStreamResponse();
}

On the client (React):

import { useAgent } from "ai/react";

export default function AgentChat() {
  const { messages, sendMessage, isLoading } = useAgent({ api: "/api/agent" });

  return (
    <div>
      {messages.map(m => <div key={m.id}>{m.content}</div>)}
      <input onKeyDown={e => e.key === "Enter" && sendMessage(e.currentTarget.value)} />
    </div>
  );
}

---

Migration Path

The common journey from learning to production:

Start with agent-blueprint patterns (understand the raw agent loop)
    ↓
Prototype with CrewAI or agent-blueprint minimal template (fast iteration, validate logic)
    ↓
Production with LangGraph (Python) or Mastra (TypeScript)
    (checkpointing, HITL, observability)
    ↓
Scale with framework + agent-blueprint production principles
    (cost optimization, reliability, graceful degradation)

Skip steps only if you understand what you're skipping. Going directly from prototype to production without addressing checkpointing and HITL is the most common cause of agent reliability failures.

---

The "Don't Make Everything Agentic" Rule

Most real applications: 30–40% genuinely needs LLM reasoning. The rest should be:

• Deterministic Python or TypeScript code
• Standard API calls
• Database queries
• Simple rule-based logic

The mistake: wrapping everything in an agent loop when it's actually just if/else.

# Wrong — using an agent for something deterministic
result = agent.run("Check if the user's subscription is active and return True or False")

# Right — just check the database
is_active = db.query("SELECT active FROM subscriptions WHERE user_id = ?", user_id)

If you can write the logic as a function without calling an LLM, write it as a function. Agents are for tasks that require reasoning over ambiguous inputs, multi-step planning, or dynamic tool selection based on context. Not for tasks that have a deterministic answer.

Identifying what needs an agent:

• Input is ambiguous or variable in structure → agent
• Steps depend on previous results in unpredictable ways → agent
• Tool selection depends on reasoning about the task → agent
• Logic is a known sequence of deterministic operations → function, not agent

Keep the LLM calls minimal. Every unnecessary agent call is latency, cost, and a failure surface.

---

Summary

Human-in-the-Loop (HITL)

In 2026, HITL is a governance requirement for enterprise agents — not an optional feature. Compliance, insurance, and liability frameworks all demand human approval gates for high-impact actions.

The Interrupt/Resume Pattern

The correct pattern is interrupt → persist → notify → wait → resume. The wrong pattern is stopping the agent and restarting from scratch. Restarting loses context, wastes LLM calls, and breaks auditability.

Agent executes
      │
      ▼
 Checkpoint reached? ──No──► Continue execution
      │
     Yes
      │
      ▼
 Persist full state to checkpoint store
      │
      ▼
 Notify human (terminal / webhook / DB)
      │
      ▼
 Wait for decision (approve / edit / reject / escalate)
      │
      ▼
 Load checkpoint, apply decision, resume from exact point

The agent must persist state before notifying. The system could restart while waiting.

---

The 4-Dimension Decision Matrix

Four dimensions determine if an action needs human approval.

from dataclasses import dataclass

@dataclass
class AgentAction:
    name: str
    tool_input: dict
    is_irreversible: bool        # Can this be undone?
    affected_records: int        # How many records / people does this touch?
    creates_legal_obligation: bool  # Does this create a legal obligation?
    model_confidence: float      # 0.0 – 1.0, how certain is the model?

def requires_human_approval(action: AgentAction) -> bool:
    """Return True if this action should pause for human review."""
    # Dimension 1 — Irreversibility
    if action.is_irreversible:
        return True
    # Dimension 2 — Blast radius
    if action.affected_records > 100:
        return True
    # Dimension 3 — Compliance exposure
    if action.creates_legal_obligation:
        return True
    # Dimension 4 — Model confidence
    if action.model_confidence < 0.85:
        return True
    return False

---

State Persistence

The agent persists its complete conversation state before handing off to a human. This enables resume-from-checkpoint even after a server restart.

from dataclasses import dataclass, field
from datetime import datetime, timedelta
import json
import uuid


@dataclass
class CheckpointState:
    session_id: str
    turn: int
    messages: list[dict]
    pending_action: dict      # the tool call awaiting approval
    context: dict             # any extra agent state (memory, scratch pad, etc.)
    created_at: str
    timeout_at: str           # auto-expire after N hours


class CheckpointStore:
    """
    Two-layer storage:
      - Redis  → fast lookup during active sessions
      - PostgreSQL / SQLite → durable record for audit trail

    For simple deployments, disk-backed JSON is enough.
    """

    def __init__(self, path: str = "/tmp/hitl_checkpoints"):
        import os
        self.path = path
        os.makedirs(path, exist_ok=True)

    def save(self, state: CheckpointState) -> str:
        checkpoint_id = str(uuid.uuid4())
        file_path = f"{self.path}/{checkpoint_id}.json"
        with open(file_path, "w") as f:
            json.dump(state.__dict__, f, indent=2)
        return checkpoint_id

    def load(self, checkpoint_id: str) -> CheckpointState:
        file_path = f"{self.path}/{checkpoint_id}.json"
        with open(file_path) as f:
            data = json.load(f)
        return CheckpointState(**data)

    def list_pending(self) -> list[CheckpointState]:
        import os
        states = []
        for fname in os.listdir(self.path):
            if fname.endswith(".json"):
                cid = fname.replace(".json", "")
                states.append(self.load(cid))
        return states

    def delete(self, checkpoint_id: str) -> None:
        import os
        file_path = f"{self.path}/{checkpoint_id}.json"
        if os.path.exists(file_path):
            os.remove(file_path)

    def write_decision(self, checkpoint_id: str, decision: dict) -> None:
        file_path = f"{self.path}/{checkpoint_id}.decision.json"
        with open(file_path, "w") as f:
            json.dump(decision, f)

    def read_decision(self, checkpoint_id: str) -> dict | None:
        import os
        file_path = f"{self.path}/{checkpoint_id}.decision.json"
        if not os.path.exists(file_path):
            return None
        with open(file_path) as f:
            return json.load(f)

---

The Four Approval Outcomes

from enum import Enum
from dataclasses import dataclass


class ApprovalDecision(Enum):
    APPROVE   = "approve"    # Execute exactly as planned
    EDIT      = "edit"       # Human modified the action — execute modified version
    REJECT    = "reject"     # Cancel this action, return error to agent
    ESCALATE  = "escalate"   # Send to a higher authority (manager, security team)


@dataclass
class ApprovalResult:
    outcome: ApprovalDecision
    modified_input: dict | None = None   # set when outcome == EDIT
    reason: str = ""

How each outcome flows back into the agent loop:

• APPROVE — pass tool_input to the tool unchanged, inject the tool result as normal.
• EDIT — replace tool_input with decision.modified_input, then execute.
• REJECT — inject a synthetic tool result: {"error": "Action rejected by human: <reason>"}.

The agent receives this and must decide what to do next (retry differently, abort, ask user).

• ESCALATE — same as REJECT for the current agent turn, but the checkpoint is forwarded

to a secondary approval queue (e.g., a Slack channel for the security team).

---

Python Implementation — HITLMiddleware

import asyncio
import time
from dataclasses import dataclass
from datetime import datetime, timedelta
from fnmatch import fnmatch
from typing import Callable


class HITLMiddleware:
    """
    Wraps any tool call with an interrupt/resume approval gate.

    Usage:
        hitl = HITLMiddleware(store=CheckpointStore(), notification=TerminalNotification())
        # In your agent loop — before every tool execution:
        if hitl.requires_approval(tool_name, tool_input):
            checkpoint_id = hitl.create_checkpoint(messages, tool_use)
            hitl.notify(checkpoint_id, f"Waiting for approval: {tool_name}")
            decision = await hitl.wait_for_approval(checkpoint_id, timeout=3600)
            ...
    """

    def __init__(
        self,
        store: CheckpointStore,
        notification: "NotificationBackend",
        rules: dict | None = None,
    ):
        self.store = store
        self.notification = notification
        self.rules = rules or {
            "auto_approve": ["Read", "Glob", "Grep", "Bash(git status*)", "Bash(git log*)"],
            "always_ask":   ["Bash(git push*)", "Bash(rm *)", "Write"],
            "auto_reject":  ["Bash(rm -rf /*)"],
        }

    def requires_approval(self, tool_name: str, tool_input: dict) -> bool:
        """True if this tool call must pause for human review."""
        if self._matches(tool_name, tool_input, self.rules["auto_reject"]):
            return True  # auto-reject also pauses to tell the human
        if self._matches(tool_name, tool_input, self.rules["auto_approve"]):
            return False
        if self._matches(tool_name, tool_input, self.rules["always_ask"]):
            return True
        # Fall back to 4-dimension matrix
        action = self._infer_action(tool_name, tool_input)
        return requires_human_approval(action)

    def create_checkpoint(
        self,
        messages: list[dict],
        pending_action: dict,
        context: dict | None = None,
        timeout_hours: int = 8,
    ) -> str:
        now = datetime.utcnow()
        state = CheckpointState(
            session_id=str(__import__("uuid").uuid4()),
            turn=len(messages),
            messages=messages,
            pending_action=pending_action,
            context=context or {},
            created_at=now.isoformat(),
            timeout_at=(now + timedelta(hours=timeout_hours)).isoformat(),
        )
        return self.store.save(state)

    def notify(self, checkpoint_id: str, message: str) -> None:
        state = self.store.load(checkpoint_id)
        asyncio.get_event_loop().run_until_complete(
            self.notification.notify(checkpoint_id, state.pending_action)
        )

    async def wait_for_approval(
        self, checkpoint_id: str, timeout: int = 3600, poll_interval: int = 2
    ) -> ApprovalResult:
        """Poll the checkpoint store until a decision arrives or timeout expires."""
        deadline = time.time() + timeout
        while time.time() < deadline:
            raw = self.store.read_decision(checkpoint_id)
            if raw:
                self.store.delete(checkpoint_id)
                return ApprovalResult(
                    outcome=ApprovalDecision(raw["outcome"]),
                    modified_input=raw.get("modified_input"),
                    reason=raw.get("reason", ""),
                )
            await asyncio.sleep(poll_interval)
        # Timeout — treat as rejection
        self.store.delete(checkpoint_id)
        return ApprovalResult(outcome=ApprovalDecision.REJECT, reason="Approval timed out")

    def _matches(self, tool_name: str, tool_input: dict, rules: list[str]) -> bool:
        for rule in rules:
            if "(" in rule:
                rule_tool, pattern = rule.split("(", 1)
                pattern = pattern.rstrip(")")
                if rule_tool == tool_name:
                    cmd = tool_input.get("command", str(tool_input))
                    if fnmatch(cmd, pattern):
                        return True
            elif rule == tool_name:
                return True
        return False

    def _infer_action(self, tool_name: str, tool_input: dict) -> AgentAction:
        irreversible = tool_name in ("Write", "Edit", "Bash")
        return AgentAction(
            name=tool_name,
            tool_input=tool_input,
            is_irreversible=irreversible,
            affected_records=0,
            creates_legal_obligation=False,
            model_confidence=0.90,
        )

Agent Loop Integration

async def run_agent_loop(messages: list[dict], hitl: HITLMiddleware) -> list[dict]:
    import anthropic

    client = anthropic.Anthropic()

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            messages=messages,
            tools=TOOLS,
        )

        if response.stop_reason == "end_turn":
            break

        for block in response.content:
            if block.type != "tool_use":
                continue

            tool_name  = block.name
            tool_input = block.input

            # HITL gate
            if hitl.requires_approval(tool_name, tool_input):
                checkpoint_id = hitl.create_checkpoint(messages, {"name": tool_name, "input": tool_input})
                hitl.notify(checkpoint_id, f"Waiting for approval: {tool_name}")
                decision = await hitl.wait_for_approval(checkpoint_id, timeout=3600)

                if decision.outcome == ApprovalDecision.REJECT:
                    tool_result = {"error": f"Action rejected: {decision.reason}"}
                elif decision.outcome == ApprovalDecision.EDIT:
                    tool_input  = decision.modified_input
                    tool_result = execute_tool(tool_name, tool_input)
                else:
                    tool_result = execute_tool(tool_name, tool_input)
            else:
                tool_result = execute_tool(tool_name, tool_input)

            messages.append({"role": "user", "content": [{"type": "tool_result", "tool_use_id": block.id, "content": str(tool_result)}]})

    return messages

---

TypeScript Equivalent

import Anthropic from "@anthropic-ai/sdk";

interface ApprovalResult {
  outcome: "approve" | "edit" | "reject" | "escalate";
  modifiedInput?: Record<string, unknown>;
  reason?: string;
}

class HITLMiddleware {
  private store: Map<string, unknown> = new Map();
  private decisions: Map<string, ApprovalResult> = new Map();

  requiresApproval(toolName: string, _toolInput: Record<string, unknown>): boolean {
    const alwaysAsk = ["Write", "Edit"];
    return alwaysAsk.includes(toolName);
  }

  createCheckpoint(messages: unknown[], pendingAction: unknown): string {
    const id = crypto.randomUUID();
    this.store.set(id, { messages, pendingAction, createdAt: new Date().toISOString() });
    return id;
  }

  notify(checkpointId: string, message: string): void {
    console.log(`\n[HITL] ${message}`);
    console.log(`[HITL] Checkpoint: ${checkpointId}`);
    console.log(`[HITL] Approve with: hitl.submitDecision("${checkpointId}", "approve")`);
  }

  submitDecision(checkpointId: string, result: ApprovalResult): void {
    this.decisions.set(checkpointId, result);
  }

  async waitForApproval(checkpointId: string, timeoutMs = 3_600_000): Promise<ApprovalResult> {
    const deadline = Date.now() + timeoutMs;
    while (Date.now() < deadline) {
      const decision = this.decisions.get(checkpointId);
      if (decision) {
        this.store.delete(checkpointId);
        this.decisions.delete(checkpointId);
        return decision;
      }
      await new Promise((r) => setTimeout(r, 2000));
    }
    return { outcome: "reject", reason: "Approval timed out" };
  }
}

---

Notification Patterns

from typing import Protocol


class NotificationBackend(Protocol):
    async def notify(self, checkpoint_id: str, action: dict) -> None: ...
    async def poll_for_decision(self, checkpoint_id: str) -> ApprovalResult: ...


class TerminalNotification:
    """Prints to stdout. Human types their decision in the terminal."""

    async def notify(self, checkpoint_id: str, action: dict) -> None:
        print(f"\n{'='*60}")
        print(f"[HITL] Action requires approval")
        print(f"  checkpoint : {checkpoint_id}")
        print(f"  tool       : {action.get('name')}")
        print(f"  input      : {action.get('input')}")
        print(f"  decide     : approve | edit | reject | escalate")
        print(f"{'='*60}\n")

    async def poll_for_decision(self, checkpoint_id: str) -> ApprovalResult:
        raw = input("Decision [approve/edit/reject/escalate]: ").strip().lower()
        modified = None
        if raw == "edit":
            modified = eval(input("Modified input (dict): "))
        return ApprovalResult(outcome=ApprovalDecision(raw), modified_input=modified)


class WebhookNotification:
    """POSTs to a URL — works for Slack, Teams, email relay, custom UI."""

    def __init__(self, webhook_url: str, poll_base_url: str):
        self.webhook_url  = webhook_url
        self.poll_base_url = poll_base_url

    async def notify(self, checkpoint_id: str, action: dict) -> None:
        import httpx
        async with httpx.AsyncClient() as client:
            await client.post(self.webhook_url, json={
                "checkpoint_id": checkpoint_id,
                "action":        action,
                "approve_url":   f"{self.poll_base_url}/approve/{checkpoint_id}",
                "reject_url":    f"{self.poll_base_url}/reject/{checkpoint_id}",
            })

    async def poll_for_decision(self, checkpoint_id: str) -> ApprovalResult:
        import httpx
        async with httpx.AsyncClient() as client:
            r = await client.get(f"{self.poll_base_url}/decision/{checkpoint_id}")
            data = r.json()
            return ApprovalResult(
                outcome=ApprovalDecision(data["outcome"]),
                modified_input=data.get("modified_input"),
                reason=data.get("reason", ""),
            )


class DatabaseNotification:
    """Writes to a DB table. Your app polls and renders a UI for the approver."""

    def __init__(self, connection_string: str):
        self.dsn = connection_string

    async def notify(self, checkpoint_id: str, action: dict) -> None:
        import asyncpg, json
        conn = await asyncpg.connect(self.dsn)
        await conn.execute(
            "INSERT INTO hitl_pending (id, action, status) VALUES ($1, $2, 'pending')",
            checkpoint_id, json.dumps(action)
        )
        await conn.close()

    async def poll_for_decision(self, checkpoint_id: str) -> ApprovalResult:
        import asyncpg
        conn = await asyncpg.connect(self.dsn)
        while True:
            row = await conn.fetchrow(
                "SELECT status, modified_input, reason FROM hitl_pending WHERE id = $1",
                checkpoint_id,
            )
            if row and row["status"] != "pending":
                await conn.close()
                return ApprovalResult(
                    outcome=ApprovalDecision(row["status"]),
                    modified_input=row["modified_input"],
                    reason=row["reason"] or "",
                )
            await asyncio.sleep(2)

---

Pre-Configured Approval Rules

Users define rules once. The agent never interrupts for routine, low-risk calls.

{
  "hitl": {
    "auto_approve": [
      "Read",
      "Glob",
      "Grep",
      "Bash(git status*)",
      "Bash(git log*)",
      "Bash(git diff*)"
    ],
    "always_ask": [
      "Bash(git push*)",
      "Bash(rm *)",
      "Write",
      "Edit"
    ],
    "auto_reject": [
      "Bash(rm -rf /*)"
    ]
  }
}

Rule matching order: auto_reject → always_ask → auto_approve → 4-dimension matrix.

---

The Tiered Delegation Model

Three tiers, chosen per task. Matches the Kilo Speed framework.

from enum import Enum
from dataclasses import dataclass, field


class AgentTier(Enum):
    AUTONOMOUS   = "tier1"
    CHECKPOINTS  = "tier2"
    PAIR         = "tier3"


@dataclass
class HITLConfig:
    mode: str                    # auto_allow | milestone | ask_always
    milestones: list[str] = field(default_factory=list)


def build_hitl_config(tier: AgentTier) -> HITLConfig:
    return {
        AgentTier.AUTONOMOUS:  HITLConfig(mode="auto_allow"),
        AgentTier.CHECKPOINTS: HITLConfig(
            mode="milestone",
            milestones=["planning_done", "before_deploy"],
        ),
        AgentTier.PAIR: HITLConfig(mode="ask_always"),
    }[tier]


async def run_agent(task: str, tier: AgentTier = AgentTier.AUTONOMOUS) -> None:
    config = build_hitl_config(tier)
    store  = CheckpointStore()
    notif  = TerminalNotification()

    if config.mode == "auto_allow":
        rules = {"auto_approve": ["*"], "always_ask": [], "auto_reject": []}
    elif config.mode == "ask_always":
        rules = {"auto_approve": [], "always_ask": ["*"], "auto_reject": []}
    else:
        # milestone mode: auto-approve everything, pause at named checkpoints
        rules = {
            "auto_approve": ["Read", "Glob", "Grep", "Bash(git *)"],
            "always_ask":   ["Write", "Edit", "Bash(git push*)"],
            "auto_reject":  [],
        }

    hitl = HITLMiddleware(store=store, notification=notif, rules=rules)
    messages = [{"role": "user", "content": task}]
    await run_agent_loop(messages, hitl)

Milestone pauses in Tier 2 are implemented by injecting named checkpoint calls at key moments in the agent logic (e.g., after the planning step, before any deployment command). The agent calls hitl.create_checkpoint(...) explicitly at those points, independent of tool-level rules.

MCP Integration

Connect external tools to your agent via the Model Context Protocol.

What MCP Is

MCP servers expose tools over a standard JSON-RPC protocol via stdio (subprocess) or SSE (HTTP). Your agent connects to them at startup, lists their available tools, and registers them alongside built-in tools. From the agent loop's perspective, an MCP tool is indistinguishable from any other Tool subclass — it goes through the same permission checks and execution pipeline.

Connecting to an MCP Server (Python)

import asyncio
from contextlib import asynccontextmanager
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.client.sse import sse_client

@asynccontextmanager
async def connect_stdio(command: str, args: list[str], env: dict[str, str] | None = None):
    """Connect to an MCP server running as a subprocess."""
    params = StdioServerParameters(command=command, args=args, env=env)
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            yield session

@asynccontextmanager
async def connect_sse(url: str):
    """Connect to an MCP server over HTTP/SSE."""
    async with sse_client(url) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            yield session

async def list_server_tools(session: ClientSession) -> list[dict]:
    """Return raw MCP tool descriptors from the server."""
    response = await session.list_tools()
    return [
        {
            "name": tool.name,
            "description": tool.description or "",
            "input_schema": tool.inputSchema or {"properties": {}, "required": []},
        }
        for tool in response.tools
    ]

MCPTool Adapter (Python)

Wraps a single MCP server tool into the agent's Tool interface so it slots into the registry.

from typing import Any
from mcp import ClientSession

# Tool and ToolResult are defined in tool-system.md
class MCPTool(Tool):
    """Wraps an MCP server tool into the agent's Tool interface."""

    def __init__(self, session: ClientSession, server_name: str, descriptor: dict):
        self._session = session
        self._server_name = server_name
        self.name = f"mcp__{server_name}__{descriptor['name']}"
        self.description = descriptor["description"]
        self.input_schema = descriptor["input_schema"]
        self.is_read_only = False
        self.is_concurrency_safe = False

    async def call(self, input: dict, context: dict) -> ToolResult:
        # Strip the mcp__servername__ prefix to get the bare tool name
        bare_name = self.name.split("__", 2)[2]
        try:
            result = await self._session.call_tool(bare_name, arguments=input)
            parts = [
                item.text if hasattr(item, "text") else str(item)
                for item in result.content
            ]
            output = "\n".join(parts)
            if result.isError:
                return ToolResult(success=False, error=output)
            return ToolResult(success=True, content=output)
        except Exception as e:
            return ToolResult(success=False, error=f"MCP call failed: {e}")

MCPTool Adapter (TypeScript)

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";

// Tool and ToolResult mirror the Python interface
interface ToolResult {
  success: boolean;
  content: string;
  error: string;
}

interface Tool {
  name: string;
  description: string;
  inputSchema: Record<string, unknown>;
  call(input: Record<string, unknown>): Promise<ToolResult>;
}

export async function createStdioClient(
  command: string,
  args: string[],
  env?: Record<string, string>
): Promise<Client> {
  const transport = new StdioClientTransport({ command, args, env });
  const client = new Client({ name: "agent-blueprint", version: "1.0.0" });
  await client.connect(transport);
  return client;
}

export async function createSSEClient(url: string): Promise<Client> {
  const transport = new SSEClientTransport(new URL(url));
  const client = new Client({ name: "agent-blueprint", version: "1.0.0" });
  await client.connect(transport);
  return client;
}

export class MCPTool implements Tool {
  name: string;
  description: string;
  inputSchema: Record<string, unknown>;
  private client: Client;
  private bareName: string;

  constructor(client: Client, serverName: string, descriptor: {
    name: string;
    description: string;
    inputSchema: Record<string, unknown>;
  }) {
    this.client = client;
    this.bareName = descriptor.name;
    this.name = `mcp__${serverName}__${descriptor.name}`;
    this.description = descriptor.description;
    this.inputSchema = descriptor.inputSchema;
  }

  async call(input: Record<string, unknown>): Promise<ToolResult> {
    try {
      const result = await this.client.callTool({
        name: this.bareName,
        arguments: input,
      });
      const parts = (result.content as Array<{ type: string; text?: string }>)
        .map((item) => (item.type === "text" ? (item.text ?? "") : JSON.stringify(item)));
      const output = parts.join("\n");
      return result.isError
        ? { success: false, content: "", error: output }
        : { success: true, content: output, error: "" };
    } catch (err) {
      return { success: false, content: "", error: `MCP call failed: ${err}` };
    }
  }
}

export async function loadMCPTools(
  client: Client,
  serverName: string
): Promise<MCPTool[]> {
  const { tools } = await client.listTools();
  return tools.map(
    (t) =>
      new MCPTool(client, serverName, {
        name: t.name,
        description: t.description ?? "",
        inputSchema: (t.inputSchema as Record<string, unknown>) ?? {},
      })
  );
}

MCP Server Registry (Python)

Manages multiple MCP server connections and flattens their tools into one list.

import asyncio
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class MCPServerRegistry:
    def __init__(self):
        self._stack = AsyncExitStack()
        self._tools: list[MCPTool] = []
        self._sessions: dict[str, ClientSession] = {}

    async def __aenter__(self):
        await self._stack.__aenter__()
        return self

    async def __aexit__(self, *exc):
        await self._stack.__aexit__(*exc)

    async def connect_server(
        self,
        name: str,
        command: list[str],
        args: list[str] | None = None,
        env: dict[str, str] | None = None,
    ) -> list[MCPTool]:
        """Connect to an MCP server and register its tools. Returns the new tools."""
        params = StdioServerParameters(
            command=command[0],
            args=(command[1:] + (args or [])),
            env=env,
        )
        read, write = await self._stack.enter_async_context(stdio_client(params))
        session = await self._stack.enter_async_context(ClientSession(read, write))
        await session.initialize()

        self._sessions[name] = session
        descriptors = await list_server_tools(session)
        new_tools = [MCPTool(session, name, d) for d in descriptors]
        self._tools.extend(new_tools)
        return new_tools

    def get_all_tools(self) -> list[MCPTool]:
        return list(self._tools)

    def get_tools_for_server(self, name: str) -> list[MCPTool]:
        prefix = f"mcp__{name}__"
        return [t for t in self._tools if t.name.startswith(prefix)]

    async def disconnect_all(self):
        await self._stack.aclose()
        self._tools.clear()
        self._sessions.clear()

Integration with ToolRegistry

Drop MCP tools into the existing registry alongside built-in tools.

from tool_system import create_default_tool_registry

async def build_registry_with_mcp() -> tuple[ToolRegistry, MCPServerRegistry]:
    registry = create_default_tool_registry()

    mcp = MCPServerRegistry()
    await mcp.__aenter__()

    await mcp.connect_server(
        "filesystem",
        ["npx", "-y", "@modelcontextprotocol/server-filesystem", "."],
    )
    await mcp.connect_server(
        "github",
        ["npx", "-y", "@modelcontextprotocol/server-github"],
        env={"GITHUB_PERSONAL_ACCESS_TOKEN": os.environ["GITHUB_TOKEN"]},
    )

    for tool in mcp.get_all_tools():
        registry.register(tool)

    # mcp must stay alive for the duration of the agent session
    return registry, mcp


# Usage in your main agent loop:
async def main():
    registry, mcp = await build_registry_with_mcp()
    try:
        await run_agent(registry)
    finally:
        await mcp.disconnect_all()

Common MCP Servers

Set required environment variables before spawning the server process. For example, brave-search requires BRAVE_API_KEY and github requires GITHUB_PERSONAL_ACCESS_TOKEN.

MCP Tools in the Permission System

MCP tools follow the naming convention mcp__<servername>__<toolname>. The permission system treats them identically to built-in tools — the same allow/deny/ask rules apply.

Allow read-only filesystem access only

{
  "permissions": {
    "allow": [
      "mcp__filesystem__read_file",
      "mcp__filesystem__list_directory",
      "mcp__filesystem__get_file_info"
    ],
    "deny": [
      "mcp__filesystem__write_file",
      "mcp__filesystem__create_directory",
      "mcp__filesystem__move_file",
      "mcp__filesystem__delete_file"
    ]
  }
}

Allow all tools from a server with a glob prefix

The matches_rule function in permission-system.md supports prefix matching. Add wildcard support for MCP server names:

def matches_mcp_rule(tool_name: str, rule: str) -> bool:
    """Match rules of the form 'mcp__server' (all tools) or 'mcp__server__tool'."""
    if not rule.startswith("mcp__"):
        return False
    # "mcp__github" matches "mcp__github__create_issue", "mcp__github__list_repos", etc.
    if tool_name.startswith(rule + "__") or tool_name == rule:
        return True
    return fnmatch(tool_name, rule)

Example rule set for a code-review agent:

{
  "permissions": {
    "allow": [
      "mcp__filesystem__read_file",
      "mcp__filesystem__list_directory",
      "mcp__github__get_pull_request",
      "mcp__github__list_pull_request_files",
      "mcp__brave-search__search"
    ],
    "deny": [
      "mcp__github__merge_pull_request",
      "mcp__github__delete_branch"
    ],
    "ask": [
      "mcp__github__create_issue",
      "mcp__github__create_pull_request_review"
    ]
  }
}

Deny all MCP tools globally (and selectively re-enable)

{
  "permissions": {
    "deny": ["mcp__*"],
    "allow": ["mcp__filesystem__read_file", "mcp__filesystem__list_directory"]
  }
}

Note: process allow before deny in your rule matching, or invert the order depending on which should take precedence in your permission system implementation.