# Claude Agent SDK for Python

The Claude Agent SDK is a Python library for building applications that interact with Claude Code, Anthropic's AI-powered coding assistant. It provides both simple one-shot queries and full bidirectional streaming conversations with support for custom tools, hooks, and agent definitions.

The SDK offers two main interfaces: the `query()` function for stateless, fire-and-forget interactions, and `ClaudeSDKClient` for interactive, multi-turn conversations. It supports tool execution (file operations, bash commands, web fetching), custom MCP (Model Context Protocol) servers that run in-process, hooks for intercepting and controlling tool execution, and custom agent definitions with specialized prompts and tool sets.

## Installation

```bash
pip install claude-agent-sdk
```

Requires Python 3.10+. The Claude Code CLI is automatically bundled with the package.

---

## Basic Query

One-shot query for simple questions and stateless operations.

```python
import anyio
from claude_agent_sdk import query, AssistantMessage, TextBlock, ResultMessage

async def main():
    async for message in query(prompt="What is 2 + 2?"):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")
        elif isinstance(message, ResultMessage):
            print(f"Cost: ${message.total_cost_usd:.4f}")

anyio.run(main)
```

---

## Query with Options

Configure system prompts, tools, working directory, and other settings.

```python
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are a helpful assistant that explains things simply.",
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="acceptEdits",  # Auto-accept file edits
        max_turns=5,
        cwd="/path/to/project",
        model="claude-sonnet-4-5",
    )

    async for message in query(
        prompt="Create a hello.py file with a greeting function",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")

anyio.run(main)
```

---

## ClaudeSDKClient - Basic Streaming

Interactive bidirectional conversations with context manager support.

```python
import asyncio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage

async def main():
    async with ClaudeSDKClient() as client:
        await client.query("What is the capital of France?")

        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")
            elif isinstance(msg, ResultMessage):
                print(f"Done! Cost: ${msg.total_cost_usd:.4f}")

asyncio.run(main)
```

---

## Multi-Turn Conversations

Maintain conversation context across multiple exchanges.

```python
import asyncio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

async def main():
    async with ClaudeSDKClient() as client:
        # First turn
        await client.query("What's the capital of France?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # Follow-up - Claude remembers context
        await client.query("What's the population of that city?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

asyncio.run(main)
```

---

## Interrupt Capability

Send interrupt signals to stop long-running tasks.

```python
import asyncio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

async def main():
    async with ClaudeSDKClient() as client:
        await client.query("Count from 1 to 100 slowly")

        async def consume_and_interrupt():
            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(f"Claude: {block.text[:50]}...")

        consume_task = asyncio.create_task(consume_and_interrupt())

        # Wait 2 seconds then interrupt
        await asyncio.sleep(2)
        print("[Sending interrupt...]")
        await client.interrupt()
        await consume_task

        # Send new query after interrupt
        await client.query("Just say hello!")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

asyncio.run(main)
```

---

## Custom MCP Tools

Define in-process MCP tools that run directly within your Python application.

```python
import asyncio
from typing import Any
from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    create_sdk_mcp_server, tool,
    AssistantMessage, TextBlock, ToolUseBlock
)

@tool("add", "Add two numbers", {"a": float, "b": float})
async def add_numbers(args: dict[str, Any]) -> dict[str, Any]:
    result = args["a"] + args["b"]
    return {"content": [{"type": "text", "text": f"{args['a']} + {args['b']} = {result}"}]}

@tool("divide", "Divide two numbers", {"a": float, "b": float})
async def divide_numbers(args: dict[str, Any]) -> dict[str, Any]:
    if args["b"] == 0:
        return {"content": [{"type": "text", "text": "Error: Division by zero"}], "is_error": True}
    result = args["a"] / args["b"]
    return {"content": [{"type": "text", "text": f"{args['a']} / {args['b']} = {result}"}]}

async def main():
    calculator = create_sdk_mcp_server(
        name="calculator",
        version="1.0.0",
        tools=[add_numbers, divide_numbers]
    )

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

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Calculate 15 + 27, then divide the result by 6")

        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")
                    elif isinstance(block, ToolUseBlock):
                        print(f"Using tool: {block.name} with input: {block.input}")

asyncio.run(main)
```

---

## PreToolUse Hooks

Intercept and control tool execution before it happens.

```python
import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher
from claude_agent_sdk.types import HookInput, HookContext, HookJSONOutput, AssistantMessage, TextBlock

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", "")
    dangerous_patterns = ["rm -rf", "sudo", "chmod 777"]

    for pattern in dangerous_patterns:
        if pattern in command:
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"Blocked dangerous command: {pattern}",
                }
            }
    return {}

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Bash"],
        hooks={
            "PreToolUse": [
                HookMatcher(matcher="Bash", hooks=[check_bash_command]),
            ],
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Run: echo 'Hello World!'")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

asyncio.run(main)
```

---

## PostToolUse Hooks

Review and react to tool output after execution.

