Skip to content

Latest commit

 

History

History
1133 lines (901 loc) · 33 KB

claude-agent-sdk-tutorial.md

File metadata and controls

1133 lines (901 loc) · 33 KB

Claude Agent SDK Tutorial

A practical guide to building AI agents and applying them to real-world workflows.


Table of Contents

  1. What Is the Claude Agent SDK?
  2. Installation & Setup
  3. Core Concepts
  4. Building Your First Agent
  5. Working with Tools
  6. Creating Custom Tools
  7. Multi-Agent Orchestration
  8. Guardrails & Permissions
  9. Streaming Responses
  10. Session & Context Management
  11. Applying the SDK to Workflows
  12. Best Practices
  13. Reference

1. What Is the Claude Agent SDK?

The Claude Agent SDK is a framework from Anthropic that lets you build AI agents programmatically using Python or TypeScript. It exposes the same tools, agent loop, and context management that power Claude Code as a library you can embed in your own applications.

What agents built with this SDK can do:

  • Read, write, and edit files autonomously
  • Execute terminal commands
  • Search the web and fetch content
  • Spawn specialized sub-agents for focused tasks
  • Integrate with external APIs via MCP (Model Context Protocol)
  • Maintain conversational context across multiple turns

2. Installation & Setup

Python

pip install claude-agent-sdk

TypeScript

npm install @anthropic-ai/claude-agent-sdk

Authentication

Set your Anthropic API key as an environment variable:

export ANTHROPIC_API_KEY=your-api-key

Get your API key at platform.claude.com.

Cloud provider alternatives:

Provider Environment Variable
AWS Bedrock CLAUDE_CODE_USE_BEDROCK=1
Google Vertex AI CLAUDE_CODE_USE_VERTEX=1
Microsoft Azure Foundry CLAUDE_CODE_USE_FOUNDRY=1

Each requires its respective cloud credentials to be configured.


3. Core Concepts

Agents

An agent is an autonomous loop that:

  1. Gathers context - reads files, searches code, queries data
  2. Takes action - executes tools like file edits or commands
  3. Verifies work - checks if the action achieved the goal
  4. Repeats - iterates until the task is complete

Tools

Tools are capabilities the agent can invoke. The SDK provides built-in tools and supports custom ones via MCP.

Subagents

Specialized agent instances spawned by a main agent to handle focused subtasks. They run in isolated contexts with restricted tool access.

Guardrails

Permission controls that determine what tools an agent can use and under what conditions. Implemented through permission modes, tool allowlists, and custom permission callbacks.


4. Building Your First Agent

One-Shot Query (Python)

The simplest way to run an agent. Send a prompt, get results.

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Glob", "Grep"],
        cwd="/path/to/your/project",
        permission_mode="bypassPermissions"
    )

    async for message in query(
        prompt="Find all TODO comments in Python files and create a summary",
        options=options
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

One-Shot Query (TypeScript)

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

async function main() {
  for await (const message of query({
    prompt: "Find all TODO comments in Python files and create a summary",
    options: {
      allowedTools: ["Read", "Glob", "Grep"],
      cwd: "/path/to/your/project"
    }
  })) {
    if ("result" in message) {
      console.log(message.result);
    }
  }
}

main();

Multi-Turn Conversation (Python)

For tasks that require back-and-forth interaction, use ClaudeSDKClient:

import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock
)

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Edit", "Bash"],
        permission_mode="acceptEdits"
    )

    async with ClaudeSDKClient(options=options) as client:
        # First turn
        await client.query("Create a Python web server")
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # Follow-up — Claude remembers context from the first turn
        await client.query("Add authentication to the server")
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

asyncio.run(main())

5. Working with Tools

Built-in Tools

Tool Purpose Key Parameters
Read Read files from the filesystem file_path, offset, limit
Write Create new files file_path, content
Edit Modify existing files (find-and-replace) file_path, old_string, new_string
Bash Execute terminal commands command, timeout
Glob Find files by pattern pattern, path
Grep Search file contents with regex pattern, path, glob
WebSearch Search the internet query
WebFetch Fetch and analyze web page content url, prompt
Task Spawn subagents description, prompt, subagent_type
AskUserQuestion Request user input questions

Tool Use Patterns

Single tool — the agent determines one tool solves the problem:

async for message in query(
    prompt="How many lines are in README.md?",
    options=ClaudeAgentOptions(allowed_tools=["Read"])
):
    print(message)

