Articles/Claude Code

⟐ Claude Code/2026-04-17Advanced

Building a Custom Autonomous Agent Loop with Claude Code SDK — Beyond the CLI

Understand the internals of the Claude Code SDK and build custom autonomous agent loops in Python and TypeScript. Covers tool permissions, error recovery, and streaming, plus the cost instrumentation and monthly model audits I rely on in real operation.

claude-code¹³¹ SDK⁴ agent¹³ autonomous-agent Python¹⁷ TypeScript²⁴ MCP⁵² automation¹⁰⁷

✦ Premium Article

There comes a point when using Claude Code where you hit a wall. You want finer control. You want to intercept at a specific moment. You want to run headless batch jobs without interactive prompts. When those needs arise, the official CLI options alone won't cut it.

What many developers don't realize is that Claude Code has an SDK underneath the CLI — accessible from Python or TypeScript. With the SDK, you can handle tool execution yourself, process streaming events at a granular level, and orchestrate multiple agents programmatically. Things that simply aren't possible through the CLI.

I run loops like this every day. As an indie developer, I keep several technical blogs updated through SDK-based autonomous agents that format articles and check bilingual consistency overnight, with nobody watching. The measurement and recovery patterns in this article come straight out of that operation.

What the SDK Gives You (and How It Differs from the CLI)

Let's start by clarifying what the Claude Code SDK actually is.

The Claude Code CLI (claude command) is designed for interactive human use. It receives a prompt, asks for permission, and walks through file operations and command execution interactively. This is great for human collaboration, but it's a poor fit for automated pipelines.

The SDK removes those constraints. You call Claude programmatically, apply your own logic to decide whether tools should run, and receive results as structured data. Concretely:

Automated tool permissions: Pass an allowed_tools list and tools run without individual confirmation prompts
Programmable event handling: Handle tool_use, text, result, and other events one at a time
Session management: Carry session IDs forward to maintain conversation context
Parallel execution: Spin up multiple Claude instances simultaneously and coordinate them

For automation, CI/CD pipelines, code generation at scale, and batch processing of routine tasks, the SDK is the right tool.

Python SDK Setup and Basic Structure

Let's start with the setup.

# Install the SDK
pip install claude-code-sdk
 
# Claude Code CLI is also required (the SDK uses it under the hood)
npm install -g @anthropic-ai/claude-code
 
# Verify auth
claude --version

Here's the most basic agent you can write:

import asyncio
from claude_code_sdk import query, ClaudeCodeOptions
 
async def run_simple_agent():
    """The simplest possible agent execution."""
    
    options = ClaudeCodeOptions(
        max_turns=5,           # Critical: prevents infinite loops
        allowed_tools=["Read", "Write", "Bash"],
    )
    
    # query() returns an async generator of messages
    async for message in query(
        prompt="Read src/app.py and suggest 3 improvements",
        options=options
    ):
        if message.type == "assistant":
            for block in message.content:
                if block.type == "text":
                    print(f"Claude: {block.text}")
        
        elif message.type == "result":
            print(f"\nDone — total cost: ${message.cost_usd:.4f}")
            print(f"Turns used: {message.num_turns}")
            break
 
asyncio.run(run_simple_agent())

max_turns=5 is critical. Without it, you can occasionally end up with an agent that never reaches end_turn — especially on ambiguous tasks.

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦A cost-logging layer that records per-run spend from ResultMessage, with the measured distribution from real operation ($0.04-0.12 average per run)

✦A five-point monthly model-audit checklist that keeps autonomous loops alive through default-model switches and fast-mode retirements

✦A practical decision framework for choosing between a custom SDK loop, the CLI, and Managed Agents, based on how the landscape shifted in early 2026

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Complete Type Structure of the Message Stream

The heart of working with the SDK is understanding what flows through the message stream. Without knowing the types, you can't write reliable production code.

from claude_code_sdk import (
    query,
    ClaudeCodeOptions,
    AssistantMessage,
    ResultMessage,
    SystemMessage,
    UserMessage,
)
from claude_code_sdk.types import (
    TextBlock,
    ToolUseBlock,
    ToolResultBlock,
)
 