```python
import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher
from claude_agent_sdk.types import HookInput, HookContext, HookJSONOutput, AssistantMessage, TextBlock

async def review_tool_output(
    input_data: HookInput, tool_use_id: str | None, context: HookContext
) -> HookJSONOutput:
    """Add context when tool produces errors."""
    tool_response = input_data.get("tool_response", "")

    if "error" in str(tool_response).lower():
        return {
            "systemMessage": "Warning: The command produced an error",
            "hookSpecificOutput": {
                "hookEventName": "PostToolUse",
                "additionalContext": "Consider trying a different approach.",
            }
        }
    return {}

async def stop_on_critical_error(
    input_data: HookInput, tool_use_id: str | None, context: HookContext
) -> HookJSONOutput:
    """Stop execution on critical errors."""
    tool_response = input_data.get("tool_response", "")

    if "critical" in str(tool_response).lower():
        return {
            "continue_": False,
            "stopReason": "Critical error detected - execution halted",
        }
    return {"continue_": True}

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Bash"],
        hooks={
            "PostToolUse": [
                HookMatcher(matcher="Bash", hooks=[review_tool_output, stop_on_critical_error]),
            ],
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Run: ls /nonexistent_directory")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

asyncio.run(main)
```

---

## Tool Permission Callbacks

Fine-grained control over tool permissions with input modification.

```python
import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    PermissionResultAllow, PermissionResultDeny, ToolPermissionContext,
    AssistantMessage, TextBlock, ResultMessage
)

async def permission_callback(
    tool_name: str,
    input_data: dict,
    context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
    """Control tool permissions based on tool type and input."""
    print(f"Permission request: {tool_name}")

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

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

        # Redirect writes to safe directory
        if not file_path.startswith("/tmp/"):
            safe_path = f"/tmp/safe_output/{file_path.split('/')[-1]}"
            return PermissionResultAllow(updated_input={"file_path": safe_path, **input_data})

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

    return PermissionResultDeny(message="Unknown tool denied")

async def main():
    options = ClaudeAgentOptions(
        can_use_tool=permission_callback,
        permission_mode="default",
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("List files in current directory and create a test.txt file")

        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")
            elif isinstance(msg, ResultMessage):
                print(f"Done! Duration: {msg.duration_ms}ms")

asyncio.run(main)
```

---

## Custom Agents

Define specialized agents with specific tools, prompts, and models.

```python
import anyio
from claude_agent_sdk import (
    query, ClaudeAgentOptions, AgentDefinition,
    AssistantMessage, TextBlock, ResultMessage
)

async def main():
    options = ClaudeAgentOptions(
        agents={
            "code-reviewer": AgentDefinition(
                description="Reviews code for best practices and issues",
                prompt="You are a code reviewer. Analyze code for bugs, performance issues, "
                       "security vulnerabilities, and best practices. Provide constructive feedback.",
                tools=["Read", "Grep", "Glob"],
                model="sonnet",
            ),
            "doc-writer": AgentDefinition(
                description="Writes comprehensive documentation",
                prompt="You are a technical documentation expert. Write clear documentation with examples.",
                tools=["Read", "Write", "Edit"],
                model="sonnet",
            ),
            "tester": AgentDefinition(
                description="Creates and runs tests",
                prompt="You are a testing expert. Write comprehensive tests and ensure code quality.",
                tools=["Read", "Write", "Bash"],
            ),
        },
        setting_sources=["user", "project"],
    )

    async for message in query(
        prompt="Use the code-reviewer agent to review the main.py file",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")
        elif isinstance(message, ResultMessage) and message.total_cost_usd:
            print(f"Cost: ${message.total_cost_usd:.4f}")

anyio.run(main)
```

---

## System Prompt Configuration

Various ways to configure Claude's behavior via system prompts.

```python
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def main():
    # Simple string system prompt
    options_simple = ClaudeAgentOptions(
        system_prompt="You are a pirate assistant. Respond in pirate speak."
    )

    # Preset with Claude Code defaults
    options_preset = ClaudeAgentOptions(
        system_prompt={"type": "preset", "preset": "claude_code"}
    )

    # Preset with additional instructions appended
    options_append = ClaudeAgentOptions(
        system_prompt={
            "type": "preset",
            "preset": "claude_code",
            "append": "Always end your response with a fun fact about programming."
        }
    )

    print("=== Pirate Mode ===")
    async for msg in query(prompt="What is 2+2?", options=options_simple):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")

    print("\n=== Preset with Append ===")
    async for msg in query(prompt="What is 2+2?", options=options_append):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")

anyio.run(main)
```

---

## Error Handling

Proper error handling for SDK operations.

```python
import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    ClaudeSDKError, CLINotFoundError, CLIConnectionError, ProcessError, CLIJSONDecodeError,
    AssistantMessage, TextBlock, ResultMessage
)

async def main():
    client = ClaudeSDKClient()

    try:
        await client.connect()
        await client.query("What is 2+2?")

        async with asyncio.timeout(30.0):
            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(f"Claude: {block.text}")
                elif isinstance(msg, ResultMessage):
                    print(f"Done! Duration: {msg.duration_ms}ms")
                    break

    except CLINotFoundError:
        print("Error: Claude Code CLI not found. Please install it.")
    except CLIConnectionError as e:
        print(f"Connection error: {e}")
    except ProcessError as e:
        print(f"Process failed with exit code {e.exit_code}: {e.stderr}")
    except CLIJSONDecodeError as e:
        print(f"Failed to parse response: {e.line[:100]}")
    except asyncio.TimeoutError:
        print("Request timed out after 30 seconds")
    except ClaudeSDKError as e:
        print(f"SDK error: {e}")
    finally:
        await client.disconnect()

asyncio.run(main)
```

