Claude Agent SDK

# Python
pip install claude-agent-sdk

# Direct API
export ANTHROPIC_API_KEY=your-api-key

# Amazon Bedrock
export CLAUDE_CODE_USE_BEDROCK=1
# + configure AWS credentials

# Claude Platform on AWS
export CLAUDE_CODE_USE_ANTHROPIC_AWS=1
export ANTHROPIC_AWS_WORKSPACE_ID=your-workspace-id
# + configure AWS credentials

# Google Vertex AI
export CLAUDE_CODE_USE_VERTEX=1
# + configure Google Cloud credentials

# Microsoft Azure AI Foundry
export CLAUDE_CODE_USE_FOUNDRY=1
# + configure Azure credentials

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())

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Edit", "Bash"],   # Auto-approve these tools
    disallowed_tools=["Write"],               # Block these tools entirely
    permission_mode="acceptEdits",            # How ungoverned tools are handled
    model="claude-sonnet-4-6",               # Pin a specific model
    max_turns=30,                             # Max tool-use round trips
    max_budget_usd=1.00,                      # Max spend before stopping
    effort="high",                            # Reasoning depth
    setting_sources=["project"],              # Load CLAUDE.md, skills, hooks
    resume="session-id-here",                 # Resume a specific past session
    continue_conversation=True,               # Resume the most recent session
    fork_session=True,                        # Fork instead of overwrite
    hooks={...},                              # Lifecycle callbacks
    mcp_servers={...},                        # MCP server connections
    agents={...},                             # Custom subagent definitions
    system_prompt="...",                      # Append to base system prompt
)

from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, ResultMessage, SystemMessage
)

async for message in query(prompt="Summarize this project"):
    if isinstance(message, SystemMessage) and message.subtype == "init":
        session_id = message.data["session_id"]

    elif isinstance(message, AssistantMessage):
        print(f"Turn: {len(message.content)} content blocks")

    elif isinstance(message, ResultMessage):
        if message.subtype == "success":
            print(message.result)
        else:
            print(f"Stopped: {message.subtype}")
        print(f"Cost: ${message.total_cost_usd:.4f}")

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, ResultMessage

async def main():
    options = ClaudeAgentOptions(allowed_tools=["Read", "Edit", "Glob"])

    async with ClaudeSDKClient(options=options) as client:
        # First query — creates a new session
        await client.query("Analyse the auth module")
        async for message in client.receive_response():
            if isinstance(message, ResultMessage):
                print(message.result)

        # Second query — automatically continues the same session
        await client.query("Now refactor it to use JWT")
        async for message in client.receive_response():
            if isinstance(message, ResultMessage):
                print(message.result)

asyncio.run(main())

session_id = None

# Step 1: Capture the session ID
async for message in query(
    prompt="Analyse the auth module and suggest improvements",
    options=ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep"]),
):
    if isinstance(message, ResultMessage):
        session_id = message.session_id

# Step 2: Resume with a follow-up prompt
async for message in query(
    prompt="Now implement the refactoring you suggested",
    options=ClaudeAgentOptions(
        resume=session_id,
        allowed_tools=["Read", "Edit", "Write"],
    ),
):
    if isinstance(message, ResultMessage) and message.subtype == "success":
        print(message.result)

# Fork from session_id to explore OAuth2 instead of JWT
forked_id = None
async for message in query(
    prompt="Instead of JWT, implement OAuth2 for the auth module",
    options=ClaudeAgentOptions(resume=session_id, fork_session=True),
):
    if isinstance(message, ResultMessage):
        forked_id = message.session_id  # New ID for the fork

# Original session_id still intact — resume it to continue the JWT thread
async for message in query(
    prompt="Continue with the JWT approach",
    options=ClaudeAgentOptions(resume=session_id),
):
    ...

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

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

async def redirect_to_sandbox(input_data, tool_use_id, context):
    if input_data["tool_name"] == "Write":
        original_path = input_data["tool_input"].get("file_path", "")
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "allow",
                "updatedInput": {
                    **input_data["tool_input"],
                    "file_path": f"/sandbox{original_path}",
                },
            }
        }
    return {}

