Claude Agent Sdk

Claude Agent SDK

The Agent SDK provides Claude with autonomous tool execution. Unlike the Messages API where you implement the tool loop, agents execute tools automatically.

Quick Start

Python:

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    async for message in query(
        prompt="Find and fix the bug in auth.py",
        options=ClaudeAgentOptions(allowed_tools=["Read", "Edit", "Bash"])
    ):
        print(message)

asyncio.run(main())

TypeScript:

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

for await (const message of query({
  prompt: "Find and fix the bug in auth.py",
  options: { allowedTools: ["Read", "Edit", "Bash"] }
})) {
  console.log(message);
}

When to Use Agent SDK vs Messages API

Core APIs

query() - One-off Tasks

Creates new session each time. Best for independent tasks.

async for message in query(
    prompt="Your task here",
    options=ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Edit", "Bash"],
        permission_mode="acceptEdits"  # Auto-approve edits
    )
):
    if hasattr(message, 'result'):
        print(message.result)

ClaudeSDKClient - Conversations

Maintains context across exchanges. Use for interactive applications.

async with ClaudeSDKClient(options=options) as client:
    await client.connect()                    # Establish connection
    await client.query("Read the auth module")
    async for msg in client.receive_messages():  # Stream all messages
        print(msg)

    await client.query("Now refactor it")     # Remembers context
    async for msg in client.receive_messages():
        print(msg)

    await client.interrupt()                  # Cancel current operation
    await client.rewind_files()               # Restore file checkpoints
    status = await client.get_mcp_status()    # Check MCP server status
    await client.disconnect()                 # Clean up

Built-in Tools

Custom Tools

Custom tools use in-process MCP servers. Tool names follow: mcp__{server}__{tool}

Python:

from claude_agent_sdk import tool, create_sdk_mcp_server

@tool("get_weather", "Get temperature for coordinates",
      {"latitude": float, "longitude": float})
async def get_weather(args):
    # Fetch weather data
    return {"content": [{"type": "text", "text": f"72°F"}]}

server = create_sdk_mcp_server(
    name="weather",
    version="1.0.0",
    tools=[get_weather]
)

options = ClaudeAgentOptions(
    mcp_servers={"weather": server},
    allowed_tools=["mcp__weather__get_weather"]
)

TypeScript:

import { tool, createSdkMcpServer } from "@anthropic-ai/claude-agent-sdk";
import { z } from "zod";

const server = createSdkMcpServer({
  name: "weather",
  version: "1.0.0",
  tools: [
    tool("get_weather", "Get temperature", {
      latitude: z.number(),
      longitude: z.number()
    }, async (args) => ({
      content: [{ type: "text", text: "72°F" }]
    }))
  ]
});

Important: Custom MCP tools require streaming input mode (async generator).

See custom-tools.md for complete patterns.

Hooks System

Hooks intercept agent execution at lifecycle points.

*PostToolUseFailure: Present in Python SDK types.py (HookEvent Literal) but not yet in the official HookInput union type or documentation. May work in recent SDK versions (>= ~0.1.26). Test with your specific version before relying on it.

Permission Decisions

Hooks can return permission decisions (checked in order):

1. deny - Block immediately
2. ask - Require user approval
3. allow - Auto-approve

Example: Block .env modifications

async def protect_env(input_data, tool_use_id, context):
    file_path = input_data['tool_input'].get('file_path', '')
    if file_path.endswith('.env'):
        return {
            'hookSpecificOutput': {
                'hookEventName': 'PreToolUse',
                'permissionDecision': 'deny',
                'permissionDecisionReason': 'Cannot modify .env files'
            }
        }
    return {}

options = ClaudeAgentOptions(
    hooks={'PreToolUse': [HookMatcher(matcher='Write|Edit', hooks=[protect_env])]}
)

See hooks-reference.md for all hooks and patterns.

Subagents

Spawn specialized agents for focused subtasks with parallelization.

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Glob", "Grep", "Task"],
    agents={
        "code-reviewer": AgentDefinition(
            description="Expert code reviewer for quality and security",
            prompt="Analyze code quality and suggest improvements",
            tools=["Read", "Glob", "Grep"],
            model="sonnet"  # "sonnet" | "opus" | "haiku" | "inherit"
        ),
        "test-writer": AgentDefinition(
            description="Test generation specialist",
            prompt="Write comprehensive unit tests",
            tools=["Read", "Write", "Bash"],
            model="sonnet"
        )
    }
)

See multi-agent.md for orchestration patterns.

Session Management

Resume sessions to maintain context across queries:

session_id = None

