2460 Agent Sdk Python 6152fc1a

---
title: Agent SDK reference - Python
source_url: https://platform.claude.com/docs/en/agent-sdk/python
---

## Installation

    pip install claude-agent-sdk

## Choosing Between `query()` and `ClaudeSDKClient`

The Python SDK provides two ways to interact with Claude Code:

### Quick Comparison

| Feature | `query()` | `ClaudeSDKClient` |
|----|----|----|
| **Session** | Creates new session each time | Reuses same session |
| **Conversation** | Single exchange | Multiple exchanges in same context |
| **Connection** | Managed automatically | Manual control |
| **Streaming Input** | ✅ Supported | ✅ Supported |
| **Interrupts** | ❌ Not supported | ✅ Supported |
| **Hooks** | ❌ Not supported | ✅ Supported |
| **Custom Tools** | ❌ Not supported | ✅ Supported |
| **Continue Chat** | ❌ New session each time | ✅ Maintains conversation |
| **Use Case** | One-off tasks | Continuous conversations |

### When to Use `query()` (New Session Each Time)

**Best for:**

- One-off questions where you don't need conversation history
- Independent tasks that don't require context from previous exchanges
- Simple automation scripts
- When you want a fresh start each time

### When to Use `ClaudeSDKClient` (Continuous Conversation)

**Best for:**

- **Continuing conversations** - When you need Claude to remember context
- **Follow-up questions** - Building on previous responses
- **Interactive applications** - Chat interfaces, REPLs
- **Response-driven logic** - When next action depends on Claude's response
- **Session control** - Managing conversation lifecycle explicitly

## Functions

### `query()`

Creates a new session for each interaction with Claude Code. Returns an async iterator that yields messages as they arrive. Each call to `query()` starts fresh with no memory of previous interactions.

    async def query(
        *,
        prompt: str | AsyncIterable[dict[str, Any]],
        options: ClaudeAgentOptions | None = None
    ) -> AsyncIterator[Message]

#### Parameters

| Parameter | Type | Description |
|----|----|----|
| `prompt` | `str | AsyncIterable[dict]` | The input prompt as a string or async iterable for streaming mode |
| `options` | `ClaudeAgentOptions | None` | Optional configuration object (defaults to `ClaudeAgentOptions()` if None) |

#### Returns

Returns an `AsyncIterator[Message]` that yields messages from the conversation.

#### Example - With options

    import asyncio
    from claude_agent_sdk import query, ClaudeAgentOptions

    async def main():
        options = ClaudeAgentOptions(
            system_prompt="You are an expert Python developer",
            permission_mode='acceptEdits',
            cwd="/home/user/project"
        )

        async for message in query(
            prompt="Create a Python web server",
            options=options
        ):
            print(message)

    asyncio.run(main())

### `tool()`

Decorator for defining MCP tools with type safety.

    def tool(
        name: str,
        description: str,
        input_schema: type | dict[str, Any]
    ) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]

#### Parameters

| Parameter | Type | Description |
|----|----|----|
| `name` | `str` | Unique identifier for the tool |
| `description` | `str` | Human-readable description of what the tool does |
| `input_schema` | `type | dict[str, Any]` | Schema defining the tool's input parameters (see below) |

#### Input Schema Options

1. **Simple type mapping** (recommended):

        {"text": str, "count": int, "enabled": bool}

2. **JSON Schema format** (for complex validation):

        {
            "type": "object",
            "properties": {
                "text": {"type": "string"},
                "count": {"type": "integer", "minimum": 0}
            },
            "required": ["text"]
        }

#### Returns

A decorator function that wraps the tool implementation and returns an `SdkMcpTool` instance.

#### Example

    from claude_agent_sdk import tool
    from typing import Any

    @tool("greet", "Greet a user", {"name": str})
    async def greet(args: dict[str, Any]) -> dict[str, Any]:
        return {
            "content": [{
                "type": "text",
                "text": f"Hello, {args['name']}!"
            }]
        }

### `create_sdk_mcp_server()`

Create an in-process MCP server that runs within your Python application.

    def create_sdk_mcp_server(
        name: str,
        version: str = "1.0.0",
        tools: list[SdkMcpTool[Any]] | None = None
    ) -> McpSdkServerConfig

#### Parameters

| Parameter | Type | Default | Description |
|----|----|----|----|
| `name` | `str` | \- | Unique identifier for the server |
| `version` | `str` | `"1.0.0"` | Server version string |
| `tools` | `list[SdkMcpTool[Any]] | None` | `None` | List of tool functions created with `@tool` decorator |

#### Returns

Returns an `McpSdkServerConfig` object that can be passed to `ClaudeAgentOptions.mcp_servers`.

#### Example

    from claude_agent_sdk import tool, create_sdk_mcp_server

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

    @tool("multiply", "Multiply two numbers", {"a": float, "b": float})
    async def multiply(args):
        return {
            "content": [{
                "type": "text",
                "text": f"Product: {args['a'] * args['b']}"
            }]
        }

    calculator = create_sdk_mcp_server(
        name="calculator",
        version="2.0.0",
        tools=[add, multiply]  # Pass decorated functions
    )

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

## Classes

### `ClaudeSDKClient`

**Maintains a conversation session across multiple exchanges.** This is the Python equivalent of how the TypeScript SDK's `query()` function works internally - it creates a client object that can continue conversations.

#### Key Features