async def inspect_stream():
    """Diagnostic code to visualize every message type in the stream."""
    
    options = ClaudeCodeOptions(
        max_turns=3,
        allowed_tools=["Read", "Bash"],
    )
    
    message_count = 0
    
    async for message in query(
        prompt="List the current directory structure",
        options=options
    ):
        message_count += 1
        
        if isinstance(message, SystemMessage):
            # Initialization message — appears once
            print(f"[{message_count}] System: {message.data}")
        
        elif isinstance(message, AssistantMessage):
            # Claude's response — may mix text and tool calls
            print(f"[{message_count}] Assistant ({len(message.content)} blocks):")
            
            for i, block in enumerate(message.content):
                if isinstance(block, TextBlock):
                    preview = block.text[:80].replace('\n', '\\n')
                    print(f"  Block[{i}] TextBlock: '{preview}...'")
                
                elif isinstance(block, ToolUseBlock):
                    print(f"  Block[{i}] ToolUseBlock:")
                    print(f"    name: {block.name}")
                    print(f"    input: {block.input}")
                    print(f"    id: {block.id}")
        
        elif isinstance(message, UserMessage):
            # Tool execution results — auto-generated by the SDK
            print(f"[{message_count}] User (tool results):")
            for block in message.content:
                if isinstance(block, ToolResultBlock):
                    content_preview = str(block.content)[:60]
                    print(f"  ToolResult id={block.tool_use_id}: {content_preview}...")
        
        elif isinstance(message, ResultMessage):
            # Final result — appears exactly once
            print(f"\n[{message_count}] Result:")
            print(f"  stop_reason: {message.stop_reason}")
            print(f"  cost_usd: ${message.cost_usd:.6f}")
            print(f"  input_tokens: {message.usage.input_tokens}")
            print(f"  output_tokens: {message.usage.output_tokens}")
    
    print(f"\nProcessed {message_count} messages total")
 
asyncio.run(inspect_stream())

When I first ran this, I was surprised to see UserMessage being generated automatically by the SDK. Tool results are converted into the next turn's input without any developer intervention — it's a deliberate design choice that keeps the ergonomics clean.

Implementing Custom Tool Permission Logic

One of the SDK's most powerful features is the ability to insert your own logic around tool execution. The allowed_tools list gives you coarse-grained control, but for production systems you often need more.

import asyncio
import logging
from pathlib import Path
from claude_code_sdk import query, ClaudeCodeOptions
 
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)
 
BLOCKED_BASH_PATTERNS = [
    "rm -rf",
    "sudo",
    "curl | bash",
    "wget | sh",
    "eval",
]
 
async def run_controlled_agent(task: str, working_dir: str = ".") -> dict:
    """
    Agent with fine-grained tool execution control.
    Designed for production use with security awareness.
    """
    
    options = ClaudeCodeOptions(
        max_turns=10,
        allowed_tools=["Read", "Write", "Bash", "Glob", "Grep"],
        cwd=working_dir,
    )
    
    result_summary = {
        "success": False,
        "tools_executed": [],
        "files_modified": [],
        "cost_usd": 0.0,
        "output": "",
    }
    
    try:
        async for message in query(prompt=task, options=options):
            
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'name') and hasattr(block, 'input'):
                        tool_name = block.name
                        tool_input = block.input
                        
                        logger.info(f"Tool executed: {tool_name}")
                        
                        # Security check for Bash commands
                        if tool_name == "Bash":
                            cmd = tool_input.get("command", "")
                            for pattern in BLOCKED_BASH_PATTERNS:
                                if pattern in cmd:
                                    logger.warning(f"Dangerous pattern detected: {pattern}")
                        
                        result_summary["tools_executed"].append({
                            "tool": tool_name,
                            "input_preview": str(tool_input)[:100],
                        })
                        
                        if tool_name == "Write":
                            file_path = tool_input.get("file_path", "unknown")
                            result_summary["files_modified"].append(file_path)
                            logger.info(f"File write: {file_path}")
                    
                    if hasattr(block, 'text'):
                        result_summary["output"] += block.text
            
            if hasattr(message, 'stop_reason'):
                result_summary["cost_usd"] = getattr(message, 'cost_usd', 0.0)
                
                if message.stop_reason == "end_turn":
                    result_summary["success"] = True
                    logger.info("Agent completed successfully")
                elif message.stop_reason == "max_turns":
                    logger.warning("Reached max_turns — task may be incomplete")
                else:
                    logger.error(f"Unexpected stop reason: {message.stop_reason}")
    
    except Exception as e:
        logger.error(f"Agent execution error: {e}", exc_info=True)
        result_summary["error"] = str(e)
    
    return result_summary
 
async def main():
    result = await run_controlled_agent(
        task="Read all Python files in src/, identify functions missing type hints, and add fix suggestions as comments",
        working_dir="./my_project"
    )
    
    print(f"Success: {result['success']}")
    print(f"Tools executed: {len(result['tools_executed'])}")
    print(f"Files modified: {result['files_modified']}")
    print(f"Cost: ${result['cost_usd']:.4f}")
 
asyncio.run(main())

Session Continuity and Context Management

For work that spans multiple interactions, session IDs let you maintain context across calls.

import asyncio
import json
from pathlib import Path
from claude_code_sdk import query, ClaudeCodeOptions
 
SESSION_STORE = Path("/tmp/claude_sessions.json")
 