Parallel tools — independent tools execute simultaneously:

async for message in query(
    prompt="Get the latest news, stock prices, and weather",
    options=ClaudeAgentOptions(allowed_tools=["WebSearch", "WebFetch"])
):
    print(message)

Sequential/dependent tools — output of one becomes input to the next:

async for message in query(
    prompt="Find all config files, read them, and check for hardcoded secrets",
    options=ClaudeAgentOptions(allowed_tools=["Glob", "Read", "Grep"])
):
    print(message)

6. Creating Custom Tools

Custom tools are defined via MCP servers that your agent can invoke.

Python

from claude_agent_sdk import tool, create_sdk_mcp_server, query, ClaudeAgentOptions

@tool(
    "get_weather",
    "Get current temperature for a location",
    {"latitude": float, "longitude": float}
)
async def get_weather(args):
    import aiohttp
    url = (
        f"https://api.open-meteo.com/v1/forecast?"
        f"latitude={args['latitude']}&longitude={args['longitude']}"
        f"&current=temperature_2m&temperature_unit=fahrenheit"
    )
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.json()

    temp = data["current"]["temperature_2m"]
    return {"content": [{"type": "text", "text": f"Temperature: {temp}°F"}]}


# Bundle tools into an MCP server
weather_server = create_sdk_mcp_server(
    name="weather-tools",
    version="1.0.0",
    tools=[get_weather]
)

# Use it in a query
async def main():
    async for message in query(
        prompt="What's the weather in San Francisco?",
        options=ClaudeAgentOptions(
            mcp_servers={"weather": weather_server},
            allowed_tools=["mcp__weather__get_weather"]
        )
    ):
        if hasattr(message, "result"):
            print(message.result)

TypeScript

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

const weatherServer = createSdkMcpServer({
  name: "weather-tools",
  version: "1.0.0",
  tools: [
    tool(
      "get_weather",
      "Get current temperature for a location",
      {
        latitude: z.number().describe("Latitude coordinate"),
        longitude: z.number().describe("Longitude coordinate")
      },
      async (args) => {
        const res = await fetch(
          `https://api.open-meteo.com/v1/forecast?latitude=${args.latitude}&longitude=${args.longitude}&current=temperature_2m&temperature_unit=fahrenheit`
        );
        const data = await res.json();
        return {
          content: [{
            type: "text",
            text: `Temperature: ${data.current.temperature_2m}°F`
          }]
        };
      }
    )
  ]
});

