pip install claude-agent-sdkquery() and ClaudeSDKClientThe Python SDK provides two ways to interact with Claude Code:
| 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 | ✅ Supported | ✅ Supported |
| Custom Tools | ✅ Supported | ✅ Supported |
| Continue Chat | ❌ New session each time | ✅ Maintains conversation |
| Use Case | One-off tasks | Continuous conversations |
query() (new session each time)Best for:
ClaudeSDKClient (continuous conversation)Best for:
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,
transport: Transport | None = None
) -> AsyncIterator[Message]| 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 an AsyncIterator[Message] that yields messages from the conversation.
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],
annotations: ToolAnnotations | None = None
) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]| 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 (e.g., readOnlyHint, destructiveHint, openWorldHint). Imported from mcp.types |
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"],
}A decorator function that wraps the tool implementation and returns an SdkMcpTool instance.
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| 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 an McpSdkServerConfig object that can be passed to ClaudeAgentOptions.mcp_servers.
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"],
)ClaudeSDKClientMaintains 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.
query() calls@tool decorator) and hooksclass 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) -> dict[str, Any]
async def get_server_info(self) -> dict[str, Any] | None
async def disconnect(self) -> None| 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 |
get_server_info() | Get server information including session ID and capabilities |
disconnect() | Disconnect from Claude |
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
breakto 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.
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())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())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())from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
from claude_agent_sdk.types import (
PermissionResultAllow,
PermissionResultDeny,
ToolPermissionContext,
)
async def custom_permission_handler(
tool_name: str, input_data: dict, context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
"""Custom logic for tool permissions."""
# Block writes to system directories
if tool_name == "Write" and input_data.get("file_path", "").startswith("/system/"):
return PermissionResultDeny(
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 PermissionResultAllow(
updated_input={**input_data, "file_path": safe_path}
)
# Allow everything else
return PermissionResultAllow(updated_input=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())SdkMcpToolDefinition 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]]]
annotations: ToolAnnotations | None = None| 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 |
annotations | ToolAnnotations | None | Optional MCP tool annotations (e.g., readOnlyHint, destructiveHint, openWorldHint). From mcp.types |
TransportAbstract base class for custom transport implementations. Use this to communicate with the Claude process over a custom channel (for example, a remote connection instead of a local subprocess).
This is a low-level internal API. The interface may change in future releases. Custom implementations must be updated to match any interface changes.
from abc import ABC, abstractmethod
from collections.abc import AsyncIterator
from typing import Any
class Transport(ABC):
@abstractmethod
async def connect(self) -> None: ...
@abstractmethod
async def write(self, data: str) -> None: ...
@abstractmethod
def read_messages(self) -> AsyncIterator[dict[str, Any]]: ...
@abstractmethod
async def close(self) -> None: ...
@abstractmethod
def is_ready(self) -> bool: ...
@abstractmethod
async def end_input(self) -> None: ...| Method | Description |
|---|---|
connect() | Connect the transport and prepare for communication |
write(data) | Write raw data (JSON + newline) to the transport |
read_messages() | Async iterator that yields parsed JSON messages |
close() | Close the connection and clean up resources |
is_ready() | Returns True if the transport can send and receive |
end_input() | Close the input stream (for example, close stdin for subprocess transports) |
Import: from claude_agent_sdk import Transport
ClaudeAgentOptionsConfiguration dataclass for Claude Code queries.
@dataclass
class ClaudeAgentOptions:
tools: list[str] | ToolsPreset | None = None
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
max_budget_usd: float | None = None
disallowed_tools: list[str] = field(default_factory=list)
model: str | None = None
fallback_model: str | None = None
betas: list[SdkBeta] = field(default_factory=list)
output_format: dict[str, Any] | None = None
permission_prompt_tool_name: str | None = None
cwd: str | Path | None = None
cli_path: 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
sandbox: SandboxSettings | None = None
plugins: list[SdkPluginConfig] = field(default_factory=list)
max_thinking_tokens: int | None = None # Deprecated: use thinking instead
thinking: ThinkingConfig | None = None
effort: Literal["low", "medium", "high", "max"] | None = None
enable_file_checkpointing: bool = False| Property | Type | Default | Description |
|---|---|---|---|
tools | list[str] | ToolsPreset | None | None | Tools configuration. Use {"type": "preset", "preset": "claude_code"} for Claude Code's default tools |
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 |
max_budget_usd | float | None | None | Maximum budget in USD for the session |
disallowed_tools | list[str] | [] | List of disallowed tool names |
enable_file_checkpointing | bool | False | Enable file change tracking for rewinding. See File checkpointing |
model | str | None | None | Claude model to use |
fallback_model | str | None | None | Fallback model to use if the primary model fails |
betas | list[SdkBeta] | [] | Beta features to enable. See SdkBeta for available options |
output_format | dict[str, Any] | None | None | Output format for structured responses (e.g., {"type": "json_schema", "schema": {...}}). See 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 |
cli_path | str | Path | None | None | Custom path to the Claude Code CLI executable |
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. See Permission types for details |
hooks | dict[HookEvent, list[HookMatcher]] | None | None | Hook configurations for intercepting events |
user | str | None | None | User identifier |
include_partial_messages | bool | False | Include partial message streaming events. When enabled, StreamEvent messages are yielded |
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 for details |
sandbox | SandboxSettings | None | None | Configure sandbox behavior programmatically. See Sandbox settings 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 |
max_thinking_tokens | int | None | None | Deprecated - Maximum tokens for thinking blocks. Use thinking instead |
thinking | ThinkingConfig | None | None | Controls extended thinking behavior. Takes precedence over max_thinking_tokens |
effort | Literal["low", "medium", "high", "max"] | None | None | Effort level for thinking depth |
OutputFormatConfiguration for structured output validation. Pass this as a dict to the output_format field on ClaudeAgentOptions:
# Expected dict shape for output_format
{
"type": "json_schema",
"schema": {...}, # Your JSON Schema definition
}| Field | Required | Description |
|---|---|---|
type | Yes | Must be "json_schema" for JSON Schema validation |
schema | Yes | JSON Schema definition for output validation |
SystemPromptPresetConfiguration 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 |
SettingSourceControls 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 |
When setting_sources is omitted or None, the SDK does not load any filesystem settings. This provides isolation for SDK applications.
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)When multiple sources are loaded, settings are merged with this precedence (highest to lowest):
.claude/settings.local.json).claude/settings.json)~/.claude/settings.json)Programmatic options (like agents, allowed_tools) always override filesystem settings.
AgentDefinitionConfiguration 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 |
PermissionModePermission 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)
]CanUseToolType alias for tool permission callback functions.
CanUseTool = Callable[
[str, dict[str, Any], ToolPermissionContext], Awaitable[PermissionResult]
]The callback receives:
tool_name: Name of the tool being calledinput_data: The tool's input parameterscontext: A ToolPermissionContext with additional informationReturns a PermissionResult (either PermissionResultAllow or PermissionResultDeny).
ToolPermissionContextContext information passed to tool permission callbacks.
@dataclass
class ToolPermissionContext:
signal: Any | None = None # Future: abort signal support
suggestions: list[PermissionUpdate] = field(default_factory=list)| Field | Type | Description |
|---|---|---|
signal | Any | None | Reserved for future abort signal support |
suggestions | list[PermissionUpdate] | Permission update suggestions from the CLI |
PermissionResultUnion type for permission callback results.
PermissionResult = PermissionResultAllow | PermissionResultDenyPermissionResultAllowResult indicating the tool call should be allowed.
@dataclass
class PermissionResultAllow:
behavior: Literal["allow"] = "allow"
updated_input: dict[str, Any] | None = None
updated_permissions: list[PermissionUpdate] | None = None| Field | Type | Default | Description |
|---|---|---|---|
behavior | Literal["allow"] | "allow" | Must be "allow" |
updated_input | dict[str, Any] | None | None | Modified input to use instead of original |
updated_permissions | list[PermissionUpdate] | None | None | Permission updates to apply |
PermissionResultDenyResult indicating the tool call should be denied.
@dataclass
class PermissionResultDeny:
behavior: Literal["deny"] = "deny"
message: str = ""
interrupt: bool = False| Field | Type | Default | Description |
|---|---|---|---|
behavior | Literal["deny"] | "deny" | Must be "deny" |
message | str | "" | Message explaining why the tool was denied |
interrupt | bool | False | Whether to interrupt the current execution |
PermissionUpdateConfiguration for updating permissions programmatically.
@dataclass
class PermissionUpdate:
type: Literal[
"addRules",
"replaceRules",
"removeRules",
"setMode",
"addDirectories",
"removeDirectories",
]
rules: list[PermissionRuleValue] | None = None
behavior: Literal["allow", "deny", "ask"] | None = None
mode: PermissionMode | None = None
directories: list[str] | None = None
destination: (
Literal["userSettings", "projectSettings", "localSettings", "session"] | None
) = None| Field | Type | Description |
|---|---|---|
type | Literal[...] | The type of permission update operation |
rules | list[PermissionRuleValue] | None | Rules for add/replace/remove operations |
behavior | Literal["allow", "deny", "ask"] | None | Behavior for rule-based operations |
mode | PermissionMode | None | Mode for setMode operation |
directories | list[str] | None | Directories for add/remove directory operations |
destination | Literal[...] | None | Where to apply the permission update |
PermissionRuleValueA rule to add, replace, or remove in a permission update.
@dataclass
class PermissionRuleValue:
tool_name: str
rule_content: str | None = NoneToolsPresetPreset tools configuration for using Claude Code's default tool set.
class ToolsPreset(TypedDict):
type: Literal["preset"]
preset: Literal["claude_code"]ThinkingConfigControls extended thinking behavior. A union of three configurations:
class ThinkingConfigAdaptive(TypedDict):
type: Literal["adaptive"]
class ThinkingConfigEnabled(TypedDict):
type: Literal["enabled"]
budget_tokens: int
class ThinkingConfigDisabled(TypedDict):
type: Literal["disabled"]
ThinkingConfig = ThinkingConfigAdaptive | ThinkingConfigEnabled | ThinkingConfigDisabled| Variant | Fields | Description |
|---|---|---|
adaptive | type | Claude adaptively decides when to think |
enabled | type, budget_tokens | Enable thinking with a specific token budget |
disabled | type | Disable thinking |
SdkBetaLiteral type for SDK beta features.
SdkBeta = Literal["context-1m-2025-08-07"]Use with the betas field in ClaudeAgentOptions to enable beta features.
McpSdkServerConfigConfiguration for SDK MCP servers created with create_sdk_mcp_server().
class McpSdkServerConfig(TypedDict):
type: Literal["sdk"]
name: str
instance: Any # MCP Server instanceMcpServerConfigUnion type for MCP server configurations.
McpServerConfig = (
McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig
)McpStdioServerConfigclass McpStdioServerConfig(TypedDict):
type: NotRequired[Literal["stdio"]] # Optional for backwards compatibility
command: str
args: NotRequired[list[str]]
env: NotRequired[dict[str, str]]McpSSEServerConfigclass McpSSEServerConfig(TypedDict):
type: Literal["sse"]
url: str
headers: NotRequired[dict[str, str]]McpHttpServerConfigclass McpHttpServerConfig(TypedDict):
type: Literal["http"]
url: str
headers: NotRequired[dict[str, str]]SdkPluginConfigConfiguration 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.
MessageUnion type of all possible messages.
Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEventUserMessageUser input message.
@dataclass
class UserMessage:
content: str | list[ContentBlock]
uuid: str | None = None
parent_tool_use_id: str | None = None
tool_use_result: dict[str, Any] | None = None| Field | Type | Description |
|---|---|---|
content | str | list[ContentBlock] | Message content as text or content blocks |
uuid | str | None | Unique message identifier |
parent_tool_use_id | str | None | Tool use ID if this message is a tool result response |
tool_use_result | dict[str, Any] | None | Tool result data if applicable |
AssistantMessageAssistant response message with content blocks.
@dataclass
class AssistantMessage:
content: list[ContentBlock]
model: str
parent_tool_use_id: str | None = None
error: AssistantMessageError | None = None| Field | Type | Description |
|---|---|---|
content | list[ContentBlock] | List of content blocks in the response |
model | str | Model that generated the response |
parent_tool_use_id | str | None | Tool use ID if this is a nested response |
error | AssistantMessageError | None | Error type if the response encountered an error |
AssistantMessageErrorPossible error types for assistant messages.
AssistantMessageError = Literal[
"authentication_failed",
"billing_error",
"rate_limit",
"invalid_request",
"server_error",
"unknown",
]SystemMessageSystem message with metadata.
@dataclass
class SystemMessage:
subtype: str
data: dict[str, Any]ResultMessageFinal 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
structured_output: Any = NoneThe usage dict contains the following keys when present:
| Key | Type | Description |
|---|---|---|
input_tokens | int | Total input tokens consumed. |
output_tokens | int | Total output tokens generated. |
cache_creation_input_tokens | int | Tokens used to create new cache entries. |
cache_read_input_tokens | int | Tokens read from existing cache entries. |
StreamEventStream event for partial message updates during streaming. Only received when include_partial_messages=True in ClaudeAgentOptions. Import via from claude_agent_sdk.types import StreamEvent.
@dataclass
class StreamEvent:
uuid: str
session_id: str
event: dict[str, Any] # The raw Claude API stream event
parent_tool_use_id: str | None = None| Field | Type | Description |
|---|---|---|
uuid | str | Unique identifier for this event |
session_id | str | Session identifier |
event | dict[str, Any] | The raw Claude API stream event data |
parent_tool_use_id | str | None | Parent tool use ID if this event is from a subagent |
ContentBlockUnion type of all content blocks.
ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlockTextBlockText content block.
@dataclass
class TextBlock:
text: strThinkingBlockThinking content block (for models with thinking capability).
@dataclass
class ThinkingBlock:
thinking: str
signature: strToolUseBlockTool use request block.
@dataclass
class ToolUseBlock:
id: str
name: str
input: dict[str, Any]ToolResultBlockTool execution result block.
@dataclass
class ToolResultBlock:
tool_use_id: str
content: str | list[dict[str, Any]] | None = None
is_error: bool | None = NoneClaudeSDKErrorBase exception class for all SDK errors.
class ClaudeSDKError(Exception):
"""Base error for Claude SDK."""CLINotFoundErrorRaised 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
"""CLIConnectionErrorRaised when connection to Claude Code fails.
class CLIConnectionError(ClaudeSDKError):
"""Failed to connect to Claude Code."""ProcessErrorRaised 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 = stderrCLIJSONDecodeErrorRaised 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_errorFor a comprehensive guide on using hooks with examples and common patterns, see the Hooks guide.
HookEventSupported hook event types.
HookEvent = Literal[
"PreToolUse", # Called before tool execution
"PostToolUse", # Called after tool execution
"PostToolUseFailure", # Called when a tool execution fails
"UserPromptSubmit", # Called when user submits a prompt
"Stop", # Called when stopping execution
"SubagentStop", # Called when a subagent stops
"PreCompact", # Called before message compaction
"Notification", # Called for notification events
"SubagentStart", # Called when a subagent starts
"PermissionRequest", # Called when a permission decision is needed
]HookCallbackType definition for hook callback functions.
HookCallback = Callable[[HookInput, str | None, HookContext], Awaitable[HookJSONOutput]]Parameters:
input: Strongly-typed hook input with discriminated unions based on hook_event_name (see Hook input types)tool_use_id: Optional tool use identifier (for tool-related hooks)context: Hook context with additional informationReturns a HookJSONOutput that may contain:
decision: "block" to block the actionsystemMessage: System message to add to the transcripthookSpecificOutput: Hook-specific output dataHookContextContext information passed to hook callbacks.
class HookContext(TypedDict):
signal: Any | None # Future: abort signal supportHookMatcherConfiguration 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
timeout: float | None = (
None # Timeout in seconds for all hooks in this matcher (default: 60)
)HookInputUnion type of all hook input types. The actual type depends on the hook_event_name field.
HookInput = (
PreToolUseHookInput
| PostToolUseHookInput
| PostToolUseFailureHookInput
| UserPromptSubmitHookInput
| StopHookInput
| SubagentStopHookInput
| PreCompactHookInput
| NotificationHookInput
| SubagentStartHookInput
| PermissionRequestHookInput
)BaseHookInputBase fields present in all hook input types.
class BaseHookInput(TypedDict):
session_id: str
transcript_path: str
cwd: str
permission_mode: NotRequired[str]| Field | Type | Description |
|---|---|---|
session_id | str | Current session identifier |
transcript_path | str | Path to the session transcript file |
cwd | str | Current working directory |
permission_mode | str (optional) | Current permission mode |
PreToolUseHookInputInput data for PreToolUse hook events.
class PreToolUseHookInput(BaseHookInput):
hook_event_name: Literal["PreToolUse"]
tool_name: str
tool_input: dict[str, Any]
tool_use_id: str| Field | Type | Description |
|---|---|---|
hook_event_name | Literal["PreToolUse"] | Always "PreToolUse" |
tool_name | str | Name of the tool about to be executed |
tool_input | dict[str, Any] | Input parameters for the tool |
tool_use_id | str | Unique identifier for this tool use |
PostToolUseHookInputInput data for PostToolUse hook events.
class PostToolUseHookInput(BaseHookInput):
hook_event_name: Literal["PostToolUse"]
tool_name: str
tool_input: dict[str, Any]
tool_response: Any
tool_use_id: str| Field | Type | Description |
|---|---|---|
hook_event_name | Literal["PostToolUse"] | Always "PostToolUse" |
tool_name | str | Name of the tool that was executed |
tool_input | dict[str, Any] | Input parameters that were used |
tool_response | Any | Response from the tool execution |
tool_use_id | str | Unique identifier for this tool use |
PostToolUseFailureHookInputInput data for PostToolUseFailure hook events. Called when a tool execution fails.
class PostToolUseFailureHookInput(BaseHookInput):
hook_event_name: Literal["PostToolUseFailure"]
tool_name: str
tool_input: dict[str, Any]
tool_use_id: str
error: str
is_interrupt: NotRequired[bool]| Field | Type | Description |
|---|---|---|
hook_event_name | Literal["PostToolUseFailure"] | Always "PostToolUseFailure" |
tool_name | str | Name of the tool that failed |
tool_input | dict[str, Any] | Input parameters that were used |
tool_use_id | str | Unique identifier for this tool use |
error | str | Error message from the failed execution |
is_interrupt | bool (optional) | Whether the failure was caused by an interrupt |
UserPromptSubmitHookInputInput data for UserPromptSubmit hook events.
class UserPromptSubmitHookInput(BaseHookInput):
hook_event_name: Literal["UserPromptSubmit"]
prompt: str| Field | Type | Description |
|---|---|---|
hook_event_name | Literal["UserPromptSubmit"] | Always "UserPromptSubmit" |
prompt | str | The user's submitted prompt |
StopHookInputInput data for Stop hook events.
class StopHookInput(BaseHookInput):
hook_event_name: Literal["Stop"]
stop_hook_active: bool| Field | Type | Description |
|---|---|---|
hook_event_name | Literal["Stop"] | Always "Stop" |
stop_hook_active | bool | Whether the stop hook is active |
SubagentStopHookInputInput data for SubagentStop hook events.
class SubagentStopHookInput(BaseHookInput):
hook_event_name: Literal["SubagentStop"]
stop_hook_active: bool
agent_id: str
agent_transcript_path: str
agent_type: str| Field | Type | Description |
|---|---|---|
hook_event_name | Literal["SubagentStop"] | Always "SubagentStop" |
stop_hook_active | bool | Whether the stop hook is active |
agent_id | str | Unique identifier for the subagent |
agent_transcript_path | str | Path to the subagent's transcript file |
agent_type | str | Type of the subagent |
PreCompactHookInputInput data for PreCompact hook events.
class PreCompactHookInput(BaseHookInput):
hook_event_name: Literal["PreCompact"]
trigger: Literal["manual", "auto"]
custom_instructions: str | None| Field | Type | Description |
|---|---|---|
hook_event_name | Literal["PreCompact"] | Always "PreCompact" |
trigger | Literal["manual", "auto"] | What triggered the compaction |
custom_instructions | str | None | Custom instructions for compaction |
NotificationHookInputInput data for Notification hook events.
class NotificationHookInput(BaseHookInput):
hook_event_name: Literal["Notification"]
message: str
title: NotRequired[str]
notification_type: str| Field | Type | Description |
|---|---|---|
hook_event_name | Literal["Notification"] | Always "Notification" |
message | str | Notification message content |
title | str (optional) | Notification title |
notification_type | str | Type of notification |
SubagentStartHookInputInput data for SubagentStart hook events.
class SubagentStartHookInput(BaseHookInput):
hook_event_name: Literal["SubagentStart"]
agent_id: str
agent_type: str| Field | Type | Description |
|---|---|---|
hook_event_name | Literal["SubagentStart"] | Always "SubagentStart" |
agent_id | str | Unique identifier for the subagent |
agent_type | str | Type of the subagent |
PermissionRequestHookInputInput data for PermissionRequest hook events. Allows hooks to handle permission decisions programmatically.
class PermissionRequestHookInput(BaseHookInput):
hook_event_name: Literal["PermissionRequest"]
tool_name: str
tool_input: dict[str, Any]
permission_suggestions: NotRequired[list[Any]]| Field | Type | Description |
|---|---|---|
hook_event_name | Literal["PermissionRequest"] | Always "PermissionRequest" |
tool_name | str | Name of the tool requesting permission |
tool_input | dict[str, Any] | Input parameters for the tool |
permission_suggestions | list[Any] (optional) | Suggested permission updates from the CLI |
HookJSONOutputUnion type for hook callback return values.
HookJSONOutput = AsyncHookJSONOutput | SyncHookJSONOutputSyncHookJSONOutputSynchronous hook output with control and decision fields.
class SyncHookJSONOutput(TypedDict):
# Control fields
continue_: NotRequired[bool] # Whether to proceed (default: True)
suppressOutput: NotRequired[bool] # Hide stdout from transcript
stopReason: NotRequired[str] # Message when continue is False
# Decision fields
decision: NotRequired[Literal["block"]]
systemMessage: NotRequired[str] # Warning message for user
reason: NotRequired[str] # Feedback for Claude
# Hook-specific output
hookSpecificOutput: NotRequired[HookSpecificOutput]Use continue_ (with underscore) in Python code. It is automatically converted to continue when sent to the CLI.
HookSpecificOutputA TypedDict containing the hook event name and event-specific fields. The shape depends on the hookEventName value. For full details on available fields per hook event, see Control execution with hooks.
A discriminated union of event-specific output types. The hookEventName field determines which fields are valid.
class PreToolUseHookSpecificOutput(TypedDict):
hookEventName: Literal["PreToolUse"]
permissionDecision: NotRequired[Literal["allow", "deny", "ask"]]
permissionDecisionReason: NotRequired[str]
updatedInput: NotRequired[dict[str, Any]]
additionalContext: NotRequired[str]
class PostToolUseHookSpecificOutput(TypedDict):
hookEventName: Literal["PostToolUse"]
additionalContext: NotRequired[str]
updatedMCPToolOutput: NotRequired[Any]
class PostToolUseFailureHookSpecificOutput(TypedDict):
hookEventName: Literal["PostToolUseFailure"]
additionalContext: NotRequired[str]
class UserPromptSubmitHookSpecificOutput(TypedDict):
hookEventName: Literal["UserPromptSubmit"]
additionalContext: NotRequired[str]
class NotificationHookSpecificOutput(TypedDict):
hookEventName: Literal["Notification"]
additionalContext: NotRequired[str]
class SubagentStartHookSpecificOutput(TypedDict):
hookEventName: Literal["SubagentStart"]
additionalContext: NotRequired[str]
class PermissionRequestHookSpecificOutput(TypedDict):
hookEventName: Literal["PermissionRequest"]
decision: dict[str, Any]
HookSpecificOutput = (
PreToolUseHookSpecificOutput
| PostToolUseHookSpecificOutput
| PostToolUseFailureHookSpecificOutput
| UserPromptSubmitHookSpecificOutput
| NotificationHookSpecificOutput
| SubagentStartHookSpecificOutput
| PermissionRequestHookSpecificOutput
)AsyncHookJSONOutputAsync hook output that defers hook execution.
class AsyncHookJSONOutput(TypedDict):
async_: Literal[True] # Set to True to defer execution
asyncTimeout: NotRequired[int] # Timeout in millisecondsUse async_ (with underscore) in Python code. It is automatically converted to async when sent to the CLI.
This example registers two hooks: one that blocks dangerous bash commands like rm -rf /, and another that logs all tool usage for auditing. The security hook only runs on Bash commands (via the matcher), while the logging hook runs on all tools.
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], timeout=120
), # 2 min for validation
HookMatcher(
hooks=[log_tool_use]
), # Applies to all tools (default 60s timeout)
],
"PostToolUse": [HookMatcher(hooks=[log_tool_use])],
}
)
async for message in query(prompt="Analyze this codebase", options=options):
print(message)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.
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
}Tool name: AskUserQuestion
Asks the user clarifying questions during execution. See Handle approvals and user input for usage details.
Input:
{
"questions": [ # Questions to ask the user (1-4 questions)
{
"question": str, # The complete question to ask the user
"header": str, # Very short label displayed as a chip/tag (max 12 chars)
"options": [ # The available choices (2-4 options)
{
"label": str, # Display text for this option (1-5 words)
"description": str, # Explanation of what this option means
}
],
"multiSelect": bool, # Set to true to allow multiple selections
}
],
"answers": dict | None, # User answers populated by the permission system
}Output:
{
"questions": [ # The questions that were asked
{
"question": str,
"header": str,
"options": [{"label": str, "description": str}],
"multiSelect": bool,
}
],
"answers": dict[str, str], # Maps question text to answer string
# Multi-select answers are comma-separated
}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
}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
}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
}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
}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
}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
}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
}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
}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,
}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},
}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
}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
}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
}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,
}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,
}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 = 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 - the session retains all previous messages
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())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 a timestamp as additional context for Claude to see
from datetime import datetime
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
return {
"hookSpecificOutput": {
"hookEventName": "UserPromptSubmit",
"additionalContext": f"[Submitted at {timestamp}] Original prompt: {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())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
async for message in client.receive_response():
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("Completed tool execution")
elif isinstance(block, TextBlock):
print(f"Claude says: {block.text[:100]}...")
print("Task completed!")
asyncio.run(monitor_progress())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())from claude_agent_sdk import query, CLINotFoundError, ProcessError, CLIJSONDecodeError
try:
async for message in query(prompt="Hello"):
print(message)
except CLINotFoundError:
print("Claude Code CLI not found. Try reinstalling: pip install --force-reinstall claude-agent-sdk")
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}")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())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())SandboxSettingsConfiguration for sandbox behavior. Use this to enable command sandboxing and configure network restrictions programmatically.
class SandboxSettings(TypedDict, total=False):
enabled: bool
autoAllowBashIfSandboxed: bool
excludedCommands: list[str]
allowUnsandboxedCommands: bool
network: SandboxNetworkConfig
ignoreViolations: SandboxIgnoreViolations
enableWeakerNestedSandbox: bool| Property | Type | Default | Description |
|---|---|---|---|
enabled | bool | False | Enable sandbox mode for command execution |
autoAllowBashIfSandboxed | bool | True | Auto-approve bash commands when sandbox is enabled |
excludedCommands | list[str] | [] | Commands that always bypass sandbox restrictions (e.g., ["docker"]). These run unsandboxed automatically without model involvement |
allowUnsandboxedCommands | bool | True | Allow the model to request running commands outside the sandbox. When True, the model can set dangerouslyDisableSandbox in tool input, which falls back to the permissions system |
network | SandboxNetworkConfig | None | Network-specific sandbox configuration |
ignoreViolations | SandboxIgnoreViolations | None | Configure which sandbox violations to ignore |
enableWeakerNestedSandbox | bool | False | Enable a weaker nested sandbox for compatibility |
Filesystem and network access restrictions are NOT configured via sandbox settings. Instead, they are derived from permission rules:
Use sandbox settings for command execution sandboxing, and permission rules for filesystem and network access control.
from claude_agent_sdk import query, ClaudeAgentOptions, SandboxSettings
sandbox_settings: SandboxSettings = {
"enabled": True,
"autoAllowBashIfSandboxed": True,
"network": {"allowLocalBinding": True},
}
async for message in query(
prompt="Build and test my project",
options=ClaudeAgentOptions(sandbox=sandbox_settings),
):
print(message)Unix socket security: The allowUnixSockets option can grant access to powerful system services. For example, allowing /var/run/docker.sock effectively grants full host system access through the Docker API, bypassing sandbox isolation. Only allow Unix sockets that are strictly necessary and understand the security implications of each.
SandboxNetworkConfigNetwork-specific configuration for sandbox mode.
class SandboxNetworkConfig(TypedDict, total=False):
allowLocalBinding: bool
allowUnixSockets: list[str]
allowAllUnixSockets: bool
httpProxyPort: int
socksProxyPort: int| Property | Type | Default | Description |
|---|---|---|---|
allowLocalBinding | bool | False | Allow processes to bind to local ports (e.g., for dev servers) |
allowUnixSockets | list[str] | [] | Unix socket paths that processes can access (e.g., Docker socket) |
allowAllUnixSockets | bool | False | Allow access to all Unix sockets |
httpProxyPort | int | None | HTTP proxy port for network requests |
socksProxyPort | int | None | SOCKS proxy port for network requests |
SandboxIgnoreViolationsConfiguration for ignoring specific sandbox violations.
class SandboxIgnoreViolations(TypedDict, total=False):
file: list[str]
network: list[str]| Property | Type | Default | Description |
|---|---|---|---|
file | list[str] | [] | File path patterns to ignore violations for |
network | list[str] | [] | Network patterns to ignore violations for |
When allowUnsandboxedCommands is enabled, the model can request to run commands outside the sandbox by setting dangerouslyDisableSandbox: True in the tool input. These requests fall back to the existing permissions system, meaning your can_use_tool handler will be invoked, allowing you to implement custom authorization logic.
excludedCommands vs allowUnsandboxedCommands:
excludedCommands: A static list of commands that always bypass the sandbox automatically (e.g., ["docker"]). The model has no control over this.allowUnsandboxedCommands: Lets the model decide at runtime whether to request unsandboxed execution by setting dangerouslyDisableSandbox: True in the tool input.from claude_agent_sdk import (
query,
ClaudeAgentOptions,
HookMatcher,
PermissionResultAllow,
PermissionResultDeny,
ToolPermissionContext,
)
async def can_use_tool(
tool: str, input: dict, context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
# Check if the model is requesting to bypass the sandbox
if tool == "Bash" and input.get("dangerouslyDisableSandbox"):
# The model is requesting to run this command outside the sandbox
print(f"Unsandboxed command requested: {input.get('command')}")
if is_command_authorized(input.get("command")):
return PermissionResultAllow()
return PermissionResultDeny(
message="Command not authorized for unsandboxed execution"
)
return PermissionResultAllow()
# Required: dummy hook keeps the stream open for can_use_tool
async def dummy_hook(input_data, tool_use_id, context):
return {"continue_": True}
async def prompt_stream():
yield {
"type": "user",
"message": {"role": "user", "content": "Deploy my application"},
}
async def main():
async for message in query(
prompt=prompt_stream(),
options=ClaudeAgentOptions(
sandbox={
"enabled": True,
"allowUnsandboxedCommands": True, # Model can request unsandboxed execution
},
permission_mode="default",
can_use_tool=can_use_tool,
hooks={"PreToolUse": [HookMatcher(matcher=None, hooks=[dummy_hook])]},
),
):
print(message)This pattern enables you to:
Commands running with dangerouslyDisableSandbox: True have full system access. Ensure your can_use_tool handler validates these requests carefully.
If permission_mode is set to bypassPermissions and allow_unsandboxed_commands is enabled, the model can autonomously execute commands outside the sandbox without any approval prompts. This combination effectively allows the model to escape sandbox isolation silently.
Was this page helpful?