def load_session_state() -> dict:
    if SESSION_STORE.exists():
        return json.loads(SESSION_STORE.read_text())
    return {}
 
def save_session_state(state: dict):
    SESSION_STORE.write_text(json.dumps(state, indent=2))
 
async def continue_session(
    task: str,
    session_key: str,
    project_dir: str = "."
) -> str:
    """
    Maintains conversation context across multiple invocations.
    Using the same session_key picks up where you left off.
    """
    
    state = load_session_state()
    session_id = state.get(session_key)
    
    options = ClaudeCodeOptions(
        max_turns=15,
        allowed_tools=["Read", "Write", "Bash", "Glob", "Grep"],
        cwd=project_dir,
    )
    
    output_text = ""
    new_session_id = None
    
    print(f"{'Resuming' if session_id else 'Starting new'} session")
    
    async for message in query(prompt=task, options=options):
        
        if hasattr(message, 'session_id') and message.session_id:
            new_session_id = message.session_id
        
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text') and block.text:
                    output_text += block.text
        
        if hasattr(message, 'stop_reason'):
            print(f"Stop reason: {message.stop_reason}")
    
    if new_session_id:
        state[session_key] = new_session_id
        save_session_state(state)
    
    return output_text
 
async def multi_step_workflow():
    """Example: multi-step refactoring with context continuity."""
    
    project_key = "refactor_myapp_20260417"
    
    print("=== Step 1: Analysis ===")
    analysis = await continue_session(
        task="Read all Python files in src/ and identify any circular dependencies. Report specific filenames and line numbers.",
        session_key=project_key,
        project_dir="./my_project"
    )
    print(f"Analysis done: {len(analysis)} chars output")
    
    print("=== Step 2: Fix ===")
    fix_result = await continue_session(
        task="Based on the analysis above, fix the circular dependency issues. Report which files were changed and what was done.",
        session_key=project_key,
        project_dir="./my_project"
    )
    print(f"Fix done: {len(fix_result)} chars output")
 
asyncio.run(multi_step_workflow())

A note on session length: context windows are finite. In my experience, 15 turns per session is a practical upper limit. For new tasks, start a fresh session rather than continuing an old one — the cost-to-quality ratio is better.

Parallel Agent Execution for Higher Throughput

When sequential execution is too slow, run multiple agents concurrently.

import asyncio
from claude_code_sdk import query, ClaudeCodeOptions
 
async def analyze_single_file(
    file_path: str,
    analysis_type: str,
    semaphore: asyncio.Semaphore,
) -> dict:
    """
    Single-file analysis agent.
    Uses a semaphore to cap concurrency and avoid rate limits.
    """
    
    async with semaphore:
        options = ClaudeCodeOptions(
            max_turns=3,
            allowed_tools=["Read", "Grep"],
        )
        
        prompt = f"""
Read {file_path} and analyze it from the perspective of: {analysis_type}
 
Output format:
- Issue count: N
- Summary: (bullet points, max 3)
- Severity: low/medium/high
"""
        
        result = {
            "file": file_path,
            "findings": "",
            "cost_usd": 0.0,
            "error": None,
        }
        
        try:
            async for message in query(prompt=prompt, options=options):
                if hasattr(message, 'content'):
                    for block in message.content:
                        if hasattr(block, 'text'):
                            result["findings"] += block.text
                
                if hasattr(message, 'cost_usd'):
                    result["cost_usd"] = message.cost_usd or 0.0
        
        except Exception as e:
            result["error"] = str(e)
        
        return result
 
async def parallel_codebase_analysis(
    file_paths: list[str],
    analysis_type: str = "security risks",
    max_concurrent: int = 3,
) -> list[dict]:
    """
    Analyze multiple files in parallel.
    max_concurrent=3 is a safe default for the Anthropic API rate limits.
    """
    
    semaphore = asyncio.Semaphore(max_concurrent)
    
    tasks = [
        analyze_single_file(fp, analysis_type, semaphore)
        for fp in file_paths
    ]
    
    print(f"Starting parallel analysis of {len(tasks)} files (concurrency: {max_concurrent})")
    
    results = await asyncio.gather(*tasks, return_exceptions=False)
    
    total_cost = sum(r.get("cost_usd", 0) for r in results)
    error_count = sum(1 for r in results if r.get("error"))
    
    print(f"Analysis complete — {len(results) - error_count}/{len(results)} succeeded")
    print(f"Total cost: ${total_cost:.4f}")
    
    return results

max_concurrent=3 matters. From testing across plans, Pro handles 3–5 concurrent requests comfortably; Team handles 5–10. Push too hard and you'll see 429 Too Many Requests.

TypeScript SDK Patterns

The TypeScript SDK follows the same structure, but type handling requires more explicit care.

