13 KiB
| title | aliases | tags | sources | created | updated | ||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| Agent SDK Python API Reference |
|
|
|
2026-04-17 | 2026-04-17 |
Installation
pip install claude-agent-sdk
query() vs ClaudeSDKClient
Two interaction modes — choose based on whether you need conversation continuity.
| Feature | query() |
ClaudeSDKClient |
|---|---|---|
| Session | New each call | Reuses same session |
| Conversation | Single exchange | Multi-turn context |
| Interrupts | No | Yes |
| Hooks | Yes | Yes |
| Custom Tools | Yes | Yes |
| Use Case | One-off tasks | Continuous conversations |
Use query() for independent tasks, automation scripts, no history needed.
Use ClaudeSDKClient for chat interfaces, follow-up questions, response-driven logic.
Core Functions
query()
async def query(
*,
prompt: str | AsyncIterable[dict[str, Any]],
options: ClaudeAgentOptions | None = None,
transport: Transport | None = None
) -> AsyncIterator[Message]
Creates a new session per call. Returns an AsyncIterator[Message].
async for message in query(prompt="Create a Python web server", options=options):
print(message)
tool() decorator
Defines MCP tools with type safety.
@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']}!"}]}
Input schema options:
- Simple:
{"text": str, "count": int} - Full JSON Schema:
{"type": "object", "properties": {...}, "required": [...]}
ToolAnnotations (optional hints, not security decisions):
readOnlyHint— tool doesn't modify environmentdestructiveHint— tool may perform destructive updatesidempotentHint— repeated calls have no extra effectopenWorldHint— tool interacts with external entities
create_sdk_mcp_server()
def create_sdk_mcp_server(
name: str,
version: str = "1.0.0",
tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig
Creates an in-process MCP server. Pass result to ClaudeAgentOptions.mcp_servers.
calculator = create_sdk_mcp_server(name="calculator", tools=[add, multiply])
options = ClaudeAgentOptions(
mcp_servers={"calc": calculator},
allowed_tools=["mcp__calc__add", "mcp__calc__multiply"],
)
Session Management (synchronous)
| Function | Description |
|---|---|
list_sessions(directory, limit, include_worktrees) |
List past sessions with metadata |
get_session_messages(session_id, directory, limit, offset) |
Get messages from a past session |
get_session_info(session_id, directory) |
Metadata for a single session by ID |
rename_session(session_id, title, directory) |
Set custom title; most recent wins |
tag_session(session_id, tag, directory) |
Tag a session; pass None to clear |
SDKSessionInfo fields: session_id, summary, last_modified, custom_title, first_prompt, git_branch, cwd, tag, created_at.
ClaudeSDKClient
Maintains conversation context across multiple exchanges.
async with ClaudeSDKClient() as client:
await client.query("What's the capital of France?")
async for message in client.receive_response():
...
await client.query("What's the population of that city?") # retains context
async for message in client.receive_response():
...
Key Methods
| Method | Description |
|---|---|
connect(prompt) |
Connect with optional initial prompt |
query(prompt, session_id) |
Send request in streaming mode |
receive_messages() |
All messages as async iterator |
receive_response() |
Messages until (and including) ResultMessage |
interrupt() |
Send stop signal (streaming mode only) |
set_permission_mode(mode) |
Change permission mode mid-session |
set_model(model) |
Change model; None resets to default |
rewind_files(user_message_id) |
Restore files to state at message (requires enable_file_checkpointing=True) |
get_mcp_status() |
Status of all MCP servers |
reconnect_mcp_server(name) |
Retry failed MCP server |
toggle_mcp_server(name, enabled) |
Enable/disable MCP server mid-session |
stop_task(task_id) |
Stop a background task |
disconnect() |
End session |
Interrupt pattern
interrupt() sends a stop signal but does not clear the buffer. You must drain the interrupted task's messages (including its ResultMessage with subtype="error_during_execution") before reading the new response.
await client.interrupt()
async for message in client.receive_response(): # drain interrupted task
if isinstance(message, ResultMessage): break
await client.query("Just say hello instead")
async for message in client.receive_response(): # now get new response
...
ClaudeAgentOptions
Key configuration fields:
| Field | Type | Description |
|---|---|---|
allowed_tools |
list[str] |
Auto-approve these tools (doesn't restrict others) |
disallowed_tools |
list[str] |
Always deny; overrides allowed_tools and bypassPermissions |
permission_mode |
PermissionMode |
"default", "acceptEdits", "plan", "dontAsk", "bypassPermissions" |
system_prompt |
str | SystemPromptPreset |
Custom string or {"type":"preset","preset":"claude_code"} |
mcp_servers |
dict[str, McpServerConfig] |
MCP server configs or path to config file |
cwd |
str | Path |
Working directory |
model |
str |
Claude model to use |
max_turns |
int |
Max agentic turns (tool-use round trips) |
max_budget_usd |
float |
Stop when cost estimate reaches this value |
can_use_tool |
CanUseTool |
Custom permission callback |
hooks |
dict[HookEvent, list[HookMatcher]] |
Hook configurations |
thinking |
ThinkingConfig |
{"type":"adaptive"}, {"type":"enabled","budget_tokens":N}, {"type":"disabled"} |
effort |
Literal["low","medium","high","xhigh","max"] |
Thinking depth |
enable_file_checkpointing |
bool |
Enable file rewinding |
setting_sources |
list[SettingSource] |
Which filesystem settings to load ("user", "project", "local"); [] disables all |
agents |
dict[str, AgentDefinition] |
Programmatic subagents |
sandbox |
SandboxSettings |
Sandbox configuration |
output_format |
dict |
{"type":"json_schema","schema":{...}} for structured output |
fork_session |
bool |
When resuming, fork instead of continuing original |
include_partial_messages |
bool |
Emit StreamEvent for partial updates |
SettingSource / settings_sources
# Disable all filesystem settings (CI/SDK-only apps)
ClaudeAgentOptions(setting_sources=[])
# Load only shared team settings
ClaudeAgentOptions(setting_sources=["project"])
# Load project settings to include CLAUDE.md
ClaudeAgentOptions(
system_prompt={"type":"preset","preset":"claude_code"},
setting_sources=["project"]
)
Settings precedence: local > project > user. Programmatic options override filesystem settings. Managed policy settings win over everything.
Permission System
CanUseTool callback
async def custom_permission(
tool_name: str, input_data: dict, context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
if tool_name == "Write" and "config" in input_data.get("file_path", ""):
return PermissionResultAllow(updated_input={**input_data, "file_path": f"./sandbox/{input_data['file_path']}"})
return PermissionResultAllow(updated_input=input_data)
PermissionResultAllow(updated_input=...)— allow, optionally modify inputPermissionResultDeny(message="...", interrupt=True)— deny, optionally interrupt
Message Types
Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent | RateLimitEvent
ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock
ResultMessage key fields
subtype—"success"or"error_during_execution"total_cost_usd— client-side cost estimateusage—{input_tokens, output_tokens, cache_creation_input_tokens, cache_read_input_tokens}model_usage— per-model breakdown (camelCase keys:inputTokens,costUSD, etc.)session_id,num_turns,duration_ms
Background task messages
TaskStartedMessage— emitted when background task starts;task_type:"local_bash","local_agent","remote_agent"TaskProgressMessage— periodic updates withTaskUsage(tokens, tool_uses, duration_ms)TaskNotificationMessage— completion;status:"completed","failed","stopped"
RateLimitEvent
status: "allowed", "allowed_warning", "rejected". Use to warn users or back off.
Hook System
HookEvent = Literal[
"PreToolUse", "PostToolUse", "PostToolUseFailure",
"UserPromptSubmit", "Stop", "SubagentStop", "PreCompact",
"Notification", "SubagentStart", "PermissionRequest"
]
options = ClaudeAgentOptions(
hooks={
"PreToolUse": [
HookMatcher(matcher="Bash", hooks=[validate_bash], timeout=120),
HookMatcher(hooks=[log_all_tools]), # matches all tools
]
}
)
Hook callback signature: async def hook(input_data, tool_use_id, context) -> HookJSONOutput
Return values:
{}— no-op{"continue_": False, "stopReason": "..."}— stop execution{"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny", ...}}{"async_": True, "asyncTimeout": 5000}— defer execution
Note: Python SDK hooks don't yet support SessionStart, SessionEnd, Setup (TypeScript only).
Sandbox Settings
SandboxSettings = {
"enabled": True,
"autoAllowBashIfSandboxed": True,
"excludedCommands": ["docker"], # always bypass sandbox, no model involvement
"allowUnsandboxedCommands": True, # model can request bypass via dangerouslyDisableSandbox
"network": {
"allowLocalBinding": True,
"allowUnixSockets": ["/var/run/docker.sock"], # ⚠️ grants full Docker/host access
}
}
Security:
bypassPermissions+allowUnsandboxedCommands=Truelets the model escape the sandbox silently.
Built-in Tool Reference (Input/Output shapes)
| Tool | Key Input | Key Output |
|---|---|---|
Bash |
command, timeout, run_in_background |
output, exitCode, shellId |
Read |
file_path, offset, limit |
content, total_lines |
Write |
file_path, content |
bytes_written |
Edit |
file_path, old_string, new_string, replace_all |
replacements |
Glob |
pattern, path |
matches, count |
Grep |
pattern, path, glob, output_mode |
matches or files |
WebFetch |
url, prompt |
response, status_code |
WebSearch |
query, allowed_domains |
results, total_results |
Agent |
description, prompt, subagent_type |
result, usage, total_cost_usd |
Monitor |
command, description, timeout_ms, persistent |
taskId |
TodoWrite |
todos[] with content, status |
stats |
Type System Notes
@dataclasstypes (e.g.ResultMessage,TextBlock) → attribute access:msg.resultTypedDicttypes (e.g.ThinkingConfigEnabled,McpStdioServerConfig) → dict access:config["budget_tokens"]
Error Types
| Exception | When |
|---|---|
CLINotFoundError |
Claude Code CLI not installed |
CLIConnectionError |
Connection to Claude Code failed |
ProcessError |
Claude Code process failed (exit_code, stderr attrs) |
CLIJSONDecodeError |
JSON parsing failed (line, original_error attrs) |
Key Takeaways
query()= stateless one-shot;ClaudeSDKClient= stateful multi-turn conversation- After
interrupt(), drain the buffer withreceive_response()before sending a new query disallowed_toolsoverrides everything includingbypassPermissionsallowed_toolsauto-approves but does NOT restrict — usedisallowed_toolsto blocksetting_sources=[]disables all filesystem settings (good for CI and SDK-only apps)ThinkingConfigandMcpServerConfigvariants areTypedDict(dict at runtime, not objects)SdkBeta: "context-1m-2025-08-07"is retired after 2026-04-30; Claude Sonnet/Opus 4.6+ have 1M context natively- Sandbox +
allowUnixSocketsfor Docker socket grants full host access — high risk - Hook callbacks use
continue_andasync_(with underscores) in Python; SDK converts tocontinue/asyncfor CLI