from datetime import datetime

async def log_file_change(input_data, tool_use_id, context):
    file_path = input_data.get("tool_input", {}).get("file_path", "unknown")
    with open("./audit.log", "a") as f:
        f.write(f"{datetime.now()}: modified {file_path}\n")
    return {}

options = ClaudeAgentOptions(
    permission_mode="acceptEdits",
    hooks={
        "PostToolUse": [HookMatcher(matcher="Edit|Write", hooks=[log_file_change])]
    },
)

import asyncio, json, urllib.request

async def notify_slack(input_data, tool_use_id, context):
    data = json.dumps({"text": f"Agent: {input_data.get('message', '')}"}).encode()
    req = urllib.request.Request(
        "https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
        data=data, headers={"Content-Type": "application/json"}, method="POST",
    )
    try:
        await asyncio.to_thread(urllib.request.urlopen, req)
    except Exception as e:
        print(f"Slack notification failed: {e}")
    return {}

options = ClaudeAgentOptions(
    hooks={"Notification": [HookMatcher(hooks=[notify_slack])]}
)

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(hooks=[rate_limiter]),        # 1. Check rate limits
            HookMatcher(hooks=[authorization_check]), # 2. Verify permissions
            HookMatcher(hooks=[input_sanitizer]),     # 3. Sanitize inputs
            HookMatcher(hooks=[audit_logger]),        # 4. Log the action
        ]
    }
)

hooks={
    "PreToolUse": [
        HookMatcher(matcher="Write|Edit|Delete", hooks=[file_security_hook]),  # File tools
        HookMatcher(matcher="^mcp__", hooks=[mcp_audit_hook]),                 # All MCP tools
        HookMatcher(hooks=[global_logger]),                                    # Everything
    ]
}

from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Glob", "Grep", "Agent"],  # Include "Agent" to enable subagents
    agents={
        "code-reviewer": AgentDefinition(
            description="Expert code reviewer for quality and security reviews.",
            prompt="Analyse code quality and suggest improvements. Focus on security, performance and maintainability.",
            tools=["Read", "Glob", "Grep"],
        ),
        "test-writer": AgentDefinition(
            description="Writes comprehensive unit tests for Python functions.",
            prompt="Write pytest unit tests with good coverage. Include edge cases and error conditions.",
            tools=["Read", "Write", "Bash"],
        ),
    },
)

async for message in query(
    prompt="Use the code-reviewer agent to review auth.py, then use test-writer to add tests",
    options=options,
):
    if hasattr(message, "result"):
        print(message.result)

async def subagent_tracker(input_data, tool_use_id, context):
    print(f"[SUBAGENT] Completed: {input_data['agent_id']}")
    print(f"  Transcript: {input_data['agent_transcript_path']}")
    return {}

options = ClaudeAgentOptions(
    hooks={"SubagentStop": [HookMatcher(hooks=[subagent_tracker])]}
)

from claude_agent_sdk import query, ClaudeAgentOptions

async def get_database_schema(table_name: str) -> str:
    """Return the schema for a database table."""
    # Your implementation here
    return f"CREATE TABLE {table_name} (id INT PRIMARY KEY, name VARCHAR(255));"

options = ClaudeAgentOptions(
    tools=[get_database_schema],
    allowed_tools=["get_database_schema", "Read", "Bash"],
)

async for message in query(
    prompt="Analyse the users table schema and suggest indexes",
    options=options,
):
    if hasattr(message, "result"):
        print(message.result)

options = ClaudeAgentOptions(
    mcp_servers={
        # Browser automation
        "playwright": {
            "command": "npx",
            "args": ["@playwright/mcp@latest"]
        },
        # File system over network
        "filesystem": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/files"]
        },
        # SQLite database
        "sqlite": {
            "command": "uvx",
            "args": ["mcp-server-sqlite", "--db-path", "./data.db"]
        },
    }
)

options = ClaudeAgentOptions(
    allowed_tools=["ToolSearch"],  # Let Claude discover tools as needed
    mcp_servers={...},
)

# Allow only npm commands via Bash
allowed_tools=["Bash(npm:*)"]

