pip install claude-agent-sdk

# Claude Code CLI is bundled - no separate install needed

from claude_agent_sdk import ClaudeAgentOptions

options = ClaudeAgentOptions(
    # Tools
    tools=["Bash", "Read", "Write"],        # Available tools
    allowed_tools=["mcp__calc__add"],       # Explicitly allowed
    disallowed_tools=["Bash"],              # Explicitly blocked

    # Prompts & Behavior
    system_prompt="You are a helpful assistant",
    max_turns=10,                           # Max conversation turns
    max_budget_usd=5.0,                     # Cost limit
    max_thinking_tokens=8000,               # Reasoning token limit

    # Models
    model="claude-sonnet-4-5",
    fallback_model="claude-haiku-3-5",

    # Permissions
    permission_mode="default",              # or "acceptEdits", "bypassPermissions"
    can_use_tool=my_permission_callback,    # Custom permission function

    # Sessions
    continue_conversation=True,
    resume="session-id-123",                # Resume previous session
    fork_session=True,                      # Branch from session

    # MCP Servers
    mcp_servers={"calc": calculator_server},

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

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

    # Working directory
    cwd="/path/to/project",
)

Content Block Types

Defining Tools

from claude_agent_sdk import tool, create_sdk_mcp_server

@tool("add", "Add two numbers", {"a": float, "b": float})
async def add(args):
    result = args["a"] + args["b"]
    return {
        "content": [{"type": "text", "text": f"Result: {result}"}]
    }

@tool("multiply", "Multiply two numbers", {"a": float, "b": float})
async def multiply(args):
    result = args["a"] * args["b"]
    return {
        "content": [{"type": "text", "text": f"Result: {result}"}],
        "is_error": False  # Optional
    }

# Create server
calculator = create_sdk_mcp_server(
    name="calculator",
    version="1.0.0",
    tools=[add, multiply]
)

Using Tools

options = ClaudeAgentOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=["mcp__calc__add", "mcp__calc__multiply"]
)

async with ClaudeSDKClient(options=options) as client:
    await client.query("Calculate 15 * 8 + 12")
    async for msg in client.receive_response():
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text)

# Tool naming: mcp__<server>__<tool>
# e.g., mcp__calc__add

Tool Input Schema Options

# Simple dict
{"name": str, "age": int, "score": float}

# Full JSON Schema
{
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer", "minimum": 0}
    },
    "required": ["name"]
}

# TypedDict
from typing_extensions import TypedDict

class MyInput(TypedDict):
    name: str
    age: int

Tool Return Format

# Text result
return {
    "content": [{"type": "text", "text": "Result here"}]
}

# Image result
return {
    "content": [{
        "type": "image",
        "data": "base64_encoded_data",
        "mimeType": "image/png"
    }]
}

# Error result
return {
    "content": [{"type": "text", "text": "Error: something failed"}],
    "is_error": True
}

Permission Callback Pattern

from claude_agent_sdk import (
    PermissionResultAllow,
    PermissionResultDeny,
    ToolPermissionContext
)

async def my_permission_callback(
    tool_name: str,
    input_data: dict,
    context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
    """Control which tools Claude can use."""

    # Allow read-only operations
    if tool_name in ["Read", "Glob", "Grep"]:
        return PermissionResultAllow()

    # Block writes to system directories
    if tool_name in ["Write", "Edit"]:
        path = input_data.get("file_path", "")
        if path.startswith("/etc/") or path.startswith("/usr/"):
            return PermissionResultDeny(
                message=f"Cannot write to system directory: {path}"
            )
        return PermissionResultAllow()

    # Modify inputs (redirect to safe location)
    if tool_name == "Write":
        safe_path = f"./output/{input_data['file_path'].split('/')[-1]}"
        return PermissionResultAllow(
            updated_input={**input_data, "file_path": safe_path}
        )

    # Check bash commands
    if tool_name == "Bash":
        command = input_data.get("command", "")
        dangerous = ["rm -rf", "sudo", "chmod 777"]
        for pattern in dangerous:
            if pattern in command:
                return PermissionResultDeny(
                    message=f"Dangerous pattern: {pattern}"
                )
        return PermissionResultAllow()

    # Default: deny
    return PermissionResultDeny(message="Tool not allowed")

# Use callback
options = ClaudeAgentOptions(
    can_use_tool=my_permission_callback,
    permission_mode="default"  # Required for callbacks
)

Hook Events

Hook Implementation

from claude_agent_sdk import HookMatcher, HookInput, HookContext, HookJSONOutput

async def check_bash_command(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """Block dangerous bash commands."""
    if input_data["tool_name"] != "Bash":
        return {}

    command = input_data["tool_input"].get("command", "")

    if "rm -rf" in command:
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": "Dangerous command blocked"
            }
        }

    return {}

async def review_output(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """Review tool output."""
    response = input_data.get("tool_response", "")

    if "error" in str(response).lower():
        return {
            "systemMessage": "Tool produced an error",
            "hookSpecificOutput": {
                "hookEventName": "PostToolUse",
                "additionalContext": "Consider a different approach"
            }
        }

    return {}

# Configure hooks
options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(
                matcher="Bash",  # Match specific tool
                hooks=[check_bash_command],
                timeout=30.0
            )
        ],
        "PostToolUse": [
            HookMatcher(
                matcher=None,  # Match all tools
                hooks=[review_output]
            )
        ]
    }
)

