Agent SDK reference - Python - Claude Code Docs

Installation

Install the package into a virtual environment. On recent Debian, Ubuntu, and Homebrew Python installs, running pip install against system Python fails with error: externally-managed-environment.

python3 -m venv .venv
source .venv/bin/activate
pip install claude-agent-sdk

For uv, Windows PowerShell, and API key setup, see Setup in the Agent SDK quickstart.

Choosing between `query()` and `ClaudeSDKClient`

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

Feature	`query()`	`ClaudeSDKClient`
Session	Creates a new session by default	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	✅ Supported	✅ Supported
Custom Tools	✅ Supported	✅ Supported
Continue Chat	Manual via `continue_conversation` or `resume`	✅ Automatic
Use Case	One-off tasks	Continuous conversations

Use ClaudeSDKClient for interactive applications such as chat interfaces, or when the next action depends on Claude’s response.

Functions

Signature blocks and bare async for / async with fragments on this page are illustrative. To run them, wrap the body in async def main(): ... and call asyncio.run(main()).

`query()`

Creates a new session for each interaction with Claude Code by default. Returns an async iterator that yields messages as they arrive. Each call to query() starts fresh with no memory of previous interactions unless you pass continue_conversation=True or resume in ClaudeAgentOptions. See Sessions.

async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeAgentOptions | None = None,
    transport: Transport | 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)
`transport`	`Transport \| None`	Optional custom transport for communicating with the CLI process

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

    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],
    annotations: ToolAnnotations | None = None
) -> 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)
`annotations`	`ToolAnnotations` `\| None`	Optional MCP tool annotations providing behavioral hints to clients

Input schema options

Simple type mapping (recommended):

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

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']}!"}]}

`ToolAnnotations`

Re-exported from mcp.types (also available as from claude_agent_sdk import ToolAnnotations). All fields are optional hints; clients should not rely on them for security decisions.

Field	Type	Default	Description
`title`	`str \| None`	`None`	Human-readable title for the tool
`readOnlyHint`	`bool \| None`	`False`	If `True`, the tool does not modify its environment
`destructiveHint`	`bool \| None`	`True`	If `True`, the tool may perform destructive updates (only meaningful when `readOnlyHint` is `False`)
`idempotentHint`	`bool \| None`	`False`	If `True`, repeated calls with the same arguments have no additional effect (only meaningful when `readOnlyHint` is `False`)
`openWorldHint`	`bool \| None`	`True`	If `True`, the tool interacts with external entities (for example, web search). If `False`, the tool’s domain is closed (for example, a memory tool)

from claude_agent_sdk import tool, ToolAnnotations
from typing import Any


@tool(
    "search",
    "Search the web",
    {"query": str},
    annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True),
)
async def search(args: dict[str, Any]) -> dict[str, Any]:
    return {"content": [{"type": "text", "text": f"Results for: {args['query']}"}]}