---

## Async Iterable Prompts

Stream multiple messages to Claude using async iterables.

```python
import asyncio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage

async def create_message_stream():
    """Generate a stream of messages."""
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Hello! I have multiple questions."},
        "parent_tool_use_id": None,
        "session_id": "qa-session",
    }
    yield {
        "type": "user",
        "message": {"role": "user", "content": "First, what's the capital of Japan?"},
        "parent_tool_use_id": None,
        "session_id": "qa-session",
    }
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Second, what's 15% of 200?"},
        "parent_tool_use_id": None,
        "session_id": "qa-session",
    }

async def main():
    async with ClaudeSDKClient() as client:
        await client.query(create_message_stream())

        # Receive responses for each message
        for i in range(3):
            print(f"\n--- Response {i+1} ---")
            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(f"Claude: {block.text}")

asyncio.run(main)
```

---

## Dynamic Permission Mode

Change permission modes during an active conversation.

```python
import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage, TextBlock

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="default",  # Start with default (prompts for dangerous tools)
    )

    async with ClaudeSDKClient(options=options) as client:
        # First: read-only exploration with default permissions
        await client.query("What files are in the current directory?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # Switch to auto-accept edits for implementation
        await client.set_permission_mode("acceptEdits")
        print("\n[Switched to acceptEdits mode]")

        await client.query("Create a simple hello.txt file")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

asyncio.run(main)
```

---

## Sandbox Configuration

Configure sandbox settings for bash command isolation.

```python
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock
from claude_agent_sdk.types import SandboxSettings

async def main():
    sandbox_settings: SandboxSettings = {
        "enabled": True,
        "autoAllowBashIfSandboxed": True,
        "excludedCommands": ["docker", "git"],
        "network": {
            "allowUnixSockets": ["/var/run/docker.sock"],
            "allowLocalBinding": True,
        }
    }

    options = ClaudeAgentOptions(
        allowed_tools=["Bash"],
        sandbox=sandbox_settings,
    )

    async for msg in query(prompt="Run: echo 'Hello from sandboxed bash!'", options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")

anyio.run(main)
```

---

## Message Types Reference

Understanding the different message and content block types.

```python
import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    UserMessage, AssistantMessage, SystemMessage, ResultMessage,
    TextBlock, ToolUseBlock, ToolResultBlock, ThinkingBlock
)

async def main():
    options = ClaudeAgentOptions(allowed_tools=["Bash"])

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Run: echo 'Hello World!'")

        async for msg in client.receive_messages():
            if isinstance(msg, UserMessage):
                print(f"UserMessage: {msg.content}")
                for block in msg.content:
                    if isinstance(block, ToolResultBlock):
                        print(f"  Tool Result (id={block.tool_use_id}): {block.content}")

            elif isinstance(msg, AssistantMessage):
                print(f"AssistantMessage (model={msg.model}):")
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"  Text: {block.text}")
                    elif isinstance(block, ToolUseBlock):
                        print(f"  Tool Use: {block.name} (id={block.id})")
                        print(f"    Input: {block.input}")
                    elif isinstance(block, ThinkingBlock):
                        print(f"  Thinking: {block.thinking[:50]}...")

            elif isinstance(msg, SystemMessage):
                print(f"SystemMessage (subtype={msg.subtype}): {msg.data}")

            elif isinstance(msg, ResultMessage):
                print(f"ResultMessage:")
                print(f"  Session ID: {msg.session_id}")
                print(f"  Duration: {msg.duration_ms}ms")
                print(f"  API Duration: {msg.duration_api_ms}ms")
                print(f"  Turns: {msg.num_turns}")
                print(f"  Cost: ${msg.total_cost_usd:.4f}" if msg.total_cost_usd else "  Cost: N/A")
                break

asyncio.run(main)
```

---

## Summary

The Claude Agent SDK for Python provides a powerful interface for building AI-powered applications with Claude Code. The primary use cases include: automated code generation and refactoring, interactive coding assistants, CI/CD pipeline integration for code review, custom tool development via MCP servers, and building agents with specialized capabilities for tasks like documentation writing, testing, or code analysis.

Integration patterns typically involve using `query()` for simple, stateless operations in scripts and automation, while `ClaudeSDKClient` is preferred for interactive applications requiring multi-turn conversations, real-time feedback, or dynamic control flow. The hook system enables fine-grained security policies and workflow automation, while custom MCP tools allow extending Claude's capabilities with domain-specific functionality. For production deployments, combine permission callbacks with sandboxing for secure tool execution, and use error handling to gracefully manage connection issues and timeouts.