- **Session Continuity**: Maintains conversation context across multiple `query()` calls
- **Same Conversation**: Claude remembers previous messages in the session
- **Interrupt Support**: Can stop Claude mid-execution
- **Explicit Lifecycle**: You control when the session starts and ends
- **Response-driven Flow**: Can react to responses and send follow-ups
- **Custom Tools & Hooks**: Supports custom tools (created with `@tool` decorator) and hooks

    class ClaudeSDKClient:
        def **init**(self, options: ClaudeAgentOptions | None = None)
        async def connect(self, prompt: str | AsyncIterable[dict] | None = None) -> None
        async def query(self, prompt: str | AsyncIterable[dict], session_id: str = "default") -> None
        async def receive_messages(self) -> AsyncIterator[Message]
        async def receive_response(self) -> AsyncIterator[Message]
        async def interrupt(self) -> None
        async def disconnect(self) -> None

#### Methods

| Method | Description |
|----|----|
| `__init__(options)` | Initialize the client with optional configuration |
| `connect(prompt)` | Connect to Claude with an optional initial prompt or message stream |
| `query(prompt, session_id)` | Send a new request in streaming mode |
| `receive_messages()` | Receive all messages from Claude as an async iterator |
| `receive_response()` | Receive messages until and including a ResultMessage |
| `interrupt()` | Send interrupt signal (only works in streaming mode) |
| `disconnect()` | Disconnect from Claude |

#### Context Manager Support

The client can be used as an async context manager for automatic connection management:

    async with ClaudeSDKClient() as client:
        await client.query("Hello Claude")
        async for message in client.receive_response():
            print(message)

> **Important:** When iterating over messages, avoid using `break` to exit early as this can cause asyncio cleanup issues. Instead, let the iteration complete naturally or use flags to track when you've found what you need.

#### Example - Continuing a conversation

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

    async def main():
        async with ClaudeSDKClient() as client:
            # First question
            await client.query("What's the capital of France?")

            # Process response
            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 question - Claude remembers the previous context
            await client.query("What's the population of that city?")

            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}")

            # Another follow-up - still in the same conversation
            await client.query("What are some famous landmarks there?")

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

#### Example - Streaming input with ClaudeSDKClient

    import asyncio
    from claude_agent_sdk import ClaudeSDKClient

    async def message_stream():
        """Generate messages dynamically."""
        yield {"type": "text", "text": "Analyze the following data:"}
        await asyncio.sleep(0.5)
        yield {"type": "text", "text": "Temperature: 25°C"}
        await asyncio.sleep(0.5)
        yield {"type": "text", "text": "Humidity: 60%"}
        await asyncio.sleep(0.5)
        yield {"type": "text", "text": "What patterns do you see?"}

    async def main():
        async with ClaudeSDKClient() as client:
            # Stream input to Claude
            await client.query(message_stream())

            # Process response
            async for message in client.receive_response():
                print(message)

            # Follow-up in same session
            await client.query("Should we be concerned about these readings?")

            async for message in client.receive_response():
                print(message)

    asyncio.run(main())

#### Example - Using interrupts

    import asyncio
    from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

    async def interruptible_task():
        options = ClaudeAgentOptions(
            allowed_tools=["Bash"],
            permission_mode="acceptEdits"
        )

        async with ClaudeSDKClient(options=options) as client:
            # Start a long-running task
            await client.query("Count from 1 to 100 slowly")

            # Let it run for a bit
            await asyncio.sleep(2)

            # Interrupt the task
            await client.interrupt()
            print("Task interrupted!")

            # Send a new command
            await client.query("Just say hello instead")

            async for message in client.receive_response():
                # Process the new response
                pass

    asyncio.run(interruptible_task())

#### Example - Advanced permission control

    from claude_agent_sdk import (
        ClaudeSDKClient,
        ClaudeAgentOptions
    )

    async def custom_permission_handler(
        tool_name: str,
        input_data: dict,
        context: dict
    ):
        """Custom logic for tool permissions."""

        # Block writes to system directories
        if tool_name == "Write" and input_data.get("file_path", "").startswith("/system/"):
            return {
                "behavior": "deny",
                "message": "System directory write not allowed",
                "interrupt": True
            }

        # Redirect sensitive file operations
        if tool_name in ["Write", "Edit"] and "config" in input_data.get("file_path", ""):
            safe_path = f"./sandbox/{input_data['file_path']}"
            return {
                "behavior": "allow",
                "updatedInput": {**input_data, "file_path": safe_path}
            }

        # Allow everything else
        return {
            "behavior": "allow",
            "updatedInput": input_data
        }

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

        async with ClaudeSDKClient(options=options) as client:
            await client.query("Update the system config file")

            async for message in client.receive_response():
                # Will use sandbox path instead
                print(message)

    asyncio.run(main())

## Types

### `SdkMcpTool`

Definition for an SDK MCP tool created with the `@tool` decorator.

    @dataclass
    class SdkMcpTool(Generic[T]):
        name: str
        description: str
        input_schema: type[T] | dict[str, Any]
        handler: Callable[[T], Awaitable[dict[str, Any]]]

| Property | Type | Description |
|----|----|----|
| `name` | `str` | Unique identifier for the tool |
| `description` | `str` | Human-readable description |
| `input_schema` | `type[T] | dict[str, Any]` | Schema for input validation |
| `handler` | `Callable[[T], Awaitable[dict[str, Any]]]` | Async function that handles tool execution |

### `ClaudeAgentOptions`