for await (const message of query({
  prompt: "What's the weather in San Francisco?",
  options: {
    mcpServers: { weather: weatherServer },
    allowedTools: ["mcp__weather__get_weather"]
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

External MCP Server Integration

You can also connect to externally-running MCP servers:

options = ClaudeAgentOptions(
    mcp_servers={
        "playwright": {
            "command": "npx",
            "args": ["@playwright/mcp@latest"]
        },
        "database": {
            "command": "npx",
            "args": ["@modelcontextprotocol/server-postgres", "postgresql://localhost/mydb"]
        }
    }
)

7. Multi-Agent Orchestration

Multi-agent systems let you break complex tasks into specialized subtasks handled by dedicated agents.

Defining Subagents

from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Glob", "Grep", "Task"],
        agents={
            "security-reviewer": AgentDefinition(
                description="Security-focused code reviewer",
                prompt="""You are a security specialist. Review code for:
                - SQL injection, XSS, CSRF vulnerabilities
                - Authentication and authorization flaws
                - Sensitive data exposure
                - Dependency vulnerabilities""",
                tools=["Read", "Grep", "Glob"],
                model="sonnet"
            ),
            "performance-optimizer": AgentDefinition(
                description="Performance analysis specialist",
                prompt="""You are a performance engineer. Analyze code for:
                - Database query optimization opportunities
                - Algorithm complexity issues
                - Memory leaks
                - Inefficient loops and data structures""",
                tools=["Read", "Grep", "Glob"],
                model="sonnet"
            ),
            "test-runner": AgentDefinition(
                description="Test execution and analysis",
                prompt="Execute tests and provide detailed analysis of failures",
                tools=["Bash", "Read", "Grep"],
                model="haiku"
            )
        }
    )

    async for message in query(
        prompt="Review and optimize this codebase. Use all three specialists.",
        options=options
    ):
        if hasattr(message, "result"):
            print(message.result)

Orchestration Patterns

Sequential — agents execute one after another, each building on the previous result:

prompt = """
1. First, use the security-reviewer to audit the authentication module
2. Then, use the performance-optimizer to check the database queries
3. Finally, use the test-runner to validate everything passes
Combine all findings into a report.
"""

Parallel — independent agents run concurrently:

prompt = """
Simultaneously:
- Have the security-reviewer examine auth.py
- Have the performance-optimizer check database.py
- Have the test-runner validate the test suite
Then combine findings into a single report.
"""

Conditional (routing) — different agents are invoked based on the situation:

prompt = """
Analyze this pull request:
- If it modifies authentication code, use the security-reviewer
- If it changes database queries, use the performance-optimizer
- Always run the test-runner
"""

Resuming a Subagent

You can resume a previously spawned subagent with its full context preserved:

# Capture session/agent ID from initial run
session_id = None

async for message in query(
    prompt="Explore the codebase architecture",
    options=ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep", "Task"])
):
    if hasattr(message, "session_id"):
        session_id = message.session_id

# Resume with full context
async for message in query(
    prompt="Now focus on the 3 most complex modules you found",
    options=ClaudeAgentOptions(resume=session_id)
):
    if hasattr(message, "result"):
        print(message.result)

8. Guardrails & Permissions

Permission Modes

Mode Behavior
default Prompts the user before potentially dangerous operations
acceptEdits Auto-approves file edits, mkdir, rm, mv, cp
plan Planning only — no tool execution
bypassPermissions Auto-approves everything (use with caution)
# Read-only agent — safe by default
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Glob", "Grep"],
    permission_mode="default"
)

# Editing agent with auto-approve for file changes
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Edit", "Bash"],
    permission_mode="acceptEdits"
)

Custom Permission Callback

For fine-grained control, implement a can_use_tool callback:

async def permission_handler(tool_name, input_data, context):
    # Block dangerous bash commands
    if tool_name == "Bash":
        command = input_data.get("command", "")
        if any(cmd in command for cmd in ["rm -rf", "DROP TABLE", "shutdown"]):
            return PermissionResultDeny(
                message="Dangerous command blocked",
                interrupt=True
            )

    # Auto-approve read operations
    if tool_name in ["Read", "Glob", "Grep"]:
        return PermissionResultAllow(updated_input=input_data)

    # Redirect writes to a sandbox directory
    if tool_name == "Write":
        safe_path = f"./sandbox/{input_data.get('file_path', '')}"
        return PermissionResultAllow(
            updated_input={**input_data, "file_path": safe_path}
        )

    # Default: ask the user
    return None

options = ClaudeAgentOptions(
    can_use_tool=permission_handler,
    allowed_tools=["Read", "Write", "Edit", "Bash"]
)

Hooks for Auditing

Log every tool invocation for audit trails:

from claude_agent_sdk import HookMatcher
from datetime import datetime

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

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [HookMatcher(hooks=[audit_log])],
        "PostToolUse": [HookMatcher(hooks=[audit_log])]
    }
)

9. Streaming Responses

Stream agent output in real time for responsive UIs.

Python

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent

async def stream_response():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash"]
    )

    async for message in query(prompt="Explain the project structure", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    print(delta.get("text", ""), end="", flush=True)

TypeScript

for await (const message of query({
  prompt: "Explain the project structure",
  options: { includePartialMessages: true }
})) {
  if (message.type === "stream_event") {
    const { event } = message;
    if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
      process.stdout.write(event.delta.text);
    }
  }
}

Tracking Tool Calls in Streams

current_tool = None

async for message in query(prompt="Analyze this codebase", options=options):
    if isinstance(message, StreamEvent):
        event = message.event
        event_type = event.get("type")

        if event_type == "content_block_start":
            block = event.get("content_block", {})
            if block.get("type") == "tool_use":
                current_tool = block.get("name")
                print(f"\n[Using {current_tool}...]", end="", flush=True)

        elif event_type == "content_block_stop":
            if current_tool:
                print(" done", flush=True)
                current_tool = None

        elif event_type == "content_block_delta":
            delta = event.get("delta", {})
            if delta.get("type") == "text_delta" and not current_tool:
                print(delta.get("text", ""), end="", flush=True)

10. Session & Context Management

Resuming Sessions

Capture a session ID from one query and resume it later to maintain full context:

