obsidian/wiki/agent-sdk/python-api-reference.md

---
title: "Agent SDK Python API Reference"
aliases: [python-sdk-reference, claude-agent-sdk-python]
tags: [agent-sdk, python, api-reference, anthropic, claude]
sources: [raw/Agent SDK reference - Python.md]
created: 2026-04-17
updated: 2026-04-17
---

## Installation

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

```python
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]`.

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

### tool() decorator

Defines MCP tools with type safety.

```python
@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 environment
- `destructiveHint` — tool may perform destructive updates
- `idempotentHint` — repeated calls have no extra effect
- `openWorldHint` — tool interacts with external entities

### create_sdk_mcp_server()

```python
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`.

```python
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.

```python
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.

```python
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

```python
# 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

```python
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 input
- `PermissionResultDeny(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 estimate
- `usage` — `{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 with `TaskUsage` (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

```python
HookEvent = Literal[
    "PreToolUse", "PostToolUse", "PostToolUseFailure",
    "UserPromptSubmit", "Stop", "SubagentStop", "PreCompact",
    "Notification", "SubagentStart", "PermissionRequest"
]
```

```python
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

```python
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=True` lets 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

- `@dataclass` types (e.g. `ResultMessage`, `TextBlock`) → attribute access: `msg.result`
- `TypedDict` types (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 with `receive_response()` before sending a new query
- `disallowed_tools` overrides everything including `bypassPermissions`
- `allowed_tools` auto-approves but does NOT restrict — use `disallowed_tools` to block
- `setting_sources=[]` disables all filesystem settings (good for CI and SDK-only apps)
- `ThinkingConfig` and `McpServerConfig` variants are `TypedDict` (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 + `allowUnixSockets` for Docker socket grants full host access — high risk
- Hook callbacks use `continue_` and `async_` (with underscores) in Python; SDK converts to `continue`/`async` for CLI

---

## Sources

- [raw/Agent SDK reference - Python.md](raw/Agent%20SDK%20reference%20-%20Python.md)

## Related

- [[wiki/agent-sdk/overview|Agent SDK Overview]]
- [[wiki/tech-patterns/ai-patterns|AI Patterns]]
- [[wiki/architecture/multi-agent-ai|Multi-Agent AI Architecture]]