`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, ClaudeAgentOptions


@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"],
)

`list_sessions()`

Lists past sessions with metadata. Filter by project directory or list sessions across all projects. Synchronous; returns immediately.

def list_sessions(
    directory: str | None = None,
    limit: int | None = None,
    offset: int = 0,
    include_worktrees: bool = True
) -> list[SDKSessionInfo]

Parameters

Parameter	Type	Default	Description
`directory`	`str \| None`	`None`	Directory to list sessions for. When omitted, returns sessions across all projects
`limit`	`int \| None`	`None`	Maximum number of sessions to return
`offset`	`int`	`0`	Number of sessions to skip from the start of the sorted results. Use with `limit` for pagination
`include_worktrees`	`bool`	`True`	When `directory` is inside a git repository, include sessions from all worktree paths

Return type: `SDKSessionInfo`

Property	Type	Description
`session_id`	`str`	Unique session identifier
`summary`	`str`	Display title: custom title, auto-generated summary, or first prompt
`last_modified`	`int`	Last modified time in milliseconds since epoch
`file_size`	`int \| None`	Session file size in bytes (`None` for remote storage backends)
`custom_title`	`str \| None`	User-set session title
`first_prompt`	`str \| None`	First meaningful user prompt in the session
`git_branch`	`str \| None`	Git branch at the end of the session
`cwd`	`str \| None`	Working directory for the session
`tag`	`str \| None`	User-set session tag (see `tag_session()`)
`created_at`	`int \| None`	Session creation time in milliseconds since epoch

Example

Print the 10 most recent sessions for a project. Results are sorted by last_modified descending, so the first item is the newest. Omit directory to search across all projects.

from claude_agent_sdk import list_sessions

for session in list_sessions(directory="/path/to/project", limit=10):
    print(f"{session.summary} ({session.session_id})")

`get_session_messages()`

Retrieves messages from a past session. Synchronous; returns immediately.

def get_session_messages(
    session_id: str,
    directory: str | None = None,
    limit: int | None = None,
    offset: int = 0
) -> list[SessionMessage]

Parameters

Parameter	Type	Default	Description
`session_id`	`str`	required	The session ID to retrieve messages for
`directory`	`str \| None`	`None`	Project directory to look in. When omitted, searches all projects
`limit`	`int \| None`	`None`	Maximum number of messages to return
`offset`	`int`	`0`	Number of messages to skip from the start

Return type: `SessionMessage`

Property	Type	Description
`type`	`Literal["user", "assistant"]`	Message role
`uuid`	`str`	Unique message identifier
`session_id`	`str`	Session identifier
`message`	`Any`	Raw message content
`parent_tool_use_id`	`None`	Reserved for future use

Example

from claude_agent_sdk import list_sessions, get_session_messages

sessions = list_sessions(limit=1)
if sessions:
    messages = get_session_messages(sessions[0].session_id)
    for msg in messages:
        print(f"[{msg.type}] {msg.uuid}")

`get_session_info()`

Reads metadata for a single session by ID without scanning the full project directory. Synchronous; returns immediately.

def get_session_info(
    session_id: str,
    directory: str | None = None,
) -> SDKSessionInfo | None

Parameters

Parameter	Type	Default	Description
`session_id`	`str`	required	UUID of the session to look up
`directory`	`str \| None`	`None`	Project directory path. When omitted, searches all project directories

Returns SDKSessionInfo, or None if the session is not found.

Example

Look up a single session’s metadata without scanning the project directory. Useful when you already have a session ID from a previous run.

from claude_agent_sdk import get_session_info

info = get_session_info("550e8400-e29b-41d4-a716-446655440000")
if info:
    print(f"{info.summary} (branch: {info.git_branch}, tag: {info.tag})")

`rename_session()`

Renames a session by appending a custom-title entry. Repeated calls are safe; the most recent title wins. Synchronous.

def rename_session(
    session_id: str,
    title: str,
    directory: str | None = None,
) -> None

Parameters

Parameter	Type	Default	Description
`session_id`	`str`	required	UUID of the session to rename
`title`	`str`	required	New title. Must be non-empty after stripping whitespace
`directory`	`str \| None`	`None`	Project directory path. When omitted, searches all project directories

Raises ValueError if session_id is not a valid UUID or title is empty; FileNotFoundError if the session cannot be found.

Example

Rename the most recent session so it’s easier to find later. The new title appears in SDKSessionInfo.custom_title on subsequent reads.

from claude_agent_sdk import list_sessions, rename_session

sessions = list_sessions(directory="/path/to/project", limit=1)
if sessions:
    rename_session(sessions[0].session_id, "Refactor auth module")

`tag_session()`

Tags a session. Pass None to clear the tag. Repeated calls are safe; the most recent tag wins. Synchronous.

def tag_session(
    session_id: str,
    tag: str | None,
    directory: str | None = None,
) -> None

Parameters

Parameter	Type	Default	Description
`session_id`	`str`	required	UUID of the session to tag
`tag`	`str \| None`	required	Tag string, or `None` to clear. Unicode-sanitized before storing
`directory`	`str \| None`	`None`	Project directory path. When omitted, searches all project directories

Raises ValueError if session_id is not a valid UUID or tag is empty after sanitization; FileNotFoundError if the session cannot be found.

Example

Tag a session, then filter by that tag on a later read. Pass None to clear an existing tag.

from claude_agent_sdk import list_sessions, tag_session

# Tag the most recent session
sessions = list_sessions(directory="/path/to/project", limit=1)
if sessions:
    tag_session(sessions[0].session_id, "needs-review")

# Later: find all sessions with that tag
for session in list_sessions(directory="/path/to/project"):
    if session.tag == "needs-review":
        print(session.summary)

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. See the comparison with query().

class ClaudeSDKClient:
    def __init__(self, options: ClaudeAgentOptions | None = None, transport: Transport | 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 set_permission_mode(self, mode: str) -> None
    async def set_model(self, model: str | None = None) -> None
    async def rewind_files(self, user_message_id: str) -> None
    async def get_mcp_status(self) -> McpStatusResponse
    async def reconnect_mcp_server(self, server_name: str) -> None
    async def toggle_mcp_server(self, server_name: str, enabled: bool) -> None
    async def stop_task(self, task_id: str) -> None
    async def get_server_info(self) -> dict[str, Any] | 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)
`set_permission_mode(mode)`	Change the permission mode for the current session
`set_model(model)`	Change the model for the current session. Pass `None` to reset to default
`rewind_files(user_message_id)`	Restore files to their state at the specified user message. Requires `enable_file_checkpointing=True`. See File checkpointing
`get_mcp_status()`	Get the status of all configured MCP servers. Returns `McpStatusResponse`
`reconnect_mcp_server(server_name)`	Retry connecting to an MCP server that failed or was disconnected
`toggle_mcp_server(server_name, enabled)`	Enable or disable an MCP server mid-session. Disabling removes its tools
`stop_task(task_id)`	Stop a running background task. A `TaskNotificationMessage` with status `"stopped"` follows in the message stream
`get_server_info()`	Get server information including session ID and capabilities
`disconnect()`	Disconnect from Claude

Context Manager Support

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

import asyncio
from claude_agent_sdk import ClaudeSDKClient


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


asyncio.run(main())

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 - the session retains 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": "user",
        "message": {"role": "user", "content": "Analyze the following data:"},
    }
    await asyncio.sleep(0.5)
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Temperature: 25°C, Humidity: 60%"},
    }
    await asyncio.sleep(0.5)
    yield {
        "type": "user",
        "message": {"role": "user", "content": "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, ResultMessage


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, using the bash sleep command")

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

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

        # Drain the interrupted task's messages (including its ResultMessage)
        async for message in client.receive_response():
            if isinstance(message, ResultMessage):
                print(f"Interrupted task: terminal_reason={message.terminal_reason!r}")
                # terminal_reason is "aborted_streaming" or "aborted_tools"
                # for interrupted turns

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

        # Now receive the new response
        async for message in client.receive_response():
            if isinstance(message, ResultMessage) and message.subtype == "success":
                print(f"New result: {message.result}")


asyncio.run(interruptible_task())

​Installation

​Choosing between query() and ClaudeSDKClient

​Functions

​query()

​Parameters

​Returns

​Example - With options

​tool()

​Parameters

​Input schema options

​Returns

​Example

​ToolAnnotations

​create_sdk_mcp_server()

​Parameters

​Returns

​Example

​list_sessions()

​Parameters

​Return type: SDKSessionInfo

​Example

​get_session_messages()

​Parameters

​Return type: SessionMessage

​Example

​get_session_info()

​Parameters

​Example

​rename_session()

​Parameters

​Example

​tag_session()

​Parameters

​Example

​Classes

​ClaudeSDKClient

​Methods

​Context Manager Support

​Example - Continuing a conversation

​Example - Streaming input with ClaudeSDKClient

​Example - Using interrupts

Installation

Choosing between `query()` and `ClaudeSDKClient`

Functions

`query()`

Parameters

Returns

Example - With options

`tool()`

Parameters

Input schema options

Returns

Example

`ToolAnnotations`

`create_sdk_mcp_server()`

Parameters

Returns

Example

`list_sessions()`

Parameters

Return type: `SDKSessionInfo`

Example

`get_session_messages()`

Parameters

Return type: `SessionMessage`

Example

`get_session_info()`

Parameters

Example

`rename_session()`

Parameters

Example

`tag_session()`

Parameters

Example

Classes

`ClaudeSDKClient`

Methods

Context Manager Support

Example - Continuing a conversation

Example - Streaming input with ClaudeSDKClient

Example - Using interrupts