# First query: capture session ID
async for message in query(prompt="Read auth.py", options=options):
    if hasattr(message, 'subtype') and message.subtype == 'init':
        session_id = message.session_id

# Resume with full context
async for message in query(
    prompt="Find all callers",  # "auth.py" understood from context
    options=ClaudeAgentOptions(resume=session_id)
):
    print(message)

Configuration Options

ClaudeAgentOptions(
    allowed_tools=["Read", "Edit"],      # Whitelist tools
    permission_mode="acceptEdits",        # default|acceptEdits|plan|bypassPermissions
    mcp_servers={"name": server},         # External MCP servers
    cwd="/path/to/project",               # Working directory
    max_turns=10,                         # Prevent infinite loops
    resume="session-id",                  # Resume session
    hooks={...},                          # Lifecycle hooks
    agents={...},                         # Subagent definitions
    can_use_tool=permission_handler,      # Programmatic permission callback (requires streaming input)
    model="claude-sonnet-4-5-20250929",   # Model selection (or "claude-sonnet-4-5", "claude-opus-4-5", "claude-haiku-4-5")
    fallback_model="claude-haiku-4-5",    # Fallback if primary model unavailable
    system_prompt="Custom instructions",  # Override system prompt
    max_budget_usd=1.0,                   # Maximum spend limit
    max_thinking_tokens=10000,            # Extended thinking budget
    enable_file_checkpointing=True,       # Enable file state snapshots for rewind
    output_format={"type": "json", "schema": {...}},  # Structured JSON output
    plugins=[...],                        # Plugin extensions
    sandbox=True,                         # Run in sandboxed environment
    setting_sources=[...],               # Custom setting sources
    env={"API_KEY": "..."}                # Environment variables
)

Security Best Practices

1. Deny-by-default: Only whitelist necessary tools
2. Confirm sensitive actions: Use hooks for git push, infrastructure changes
3. Block dangerous commands: rm -rf, sudo, etc.
4. Protect secrets: Never expose credentials in agent-visible context
5. Use short-lived tokens: Rotate credentials aggressively
6. Log all tool invocations: Audit trail with timestamps

Error Handling

from claude_agent_sdk import (
    CLINotFoundError,      # Claude Code not installed
    CLIConnectionError,    # Connection issues
    ProcessError,          # Process failed
    CLIJSONDecodeError     # JSON parsing issues
)

try:
    async for message in query(prompt="Task", options=options):
        pass
except ProcessError as e:
    print(f"Failed with exit code: {e.exit_code}")

Message Types

from claude_agent_sdk import ResultMessage, AssistantMessage

# Stream yields different message types
async for message in query(prompt="Task", options=options):
    if isinstance(message, AssistantMessage):
        print(message.content)
    elif isinstance(message, ResultMessage):
        if message.subtype == "success":
            print(f"Completed: {message.result}")
        print(f"Cost: ${message.total_cost_usd}")
        break  # IMPORTANT: query() does not auto-terminate after ResultMessage

Authentication

• Anthropic API: export ANTHROPIC_API_KEY=your-key
• Amazon Bedrock: CLAUDE_CODE_USE_BEDROCK=1 + AWS credentials
• Google Vertex AI: CLAUDE_CODE_USE_VERTEX=1 + GCP credentials
• Microsoft Azure AI Foundry: CLAUDE_CODE_USE_FOUNDRY=1 + Azure credentials

OpenRouter

Route Agent SDK requests through OpenRouter for access to multiple model providers, fallbacks, and cost optimization:

export ANTHROPIC_BASE_URL="https://openrouter.ai/api"
export ANTHROPIC_AUTH_TOKEN="$OPENROUTER_API_KEY"
export ANTHROPIC_API_KEY=""  # Must be explicitly empty

The Agent SDK inherits Claude Code's model override environment variables, so you can route to different OpenRouter models:

# Route to specific models via OpenRouter
export ANTHROPIC_DEFAULT_SONNET_MODEL="anthropic/claude-sonnet-4"
export ANTHROPIC_DEFAULT_OPUS_MODEL="anthropic/claude-opus-4"

Important: ANTHROPIC_API_KEY must be set to an empty string — if omitted entirely, the SDK will fail with an authentication error. The actual auth is handled by ANTHROPIC_AUTH_TOKEN.

Structured Outputs

Use output_format for validated JSON responses:

options = ClaudeAgentOptions(
    output_format={
        "type": "json",
        "schema": {
            "type": "object",
            "properties": {
                "summary": {"type": "string"},
                "issues": {"type": "array", "items": {"type": "string"}},
                "severity": {"type": "string", "enum": ["low", "medium", "high"]}
            },
            "required": ["summary", "issues", "severity"]
        }
    }
)