Configuration dataclass for Claude Code queries.

    @dataclass
    class ClaudeAgentOptions:
        allowed_tools: list[str] = field(default_factory=list)
        system_prompt: str | SystemPromptPreset | None = None
        mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
        permission_mode: PermissionMode | None = None
        continue_conversation: bool = False
        resume: str | None = None
        max_turns: int | None = None
        disallowed_tools: list[str] = field(default_factory=list)
        model: str | None = None
        output_format: OutputFormat | None = None
        permission_prompt_tool_name: str | None = None
        cwd: str | Path | None = None
        settings: str | None = None
        add_dirs: list[str | Path] = field(default_factory=list)
        env: dict[str, str] = field(default_factory=dict)
        extra_args: dict[str, str | None] = field(default_factory=dict)
        max_buffer_size: int | None = None
        debug_stderr: Any = sys.stderr  # Deprecated
        stderr: Callable[[str], None] | None = None
        can_use_tool: CanUseTool | None = None
        hooks: dict[HookEvent, list[HookMatcher]] | None = None
        user: str | None = None
        include_partial_messages: bool = False
        fork_session: bool = False
        agents: dict[str, AgentDefinition] | None = None
        setting_sources: list[SettingSource] | None = None

| Property | Type | Default | Description |
|----|----|----|----|
| `allowed_tools` | `list[str]` | `[]` | List of allowed tool names |
| `system_prompt` | `str | SystemPromptPreset | None` | `None` | System prompt configuration. Pass a string for custom prompt, or use `{"type": "preset", "preset": "claude_code"}` for Claude Code's system prompt. Add `"append"` to extend the preset |
| `mcp_servers` | `dict[str, McpServerConfig] | str | Path` | `{}` | MCP server configurations or path to config file |
| `permission_mode` | `PermissionMode | None` | `None` | Permission mode for tool usage |
| `continue_conversation` | `bool` | `False` | Continue the most recent conversation |
| `resume` | `str | None` | `None` | Session ID to resume |
| `max_turns` | `int | None` | `None` | Maximum conversation turns |
| `disallowed_tools` | `list[str]` | `[]` | List of disallowed tool names |
| `model` | `str | None` | `None` | Claude model to use |
| `output_format` | [`OutputFormat`](https://platform.claude.com/docs/en/agent-sdk/python#outputformat) ` | None` | `None` | Define output format for agent results. See [Structured outputs](https://platform.claude.com/docs/en/agent-sdk/structured-outputs) for details |
| `permission_prompt_tool_name` | `str | None` | `None` | MCP tool name for permission prompts |
| `cwd` | `str | Path | None` | `None` | Current working directory |
| `settings` | `str | None` | `None` | Path to settings file |
| `add_dirs` | `list[str | Path]` | `[]` | Additional directories Claude can access |
| `env` | `dict[str, str]` | `{}` | Environment variables |
| `extra_args` | `dict[str, str | None]` | `{}` | Additional CLI arguments to pass directly to the CLI |
| `max_buffer_size` | `int | None` | `None` | Maximum bytes when buffering CLI stdout |
| `debug_stderr` | `Any` | `sys.stderr` | *Deprecated* - File-like object for debug output. Use `stderr` callback instead |
| `stderr` | `Callable[[str], None] | None` | `None` | Callback function for stderr output from CLI |
| `can_use_tool` | `CanUseTool | None` | `None` | Tool permission callback function |
| `hooks` | `dict[HookEvent, list[HookMatcher]] | None` | `None` | Hook configurations for intercepting events |
| `user` | `str | None` | `None` | User identifier |
| `include_partial_messages` | `bool` | `False` | Include `StreamEvent` messages with real-time text deltas. See [StreamEvent](#streamevent) for handling |
| `fork_session` | `bool` | `False` | When resuming with `resume`, fork to a new session ID instead of continuing the original session |
| `agents` | `dict[str, AgentDefinition] | None` | `None` | Programmatically defined subagents |
| `plugins` | `list[SdkPluginConfig]` | `[]` | Load custom plugins from local paths. See [Plugins](https://platform.claude.com/docs/en/agent-sdk/plugins) for details |
| `setting_sources` | `list[SettingSource] | None` | `None` (no settings) | Control which filesystem settings to load. When omitted, no settings are loaded. **Note:** Must include `"project"` to load CLAUDE.md files |

### `OutputFormat`

Configuration for structured output validation.

    class OutputFormat(TypedDict):
        type: Literal["json_schema"]
        schema: dict[str, Any]

| Field    | Required | Description                                        |
|----------|----------|----------------------------------------------------|
| `type`   | Yes      | Must be `"json_schema"` for JSON Schema validation |
| `schema` | Yes      | JSON Schema definition for output validation       |

### `SystemPromptPreset`

Configuration for using Claude Code's preset system prompt with optional additions.

    class SystemPromptPreset(TypedDict):
        type: Literal["preset"]
        preset: Literal["claude_code"]
        append: NotRequired[str]

| Field | Required | Description |
|----|----|----|
| `type` | Yes | Must be `"preset"` to use a preset system prompt |
| `preset` | Yes | Must be `"claude_code"` to use Claude Code's system prompt |
| `append` | No | Additional instructions to append to the preset system prompt |

### `SettingSource`

Controls which filesystem-based configuration sources the SDK loads settings from.

    SettingSource = Literal["user", "project", "local"]

| Value | Description | Location |
|----|----|----|
| `"user"` | Global user settings | `~/.claude/settings.json` |
| `"project"` | Shared project settings (version controlled) | `.claude/settings.json` |
| `"local"` | Local project settings (gitignored) | `.claude/settings.local.json` |

#### Default behavior

When `setting_sources` is **omitted** or **`None`**, the SDK does **not** load any filesystem settings. This provides isolation for SDK applications.

#### Why use setting_sources?

**Load all filesystem settings (legacy behavior):**

    # Load all settings like SDK v0.0.x did
    from claude_agent_sdk import query, ClaudeAgentOptions

    async for message in query(
        prompt="Analyze this code",
        options=ClaudeAgentOptions(
            setting_sources=["user", "project", "local"]  # Load all settings
        )
    ):
        print(message)

**Load only specific setting sources:**

    # Load only project settings, ignore user and local
    async for message in query(
        prompt="Run CI checks",
        options=ClaudeAgentOptions(
            setting_sources=["project"]  # Only .claude/settings.json
        )
    ):
        print(message)

**Testing and CI environments:**

    # Ensure consistent behavior in CI by excluding local settings
    async for message in query(
        prompt="Run tests",
        options=ClaudeAgentOptions(
            setting_sources=["project"],  # Only team-shared settings
            permission_mode="bypassPermissions"
        )
    ):
        print(message)

**SDK-only applications:**

    # Define everything programmatically (default behavior)
    # No filesystem dependencies - setting_sources defaults to None
    async for message in query(
        prompt="Review this PR",
        options=ClaudeAgentOptions(
            # setting_sources=None is the default, no need to specify
            agents={ /* ... */ },
            mcp_servers={ /* ... */ },
            allowed_tools=["Read", "Grep", "Glob"]
        )
    ):
        print(message)

**Loading CLAUDE.md project instructions:**

    # Load project settings to include CLAUDE.md files
    async for message in query(
        prompt="Add a new feature following project conventions",
        options=ClaudeAgentOptions(
            system_prompt={
                "type": "preset",
                "preset": "claude_code"  # Use Claude Code's system prompt
            },
            setting_sources=["project"],  # Required to load CLAUDE.md from project
            allowed_tools=["Read", "Write", "Edit"]
        )
    ):
        print(message)

#### Settings precedence

When multiple sources are loaded, settings are merged with this precedence (highest to lowest):

1. Local settings (`.claude/settings.local.json`)
2. Project settings (`.claude/settings.json`)
3. User settings (`~/.claude/settings.json`)

Programmatic options (like `agents`, `allowed_tools`) always override filesystem settings.

### `AgentDefinition`

Configuration for a subagent defined programmatically.

    @dataclass
    class AgentDefinition:
        description: str
        prompt: str
        tools: list[str] | None = None
        model: Literal["sonnet", "opus", "haiku", "inherit"] | None = None

| Field | Required | Description |
|----|----|----|
| `description` | Yes | Natural language description of when to use this agent |
| `tools` | No | Array of allowed tool names. If omitted, inherits all tools |
| `prompt` | Yes | The agent's system prompt |
| `model` | No | Model override for this agent. If omitted, uses the main model |

### `PermissionMode`

Permission modes for controlling tool execution.

    PermissionMode = Literal[
        "default",           # Standard permission behavior
        "acceptEdits",       # Auto-accept file edits
        "plan",              # Planning mode - no execution
        "bypassPermissions"  # Bypass all permission checks (use with caution)
    ]

### `McpSdkServerConfig`

Configuration for SDK MCP servers created with `create_sdk_mcp_server()`.

    class McpSdkServerConfig(TypedDict):
        type: Literal["sdk"]
        name: str
        instance: Any  # MCP Server instance

### `McpServerConfig`

Union type for MCP server configurations.

    McpServerConfig = McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig

#### `McpStdioServerConfig`

    class McpStdioServerConfig(TypedDict):
        type: NotRequired[Literal["stdio"]]  # Optional for backwards compatibility
        command: str
        args: NotRequired[list[str]]
        env: NotRequired[dict[str, str]]

#### `McpSSEServerConfig`

    class McpSSEServerConfig(TypedDict):
        type: Literal["sse"]
        url: str
        headers: NotRequired[dict[str, str]]

#### `McpHttpServerConfig`

    class McpHttpServerConfig(TypedDict):
        type: Literal["http"]
        url: str
        headers: NotRequired[dict[str, str]]

### `SdkPluginConfig`

Configuration for loading plugins in the SDK.

    class SdkPluginConfig(TypedDict):
        type: Literal["local"]
        path: str

| Field | Type | Description |
|----|----|----|
| `type` | `Literal["local"]` | Must be `"local"` (only local plugins currently supported) |
| `path` | `str` | Absolute or relative path to the plugin directory |

**Example:**

    plugins=[
        {"type": "local", "path": "./my-plugin"},
        {"type": "local", "path": "/absolute/path/to/plugin"}
    ]

For complete information on creating and using plugins, see [Plugins](https://platform.claude.com/docs/en/agent-sdk/plugins).

## Message Types

### `Message`

Union type of all possible messages.

    Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage

> **Note:** When `include_partial_messages=True`, you will also receive `StreamEvent` messages containing real-time text deltas. See [`StreamEvent`](#streamevent) below.

### `UserMessage`

User input message.

    @dataclass
    class UserMessage:
        content: str | list[ContentBlock]

### `AssistantMessage`

Assistant response message with content blocks.

    @dataclass
    class AssistantMessage:
        content: list[ContentBlock]
        model: str

### `SystemMessage`

System message with metadata.

    @dataclass
    class SystemMessage:
        subtype: str
        data: dict[str, Any]

### `ResultMessage`

Final result message with cost and usage information.

    @dataclass
    class ResultMessage:
        subtype: str
        duration_ms: int
        duration_api_ms: int
        is_error: bool
        num_turns: int
        session_id: str
        total_cost_usd: float | None = None
        usage: dict[str, Any] | None = None
        result: str | None = None

### `StreamEvent`

Streaming event message, only received when `include_partial_messages=True`. Contains raw Anthropic API streaming events for real-time text updates.

    @dataclass
    class StreamEvent:
        uuid: str
        session_id: str
        event: dict[str, Any]  # Raw Anthropic streaming event
        parent_tool_use_id: str | None = None

The `event` dict contains Anthropic streaming events. Key event types:

| Event Type | Description | Useful Fields |
|------------|-------------|---------------|
| `message_start` | Conversation begins | `message.model`, `message.usage` |
| `content_block_start` | New content block starting | `index`, `content_block.type` |
| `content_block_delta` | Incremental text update | `delta.type`, `delta.text` |
| `content_block_stop` | Content block complete | `index` |
| `message_delta` | Message metadata update | `delta.stop_reason`, `usage` |
| `message_stop` | Message complete | - |

#### Example - Handling streaming text deltas

    from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage, TextBlock

    async def stream_responses():
        options = ClaudeAgentOptions(
            include_partial_messages=True,  # Enable streaming events
            allowed_tools=["Read"]
        )

        async with ClaudeSDKClient(options=options) as client:
            await client.query("Explain Python decorators")

            async for message in client.receive_response():
                # Handle streaming deltas
                if type(message).__name__ == "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)

                # Handle complete message (also received after streaming)
                elif isinstance(message, AssistantMessage):
                    for block in message.content:
                        if isinstance(block, TextBlock):
                            print(f"\n[Complete]: {block.text}")

## Content Block Types

### `ContentBlock`

Union type of all content blocks.

    ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

### `TextBlock`

Text content block.

    @dataclass
    class TextBlock:
        text: str

### `ThinkingBlock`

Thinking content block (for models with thinking capability).

    @dataclass
    class ThinkingBlock:
        thinking: str
        signature: str

### `ToolUseBlock`

Tool use request block.

    @dataclass
    class ToolUseBlock:
        id: str
        name: str
        input: dict[str, Any]

### `ToolResultBlock`

Tool execution result block.

    @dataclass
    class ToolResultBlock:
        tool_use_id: str
        content: str | list[dict[str, Any]] | None = None
        is_error: bool | None = None

## Error Types

### `ClaudeSDKError`

Base exception class for all SDK errors.

    class ClaudeSDKError(Exception):
        """Base error for Claude SDK."""

### `CLINotFoundError`

Raised when Claude Code CLI is not installed or not found.

    class CLINotFoundError(CLIConnectionError):
        def __init__(self, message: str = "Claude Code not found", cli_path: str | None = None):
            """
            Args:
                message: Error message (default: "Claude Code not found")
                cli_path: Optional path to the CLI that was not found
            """

### `CLIConnectionError`

Raised when connection to Claude Code fails.

    class CLIConnectionError(ClaudeSDKError):
        """Failed to connect to Claude Code."""

### `ProcessError`

Raised when the Claude Code process fails.

    class ProcessError(ClaudeSDKError):
        def __init__(self, message: str, exit_code: int | None = None, stderr: str | None = None):
            self.exit_code = exit_code
            self.stderr = stderr

### `CLIJSONDecodeError`

Raised when JSON parsing fails.

    class CLIJSONDecodeError(ClaudeSDKError):
        def __init__(self, line: str, original_error: Exception):
            """
            Args:
                line: The line that failed to parse
                original_error: The original JSON decode exception
            """
            self.line = line
            self.original_error = original_error

## Hook Types

### `HookEvent`

Supported hook event types. Note that due to setup limitations, the Python SDK does not support SessionStart, SessionEnd, and Notification hooks.

    HookEvent = Literal[
        "PreToolUse",      # Called before tool execution
        "PostToolUse",     # Called after tool execution
        "UserPromptSubmit", # Called when user submits a prompt
        "Stop",            # Called when stopping execution
        "SubagentStop",    # Called when a subagent stops
        "PreCompact"       # Called before message compaction
    ]

### `HookCallback`

Type definition for hook callback functions.

    HookCallback = Callable[
        [dict[str, Any], str | None, HookContext],
        Awaitable[dict[str, Any]]
    ]

Parameters:

- `input_data`: Hook-specific input data (see [hook documentation](https://docs.claude.comhttps//code.claude.com/docs/en/hooks#hook-input))
- `tool_use_id`: Optional tool use identifier (for tool-related hooks)
- `context`: Hook context with additional information

Returns a dictionary that may contain:

- `decision`: `"block"` to block the action
- `systemMessage`: System message to add to the transcript
- `hookSpecificOutput`: Hook-specific output data

### `HookContext`

Context information passed to hook callbacks.

    @dataclass
    class HookContext:
        signal: Any | None = None  # Future: abort signal support

### `HookMatcher`

Configuration for matching hooks to specific events or tools.

    @dataclass
    class HookMatcher:
        matcher: str | None = None        # Tool name or pattern to match (e.g., "Bash", "Write|Edit")
        hooks: list[HookCallback] = field(default_factory=list)  # List of callbacks to execute

### Hook Usage Example

    from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher, HookContext
    from typing import Any

    async def validate_bash_command(
        input_data: dict[str, Any],
        tool_use_id: str | None,
        context: HookContext
    ) -> dict[str, Any]:
        """Validate and potentially block dangerous bash commands."""
        if input_data['tool_name'] == 'Bash':
            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 log_tool_use(
        input_data: dict[str, Any],
        tool_use_id: str | None,
        context: HookContext
    ) -> dict[str, Any]:
        """Log all tool usage for auditing."""
        print(f"Tool used: {input_data.get('tool_name')}")
        return {}

    options = ClaudeAgentOptions(
        hooks={
            'PreToolUse': [
                HookMatcher(matcher='Bash', hooks=[validate_bash_command]),
                HookMatcher(hooks=[log_tool_use])  # Applies to all tools
            ],
            'PostToolUse': [
                HookMatcher(hooks=[log_tool_use])
            ]
        }
    )

    async for message in query(
        prompt="Analyze this codebase",
        options=options
    ):
        print(message)

## Tool Input/Output Types

Documentation of input/output schemas for all built-in Claude Code tools. While the Python SDK doesn't export these as types, they represent the structure of tool inputs and outputs in messages.

### Task

**Tool name:** `Task`

**Input:**

    {
        "description": str,      # A short (3-5 word) description of the task
        "prompt": str,           # The task for the agent to perform
        "subagent_type": str     # The type of specialized agent to use
    }

**Output:**

    {
        "result": str,                    # Final result from the subagent
        "usage": dict | None,             # Token usage statistics
        "total_cost_usd": float | None,  # Total cost in USD
        "duration_ms": int | None         # Execution duration in milliseconds
    }

### Bash

**Tool name:** `Bash`

**Input:**

    {
        "command": str,                  # The command to execute
        "timeout": int | None,           # Optional timeout in milliseconds (max 600000)
        "description": str | None,       # Clear, concise description (5-10 words)
        "run_in_background": bool | None # Set to true to run in background
    }

**Output:**

    {
        "output": str,              # Combined stdout and stderr output
        "exitCode": int,            # Exit code of the command
        "killed": bool | None,      # Whether command was killed due to timeout
        "shellId": str | None       # Shell ID for background processes
    }

### Edit

**Tool name:** `Edit`

**Input:**

    {
        "file_path": str,           # The absolute path to the file to modify
        "old_string": str,          # The text to replace
        "new_string": str,          # The text to replace it with
        "replace_all": bool | None  # Replace all occurrences (default False)
    }

**Output:**

    {
        "message": str,      # Confirmation message
        "replacements": int, # Number of replacements made
        "file_path": str     # File path that was edited
    }

### Read

**Tool name:** `Read`

**Input:**

    {
        "file_path": str,       # The absolute path to the file to read
        "offset": int | None,   # The line number to start reading from
        "limit": int | None     # The number of lines to read
    }

**Output (Text files):**

    {
        "content": str,         # File contents with line numbers
        "total_lines": int,     # Total number of lines in file
        "lines_returned": int   # Lines actually returned
    }

**Output (Images):**

    {
        "image": str,       # Base64 encoded image data
        "mime_type": str,   # Image MIME type
        "file_size": int    # File size in bytes
    }

### Write

**Tool name:** `Write`

**Input:**

    {
        "file_path": str,  # The absolute path to the file to write
        "content": str     # The content to write to the file
    }

**Output:**

    {
        "message": str,        # Success message
        "bytes_written": int,  # Number of bytes written
        "file_path": str       # File path that was written
    }

### Glob

**Tool name:** `Glob`

**Input:**

    {
        "pattern": str,       # The glob pattern to match files against
        "path": str | None    # The directory to search in (defaults to cwd)
    }

**Output:**

    {
        "matches": list[str],  # Array of matching file paths
        "count": int,          # Number of matches found
        "search_path": str     # Search directory used
    }

### Grep

**Tool name:** `Grep`

**Input:**

    {
        "pattern": str,                    # The regular expression pattern
        "path": str | None,                # File or directory to search in
        "glob": str | None,                # Glob pattern to filter files
        "type": str | None,                # File type to search
        "output_mode": str | None,         # "content", "files_with_matches", or "count"
        "-i": bool | None,                 # Case insensitive search
        "-n": bool | None,                 # Show line numbers
        "-B": int | None,                  # Lines to show before each match
        "-A": int | None,                  # Lines to show after each match
        "-C": int | None,                  # Lines to show before and after
        "head_limit": int | None,          # Limit output to first N lines/entries
        "multiline": bool | None           # Enable multiline mode
    }

**Output (content mode):**

    {
        "matches": [
            {
                "file": str,
                "line_number": int | None,
                "line": str,
                "before_context": list[str] | None,
                "after_context": list[str] | None
            }
        ],
        "total_matches": int
    }

**Output (files_with_matches mode):**

    {
        "files": list[str],  # Files containing matches
        "count": int         # Number of files with matches
    }

### NotebookEdit

**Tool name:** `NotebookEdit`

**Input:**

    {
        "notebook_path": str,                     # Absolute path to the Jupyter notebook
        "cell_id": str | None,                    # The ID of the cell to edit
        "new_source": str,                        # The new source for the cell
        "cell_type": "code" | "markdown" | None,  # The type of the cell
        "edit_mode": "replace" | "insert" | "delete" | None  # Edit operation type
    }

**Output:**

    {
        "message": str,                              # Success message
        "edit_type": "replaced" | "inserted" | "deleted",  # Type of edit performed
        "cell_id": str | None,                       # Cell ID that was affected
        "total_cells": int                           # Total cells in notebook after edit
    }

### WebFetch

**Tool name:** `WebFetch`

**Input:**

    {
        "url": str,     # The URL to fetch content from
        "prompt": str   # The prompt to run on the fetched content
    }

**Output:**

    {
        "response": str,           # AI model's response to the prompt
        "url": str,                # URL that was fetched
        "final_url": str | None,   # Final URL after redirects
        "status_code": int | None  # HTTP status code
    }

### WebSearch

**Tool name:** `WebSearch`

**Input:**

    {
        "query": str,                        # The search query to use
        "allowed_domains": list[str] | None, # Only include results from these domains
        "blocked_domains": list[str] | None  # Never include results from these domains
    }

**Output:**

    {
        "results": [
            {
                "title": str,
                "url": str,
                "snippet": str,
                "metadata": dict | None
            }
        ],
        "total_results": int,
        "query": str
    }

### TodoWrite

**Tool name:** `TodoWrite`

**Input:**

    {
        "todos": [
            {
                "content": str,                              # The task description
                "status": "pending" | "in_progress" | "completed",  # Task status
                "activeForm": str                            # Active form of the description
            }
        ]
    }

**Output:**

    {
        "message": str,  # Success message
        "stats": {
            "total": int,
            "pending": int,
            "in_progress": int,
            "completed": int
        }
    }

### BashOutput

**Tool name:** `BashOutput`

**Input:**

    {
        "bash_id": str,       # The ID of the background shell
        "filter": str | None  # Optional regex to filter output lines
    }

**Output:**

    {
        "output": str,                                      # New output since last check
        "status": "running" | "completed" | "failed",       # Current shell status
        "exitCode": int | None                              # Exit code when completed
    }

### KillBash

**Tool name:** `KillBash`

**Input:**

    {
        "shell_id": str  # The ID of the background shell to kill
    }

**Output:**

    {
        "message": str,  # Success message
        "shell_id": str  # ID of the killed shell
    }

### ExitPlanMode

**Tool name:** `ExitPlanMode`

**Input:**

    {
        "plan": str  # The plan to run by the user for approval
    }

**Output:**

    {
        "message": str,          # Confirmation message
        "approved": bool | None  # Whether user approved the plan
    }

### ListMcpResources

**Tool name:** `ListMcpResources`

**Input:**

    {
        "server": str | None  # Optional server name to filter resources by
    }

**Output:**

    {
        "resources": [
            {
                "uri": str,
                "name": str,
                "description": str | None,
                "mimeType": str | None,
                "server": str
            }
        ],
        "total": int
    }

### ReadMcpResource

**Tool name:** `ReadMcpResource`

**Input:**

    {
        "server": str,  # The MCP server name
        "uri": str      # The resource URI to read
    }

**Output:**

    {
        "contents": [
            {
                "uri": str,
                "mimeType": str | None,
                "text": str | None,
                "blob": str | None
            }
        ],
        "server": str
    }

## Advanced Features with ClaudeSDKClient

### Building a Continuous Conversation Interface

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

    class ConversationSession:
        """Maintains a single conversation session with Claude."""

        def __init__(self, options: ClaudeAgentOptions = None):
            self.client = ClaudeSDKClient(options)
            self.turn_count = 0

        async def start(self):
            await self.client.connect()
            print("Starting conversation session. Claude will remember context.")
            print("Commands: 'exit' to quit, 'interrupt' to stop current task, 'new' for new session")

            while True:
                user_input = input(f"\n[Turn {self.turn_count + 1}] You: ")

                if user_input.lower() == 'exit':
                    break
                elif user_input.lower() == 'interrupt':
                    await self.client.interrupt()
                    print("Task interrupted!")
                    continue
                elif user_input.lower() == 'new':
                    # Disconnect and reconnect for a fresh session
                    await self.client.disconnect()
                    await self.client.connect()
                    self.turn_count = 0
                    print("Started new conversation session (previous context cleared)")
                    continue

                # Send message - Claude remembers all previous messages in this session
                await self.client.query(user_input)
                self.turn_count += 1

                # Process response
                print(f"[Turn {self.turn_count}] Claude: ", end="")
                async for message in self.client.receive_response():
                    if isinstance(message, AssistantMessage):
                        for block in message.content:
                            if isinstance(block, TextBlock):
                                print(block.text, end="")
                print()  # New line after response

            await self.client.disconnect()
            print(f"Conversation ended after {self.turn_count} turns.")

    async def main():
        options = ClaudeAgentOptions(
            allowed_tools=["Read", "Write", "Bash"],
            permission_mode="acceptEdits"
        )
        session = ConversationSession(options)
        await session.start()

    # Example conversation:
    # Turn 1 - You: "Create a file called hello.py"
    # Turn 1 - Claude: "I'll create a hello.py file for you..."
    # Turn 2 - You: "What's in that file?"
    # Turn 2 - Claude: "The hello.py file I just created contains..." (remembers!)
    # Turn 3 - You: "Add a main function to it"
    # Turn 3 - Claude: "I'll add a main function to hello.py..." (knows which file!)

    asyncio.run(main())

### Using Hooks for Behavior Modification

    from claude_agent_sdk import (
        ClaudeSDKClient,
        ClaudeAgentOptions,
        HookMatcher,
        HookContext
    )
    import asyncio
    from typing import Any

    async def pre_tool_logger(
        input_data: dict[str, Any],
        tool_use_id: str | None,
        context: HookContext
    ) -> dict[str, Any]:
        """Log all tool usage before execution."""
        tool_name = input_data.get('tool_name', 'unknown')
        print(f"[PRE-TOOL] About to use: {tool_name}")

        # You can modify or block the tool execution here
        if tool_name == "Bash" and "rm -rf" in str(input_data.get('tool_input', {})):
            return {
                'hookSpecificOutput': {
                    'hookEventName': 'PreToolUse',
                    'permissionDecision': 'deny',
                    'permissionDecisionReason': 'Dangerous command blocked'
                }
            }
        return {}

    async def post_tool_logger(
        input_data: dict[str, Any],
        tool_use_id: str | None,
        context: HookContext
    ) -> dict[str, Any]:
        """Log results after tool execution."""
        tool_name = input_data.get('tool_name', 'unknown')
        print(f"[POST-TOOL] Completed: {tool_name}")
        return {}

    async def user_prompt_modifier(
        input_data: dict[str, Any],
        tool_use_id: str | None,
        context: HookContext
    ) -> dict[str, Any]:
        """Add context to user prompts."""
        original_prompt = input_data.get('prompt', '')

        # Add timestamp to all prompts
        from datetime import datetime
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        return {
            'hookSpecificOutput': {
                'hookEventName': 'UserPromptSubmit',
                'updatedPrompt': f"[{timestamp}] {original_prompt}"
            }
        }

    async def main():
        options = ClaudeAgentOptions(
            hooks={
                'PreToolUse': [
                    HookMatcher(hooks=[pre_tool_logger]),
                    HookMatcher(matcher='Bash', hooks=[pre_tool_logger])
                ],
                'PostToolUse': [
                    HookMatcher(hooks=[post_tool_logger])
                ],
                'UserPromptSubmit': [
                    HookMatcher(hooks=[user_prompt_modifier])
                ]
            },
            allowed_tools=["Read", "Write", "Bash"]
        )

        async with ClaudeSDKClient(options=options) as client:
            await client.query("List files in current directory")

            async for message in client.receive_response():
                # Hooks will automatically log tool usage
                pass

    asyncio.run(main())

### Real-time Progress Monitoring

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

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

        async with ClaudeSDKClient(options=options) as client:
            await client.query(
                "Create 5 Python files with different sorting algorithms"
            )

            # Monitor progress in real-time
            files_created = []
            async for message in client.receive_messages():
                if isinstance(message, AssistantMessage):
                    for block in message.content:
                        if isinstance(block, ToolUseBlock):
                            if block.name == "Write":
                                file_path = block.input.get("file_path", "")
                                print(f"🔨 Creating: {file_path}")
                        elif isinstance(block, ToolResultBlock):
                            print(f"✅ Completed tool execution")
                        elif isinstance(block, TextBlock):
                            print(f"💭 Claude says: {block.text[:100]}...")

                # Check if we've received the final result
                if hasattr(message, 'subtype') and message.subtype in ['success', 'error']:
                    print(f"\n🎯 Task completed!")
                    break

    asyncio.run(monitor_progress())

## Example Usage

### Basic file operations (using query)

    from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ToolUseBlock
    import asyncio

    async def create_project():
        options = ClaudeAgentOptions(
            allowed_tools=["Read", "Write", "Bash"],
            permission_mode='acceptEdits',
            cwd="/home/user/project"
        )

        async for message in query(
            prompt="Create a Python project structure with setup.py",
            options=options
        ):
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, ToolUseBlock):
                        print(f"Using tool: {block.name}")

    asyncio.run(create_project())

### Error handling

    from claude_agent_sdk import (
        query,
        CLINotFoundError,
        ProcessError,
        CLIJSONDecodeError
    )

    try:
        async for message in query(prompt="Hello"):
            print(message)
    except CLINotFoundError:
        print("Please install Claude Code: npm install -g @anthropic-ai/claude-code")
    except ProcessError as e:
        print(f"Process failed with exit code: {e.exit_code}")
    except CLIJSONDecodeError as e:
        print(f"Failed to parse response: {e}")

### Streaming mode with client

    from claude_agent_sdk import ClaudeSDKClient
    import asyncio

    async def interactive_session():
        async with ClaudeSDKClient() as client:
            # Send initial message
            await client.query("What's the weather like?")

            # Process responses
            async for msg in client.receive_response():
                print(msg)

            # Send follow-up
            await client.query("Tell me more about that")

            # Process follow-up response
            async for msg in client.receive_response():
                print(msg)

    asyncio.run(interactive_session())

### Using custom tools with ClaudeSDKClient

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

    # Define custom tools with @tool decorator
    @tool("calculate", "Perform mathematical calculations", {"expression": str})
    async def calculate(args: dict[str, Any]) -> dict[str, Any]:
        try:
            result = eval(args["expression"], {"__builtins__": {}})
            return {
                "content": [{
                    "type": "text",
                    "text": f"Result: {result}"
                }]
            }
        except Exception as e:
            return {
                "content": [{
                    "type": "text",
                    "text": f"Error: {str(e)}"
                }],
                "is_error": True
            }

    @tool("get_time", "Get current time", {})
    async def get_time(args: dict[str, Any]) -> dict[str, Any]:
        from datetime import datetime
        current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        return {
            "content": [{
                "type": "text",
                "text": f"Current time: {current_time}"
            }]
        }

    async def main():
        # Create SDK MCP server with custom tools
        my_server = create_sdk_mcp_server(
            name="utilities",
            version="1.0.0",
            tools=[calculate, get_time]
        )

        # Configure options with the server
        options = ClaudeAgentOptions(
            mcp_servers={"utils": my_server},
            allowed_tools=[
                "mcp__utils__calculate",
                "mcp__utils__get_time"
            ]
        )

        # Use ClaudeSDKClient for interactive tool usage
        async with ClaudeSDKClient(options=options) as client:
            await client.query("What's 123 * 456?")

            # Process calculation response
            async for message in client.receive_response():
                if isinstance(message, AssistantMessage):
                    for block in message.content:
                        if isinstance(block, TextBlock):
                            print(f"Calculation: {block.text}")

            # Follow up with time query
            await client.query("What time is it now?")

            async for message in client.receive_response():
                if isinstance(message, AssistantMessage):
                    for block in message.content:
                        if isinstance(block, TextBlock):
                            print(f"Time: {block.text}")

    asyncio.run(main())

## See also

- [Python SDK guide](https://platform.claude.com/docs/en/agent-sdk/python) - Tutorial and examples
- [SDK overview](https://platform.claude.com/docs/en/agent-sdk/overview) - General SDK concepts
- [TypeScript SDK reference](https://platform.claude.com/docs/en/agent-sdk/typescript) - TypeScript SDK documentation
- [CLI reference](https://code.claude.com/docs/en/cli-reference) - Command-line interface
- [Common workflows](https://code.claude.com/docs/en/common-workflows) - Step-by-step guides