session_id = None

# First interaction
async for message in query(
    prompt="Understand the authentication module",
    options=ClaudeAgentOptions(allowed_tools=["Read", "Glob"])
):
    if hasattr(message, "subtype") and message.subtype == "init":
        session_id = message.data.get("session_id")

# Later — agent remembers the auth module analysis
async for message in query(
    prompt="Now refactor it to use JWT tokens",
    options=ClaudeAgentOptions(resume=session_id)
):
    if hasattr(message, "result"):
        print(message.result)

Forking Sessions

Create a branch of a session to explore alternative approaches without losing the original:

async for message in query(
    prompt="Try a different approach — use OAuth2 instead",
    options=ClaudeAgentOptions(
        resume=session_id,
        fork_session=True  # Creates a new session; original is untouched
    )
):
    pass

Automatic Context Compaction

The SDK automatically manages context when conversations get long:

  • When token usage approaches the budget limit, older messages are summarized
  • Core instructions (system prompts, CLAUDE.md) always persist
  • Working context for the current task is preserved
  • Transient output (intermediate results) is compacted first

No configuration needed — this happens transparently.


11. Applying the SDK to Workflows

This section shows complete, practical examples of how to use the Agent SDK in real scenarios.

Workflow 1: Automated Code Review Pipeline

A multi-agent system that reviews pull requests from three angles simultaneously.

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition

async def review_pull_request(changed_files: list[str]):
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Grep", "Glob", "Task"],
        agents={
            "security-reviewer": AgentDefinition(
                description="Security vulnerability scanner",
                prompt="""Review code for OWASP Top 10 vulnerabilities:
                - SQL injection, XSS, CSRF
                - Authentication/authorization flaws
                - Sensitive data exposure
                - Insecure deserialization
                Return a structured list of findings with severity levels.""",
                tools=["Read", "Grep", "Glob"],
                model="sonnet"
            ),
            "style-checker": AgentDefinition(
                description="Code style and best practices reviewer",
                prompt="""Check code for:
                - Naming conventions
                - Function complexity
                - Dead code
                - Missing error handling
                Return suggestions with file:line references.""",
                tools=["Read", "Grep", "Glob"],
                model="haiku"
            ),
            "test-coverage-analyzer": AgentDefinition(
                description="Test coverage gap finder",
                prompt="""Analyze whether the changed code has adequate test coverage:
                - Identify untested code paths
                - Check edge case handling
                - Verify assertion quality
                Return a list of missing test cases.""",
                tools=["Read", "Grep", "Glob"],
                model="sonnet"
            )
        }
    )

    files_list = ", ".join(changed_files)
    async for message in query(
        prompt=f"""Review these changed files: {files_list}

        Run all three reviewers in parallel:
        1. security-reviewer — scan for vulnerabilities
        2. style-checker — check code quality
        3. test-coverage-analyzer — find coverage gaps

        Combine results into a structured review report with:
        - Critical issues (must fix)
        - Warnings (should fix)
        - Suggestions (nice to have)""",
        options=options
    ):
        if hasattr(message, "result"):
            return message.result

# Usage
asyncio.run(review_pull_request(["auth.py", "database.py", "api.py"]))

Workflow 2: Research & Documentation Generator

An agent that explores a codebase and produces architecture documentation.

async def generate_architecture_docs(project_path: str):
    async for message in query(
        prompt=f"""Analyze the project at {project_path} and generate architecture documentation.

        Steps:
        1. Use the code-explorer to map the project structure
        2. Use the research-agent to look up frameworks and libraries used
        3. Combine findings into a comprehensive architecture document

        The document should cover:
        - High-level architecture overview
        - Module descriptions and responsibilities
        - Key dependencies and their roles
        - Data flow between components
        - Entry points and API surface""",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Glob", "Grep", "WebSearch", "WebFetch", "Write", "Task"],
            agents={
                "code-explorer": AgentDefinition(
                    description="Codebase structure and architecture analyzer",
                    prompt="Map project structure. Identify modules, entry points, and dependencies.",
                    tools=["Read", "Glob", "Grep"],
                    model="sonnet"
                ),
                "research-agent": AgentDefinition(
                    description="Framework and library researcher",
                    prompt="Research the frameworks and libraries found in the codebase. Document their purpose and best practices.",
                    tools=["WebSearch", "WebFetch", "Read"],
                    model="sonnet"
                )
            }
        )
    ):
        if hasattr(message, "result"):
            print(message.result)