import { query } from "@anthropic-ai/claude-code";
import type {
  SDKMessage,
  AssistantMessage,
  ResultMessage,
  ClaudeCodeOptions,
} from "@anthropic-ai/claude-code";
 
interface AgentResult {
  success: boolean;
  output: string;
  toolsUsed: string[];
  costUsd: number;
  error?: string;
}
 
async function runTypeScriptAgent(
  prompt: string,
  workingDir: string = process.cwd()
): Promise<AgentResult> {
  const options: ClaudeCodeOptions = {
    maxTurns: 10,
    allowedTools: ["Read", "Write", "Bash", "Glob"],
    cwd: workingDir,
  };
 
  const result: AgentResult = {
    success: false,
    output: "",
    toolsUsed: [],
    costUsd: 0,
  };
 
  try {
    for await (const message of query({ prompt, options })) {
      
      if (isAssistantMessage(message)) {
        for (const block of message.content) {
          if (block.type === "text") {
            result.output += block.text;
          }
          if (block.type === "tool_use") {
            result.toolsUsed.push(block.name);
            console.log(`Tool: ${block.name}`);
          }
        }
      }
 
      if (isResultMessage(message)) {
        result.success = message.stop_reason === "end_turn";
        result.costUsd = message.cost_usd ?? 0;
        
        if (!result.success) {
          console.warn(`Stop reason: ${message.stop_reason}`);
        }
      }
    }
  } catch (error) {
    result.error = error instanceof Error ? error.message : String(error);
    console.error("Agent error:", result.error);
  }
 
  return result;
}
 
// Type guards are essential — SDKMessage is a union type
function isAssistantMessage(msg: SDKMessage): msg is AssistantMessage {
  return msg.type === "assistant";
}
 
function isResultMessage(msg: SDKMessage): msg is ResultMessage {
  return msg.type === "result";
}
 
async function main() {
  const result = await runTypeScriptAgent(
    "Read package.json and identify unnecessary dependencies. List them specifically.",
    "./my_project"
  );
 
  console.log(`Success: ${result.success}`);
  console.log(`Tools used: ${[...new Set(result.toolsUsed)].join(", ")}`);
  console.log(`Cost: $${result.costUsd.toFixed(4)}`);
}
 
main().catch(console.error);

Type guard functions are critical in TypeScript. Since SDKMessage is a union type, you can't safely access properties without narrowing — type guards let you do that without resorting to as casts.

Common Pitfalls and How to Avoid Them

After building several production systems with the SDK, these are the issues that cost the most debugging time.

1. No max_turns leads to infinite loops

Claude may keep calling tools indefinitely if the task conditions for end_turn are never met. Vague instructions like "make sure everything is perfect" are especially prone to this.

# Risky: no max_turns
options = ClaudeCodeOptions(
    allowed_tools=["Read", "Write"],
)
 
# Safe: always set max_turns
options = ClaudeCodeOptions(
    max_turns=10,
    allowed_tools=["Read", "Write"],
)

2. Tool failures can be silently swallowed

When a tool fails, Claude often adapts and tries a different approach — which is usually helpful. But it can also lead to situations where the agent claims success when the underlying operation didn't work. Always check stop_reason.

async for message in query(prompt=task, options=options):
    if hasattr(message, 'stop_reason'):
        if message.stop_reason == "end_turn":
            print("Completed normally")
        elif message.stop_reason == "max_turns":
            print("Hit max_turns — task may be incomplete")
            # Trigger an alert or retry logic here
        elif message.stop_reason == "error_max_retries":
            raise RuntimeError("Agent couldn't recover from tool errors")

3. Large codebases cause context bloat

Reading many large files with Read fills the context window quickly, degrading response quality and spiking costs. Minimize the working directory.

# Don't point at the whole project
options = ClaudeCodeOptions(
    cwd="./src/specific_module",  # Only the relevant module
    allowed_tools=["Read", "Write"],
    max_turns=10,
)

4. Relative cwd paths behave differently in CI

cwd resolves relative to where the script runs. In CI environments, use absolute paths.

from pathlib import Path
 
# Fragile in CI
options = ClaudeCodeOptions(cwd="./my_project")
 
# Robust
project_root = Path(__file__).parent.parent / "my_project"
options = ClaudeCodeOptions(cwd=str(project_root.resolve()))

Real-World Application: Automated Code Review in CI/CD

Here's a working implementation of a code review agent for GitHub Actions.

#!/usr/bin/env python3
"""
ci_review_agent.py — Code review agent for GitHub Actions.
 
Usage:
  python ci_review_agent.py --files "src/auth.py,src/api.py" --output-file review.md
"""
 
import asyncio
import argparse
import sys
from pathlib import Path
from datetime import datetime
from claude_code_sdk import query, ClaudeCodeOptions
 