# Allow read-only filesystem operations
allowed_tools=["Read", "Glob", "Grep"]

# Block specific paths
disallowed_tools=["Bash(rm -rf:*)"]

async def approve_tool(tool_name, tool_input):
    print(f"Claude wants to run: {tool_name}")
    print(f"Arguments: {tool_input}")
    decision = input("Allow? (y/n): ").strip().lower()
    return decision == "y"

options = ClaudeAgentOptions(
    permission_mode="default",
    # approval_callback=approve_tool  # Framework-specific
)

## Summary instructions

When summarising this conversation, always preserve:

- The current task objective and acceptance criteria
- File paths that have been read or modified
- Test results and error messages
- Decisions made and the reasoning behind them

async def archive_transcript(input_data, tool_use_id, context):
    trigger = input_data.get("trigger")  # "manual" or "auto"
    # Save full transcript to your storage before it gets summarised
    await save_transcript_to_s3(input_data["session_id"])
    return {}

options = ClaudeAgentOptions(
    hooks={"PreCompact": [HookMatcher(hooks=[archive_transcript])]}
)

options = ClaudeAgentOptions(
    setting_sources=["project"],   # Load from .claude/ in working dir
    # setting_sources=["global"],  # Load from ~/.claude/ only
    # setting_sources=["project", "global"],  # Both
)

# Project Context

This is a Python FastAPI application. Always:

- Use type hints on all functions
- Write docstrings for public APIs
- Follow PEP 8 naming conventions
- Run `pytest` before declaring code complete

## Summary instructions

When summarising, always preserve: current task, files modified, test results.

.claude/
  skills/
    deploy/
      SKILL.md         ← Deployment workflow instructions
    code-review/
      SKILL.md         ← Code review checklist and standards
      templates/
        review.md      ← Review template

# Trigger a custom slash command
async for message in query(prompt="/deploy staging"):
    ...

# Trigger compaction
async for message in query(prompt="/compact"):
    ...

options = ClaudeAgentOptions(
    plugins=[
        {
            "name": "my-plugin",
            "mcpServers": {"my-db": {"command": "uvx", "args": ["my-db-server"]}},
            "agents": {"my-agent": {"description": "...", "prompt": "..."}},
        }
    ]
)

options = ClaudeAgentOptions(
    include_partial_messages=True,  # Enable StreamEvent messages
    allowed_tools=["Read", "Edit"],
)

async for message in query(prompt="Refactor utils.py", options=options):
    if hasattr(message, "type") and message.type == "content_block_delta":
        print(message.delta.text, end="", flush=True)  # Live text output

async for message in query(prompt="Deploy the application", options=options):
    if hasattr(message, "tool_name") and message.tool_name == "AskUserQuestion":
        # Present choices to the user and respond
        user_choice = present_choices_to_user(message.choices)
        # Framework-specific: respond with the user's choice

from pydantic import BaseModel, Field
from claude_agent_sdk import query, ClaudeAgentOptions

class BugReport(BaseModel):
    file_path: str = Field(description="Path to the file containing the bug")
    line_number: int = Field(description="Line number where the bug occurs")
    description: str = Field(description="Description of the bug")
    severity: str = Field(description="Severity: critical, high, medium, or low")
    suggested_fix: str = Field(description="Suggested code fix")

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Glob", "Grep"],
    response_format=BugReport,
)

async for message in query(prompt="Find the most critical bug in auth.py", options=options):
    if hasattr(message, "structured_response"):
        report = message.structured_response
        print(f"Bug at {report.file_path}:{report.line_number}")
        print(f"Severity: {report.severity}")
        print(f"Fix: {report.suggested_fix}")

from claude_agent_sdk import query, ClaudeAgentOptions

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Edit", "Write", "Bash"],
    # checkpointing=True  # Framework-specific; creates snapshots before each tool
)

# After the agent runs, you can revert to any prior snapshot if needed

async for message in query(prompt="Refactor the API layer", options=options):
    if isinstance(message, ResultMessage):
        if message.total_cost_usd is not None:
            print(f"Total cost: ${message.total_cost_usd:.4f}")
        print(f"Turns used: {message.num_turns}")
        print(f"Session: {message.session_id}")

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)