Workflow 3: Bug Investigation & Fix

A systematic debugging workflow using coordinated agents.

async def investigate_and_fix_bug(error_description: str):
    async for message in query(
        prompt=f"""A bug has been reported: "{error_description}"

        Phase 1 — Investigation (parallel):
        - test-analyzer: run the test suite, capture failures
        - code-inspector: search for code matching the error description

        Phase 2 — Diagnosis:
        - Correlate test failures with code analysis
        - Identify the root cause

        Phase 3 — Fix:
        - Implement the fix
        - Re-run tests to verify

        Phase 4 — Report:
        - Summarize what was wrong and what was changed""",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Edit", "Bash", "Grep", "Glob", "Task"],
            permission_mode="acceptEdits",
            agents={
                "test-analyzer": AgentDefinition(
                    description="Test execution and failure analysis",
                    prompt="Run tests. Analyze failures. Report which tests fail and why.",
                    tools=["Bash", "Read", "Grep"],
                    model="sonnet"
                ),
                "code-inspector": AgentDefinition(
                    description="Source code bug hunter",
                    prompt="Search source code for patterns related to the reported error.",
                    tools=["Read", "Grep", "Glob"],
                    model="sonnet"
                )
            }
        )
    ):
        if hasattr(message, "result"):
            print(message.result)

# Usage
asyncio.run(investigate_and_fix_bug("Login returns 500 when email contains a plus sign"))

Workflow 4: CI/CD Integration

Run the agent as part of a GitHub Actions pipeline for automated PR analysis.

import os

async def ci_pr_analysis():
    """Run as a GitHub Action step for PR analysis."""
    pr_files = os.environ.get("PR_FILES", "").split(",")

    async for message in query(
        prompt=f"Analyze these PR files for potential issues: {pr_files}",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Glob", "Grep"],
            permission_mode="bypassPermissions",  # No human in the loop in CI
            cwd=os.environ.get("GITHUB_WORKSPACE", ".")
        )
    ):
        if hasattr(message, "result"):
            # Output as GitHub Actions annotation
            print(f"::notice::{message.result}")

GitHub Actions workflow file:

- name: AI PR Review
  env:
    ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
    PR_FILES: ${{ steps.changed-files.outputs.all_changed_files }}
  run: python ci_pr_analysis.py

Workflow 5: Data Processing Pipeline

Process data in batches with context maintained across steps.

async def data_pipeline(input_dir: str, output_dir: str):
    async for message in query(
        prompt=f"""Process all CSV files in {input_dir}:

        For each file:
        1. Read and validate the data
        2. Clean: remove duplicates, fix formatting, handle missing values
        3. Transform: normalize columns, compute derived fields
        4. Write cleaned data to {output_dir}

        After all files are processed, generate a summary report
        with statistics for each file.""",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Write", "Bash", "Glob"],
            permission_mode="acceptEdits"
        )
    ):
        if hasattr(message, "result"):
            print(message.result)

Workflow 6: API Integration with Custom Tools

Build an agent that interacts with external services through custom MCP tools.

import os
import json
import aiohttp
from claude_agent_sdk import tool, create_sdk_mcp_server, query, ClaudeAgentOptions

@tool("github_create_issue", "Create a GitHub issue", {
    "repo": str,
    "title": str,
    "body": str,
    "labels": list
})
async def github_create_issue(args):
    async with aiohttp.ClientSession() as session:
        async with session.post(
            f"https://api.github.com/repos/{args['repo']}/issues",
            headers={
                "Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}",
                "Accept": "application/vnd.github.v3+json"
            },
            json={
                "title": args["title"],
                "body": args["body"],
                "labels": args.get("labels", [])
            }
        ) as response:
            data = await response.json()
            return {"content": [{"type": "text", "text": json.dumps(data, indent=2)}]}

@tool("slack_send_message", "Send a Slack message", {
    "channel": str,
    "text": str
})
async def slack_send_message(args):
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://slack.com/api/chat.postMessage",
            headers={"Authorization": f"Bearer {os.environ['SLACK_TOKEN']}"},
            json={"channel": args["channel"], "text": args["text"]}
        ) as response:
            data = await response.json()
            return {"content": [{"type": "text", "text": json.dumps(data, indent=2)}]}