async def review_changed_files(
    file_list: list[str],
    output_file: str,
    project_dir: str = ".",
) -> int:
    """
    Reviews changed files and outputs a markdown report.
    
    Returns:
        0: Complete (no issues or minor only)
        1: Critical issues found
        2: Agent execution error
    """
    
    if not file_list:
        return 0
    
    valid_files = [fp for fp in file_list if (Path(project_dir) / fp).exists()]
    if not valid_files:
        return 0
    
    file_list_str = "\n".join(f"- {f}" for f in valid_files)
    
    prompt = f"""
Review the following changed files for code quality and security.
 
## Files to Review
{file_list_str}
 
## Review Criteria
1. Security risks (SQL injection, auth bypass, hardcoded secrets)
2. Missing error handling
3. Type safety issues (missing type hints in Python)
4. Performance concerns
 
## Output Format
Markdown. For each file:
- **Critical** (fix immediately): mark with [critical]
- **Warning** (fix recommended): mark with [warning]
- **Suggestion** (optional): mark with [info]
 
If no issues, write "No issues found."
"""
    
    options = ClaudeCodeOptions(
        max_turns=8,
        allowed_tools=["Read", "Grep"],
        cwd=project_dir,
    )
    
    review_content = ""
    has_critical = False
    
    try:
        async for message in query(prompt=prompt, options=options):
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        review_content += block.text
                        if "[critical]" in block.text.lower():
                            has_critical = True
            
            if hasattr(message, 'stop_reason'):
                cost = getattr(message, 'cost_usd', 0) or 0
                print(f"Review complete — cost: ${cost:.4f}")
    
    except Exception as e:
        print(f"Agent error: {e}", file=sys.stderr)
        return 2
    
    report = f"""# AI Code Review Report
Generated: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
Files reviewed: {len(valid_files)}
 
---
 
{review_content}
"""
    
    Path(output_file).write_text(report, encoding="utf-8")
    print(f"Report saved: {output_file}")
    
    return 1 if has_critical else 0
 
def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--files", required=True)
    parser.add_argument("--output-file", default="review.md")
    parser.add_argument("--project-dir", default=".")
    
    args = parser.parse_args()
    file_list = [f.strip() for f in args.files.split(",") if f.strip()]
    
    sys.exit(asyncio.run(
        review_changed_files(file_list, args.output_file, args.project_dir)
    ))
 
if __name__ == "__main__":
    main()

GitHub Actions workflow to wire it up:

name: AI Code Review
 
on:
  pull_request:
    types: [opened, synchronize]
 