# Agent runs are now traced automatically
async for message in query(prompt="Fix the tests", options=options):
    ...

async def track_todos(input_data, tool_use_id, context):
    if input_data.get("tool_name") == "TodoWrite":
        todos = input_data["tool_input"].get("todos", [])
        completed = [t for t in todos if t.get("status") == "completed"]
        print(f"Progress: {len(completed)}/{len(todos)} tasks done")
    return {}

# Example: FastAPI wrapper around the SDK
from fastapi import FastAPI
from claude_agent_sdk import query, ClaudeAgentOptions

app = FastAPI()

@app.post("/agent/run")
async def run_agent(prompt: str, session_id: str | None = None):
    results = []
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Glob", "Grep"],
        resume=session_id,
        max_budget_usd=0.50,
    )
    async for message in query(prompt=prompt, options=options):
        if hasattr(message, "result"):
            results.append(message.result)
    return {"results": results}

# Client SDK — you implement the loop
response = client.messages.create(...)
while response.stop_reason == "tool_use":
    result = your_tool_executor(response.tool_use)
    response = client.messages.create(tool_result=result, **params)

# Agent SDK — Claude handles it
async for message in query(prompt="Fix the bug in auth.py"):
    print(message)

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Glob", "Grep"],
    permission_mode="dontAsk",
    effort="high",
    response_format=CodeReviewReport,
    max_budget_usd=0.50,
)

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Glob", "Grep", "Bash"],
    permission_mode="default",
    hooks={
        "PreToolUse": [HookMatcher(matcher="Edit|Write", hooks=[human_approval_hook])]
    },
)

options = ClaudeAgentOptions(
    allowed_tools=["WebSearch", "WebFetch"],
    permission_mode="dontAsk",
    max_turns=20,
    effort="medium",
)

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Agent"],
    agents={
        "data-collector": AgentDefinition(
            description="Collects raw data from sources",
            prompt="Gather all relevant data. Be thorough.",
            tools=["WebSearch", "WebFetch", "Read"],
        ),
        "analyst": AgentDefinition(
            description="Analyses collected data",
            prompt="Produce key insights. Be concise.",
            tools=["Read"],
        ),
    },
)

import sys

session_id = None
async for message in query(
    prompt="Run the test suite and fix any failing tests",
    options=ClaudeAgentOptions(
        allowed_tools=["Read", "Edit", "Bash"],
        permission_mode="bypassPermissions",  # Safe in CI container
        max_turns=50,
        max_budget_usd=2.00,
        effort="high",
    ),
):
    if isinstance(message, ResultMessage):
        session_id = message.session_id
        if message.subtype != "success":
            print(f"Agent failed: {message.subtype}")
            sys.exit(1)
        print(message.result)

# Single query
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async for message in query(
    prompt="Your task here",
    options=ClaudeAgentOptions(
        allowed_tools=["Read", "Edit", "Bash"],
        permission_mode="acceptEdits",
        model="claude-sonnet-4-6",
        max_turns=30,
        max_budget_usd=1.00,
        effort="high",
        setting_sources=["project"],
        resume="session-id",          # Resume specific session
        continue_conversation=True,   # Resume most recent session
        fork_session=True,            # Fork instead of modify
        hooks={...},
        mcp_servers={...},
        agents={...},
        system_prompt="...",
    ),
):
    if isinstance(message, ResultMessage):
        if message.subtype == "success":
            print(message.result)
        print(f"Cost: ${message.total_cost_usd:.4f}")
        print(f"Session: {message.session_id}")

# Multi-turn (Python)
from claude_agent_sdk import ClaudeSDKClient

async with ClaudeSDKClient(options=options) as client:
    await client.query("First prompt")
    async for msg in client.receive_response():
        ...
    await client.query("Follow-up prompt")   # Continues same session
    async for msg in client.receive_response():
        ...

# Session utilities
from claude_agent_sdk import list_sessions, get_session_messages, get_session_info

sessions = await list_sessions()
messages = await get_session_messages(session_id)
info = await get_session_info(session_id)