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]` |
options |
`ClaudeAgentOptions | 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]` |
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']}!"
}]
}
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 |
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()callsSame 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
@tooldecorator) and hooksclass 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
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.
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]` |
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` |
mcp_servers |
`dict[str, McpServerConfig] | str | Path` |
permission_mode |
`PermissionMode | None` | None |
continue_conversation |
bool |
False |
Continue the most recent conversation |
resume |
`str | None` | None |
max_turns |
`int | None` | None |
disallowed_tools |
list[str] |
[] |
List of disallowed tool names |
model |
`str | None` | None |
output_format |
OutputFormat ` |
None` | None |
permission_prompt_tool_name |
`str | None` | None |
cwd |
`str | Path | None` |
settings |
`str | None` | None |
add_dirs |
`list[str | Path]` | [] |
env |
dict[str, str] |
{} |
Environment variables |
extra_args |
`dict[str, str | None]` | {} |
max_buffer_size |
`int | None` | None |
debug_stderr |
Any |
sys.stderr |
Deprecated - File-like object for debug output. Use stderr callback instead |
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 |
Include StreamEvent messages with real-time text deltas. See 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 |
plugins |
list[SdkPluginConfig] |
[] |
Load custom plugins from local paths. See Plugins for details |
setting_sources |
`list[SettingSource] | None` | None (no settings) |
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):
- Local settings (
.claude/settings.local.json) - Project settings (
.claude/settings.json) - 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.
Message Types
Message
Union type of all possible messages.
Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage
Note: When
include_partial_messages=True, you will also receiveStreamEventmessages containing real-time text deltas. SeeStreamEventbelow.
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)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 actionsystemMessage: System message to add to the transcripthookSpecificOutput: 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
…(truncated)