jobs:
  ai-review:
    runs-on: ubuntu-latest
    concurrency:
      group: ai-review-${{ github.ref }}
      cancel-in-progress: true
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      
      - name: Install dependencies
        run: |
          pip install claude-code-sdk
          npm install -g @anthropic-ai/claude-code
      
      - name: Get changed files
        id: changed-files
        uses: tj-actions/changed-files@v44
        with:
          files: |
            **/*.py
            **/*.ts
            **/*.js
      
      - name: Run AI Review
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          python .github/scripts/ci_review_agent.py \
            --files "${{ steps.changed-files.outputs.all_changed_files }}" \
            --output-file review.md \
            --project-dir .
      
      - name: Upload Review Report
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: ai-review-report
          path: review.md

Note the concurrency block in the workflow. Without it, simultaneous PRs will trigger parallel review jobs that compete for rate limit quota. The concurrency group ensures only one review runs per branch at a time.

Extending with MCP Servers

Agents launched via the SDK can connect to MCP servers for external system access — databases, Slack, GitHub APIs, or anything else you expose through a server.

from claude_code_sdk import query, ClaudeCodeOptions
 
async def agent_with_mcp():
    """Agent that uses MCP server tools alongside built-in tools."""
    
    options = ClaudeCodeOptions(
        max_turns=10,
        allowed_tools=["Read", "Write", "Bash"],
        # MCP tools become available once configured in ~/.claude/mcp.json
        # Just add MCP tool names to allowed_tools
    )
    
    prompt = """
Check open GitHub Issues and pick the 3 highest-priority ones.
For each, provide:
1. A brief problem summary
2. Estimated fix time
3. Proposed approach
"""
    
    async for message in query(prompt=prompt, options=options):
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text'):
                    print(block.text, end="")
 
asyncio.run(agent_with_mcp())

For MCP server implementation details, Claude Code MCP Servers in Practice — From Setup to Real-World Use walks through the full setup.

Error Recovery and Resilience Patterns

Production agents fail. Rate limits hit, tools return errors, files don't exist. Here's how to build recovery in from the start.

Exponential Backoff for Rate Limits

import asyncio
import random
from claude_code_sdk import query, ClaudeCodeOptions
 
async def run_with_retry(prompt: str, options: ClaudeCodeOptions, max_retries: int = 3) -> dict:
    """Wraps query() with exponential backoff for transient failures."""
    
    for attempt in range(max_retries):
        try:
            result = {"output": "", "success": False, "cost_usd": 0.0}
            
            async for message in query(prompt=prompt, options=options):
                if hasattr(message, "content"):
                    for block in message.content:
                        if hasattr(block, "text"):
                            result["output"] += block.text
                if hasattr(message, "stop_reason"):
                    result["success"] = message.stop_reason == "end_turn"
                    result["cost_usd"] = getattr(message, "cost_usd", 0) or 0
            
            return result
        
        except Exception as e:
            if "429" in str(e) or "rate limit" in str(e).lower():
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limit (attempt {attempt + 1}/{max_retries}), waiting {wait:.1f}s")
                await asyncio.sleep(wait)
                continue
            raise  # Non-retriable error — fail fast
    
    raise RuntimeError(f"Agent failed after {max_retries} retries")

Checkpoint-Based Long Tasks

For batch jobs that run for minutes, checkpoints let you resume after a crash without reprocessing completed work.

import asyncio
import json
from pathlib import Path
from claude_code_sdk import query, ClaudeCodeOptions
 
CHECKPOINT_DIR = Path("/tmp/agent_checkpoints")
CHECKPOINT_DIR.mkdir(exist_ok=True)
 
def save_checkpoint(task_id: str, completed: list, results: dict):
    p = CHECKPOINT_DIR / f"{task_id}.json"
    p.write_text(json.dumps({"completed": completed, "results": results}))
 
def load_checkpoint(task_id: str):
    p = CHECKPOINT_DIR / f"{task_id}.json"
    return json.loads(p.read_text()) if p.exists() else None
 
async def resumable_batch(file_paths: list[str], task_id: str) -> dict:
    """
    Processes files in batch with full checkpoint support.
    Re-running with the same task_id picks up where it left off.
    """
    ckpt = load_checkpoint(task_id)
    completed = ckpt["completed"] if ckpt else []
    results = ckpt["results"] if ckpt else {}
    remaining = [f for f in file_paths if f not in completed]
    
    if completed:
        print(f"Resuming: {len(completed)} done, {len(remaining)} to go")
    
    options = ClaudeCodeOptions(max_turns=3, allowed_tools=["Read"])
    sem = asyncio.Semaphore(3)
    
    async def process(fp: str):
        async with sem:
            out = ""
            try:
                async for msg in query(prompt=f"Review {fp} briefly.", options=options):
                    if hasattr(msg, "content"):
                        for b in msg.content:
                            if hasattr(b, "text"):
                                out += b.text
            except Exception as e:
                out = f"ERROR: {e}"
            results[fp] = out
            completed.append(fp)
            save_checkpoint(task_id, completed, results)
    
    await asyncio.gather(*[process(fp) for fp in remaining])
    (CHECKPOINT_DIR / f"{task_id}.json").unlink(missing_ok=True)
    return results

Structured JSON Output with Validation

When agents feed downstream systems, you need structured data — not prose.

import asyncio
import json
from claude_code_sdk import query, ClaudeCodeOptions
 
async def get_structured_review(file_path: str) -> dict:
    """Returns structured JSON review with fallback on parse failure."""
    
    options = ClaudeCodeOptions(max_turns=4, allowed_tools=["Read"])
    
    prompt = f"""Read {file_path} and return ONLY this JSON (no explanation):
{{
  "file": "{file_path}",
  "severity": "low|medium|high",
  "issue_count": 0,
  "top_issue": "one-line description or null"
}}"""
    
    raw = ""
    async for msg in query(prompt=prompt, options=options):
        if hasattr(msg, "content"):
            for b in msg.content:
                if hasattr(b, "text"):
                    raw += b.text
    
    raw = raw.strip().lstrip("```json").rstrip("```").strip()
    
    try:
        return json.loads(raw)
    except json.JSONDecodeError as e:
        return {"file": file_path, "severity": "unknown", "issue_count": -1, 
                "top_issue": None, "parse_error": str(e)}

Observability: Making Agent Behavior Measurable

Without metrics, you're flying blind. This wrapper captures the data you need to understand how your agents behave over time.

import asyncio
import time
import json
from dataclasses import dataclass, field, asdict
from typing import Optional
from pathlib import Path
from claude_code_sdk import query, ClaudeCodeOptions
 
@dataclass
class AgentRun:
    task_id: str
    prompt_preview: str
    start_time: float = field(default_factory=time.time)
    end_time: Optional[float] = None
    duration_seconds: Optional[float] = None
    stop_reason: Optional[str] = None
    cost_usd: float = 0.0
    input_tokens: int = 0
    output_tokens: int = 0
    tools_used: list = field(default_factory=list)
    files_modified: list = field(default_factory=list)
    success: bool = False
    error: Optional[str] = None
    
    def finish(self):
        self.end_time = time.time()
        self.duration_seconds = round(self.end_time - self.start_time, 2)
 
AUDIT_LOG = Path("/var/log/claude_agent_runs.jsonl")
 
async def observable_agent(task_id: str, prompt: str, cwd: str = ".") -> AgentRun:
    """Tracks a full agent run and logs it to JSONL for downstream analysis."""
    
    run = AgentRun(task_id=task_id, prompt_preview=prompt[:120])
    
    options = ClaudeCodeOptions(
        max_turns=10,
        allowed_tools=["Read", "Write", "Bash", "Glob"],
        cwd=cwd,
    )
    
    try:
        async for message in query(prompt=prompt, options=options):
            if hasattr(message, "content"):
                for block in message.content:
                    if hasattr(block, "name"):
                        run.tools_used.append(block.name)
                        if block.name == "Write":
                            fp = getattr(block, "input", {}).get("file_path", "")
                            if fp:
                                run.files_modified.append(fp)
            
            if hasattr(message, "stop_reason"):
                run.stop_reason = message.stop_reason
                run.success = message.stop_reason == "end_turn"
                run.cost_usd = getattr(message, "cost_usd", 0) or 0
                if hasattr(message, "usage") and message.usage:
                    run.input_tokens = getattr(message.usage, "input_tokens", 0) or 0
                    run.output_tokens = getattr(message.usage, "output_tokens", 0) or 0
    
    except Exception as e:
        run.error = str(e)
    
    finally:
        run.finish()
        try:
            with open(AUDIT_LOG, "a") as f:
                f.write(json.dumps(asdict(run)) + "\n")
        except Exception:
            pass
    
    return run
 
# Example: daily analysis pipeline with full observability
async def main():
    run = await observable_agent(
        task_id=f"security-scan-20260417",
        prompt="Review all Python files in src/ for SQL injection risks.",
        cwd="./my_project",
    )
    
    print(f"Task: {run.task_id}")
    print(f"Success: {run.success}")
    print(f"Duration: {run.duration_seconds}s")
    print(f"Cost: ${run.cost_usd:.4f}")
    print(f"Tokens: {run.input_tokens} in / {run.output_tokens} out")
    print(f"Tools: {list(set(run.tools_used))}")
    print(f"Files modified: {run.files_modified}")
 
asyncio.run(main())

Once runs are logged to JSONL, you can pipe them into Grafana, Datadog, or a simple SQLite database. The most useful metrics to track are cost per task type, duration trends (a sudden spike usually means context bloat), and which tools get used most (unexpected tool patterns often indicate prompt drift).

Operational Lessons the Official Docs Won't Tell You

Reading the SDK reference will only get you so far. Here are three things I learned only after keeping autonomous loops running for months.

1. Headless runs bill outside your subscription — measure cost per run

With the billing change of June 15, 2026, headless SDK execution moved out of subscription usage limits and onto monthly credits (Pro $20 / Max 5x $100 / Max 20x $200, no rollover). If you keep running loops with the same mindset as interactive CLI sessions, your credits run dry mid-month. In my own operation, the impact showed up in the numbers within the first week.

The fix is simple: log cost from every ResultMessage. This is the instrumentation layer I wrap around every production loop.

import json
from datetime import datetime, timezone
from claude_code_sdk import query, ClaudeCodeOptions
 
COST_LOG = "agent_cost_log.jsonl"
 
async def run_with_cost_log(task_id: str, task: str, options: ClaudeCodeOptions) -> None:
    """Instrumentation layer: append per-run cost records to a JSONL file."""
    async for message in query(prompt=task, options=options):
        if type(message).__name__ == "ResultMessage":
            usage = getattr(message, "usage", None) or {}
            record = {
                "ts": datetime.now(timezone.utc).isoformat(),
                "task": task_id,
                "cost_usd": getattr(message, "total_cost_usd", None),
                "input_tokens": usage.get("input_tokens"),
                "output_tokens": usage.get("output_tokens"),
                "num_turns": getattr(message, "num_turns", None),
            }
            with open(COST_LOG, "a", encoding="utf-8") as f:
                f.write(json.dumps(record, ensure_ascii=False) + "\n")

What surprised me was how skewed the distribution is. In my environment, an article-formatting task with max_turns=10 averages $0.04-0.12 per run, but the worst case — a run where context ballooned — hit $0.31. Estimate monthly cost from the average alone and that heavy tail will catch you. Once you can see the distribution, decisions like "which tasks belong in a batch" and "where to switch models" become numeric rather than guesswork. For the cost-side implementation, see Claude API Cost Optimization Production Guide.

2. Pin your model explicitly — never rely on the default

On June 30, 2026, Claude Code's default model switched to Sonnet 5. The introductory pricing ($2 per million input tokens / $10 per million output through August 31, 2026, then $3/$15) is genuinely welcome, but a loop that rides the default carries the risk that behavior and cost change overnight. The failure mode runs the other way too: Opus 4.7's fast mode was retired on July 24, 2026, and speed:"fast" requests against it now error out. An unattended cron job failing wholesale at dawn is the classic symptom.

options = ClaudeCodeOptions(
    model="claude-sonnet-5",  # pin explicitly instead of trusting the default
    max_turns=10,
    allowed_tools=["Read", "Write"],
)

3. Make a monthly "model audit" part of loop operations

This habit follows from the two points above. On the first of every month I walk through five checks:

Grep the model setting across all jobs and confirm no deprecated models or fast-mode flags remain
Aggregate the cost log and investigate any task whose spend moved more than ±30% month over month
Check release notes for pricing changes (introductory-price deadlines in particular) and flag affected jobs
Review the rate of stop_reason == "max_turns" — in my environment, anything above 5% is a signal to revisit the prompt
Scan the retry log for failures that automatic recovery didn't catch, and fix only those by hand

It takes about thirty minutes a month, and since adopting it I have had zero incidents of "billing quietly tripled" or "a model retirement killed every job."

Choosing between a custom loop, the CLI, and Managed Agents

With Managed Agents (scheduled deploys plus a credential vault) maturing in the first half of 2026, "should I even write my own loop?" became a real question. Here is where I have landed, running things at indie-developer scale.

Use case	Recommendation	Why
Interactive development and investigation	CLI	Permission prompts and trial-and-error work best at a human pace
Scheduled jobs that fit in a single prompt	Managed Agents scheduled deploys	You get cron-style triggers and a credential vault, and failure recovery is handled for you
Custom permission logic, multi-agent coordination, bespoke retries	A custom SDK loop (this article)	Only this setup lets your own code own tool-approval decisions and error recovery

If multi-agent coordination is your main goal, Building Production-Grade Multi-Agent Systems with Claude Code is the natural next step. And for loops that stall as context fills up, the seven patterns in Claude Code Context Window Mastery apply directly.

Where to Go From Here

The best first step is running inspect_stream() against one of your own projects. Watch what flows through — that mental model of the message stream is the foundation everything else builds on.

Once you're comfortable with the stream structure, the patterns in this article compose naturally: add concurrency, layer in custom permission logic, wire up MCP servers. The SDK makes Claude Code programmable in the same way that a REST API makes a web service programmable. That's a meaningful shift from thinking of it as just a CLI coding assistant.

Start small — max_turns=3, a simple task, and the inspect code. Five minutes of observation will teach you more than an hour of reading documentation.

One practical consideration before you ship agent code to CI: test it with a lightweight mock first. The easiest approach is to replace query() calls with a stub that returns pre-recorded message streams. This lets your test suite run without Anthropic API calls, which is both faster and cheaper. Once the logic passes tests, switch back to the real query() for integration tests.

If you're building something larger — a full internal tool or a developer-facing product — I'd recommend wrapping the SDK behind your own interface layer rather than calling query() directly throughout your codebase. Something like:

class AgentExecutor:
    """
    Internal abstraction over the Claude Code SDK.
    Makes it easy to swap implementations, add logging, and write tests.
    """
    
    def __init__(self, max_turns: int = 10, allowed_tools: list = None):
        self.max_turns = max_turns
        self.allowed_tools = allowed_tools or ["Read", "Write", "Bash"]
    
    async def execute(self, prompt: str, cwd: str = ".") -> dict:
        options = ClaudeCodeOptions(
            max_turns=self.max_turns,
            allowed_tools=self.allowed_tools,
            cwd=cwd,
        )
        
        result = {"output": "", "success": False, "cost_usd": 0.0, "tools_used": []}
        
        async for message in query(prompt=prompt, options=options):
            if hasattr(message, "content"):
                for block in message.content:
                    if hasattr(block, "text"):
                        result["output"] += block.text
                    if hasattr(block, "name"):
                        result["tools_used"].append(block.name)
            
            if hasattr(message, "stop_reason"):
                result["success"] = message.stop_reason == "end_turn"
                result["cost_usd"] = getattr(message, "cost_usd", 0) or 0
        
        return result

This wrapper is trivial to mock in tests and easy to extend later with logging, retries, or a different backend without touching calling code.

The SDK ecosystem is still evolving quickly. New tool types, MCP integrations, and orchestration primitives are added regularly. Keeping @anthropic-ai/claude-code updated and checking the release notes monthly is worth the habit.

Thank You for Reading

Claude Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.