Hook Output Options

# Control execution flow
{
    "continue_": False,      # Stop after hook (default: True)
    "stopReason": "Error detected",
    "suppressOutput": True   # Hide from transcript
}

# Block/approve (PreToolUse)
{
    "hookSpecificOutput": {
        "hookEventName": "PreToolUse",
        "permissionDecision": "allow" | "deny" | "ask",
        "permissionDecisionReason": "Reason"
    }
}

# Modify inputs (PreToolUse)
{
    "hookSpecificOutput": {
        "hookEventName": "PreToolUse",
        "updatedInput": {"file_path": "/safe/path"}
    }
}

# Add context (PostToolUse)
{
    "hookSpecificOutput": {
        "hookEventName": "PostToolUse",
        "additionalContext": "Extra info for Claude"
    }
}

# Defer (async)
{
    "async_": True,
    "asyncTimeout": 5000  # milliseconds
}

from claude_agent_sdk import AgentDefinition

options = ClaudeAgentOptions(
    agents={
        "code-reviewer": AgentDefinition(
            description="Reviews code for best practices",
            prompt="""You are a code reviewer. Analyze code for:
            - Bugs and security vulnerabilities
            - Performance issues
            - Best practice violations
            Provide constructive feedback.""",
            tools=["Read", "Grep"],
            model="sonnet"
        ),
        "doc-writer": AgentDefinition(
            description="Writes comprehensive documentation",
            prompt="Write clear documentation with examples.",
            tools=["Read", "Write", "Edit"],
            model="sonnet"
        ),
        "tester": AgentDefinition(
            description="Writes and runs tests",
            prompt="Write comprehensive tests with good coverage.",
            tools=["Read", "Write", "Bash"],
            model="haiku"
        )
    }
)

# Use agent in prompt
async for msg in query(
    prompt="Use the code-reviewer agent to review src/main.py",
    options=options
):
    print(msg)

options = ClaudeAgentOptions(
    output_format={
        "type": "json_schema",
        "schema": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "email": {"type": "string", "format": "email"},
                "age": {"type": "integer", "minimum": 0}
            },
            "required": ["name", "email"]
        }
    }
)

async for msg in query(
    prompt="Extract: John Doe, john@example.com, 30",
    options=options
):
    if isinstance(msg, ResultMessage):
        print(msg.structured_output)
        # {"name": "John Doe", "email": "john@example.com", "age": 30}

Dynamic Settings

async with ClaudeSDKClient(options) as client:
    await client.query("Start with sonnet")
    async for msg in client.receive_response():
        pass

    # Switch to faster model
    await client.set_model("claude-haiku-3-5")

    # Change permissions
    await client.set_permission_mode("acceptEdits")

    await client.query("Continue with haiku")
    async for msg in client.receive_response():
        pass

from claude_agent_sdk import (
    ClaudeSDKError,      # Base error
    CLINotFoundError,    # CLI not installed
    CLIConnectionError,  # Connection issues
    ProcessError,        # Process failed
    CLIJSONDecodeError,  # JSON parse error
)

try:
    async for msg in query(prompt="Hello"):
        pass
except CLINotFoundError:
    print("Install: pip install claude-agent-sdk")
except CLIConnectionError as e:
    print(f"Connection error: {e}")
except ProcessError as e:
    print(f"Process failed (exit {e.exit_code}): {e.stderr}")
except CLIJSONDecodeError as e:
    print(f"JSON parse error: {e.line}")

Simple Query

import anyio
from claude_agent_sdk import query, AssistantMessage, TextBlock

async def main():
    async for msg in query(prompt="What is Python?"):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)

Interactive Chat

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

async def chat():
    options = ClaudeAgentOptions(
        system_prompt="You are a helpful Python expert"
    )

    async with ClaudeSDKClient(options=options) as client:
        while True:
            user_input = input("You: ")
            if user_input.lower() == "quit":
                break

            await client.query(user_input)

            print("Claude: ", end="")
            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(block.text)
                elif isinstance(msg, ResultMessage):
                    print(f"\n[Cost: ${msg.total_cost_usd:.4f}]\n")

asyncio.run(chat())

Tool + Permission + Hook

import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    tool, create_sdk_mcp_server,
    HookMatcher,
    PermissionResultAllow, PermissionResultDeny,
    AssistantMessage, TextBlock
)

# Tool
@tool("search", "Search database", {"query": str})
async def search(args):
    results = f"Found: {args['query']}"
    return {"content": [{"type": "text", "text": results}]}

server = create_sdk_mcp_server("db", tools=[search])

# Permission callback
async def check_permission(tool_name, input_data, ctx):
    if tool_name == "mcp__db__search":
        query = input_data.get("query", "")
        if "DROP" in query.upper():
            return PermissionResultDeny(message="SQL injection blocked")
    return PermissionResultAllow()

# Hook
async def log_search(input_data, tool_use_id, ctx):
    print(f"[LOG] Searching: {input_data['tool_input']}")
    return {}

options = ClaudeAgentOptions(
    mcp_servers={"db": server},
    allowed_tools=["mcp__db__search"],
    can_use_tool=check_permission,
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="mcp__db__search", hooks=[log_search])
        ]
    }
)

async def main():
    async with ClaudeSDKClient(options=options) as client:
        await client.query("Search for Python tutorials")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)

asyncio.run(main())