File Checkpointing

Enable file state snapshots for safe rollback:

options = ClaudeAgentOptions(
    enable_file_checkpointing=True,
    allowed_tools=["Read", "Edit", "Write"]
)

# Later, rewind all file changes made by the agent
async with ClaudeSDKClient(options=options) as client:
    await client.connect()
    await client.query("Refactor the auth module")
    async for msg in client.receive_messages():
        pass
    # If results are unsatisfactory:
    await client.rewind_files()  # Restores all files to pre-query state

Extended Thinking

Enable deeper reasoning with thinking token budgets:

options = ClaudeAgentOptions(
    max_thinking_tokens=10000,  # Budget for extended thinking
    allowed_tools=["Read", "Glob", "Grep"]
)

TypeScript Query Object Methods

The TypeScript SDK's Query object (returned by query()) exposes additional control methods:

const q = query({ prompt: "Analyze codebase", options });

// Control methods
await q.interrupt();              // Cancel current operation
await q.rewindFiles();            // Restore file checkpoints
await q.setPermissionMode("acceptEdits");
await q.setModel("claude-opus-4-5");
await q.setMaxThinkingTokens(10000);

// Informational methods
const commands = await q.supportedCommands();
const models = await q.supportedModels();
const mcpStatus = await q.mcpServerStatus();
const account = await q.accountInfo();

Known Issues & Troubleshooting

In-process MCP servers unreliable with subagents

create_sdk_mcp_server() can fail with "Stream closed" when used with parallel subagents due to message-queue backpressure. The SDK's internal transport gets blocked when multiple subagents compete for the MCP connection.

Recommendation: Use stdio subprocess MCP servers for production multi-agent setups. In-process servers are fine for single-agent simple use cases. (Python SDK #425, TS SDK #41)

acceptEdits does NOT auto-approve MCP tools

The acceptEdits permission mode only auto-approves file operations (Edit, Write) and filesystem Bash commands (mkdir, rm, mv, cp). MCP tools still require explicit permission resolution.

Options for MCP tool approval:

• bypassPermissions - Approves everything (use with caution)
• can_use_tool callback - Programmatic per-tool decisions (requires streaming input mode)
• PreToolUse hooks returning allow - Most reliable for headless agents

"Stream closed" in headless mode

Without a can_use_tool callback or explicit PreToolUse allow decisions, tools requiring permission trigger an interactive prompt. When running the SDK without a terminal (headless/server mode), this fails with "Stream closed" because there is no UI to display the prompt.

Workaround: Use PreToolUse hooks to return explicit allow/deny decisions for all tools that would otherwise trigger a permission prompt. See hooks-reference.md for permission interaction details.

query() does NOT auto-terminate after ResultMessage

The async for loop over query() does not end when ResultMessage arrives. Internally, query() waits for an "end" event from _read_messages(), which only fires after the transport closes. You must break explicitly after ResultMessage.

With a plain string prompt this may appear to work (the stream eventually closes), but with streaming input (async generator + stream_done pattern), omitting break causes a circular deadlock: the loop waits for the stream to close → the stream waits for the generator to finish → the generator waits for stream_done.set() in finally → finally only runs after the loop exits.

# CORRECT
async for message in query(prompt=generate_messages(), options=options):
    if isinstance(message, ResultMessage):
        result = message
        break  # Required — prevents deadlock

See custom-tools.md Known Issues for the full stream_done pattern.

Generator exhaustion causes "Stream closed"

When using query() with an async generator prompt, the SDK closes stdin after the generator finishes yielding + a 60-second timeout. For long-running agents, this causes "Stream closed" errors on hooks and can_use_tool callbacks even when permissions are correctly configured.

Fix: Keep the generator alive with an asyncio.Event until the agent completes. See custom-tools.md Known Issues for the full stream_done pattern.

Background subagents can't access MCP tools

Subagents spawned with run_in_background: true silently fail when calling MCP tools. The background process doesn't inherit MCP server connections from the parent agent. Use foreground subagents for MCP-dependent tasks. (Claude Code #13254)

can_use_tool empty arguments bug

When using the can_use_tool callback, always pass updated_input=tool_input (the original arguments) in PermissionResultAllow. Omitting updated_input causes the tool to receive empty arguments. (Python SDK #320)

Reference Files

• Python Patterns - Complete Python examples
• TypeScript Patterns - Complete TypeScript examples
• Hooks Reference - All hooks with examples
• Custom Tools - Tool creation guide
• Multi-Agent Patterns - Orchestration patterns