integrations = create_sdk_mcp_server(
    name="integrations",
    version="1.0.0",
    tools=[github_create_issue, slack_send_message]
)

async def triage_and_notify(error_log: str):
    async for message in query(
        prompt=f"""Analyze this error log and take action:

        {error_log}

        1. Determine severity (critical / warning / info)
        2. If critical: create a GitHub issue AND send a Slack alert
        3. If warning: create a GitHub issue only
        4. If info: just report the finding""",
        options=ClaudeAgentOptions(
            mcp_servers={"integrations": integrations},
            allowed_tools=[
                "mcp__integrations__github_create_issue",
                "mcp__integrations__slack_send_message"
            ]
        )
    ):
        if hasattr(message, "result"):
            print(message.result)

12. Best Practices

Principle of Least Privilege

Give each agent only the tools it needs:

# Good — read-only agent for analysis
"analyzer": AgentDefinition(
    tools=["Read", "Grep", "Glob"]  # Cannot modify anything
)

# Bad — over-privileged agent
"analyzer": AgentDefinition(
    tools=["Read", "Write", "Edit", "Bash", "WebSearch"]  # Why does an analyzer need Write and Bash?
)

Model Selection for Cost Efficiency

Use the right model tier for each task:

agents={
    # Complex reasoning → Opus
    "architect": AgentDefinition(model="opus", ...),

    # Standard coding tasks → Sonnet
    "implementer": AgentDefinition(model="sonnet", ...),

    # Simple, repetitive tasks → Haiku
    "formatter": AgentDefinition(model="haiku", ...)
}

Budget and Turn Limits

Prevent runaway costs with explicit limits:

options = ClaudeAgentOptions(
    max_turns=50,        # Stop after 50 agent loop iterations
    max_budget_usd=5.0   # Hard cost cap
)

Error Handling

from claude_agent_sdk import CLINotFoundError, ProcessError

try:
    async for message in query(prompt="...", options=options):
        print(message)
except CLINotFoundError:
    print("Claude Agent SDK CLI not found. Run: pip install claude-agent-sdk")
except ProcessError as e:
    print(f"Agent process failed (exit code {e.exit_code}): {e.stderr}")

Sandbox Mode

Run untrusted operations in a sandbox:

options = ClaudeAgentOptions(
    sandbox={
        "enabled": True,
        "autoAllowBashIfSandboxed": True,
        "network": {
            "allowLocalBinding": True,
            "allowUnixSockets": ["/var/run/docker.sock"]
        }
    }
)

File Checkpointing

Enable undo/redo for file modifications:

options = ClaudeAgentOptions(enable_file_checkpointing=True)

# Later, rewind all files to a specific point
query_result = query(prompt="...", options=options)
await query_result.rewindFiles(message_uuid)

13. Reference

ClaudeAgentOptions — Full Configuration (Python)

ClaudeAgentOptions(
    # Tool access
    allowed_tools=["Read", "Edit", "Bash"],
    disallowed_tools=[],

    # Permissions
    permission_mode="default",          # default | acceptEdits | plan | bypassPermissions
    can_use_tool=my_callback,           # Custom permission function

    # Agents
    agents={"name": AgentDefinition(...)},

    # MCP servers
    mcp_servers={"name": {...}},

    # System prompt
    system_prompt="You are a helpful assistant",
    # or: system_prompt={"type": "preset", "preset": "claude_code", "append": "extra instructions"}

    # Session management
    resume="session-id",
    continue_conversation=True,
    fork_session=False,

    # Resource limits
    max_turns=100,
    max_budget_usd=10.0,
    max_thinking_tokens=8192,

    # Execution environment
    cwd="/path/to/project",
    env={"MY_VAR": "value"},

    # Model
    model="sonnet",                     # opus | sonnet | haiku
    fallback_model="haiku",

    # Hooks
    hooks={"PreToolUse": [...], "PostToolUse": [...]},

    # Output
    output_format={"type": "json_schema", "schema": {...}},

    # Advanced
    enable_file_checkpointing=True,
    sandbox={...},
    setting_sources=["project"],        # Load CLAUDE.md files
)

Official Resources

Resource Link
Python SDK (GitHub) anthropics/claude-agent-sdk-python
TypeScript SDK (GitHub) anthropics/claude-agent-sdk-typescript
Demo Agents anthropics/claude-agent-sdk-demos
Documentation platform.claude.com/docs/en/agent-sdk
API Key Management platform.claude.com