Agent SDK 参考 - Python

Python Agent SDK 的完整 API 参考，包括所有函数、类型和类。

安装

在虚拟环境中安装该包。在最近的 Debian、Ubuntu 和 Homebrew Python 安装上，针对系统 Python 运行 pip install 会失败，出现 error: externally-managed-environment 错误。

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

有关 uv、Windows PowerShell 和 API 密钥设置，请参阅 Agent SDK 快速入门中的设置。

在 `query()` 和 `ClaudeSDKClient` 之间选择

Python SDK 提供了两种与 Claude Code 交互的方式：

对于交互式应用程序（如聊天界面）或当下一步操作取决于 Claude 的响应时，请使用 ClaudeSDKClient。

函数

`query()`

为每次与 Claude Code 的交互创建一个新会话。默认情况下返回一个异步迭代器，当消息到达时产生消息。每次调用 query() 都会重新开始，不记得之前的交互，除非你传递 continue_conversation=True 或在 ClaudeAgentOptions 中传递 resume。参见 Sessions。

async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeAgentOptions | None = None,
    transport: Transport | None = None
) -> AsyncIterator[Message]

参数

返回

返回一个 AsyncIterator[Message]，从对话中产生消息。

示例 - 带选项

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions


async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are an expert Python developer",
        permission_mode="acceptEdits",
    )

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


asyncio.run(main())

`tool()`

用于定义具有类型安全的 MCP 工具的装饰器。

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]]

参数

输入模式选项

1. 简单类型映射（推荐）：

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

2. JSON Schema 格式（用于复杂验证）：

   {
       "type": "object",
       "properties": {
           "text": {"type": "string"},
           "count": {"type": "integer", "minimum": 0},
       },
       "required": ["text"],
   }
   

返回

一个装饰器函数，包装工具实现并返回一个 SdkMcpTool 实例。

示例

from claude_agent_sdk import tool
from typing import Any


@tool("greet", "Greet a user", {"name": str})
async def greet(args: dict[str, Any]) -> dict[str, Any]:
    return {"content": [{"type": "text", "text": f"Hello, {args['name']}!"}]}

`ToolAnnotations`

工具的行为提示，作为 tool() 的 annotations 参数传递。ToolAnnotations 扩展了 MCP SDK 的 mcp.types.ToolAnnotations，添加了 maxResultSizeChars 字段，你可以用 camelCase 或 snake_case 编写每个提示：ToolAnnotations(readOnlyHint=True) 和 ToolAnnotations(read_only_hint=True) 是等价的。你也可以在 SDK 接受注解的任何地方传递普通的 mcp.types.ToolAnnotations。

snake_case 名称和类型化的 maxResultSizeChars 字段需要 Python Agent SDK 0.2.140 或更高版本。版本 0.1.31 到 0.2.139 重新导出 mcp.types.ToolAnnotations 不变。在版本 0.1.55 到 0.2.139 上，你仍然可以将 maxResultSizeChars 作为关键字参数传递：MCP 类接受额外字段，SDK 将值转发给 Claude Code。

所有字段都是可选的。客户端不应依赖这些提示做出安全决策。

from claude_agent_sdk import tool, ToolAnnotations
from typing import Any


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

`create_sdk_mcp_server()`

创建在 Python 应用程序中运行的进程内 MCP 服务器。

def create_sdk_mcp_server(
    name: str,
    version: str = "1.0.0",
    tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig

参数

返回

返回一个 McpSdkServerConfig 对象，可以传递给 ClaudeAgentOptions.mcp_servers。

示例

from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions


@tool("add", "Add two numbers", {"a": float, "b": float})
async def add(args):
    return {"content": [{"type": "text", "text": f"Sum: {args['a'] + args['b']}"}]}


@tool("multiply", "Multiply two numbers", {"a": float, "b": float})
async def multiply(args):
    return {"content": [{"type": "text", "text": f"Product: {args['a'] * args['b']}"}]}


calculator = create_sdk_mcp_server(
    name="calculator",
    version="2.0.0",
    tools=[add, multiply],  # Pass decorated functions
)

# Use with Claude
options = ClaudeAgentOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=["mcp__calc__add", "mcp__calc__multiply"],
)

`list_sessions()`

列出带有元数据的过去会话。按项目目录过滤或列出所有项目中的会话。同步；立即返回。

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

参数

返回类型：`SDKSessionInfo`

示例

打印项目的 10 个最近会话。结果按 last_modified 降序排序，所以第一项是最新的。省略 directory 以搜索所有项目。

from claude_agent_sdk import list_sessions

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

`get_session_messages()`

从过去的会话中检索消息。同步；立即返回。

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

参数

返回类型：`SessionMessage`

示例

from claude_agent_sdk import list_sessions, get_session_messages

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

`get_session_info()`

按 ID 读取单个会话的元数据，无需扫描完整项目目录。同步；立即返回。

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

参数

返回 SDKSessionInfo，如果找不到会话则返回 None。

示例

查找单个会话的元数据，无需扫描项目目录。当你已经从之前的运行中获得会话 ID 时很有用。

from claude_agent_sdk import get_session_info

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

`rename_session()`

通过追加自定义标题条目来重命名会话。重复调用是安全的；最新的标题获胜。同步。

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

参数

如果 session_id 不是有效的 UUID 或 title 为空，则抛出 ValueError；如果找不到会话，则抛出 FileNotFoundError。

示例

重命名最近的会话，使其更容易找到。新标题在后续读取时出现在 SDKSessionInfo.custom_title 中。

from claude_agent_sdk import list_sessions, rename_session

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

`tag_session()`

标记会话。传递 None 以清除标签。重复调用是安全的；最新的标签获胜。同步。

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

参数

如果 session_id 不是有效的 UUID 或 tag 在清理后为空，则抛出 ValueError；如果找不到会话，则抛出 FileNotFoundError。

示例

标记会话，然后在稍后的读取中按该标签过滤。传递 None 以清除现有标签。

from claude_agent_sdk import list_sessions, tag_session

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

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

类

`ClaudeSDKClient`

在多次交换中维持对话会话。 这是 TypeScript SDK 的 query() 函数内部工作方式的 Python 等价物 - 它创建一个可以继续对话的客户端对象。见与 query() 的比较。

class ClaudeSDKClient:
    def __init__(self, options: ClaudeAgentOptions | None = None, transport: Transport | None = None)
    async def connect(self, prompt: str | AsyncIterable[dict] | None = None) -> None
    async def query(self, prompt: str | AsyncIterable[dict], session_id: str = "default") -> None
    async def receive_messages(self) -> AsyncIterator[Message]
    async def receive_response(self) -> AsyncIterator[Message]
    async def interrupt(self) -> None
    async def set_permission_mode(self, mode: PermissionMode) -> None
    async def set_model(self, model: str | None = None) -> None
    async def rewind_files(self, user_message_id: str) -> None
    async def get_mcp_status(self) -> McpStatusResponse
    async def reconnect_mcp_server(self, server_name: str) -> None
    async def toggle_mcp_server(self, server_name: str, enabled: bool) -> None
    async def stop_task(self, task_id: str) -> None
    async def get_server_info(self) -> dict[str, Any] | None
    async def disconnect(self) -> None

方法

上下文管理器支持

客户端可以用作异步上下文管理器以自动管理连接：

import asyncio
from claude_agent_sdk import ClaudeSDKClient


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


asyncio.run(main())

重要： 迭代消息时，避免使用 break 提前退出，因为这可能导致 asyncio 清理问题。相反，让迭代自然完成或使用标志来跟踪何时找到了你需要的内容。

示例 - 继续对话

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

示例 - 使用 ClaudeSDKClient 进行流式输入

import asyncio
from claude_agent_sdk import ClaudeSDKClient


async def message_stream():
    """Generate messages dynamically."""
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Analyze the following data:"},
    }
    await asyncio.sleep(0.5)
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Temperature: 25°C, Humidity: 60%"},
    }
    await asyncio.sleep(0.5)
    yield {
        "type": "user",
        "message": {"role": "user", "content": "What patterns do you see?"},
    }


async def main():
    async with ClaudeSDKClient() as client:
        # Stream input to Claude
        await client.query(message_stream())

        # Process response
        async for message in client.receive_response():
            print(message)

        # Follow-up in same session
        await client.query("Should we be concerned about these readings?")

        async for message in client.receive_response():
            print(message)


asyncio.run(main())

示例 - 使用中断

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, ResultMessage


async def interruptible_task():
    options = ClaudeAgentOptions(allowed_tools=["Bash"], permission_mode="acceptEdits")

    async with ClaudeSDKClient(options=options) as client:
        # Start a long-running task
        await client.query("Count from 1 to 100 slowly, using the bash sleep command")

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

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

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

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

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


asyncio.run(interruptible_task())

示例 - 高级权限控制

import asyncio
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():
    # 不要同时在 allowed_tools 中列出受限工具：allow 规则在 can_use_tool 运行之前批准调用
    options = ClaudeAgentOptions(can_use_tool=custom_permission_handler)

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

类型

`SdkMcpTool`

使用 @tool 装饰器创建的 SDK MCP 工具的定义。

@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

`Transport`

自定义传输实现的抽象基类。使用此类通过自定义通道与 Claude 进程通信（例如，远程连接而不是本地子进程）。

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: ...

导入：from claude_agent_sdk import Transport

`ClaudeAgentOptions`

Claude Code 查询的配置数据类。

@dataclass
class ClaudeAgentOptions:
    tools: list[str] | ToolsPreset | None = None
    allowed_tools: list[str] = field(default_factory=list)
    system_prompt: str | SystemPromptPreset | SystemPromptCustom | SystemPromptFile | None = None
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
    strict_mcp_config: bool = False
    permission_mode: PermissionMode | None = None
    continue_conversation: bool = False
    resume: str | None = None
    session_id: 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
    include_hook_events: bool = False
    forward_subagent_text: bool = False
    fork_session: bool = False
    resume_session_at: str | None = None
    resume_drops_turn: str | None = None
    agents: dict[str, AgentDefinition] | None = None
    setting_sources: list[SettingSource] | None = None
    skills: list[str] | Literal["all"] | 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: EffortLevel | None = None
    enable_file_checkpointing: bool = False
    session_store: SessionStore | None = None
    session_store_flush: SessionStoreFlushMode = "batched"
    load_timeout_ms: int = 60_000
    task_budget: TaskBudget | None = None

处理缓慢或停滞的 API 响应

CLI 子进程读取多个环境变量，这些变量控制 API 超时和停滞检测。通过 ClaudeAgentOptions.env 传递它们：

from claude_agent_sdk import ClaudeAgentOptions

options = ClaudeAgentOptions(
    env={
        "API_TIMEOUT_MS": "120000",
        "CLAUDE_CODE_MAX_RETRIES": "2",
        "CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS": "120000",
    },
)

• API_TIMEOUT_MS：Anthropic 客户端上的每个请求超时，以毫秒为单位。默认 600000。适用于主循环和所有子代理。

• CLAUDE_CODE_MAX_RETRIES：最大 API 重试次数。默认 10，上限为 15。每次重试都有自己的 API_TIMEOUT_MS 窗口，因此最坏情况下的实际时间大约是 API_TIMEOUT_MS × (CLAUDE_CODE_MAX_RETRIES + 1) 加上退避。对于需要等待更长时间中断的无人值守运行，设置 CLAUDE_CODE_RETRY_WATCHDOG=1：它无限期重试瞬时容量错误，自 Claude Code v2.1.199 起，对其他瞬时错误将默认值提高到 300 并移除此变量的上限。

• CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS：子代理的停滞监视器。当流监视器打开时，默认值为 CLAUDE_STREAM_IDLE_TIMEOUT_MS 加 5 分钟，即 600000，除非你提高该变量。当流监视器关闭时，默认值为 600000。在 v2.1.257 之前，默认值始终为 600000。

  计时器在每个流事件时重置。停滞时，Claude Code 中止子代理并向父代理报告停滞。对于后台子代理，它也会将任务标记为失败并附加任何部分结果。

• CLAUDE_ENABLE_STREAM_WATCHDOG 与 CLAUDE_STREAM_IDLE_TIMEOUT_MS：流监视器，当标头已到达但响应体停止流式传输时中止请求。监视器对所有提供商默认启用；设置 CLAUDE_ENABLE_STREAM_WATCHDOG=0 以禁用它。CLAUDE_STREAM_IDLE_TIMEOUT_MS 默认为 300000 并被限制为该最小值。中止后，自动重试涵盖 Claude Code 所做的事情，基于响应进行的程度。

  当监视器等待 ANTHROPIC_BASE_URL 后面的网关保持打开的响应时，设置 include_partial_messages 的主机继续接收 ping StreamEvent 消息。将这些帧读作活跃性而不是在沉默时超时会话。在 v2.1.257 之前，帧在最后一个真实流事件后 5 分钟停止。

`OutputFormat`

结构化输出验证的配置。将其作为 dict 传递给 ClaudeAgentOptions 上的 output_format 字段：

# Expected dict shape for output_format
{
    "type": "json_schema",
    "schema": {...},  # Your JSON Schema definition
}

`SystemPromptPreset`

使用 Claude Code 的预设系统提示和可选添加的配置。

class SystemPromptPreset(TypedDict):
    type: Literal["preset"]
    preset: Literal["claude_code"]
    append: NotRequired[str]
    exclude_dynamic_sections: NotRequired[bool]
    snapshot: NotRequired[bool]

`SystemPromptCustom`

对象形式的自定义系统提示，等同于将字符串作为 system_prompt 传递，也可以设置 snapshot。需要 claude-agent-sdk v0.2.153 或更高版本。

class SystemPromptCustom(TypedDict):
    type: Literal["custom"]
    prompt: str
    snapshot: NotRequired[bool]

`SystemPromptFile`

从文件加载自定义系统提示而不是作为字符串传递的配置。SDK 将其映射到 CLI --system-prompt-file 标志。当提示很大时使用文件形式：SDK 在 CLI 子进程 argv 上传递字符串 system_prompt，这受到 OS 命令行长度限制的限制，然后 SDK 才能发送任何 API 请求。在 Linux 上，单个参数长于大约 128 KB 会在进程生成时失败，出现 Argument list too long。在 Windows 上，整个命令行被限制为大约 32 KB，因此字符串形式在更低的阈值处失败。

class SystemPromptFile(TypedDict):
    type: Literal["file"]
    path: str

`SettingSource`

控制 SDK 从哪些基于文件系统的配置源加载设置。

SettingSource = Literal["user", "project", "local"]

默认行为

当 setting_sources 被省略或为 None 且 skills 未设置时，query() 加载与 Claude Code CLI 相同的文件系统设置：用户、项目和本地。使用 skills 设置时，setting_sources 行描述当前默认值。无论如何都会加载托管策略设置；当会话使用组织凭证在符合条件的配置上进行身份验证时，会获取服务器管理的设置。有关更多信息，见 settingSources 不控制什么。

为什么使用 setting\_sources

禁用文件系统设置：

# Do not load user, project, or local settings from disk
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions


async def main():
    async for message in query(
        prompt="Analyze this code",
        options=ClaudeAgentOptions(
            setting_sources=[]
        ),
    ):
        print(message)


asyncio.run(main())

仅加载特定设置源：

# Load only project settings, ignore user and local
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions


async def main():
    async for message in query(
        prompt="Run CI checks",
        options=ClaudeAgentOptions(
            setting_sources=["project"]  # Only .claude/settings.json
        ),
    ):
        print(message)


asyncio.run(main())

仅 SDK 应用程序：

# Define everything programmatically.
# Pass [] to opt out of filesystem setting sources.
import asyncio
from claude_agent_sdk import AgentDefinition, ClaudeAgentOptions, query


async def main():
    async for message in query(
        prompt="Review this PR",
        options=ClaudeAgentOptions(
            setting_sources=[],
            agents={
                "code-reviewer": AgentDefinition(
                    description="Reviews code changes",
                    prompt="You are a code reviewer. Report issues in the diff.",
                ),
            },
            allowed_tools=["Read", "Grep", "Glob"],
        ),
    ):
        print(message)


asyncio.run(main())

要加载 CLAUDE.md 项目说明，在 setting_sources 中包含 "project"。见 修改系统提示 了解 CLAUDE.md 加载如何与系统提示选项交互。

设置优先级

加载多个源时，设置按此优先级合并（从高到低）：

1. 本地设置（.claude/settings.local.json）
2. 项目设置（.claude/settings.json）
3. 用户设置（~/.claude/settings.json）

编程选项（如 agents、allowed_tools 和 settings）覆盖用户、项目和本地文件系统设置。托管策略设置优先于编程选项。

`AgentDefinition`

以编程方式定义的子代理的配置。

@dataclass
class AgentDefinition:
    description: str
    prompt: str
    tools: list[str] | None = None
    disallowedTools: list[str] | None = None
    model: str | None = None
    skills: list[str] | None = None
    memory: Literal["user", "project", "local"] | None = None
    mcpServers: list[str | dict[str, Any]] | None = None
    initialPrompt: str | None = None
    maxTurns: int | None = None
    background: bool | None = None
    effort: EffortLevel | int | None = None
    permissionMode: PermissionMode | None = None

`PermissionMode`

用于控制工具执行的权限模式。

PermissionMode = Literal[
    "default",  # Standard permission behavior
    "acceptEdits",  # Auto-accept file edits
    "plan",  # Planning mode - explore without editing
    "dontAsk",  # Deny anything not pre-approved instead of prompting
    "bypassPermissions",  # Bypass permission checks; explicit ask rules still prompt (use with caution)
    "auto",  # Model classifier approves or denies permission prompts
]

`EffortLevel`

用于指导思考深度的努力级别。

EffortLevel = Literal[
    "low",  # Minimal thinking, fastest responses
    "medium",  # Moderate thinking
    "high",  # Deep reasoning
    "xhigh",  # Extended reasoning; falls back to "high" on models that don't support it
    "max",  # Maximum effort
]

`CanUseTool`

工具权限回调函数的类型别名。

CanUseTool = Callable[
    [str, dict[str, Any], ToolPermissionContext], Awaitable[PermissionResult]
]

回调接收：

• tool_name：被调用的工具的名称
• input_data：工具的输入参数
• context：带有附加信息的 ToolPermissionContext

返回 PermissionResult（PermissionResultAllow 或 PermissionResultDeny）。

回调是 SDK 对交互式权限提示的替代：它仅在权限评估流解析为提示时调用。已由 allowed_tools 条目、设置允许规则或权限模式（如 acceptEdits 或 bypassPermissions）批准的工具调用永远不会调用它。要限制每个工具调用，改用 PreToolUse hook。

允许规则不会预批准任何模式都不自动批准的操作；见 权限如何被评估 了解其中哪些到达回调以及在 dontAsk 和 auto 模式下发生什么。

`ToolPermissionContext`

传递给工具权限回调的上下文信息。

@dataclass
class ToolPermissionContext:
    signal: Any | None = None  # Future: abort signal support
    suggestions: list[PermissionUpdate] = field(default_factory=list)
    tool_use_id: str | None = None
    agent_id: str | None = None
    blocked_path: str | None = None
    decision_reason: str | None = None
    title: str | None = None
    display_name: str | None = None
    description: str | None = None

`PermissionResult`

权限回调结果的联合类型。

PermissionResult = PermissionResultAllow | PermissionResultDeny

`PermissionResultAllow`

指示应允许工具调用的结果。

@dataclass
class PermissionResultAllow:
    behavior: Literal["allow"] = "allow"
    updated_input: dict[str, Any] | None = None
    updated_permissions: list[PermissionUpdate] | None = None

`PermissionResultDeny`

指示应拒绝工具调用的结果。

@dataclass
class PermissionResultDeny:
    behavior: Literal["deny"] = "deny"
    message: str = ""
    interrupt: bool = False

`PermissionUpdate`

用于以编程方式更新权限的配置。

@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

`PermissionRuleValue`

要在权限更新中添加、替换或移除的规则。

@dataclass
class PermissionRuleValue:
    tool_name: str
    rule_content: str | None = None

`ToolsPreset`

使用 Claude Code 的默认工具集的预设工具配置。

class ToolsPreset(TypedDict):
    type: Literal["preset"]
    preset: Literal["claude_code"]

`ThinkingConfig`

控制扩展思考行为。三种配置的联合：

ThinkingDisplay = Literal["summarized", "omitted"]


class ThinkingConfigAdaptive(TypedDict):
    type: Literal["adaptive"]
    display: NotRequired[ThinkingDisplay]


class ThinkingConfigEnabled(TypedDict):
    type: Literal["enabled"]
    budget_tokens: int
    display: NotRequired[ThinkingDisplay]


class ThinkingConfigDisabled(TypedDict):
    type: Literal["disabled"]


ThinkingConfig = ThinkingConfigAdaptive | ThinkingConfigEnabled | ThinkingConfigDisabled

可选的 display 字段控制思考文本是否返回为 "summarized" 或 "omitted"。在 Claude Opus 4.7 及更高版本上，API 默认值为 "omitted"，因此设置 "summarized" 以在 ThinkingBlock 输出中接收思考内容。Claude Code 不会向 Amazon Bedrock 或 Google Cloud 的 Agent Platform 发送 display，因此在这些提供商上，Opus 4.7 及更高版本即使你将 display 设置为 "summarized" 也会返回空 ThinkingBlock 输出。

因为这些是 TypedDict 类，它们在运行时是普通字典。要么将它们构造为字典字面量，要么调用类作为构造函数；两者都产生 dict。使用 config["budget_tokens"] 访问字段，而不是 config.budget_tokens：

from claude_agent_sdk import ClaudeAgentOptions, ThinkingConfigEnabled

# Option 1: dict literal (recommended, no import needed)
options = ClaudeAgentOptions(thinking={"type": "enabled", "budget_tokens": 20000})

# Option 2: constructor-style (returns a plain dict)
config = ThinkingConfigEnabled(type="enabled", budget_tokens=20000)
print(config["budget_tokens"])  # 20000
# config.budget_tokens would raise AttributeError

`TaskBudget`

API 端任务预算（以令牌为单位），与 ClaudeAgentOptions 中的 task_budget 字段一起使用。

class TaskBudget(TypedDict):
    total: int

因为这是 TypedDict，将其作为普通字典传递，如 ClaudeAgentOptions(task_budget={"total": 50000})。

`SdkBeta`

SDK 测试功能的字面类型。

SdkBeta = Literal["context-1m-2025-08-07"]

与 ClaudeAgentOptions 中的 betas 字段一起使用以启用测试功能。

`McpSdkServerConfig`

使用 create_sdk_mcp_server() 创建的 SDK MCP 服务器的配置。

class McpSdkServerConfig(TypedDict):
    type: Literal["sdk"]
    name: str
    instance: Any  # MCP Server instance

`McpServerConfig`

MCP 服务器配置的联合类型。

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]]

`McpServerStatusConfig`

由 get_mcp_status() 报告的 MCP 服务器的配置。这是所有 McpServerConfig 传输变体加上用于通过 claude.ai 代理的服务器的仅输出 claudeai-proxy 变体的联合。

McpServerStatusConfig = (
    McpStdioServerConfig
    | McpSSEServerConfig
    | McpHttpServerConfig
    | McpSdkServerConfigStatus
    | McpClaudeAIProxyServerConfig
)

McpSdkServerConfigStatus 是 McpSdkServerConfig 的可序列化形式，仅包含 type（"sdk"）和 name（str）字段；进程内 instance 被省略。McpClaudeAIProxyServerConfig 具有 type（"claudeai-proxy"）、url（str）和 id（str）字段。

`McpStatusResponse`

来自 ClaudeSDKClient.get_mcp_status() 的响应。在 mcpServers 键下包装服务器状态列表。

class McpStatusResponse(TypedDict):
    mcpServers: list[McpServerStatus]

`McpServerStatus`

连接的 MCP 服务器的状态，包含在 McpStatusResponse 中。

class McpServerStatus(TypedDict):
    name: str
    status: McpServerConnectionStatus  # "connected" | "failed" | "needs-auth" | "pending" | "disabled"
    serverInfo: NotRequired[McpServerInfo]
    error: NotRequired[str]
    config: NotRequired[McpServerStatusConfig]
    scope: NotRequired[str]
    tools: NotRequired[list[McpToolInfo]]

`SdkPluginConfig`

SDK 中加载插件的配置。

class SdkPluginConfig(TypedDict):
    type: Literal["local"]
    path: str

示例：

plugins = [
    {"type": "local", "path": "./my-plugin"},
    {"type": "local", "path": "/absolute/path/to/plugin"},
]

有关创建和使用插件的完整信息，见 Plugins。

消息类型

`Message`

所有可能消息的联合类型。

Message = (
    UserMessage
    | AssistantMessage
    | SystemMessage
    | ResultMessage
    | StreamEvent
    | RateLimitEvent
    | ConversationResetMessage
)

`UserMessage`

用户输入消息。

@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
    origin: MessageOrigin | None = None

SDK 从 CLI 未修改地传递 tool_use_result。对于外部 MCP 服务器上的工具，其结果包含 resource_link 块，该字典具有 resourceLinks 键，保存具有 TypeScript SDKMcpResourceLink 类型键的字典列表。Claude 将每个链接作为工具结果中的一行文本接收。要呈现服务器返回的文件，请读取 resourceLinks 而不是解析该文本。resourceLinks 键需要 Python Agent SDK 0.2.150 或更高版本和 Claude Code v2.1.257 或更高版本；该 SDK 版本附带的 CLI 满足 Claude Code 要求。

当结果没有链接时，CLI 会省略该键，在来自子代理的结果上也会省略。CLI 每个结果最多保留 50 个链接，一旦列表达到 64 KiB 的序列化 JSON，就停止添加链接。使用 tool() 在进程中定义的工具永远不会产生该键，因为 SDK 在 CLI 看到结果之前将其 resource_link 块展平为文本。

`AssistantMessage`

带有内容块的助手响应消息。

@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str
    parent_tool_use_id: str | None = None
    error: AssistantMessageError | None = None
    usage: dict[str, Any] | None = None
    message_id: str | None = None
    stop_reason: str | None = None
    session_id: str | None = None
    uuid: str | None = None

`AssistantMessageError`

助手消息的可能错误类型。

AssistantMessageError = Literal[
    "authentication_failed",
    "billing_error",
    "rate_limit",
    "invalid_request",
    "server_error",
    "unknown",
]

底层 CLI 进程可以发出此 Literal 未列出的错误类型，例如 max_output_tokens。SDK 未修改地传递该值，因此将此列表之外的字符串视为处理 unknown 的方式。TypeScript SDKAssistantMessageError 类型列出了 CLI 可以发出的完整值集。

`SystemMessage`

带有元数据的系统消息。

@dataclass
class SystemMessage:
    subtype: str
    data: dict[str, Any]

`ResultMessage`

带有成本和使用信息的最终结果消息。

@dataclass
class ResultMessage:
    subtype: str
    duration_ms: int
    duration_api_ms: int
    is_error: bool
    num_turns: int
    session_id: str
    stop_reason: str | None = None
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | None = None
    structured_output: Any = None
    model_usage: dict[str, ModelUsage] | None = None
    permission_denials: list[Any] | None = None
    deferred_tool_use: DeferredToolUse | None = None
    errors: list[str] | None = None
    api_error_status: int | None = None
    uuid: str | None = None
    terminal_reason: str | None = None
    origin: MessageOrigin | None = None

subtype 字段确定填充哪些其他字段。它是 "success"、"error_during_execution"、"error_max_turns"、"error_max_budget_usd" 或 "error_max_structured_output_retries" 之一。Python 数据类将所有变体展平为一种形状，因此不适用于返回的子类型的字段为 None。

多个字段携带有关对话如何结束的诊断详情：

• is_error：当对话以错误状态结束时为 True。在 error_* 子类型上始终为 True。在 subtype="success" 上，当最终模型请求失败时为 True，这意味着代理循环完成但最后一个 API 调用返回了错误。
• api_error_status：终止 API 错误的 HTTP 状态代码。当轮次结束时没有错误时为 None。仅在 subtype="success" 上填充。
• result：在 subtype="success" 上为最终助手消息的文本，或在 error_* 子类型上为 None。当 subtype="success" 且 is_error=True 时，如果可用，此字段保存 API 错误字符串，但可能为空，因此请检查 api_error_status 和前面的 AssistantMessage 内容以获取详情。
• errors：循环级别的错误字符串，例如最大轮次消息。仅在 error_* 子类型上填充。
• terminal_reason：查询循环结束的原因，例如 "completed"、"max_turns"、"api_error"、"aborted_streaming" 或 "aborted_tools"。值为 "aborted_streaming" 或 "aborted_tools" 意味着轮次在完成前被中止。常见原因是 interrupt() 和权限回调返回 PermissionResultDeny 且 interrupt=True。在早于该字段的 CLI 版本上为 None，在本地命令（如 /voice 或 /usage）的结果上为 None，这些命令绕过查询循环，或在会话严重失败时发出的合成错误结果上为 None。镜像 TypeScript SDK 的 SDKResultMessage.terminal_reason，其列出了完整的值集。
• origin：触发此轮次的用户消息的来源。在流式输入模式中，检查此项以区分你自己的提示结果（其中 origin 为 None 或 {"kind": "human"}）与注入的轮次（如后台任务通知）的结果。需要 Python Agent SDK 0.2.137 或更高版本。

usage 字典仅涵盖主代理循环，不包括子代理和其他嵌套或辅助模型调用。在流式输入模式中，值是按轮次的。优先使用 model_usage 进行令牌和成本计费。usage 字典在存在时包含以下键：

model_usage 字典将模型名称映射到每个模型的使用情况。它涵盖通过查询管道进行的每个模型调用：主循环、子代理和内部调用（如压缩和 Workflow 代理）。该管道外的辅助调用（如权限分类器和令牌计数请求）从 model_usage 中排除。将 model_usage 视为估计值，而不是计费声明。

在流式输入模式中，model_usage 和 total_cost_usd 在轮次间是累积的，因此读取最新结果而不是跨结果求和。调用恢复会话时，也会计算从会话早期调用恢复的总计。有关重置，请参阅在流式输入模式中跟踪成本，有关清零结果，请参阅在会话崩溃后恢复总计。

model_usage 中的每个值都是 ModelUsage TypedDict，通过 from claude_agent_sdk.types import ModelUsage 导入。其键使用 camelCase，因为 SDK 从底层 CLI 进程未修改地传递该值，匹配 TypeScript ModelUsage 类型：

`StreamEvent`

流式事件，用于流式传输期间的部分消息更新。仅在 ClaudeAgentOptions 中 include_partial_messages=True 时接收。通过 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

`RateLimitEvent`

当速率限制状态更改时发出（例如，从 "allowed" 到 "allowed_warning"）。使用此来在用户达到硬限制之前警告他们，或在状态为 "rejected" 时退避。

@dataclass
class RateLimitEvent:
    rate_limit_info: RateLimitInfo
    uuid: str
    session_id: str

`RateLimitInfo`

由 RateLimitEvent 携带的速率限制状态。

RateLimitStatus = Literal["allowed", "allowed_warning", "rejected"]
RateLimitType = Literal[
    "five_hour", "seven_day", "seven_day_opus", "seven_day_sonnet", "overage"
]


@dataclass
class RateLimitInfo:
    status: RateLimitStatus
    resets_at: int | None = None
    rate_limit_type: RateLimitType | None = None
    utilization: float | None = None
    overage_status: RateLimitStatus | None = None
    overage_resets_at: int | None = None
    overage_disabled_reason: str | None = None
    raw: dict[str, Any] = field(default_factory=dict)

`ConversationResetMessage`

在不结束连接的情况下替换对话时发出，例如在 /clear 之后。有关重置如何影响后续 ResultMessage 对象上的运行总计，请参阅在流式输入模式中跟踪成本。需要 Python Agent SDK 0.2.137 或更高版本。

@dataclass
class ConversationResetMessage:
    new_conversation_id: str
    uuid: str
    session_id: str

`TaskStartedMessage`

当后台任务启动时发出。后台任务是在主轮次之外跟踪的任何内容：后台 Bash 命令、Monitor 监视、通过 Agent 工具生成的子代理或远程代理。task_type 字段告诉你是哪一个。此命名与 Task 到 Agent 工具重命名无关。

@dataclass
class TaskStartedMessage(SystemMessage):
    task_id: str
    description: str
    uuid: str
    session_id: str
    tool_use_id: str | None = None
    task_type: str | None = None

`TaskUsage`

后台任务的令牌和计时数据。

class TaskUsage(TypedDict):
    total_tokens: int
    tool_uses: int
    duration_ms: int

`TaskProgressMessage`

定期为运行的后台任务发出进度更新。

@dataclass
class TaskProgressMessage(SystemMessage):
    task_id: str
    description: str
    usage: TaskUsage
    uuid: str
    session_id: str
    tool_use_id: str | None = None
    last_tool_name: str | None = None

`TaskNotificationMessage`

当后台任务完成、失败或停止时发出。后台任务包括 run_in_background Bash 命令、Monitor 监视和后台子代理。

@dataclass
class TaskNotificationMessage(SystemMessage):
    task_id: str
    status: TaskNotificationStatus  # "completed" | "failed" | "stopped"
    output_file: str
    summary: str
    uuid: str
    session_id: str
    tool_use_id: str | None = None
    usage: TaskUsage | None = None

当 CLI 将长 MCP 工具调用移到后台时，该调用的工具结果仅保存占位符，该调用的真实结果在此消息中到达。在此类调用的 "completed" 通知上，CLI 添加 resource_links 键，列出工具通过引用返回的文件，具有与 UserMessage.tool_use_result 上的 resourceLinks 键相同的条目和限制。resource_links 键需要 Python Agent SDK 0.2.150 或更高版本和 Claude Code v2.1.257 或更高版本；该 SDK 版本附带的 CLI 满足 Claude Code 要求。

数据类没有 resource_links 字段。从消息继承自 SystemMessage 的 data 字典读取它：message.data.get("resource_links")。使用 tool_use_id 匹配通知与调用。当结果没有链接时，CLI 会省略该键，在不是 MCP 工具调用的任务的通知上也会省略。

内容块类型

`ContentBlock`

所有内容块的联合类型。

ContentBlock = (
    TextBlock
    | ThinkingBlock
    | ToolUseBlock
    | ToolResultBlock
    | ServerToolUseBlock
    | ServerToolResultBlock
)

`TextBlock`

文本内容块。

@dataclass
class TextBlock:
    text: str

`ThinkingBlock`

思考内容块（用于具有思考能力的模型）。

@dataclass
class ThinkingBlock:
    thinking: str
    signature: str

`ToolUseBlock`

工具使用请求块。

@dataclass
class ToolUseBlock:
    id: str
    name: str
    input: dict[str, Any]

`ToolResultBlock`

工具执行结果块。

@dataclass
class ToolResultBlock:
    tool_use_id: str
    content: str | list[dict[str, Any]] | None = None
    is_error: bool | None = None

错误类型

下面的类型定义了你的代码可以捕获的内容。对于与这些类型引发的错误消息相关的条目，包括每个错误的原因和修复方法，请参阅故障排除。

`ClaudeSDKError`

所有 SDK 错误的基础异常类。

class ClaudeSDKError(Exception):
    """Base error for Claude SDK."""

当单次 query() 以错误结果结束时，例如达到轮次限制错误，SDK 会在生成最终结果消息后引发 ResultError。Python Agent SDK 0.2.140 之前的版本引发的是不属于 ClaudeSDKError 子类的普通 Exception。

`CLINotFoundError`

当 Claude Code CLI 未安装或找不到时引发。

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`

当连接到 Claude Code 失败时引发。

class CLIConnectionError(ClaudeSDKError):
    """Failed to connect to Claude Code."""

`ProcessError`

当 Claude Code 进程失败时引发。

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

`ResultError`

当 Claude Code 进程因运行以错误结果结束而退出时引发，例如达到轮次限制错误或 API 错误。在最终 ResultMessage 之后引发。ResultError 是 ProcessError 的子类，因此现有的 except ProcessError 处理程序也会捕获它。其属性包含该结果消息的字段，因此你可以根据运行失败的原因进行分支，而无需解析消息文本。需要 Python Agent SDK 0.2.140 或更高版本。

class ResultError(ProcessError):
    subtype: str | None  # "error_max_turns", "error_during_execution", ...; "success" when the run ended on a failed request
    errors: list[str]  # an empty list when the result message reported none
    result: str | None
    api_error_status: int | None
    terminal_reason: str | None  # "max_turns", "api_error", ...; check this before subtype
    session_id: str | None
    data: dict[str, Any]  # the raw result message payload

要区分失败，请在检查 subtype 之前先检查 terminal_reason。当最终请求失败时，例如 API 错误，Claude Code 会报告 subtype 为 "success"，原因在 terminal_reason 中，例如 "api_error"；当你设置的限制结束运行时，例如 max_turns 或 max_budget_usd，它会报告 error_* 子类型。

`CLIJSONDecodeError`

当 JSON 解析失败时引发。

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 类型

有关使用 hooks 的综合指南，包括示例和常见模式，见 Hooks 指南。

`HookEvent`

支持的 hook 事件类型。

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
]

`HookCallback`

hook 回调函数的类型定义。

HookCallback = Callable[[HookInput, str | None, HookContext], Awaitable[HookJSONOutput]]

参数：

• input：强类型 hook 输入，具有基于 hook_event_name 的判别联合（见 HookInput）
• tool_use_id：可选工具使用标识符（用于工具相关的 hooks）
• context：带有附加信息的 hook 上下文

返回 HookJSONOutput。

`HookContext`

传递给 hook 回调的上下文信息。

class HookContext(TypedDict):
    signal: Any | None  # Future: abort signal support

`HookMatcher`

用于将 hooks 匹配到特定事件或工具的配置。

@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. When omitted, the per-event default applies:
        # 600 for most events, 30 for UserPromptSubmit
    )

`HookInput`

所有 hook 输入类型的联合类型。实际类型取决于 hook_event_name 字段。

HookInput = (
    PreToolUseHookInput
    | PostToolUseHookInput
    | PostToolUseFailureHookInput
    | UserPromptSubmitHookInput
    | StopHookInput
    | SubagentStopHookInput
    | PreCompactHookInput
    | NotificationHookInput
    | SubagentStartHookInput
    | PermissionRequestHookInput
)

`BaseHookInput`

所有 hook 输入类型中存在的基础字段。

class BaseHookInput(TypedDict):
    session_id: str
    transcript_path: str
    cwd: str
    permission_mode: NotRequired[str]

`PreToolUseHookInput`

PreToolUse hook 事件的输入数据。

class PreToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PreToolUse"]
    tool_name: str
    tool_input: dict[str, Any]
    tool_use_id: str
    agent_id: NotRequired[str]
    agent_type: NotRequired[str]

`PostToolUseHookInput`

PostToolUse hook 事件的输入数据。

class PostToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PostToolUse"]
    tool_name: str
    tool_input: dict[str, Any]
    tool_response: Any
    tool_use_id: str
    agent_id: NotRequired[str]
    agent_type: NotRequired[str]

`PostToolUseFailureHookInput`

PostToolUseFailure hook 事件的输入数据。当工具执行失败时调用。

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]
    agent_id: NotRequired[str]
    agent_type: NotRequired[str]

`UserPromptSubmitHookInput`

UserPromptSubmit hook 事件的输入数据。

class UserPromptSubmitHookInput(BaseHookInput):
    hook_event_name: Literal["UserPromptSubmit"]
    prompt: str

`StopHookInput`

Stop hook 事件的输入数据。

class StopHookInput(BaseHookInput):
    hook_event_name: Literal["Stop"]
    stop_hook_active: bool

`SubagentStopHookInput`

SubagentStop hook 事件的输入数据。

class SubagentStopHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStop"]
    stop_hook_active: bool
    agent_id: str
    agent_transcript_path: str
    agent_type: str

`PreCompactHookInput`

PreCompact hook 事件的输入数据。

class PreCompactHookInput(BaseHookInput):
    hook_event_name: Literal["PreCompact"]
    trigger: Literal["manual", "auto"]
    custom_instructions: str | None

`NotificationHookInput`

Notification hook 事件的输入数据。

class NotificationHookInput(BaseHookInput):
    hook_event_name: Literal["Notification"]
    message: str
    title: NotRequired[str]
    notification_type: str

`SubagentStartHookInput`

SubagentStart hook 事件的输入数据。

class SubagentStartHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStart"]
    agent_id: str
    agent_type: str

`PermissionRequestHookInput`

PermissionRequest hook 事件的输入数据。允许 hooks 以编程方式处理权限决策。

class PermissionRequestHookInput(BaseHookInput):
    hook_event_name: Literal["PermissionRequest"]
    tool_name: str
    tool_input: dict[str, Any]
    permission_suggestions: NotRequired[list[Any]]
    agent_id: NotRequired[str]
    agent_type: NotRequired[str]

`HookJSONOutput`

hook 回调返回值的联合类型。

HookJSONOutput = AsyncHookJSONOutput | SyncHookJSONOutput

`SyncHookJSONOutput`

具有控制和决策字段的同步 hook 输出。

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]

`HookSpecificOutput`

事件特定输出类型的判别联合。hookEventName 字段确定哪些字段有效。有关每个 hook 事件的可用字段的完整详情，见 使用 hooks 控制执行。

class PreToolUseHookSpecificOutput(TypedDict):
    hookEventName: Literal["PreToolUse"]
    permissionDecision: NotRequired[Literal["allow", "deny", "ask", "defer"]]
    permissionDecisionReason: NotRequired[str]
    updatedInput: NotRequired[dict[str, Any]]
    additionalContext: NotRequired[str]


class PostToolUseHookSpecificOutput(TypedDict):
    hookEventName: Literal["PostToolUse"]
    additionalContext: NotRequired[str]
    updatedToolOutput: NotRequired[Any]
    updatedMCPToolOutput: NotRequired[Any]  # Deprecated: use updatedToolOutput, which works for all tools


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
)

`AsyncHookJSONOutput`

延迟 hook 执行的异步 hook 输出。

class AsyncHookJSONOutput(TypedDict):
    async_: Literal[True]  # Set to True to defer execution
    asyncTimeout: NotRequired[int]  # Timeout in milliseconds

Hook 使用示例

此示例注册两个 hooks：一个阻止危险的 bash 命令（如 rm -rf /），另一个记录所有工具使用以进行审计。安全 hook 仅在 Bash 命令上运行（通过 matcher），而日志 hook 在所有工具上运行。

import asyncio
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 (per-event default timeout)
        ],
        "PostToolUse": [HookMatcher(hooks=[log_tool_use])],
    }
)

async def main():
    async for message in query(prompt="Analyze this codebase", options=options):
        print(message)


asyncio.run(main())

工具输入/输出类型

所有内置 Claude Code 工具的输入/输出模式文档。虽然 Python SDK 不将这些导出为类型，但它们代表消息中工具输入和输出的结构。

Agent

工具名称： Agent。之前的名称 Task 仍然被接受作为别名，初始化 SystemMessage 中的 tools 列表为了向后兼容将此工具报告为 Task。

输入：

{
    "description": str,  # 任务的简短描述（3-5 个单词）
    "prompt": str,  # 代理要执行的任务
    "subagent_type": str | None,  # 要使用的专门代理的类型
    "model": "sonnet" | "opus" | "haiku" | "fable" | None,  # 此代理的模型覆盖
    "run_in_background": bool | None,  # 代理默认在后台运行；设置为 False 以同步运行
    "name": str | None,  # 生成的代理的名称
    "team_name": str | None,  # 已弃用；被忽略
    "mode": "acceptEdits" | "auto" | "bypassPermissions" | "default" | "dontAsk" | "plan" | None,  # 已弃用；被忽略。子代理继承规则决定子代理的权限模式
    "isolation": "worktree" | "remote" | None,  # 代理更改的隔离模式
}

启动一个新代理来自主处理复杂的多步骤任务。

输出（状态："completed"）：

{
    "status": "completed",
    "agentId": str,  # 运行的代理的 ID
    "agentType": str | None,  # 处理任务的子代理类型
    "content": [  # 结果内容块
        {
            "type": "text",
            "text": str,
            "citations": list | None,
        }
    ],
    "resolvedModel": str | None,  # 子代理启动时的模型
    "modelsUsed": list[str] | None,  # 按顺序使用的模型，连续重复被折叠
    "totalToolUseCount": int,  # 代理进行的工具调用次数
    "totalDurationMs": int,  # 执行持续时间（毫秒）
    "totalTokens": int,  # 来自最终 API 请求的令牌计数，不是整个运行
    "usage": {  # 令牌使用统计
        "input_tokens": int,
        "output_tokens": int,
        "cache_creation_input_tokens": int | None,
        "cache_read_input_tokens": int | None,
        "server_tool_use": {"web_search_requests": int, "web_fetch_requests": int} | None,
        "service_tier": str | None,
        "cache_creation": {"ephemeral_1h_input_tokens": int, "ephemeral_5m_input_tokens": int} | None,
        "inference_geo": str | None,
        "speed": str | None,
        "iterations": Any | None,
        "output_tokens_details": {"thinking_tokens": int | None} | None,
    },
    "toolStats": {  # 运行的聚合工具活动
        "readCount": int,
        "searchCount": int,
        "bashCount": int,
        "editFileCount": int,
        "linesAdded": int,
        "linesRemoved": int,
        "otherToolCount": int,
        "frameCount": int | None,
    } | None,
    "prompt": str,  # 代理运行的提示
    "worktreePath": str | None,  # 当 Claude Code 保留子代理的 worktree 时出现
    "worktreeBranch": str | None,  # 当 Claude Code 使用 git 创建该 worktree 时出现
}

输出（状态："async_launched"）：

{
    "status": "async_launched",
    "isAsync": bool | None,  # 后台启动时为 True
    "agentId": str,  # 启动的代理的 ID
    "description": str,  # 任务描述
    "resolvedModel": str | None,  # 后台转换时使用的模型
    "modelsUsed": list[str] | None,  # 后台转换前使用的模型，按顺序，连续重复被折叠
    "prompt": str,  # 代理运行的提示
    "outputFile": str,  # 代理输出被写入的文件路径
    "canReadOutputFile": bool | None,  # 输出文件是否可以直接读取
}

输出（状态："remote_launched"）：

{
    "status": "remote_launched",
    "taskId": str,  # 远程任务的 ID
    "sessionUrl": str,  # 远程云会话的链接
    "description": str,  # 任务描述
    "prompt": str,  # 代理运行的提示
    "outputFile": str,  # 代理输出被写入的文件路径
}

返回来自子代理的结果。输出在 status 字段上进行区分："completed" 用于完成的任务，"async_launched" 用于后台任务，"remote_launched" 用于 Claude Code 分派到远程云会话的任务，其中 sessionUrl 链接到该会话，taskId 标识它。如果 Claude Code 保留了子代理的隔离 worktree，completed 变体上的 worktreePath 是找到它的位置，worktreeBranch 是当 Claude Code 使用 git 创建 worktree 时的分支。

在 completed 变体上，resolvedModel 命名子代理启动时的模型，当应用 availableModels 或其他覆盖时，它可能与请求的 model 输入不同。此字段需要 Claude Code v2.1.174 或更高版本。在 async_launched 变体上，resolvedModel 命名代理移到后台时使用的模型，因此在后台转换之前发生的交换会反映在那里。两个变体上的 modelsUsed 字段按顺序列出使用的模型，连续重复被折叠；仅当模型在运行中被交换时才设置。modelsUsed 和后台转换时的 resolvedModel 行为需要 Claude Code v2.1.212 或更高版本。

Claude Code 从子代理的最终 API 请求而不是整个运行中填充 usage 和 totalTokens。当存在时，usage 中 output_tokens_details 下的 thinking_tokens 是该请求的输出令牌中是思考令牌的数量。output_tokens_details 键需要 Python SDK v0.2.136 或更高版本，它捆绑了 Claude Code v2.1.228。

AskUserQuestion

工具名称： AskUserQuestion

在执行期间向用户提出澄清问题。见 处理批准和用户输入 了解使用详情。

输入：

{
    "questions": [  # 要向用户提出的问题（1-4 个问题）
        {
            "question": str,  # 要向用户提出的完整问题
            "header": str,  # 显示为芯片/标签的非常简短的标签（最多 12 个字符）
            "options": [  # 可用的选择（2-4 个选项）
                {
                    "label": str,  # 此选项的显示文本（1-5 个单词）
                    "description": str,  # 此选项含义的说明
                    "preview": str | None,  # 当选项被聚焦时呈现的预览内容
                }
            ],
            "multiSelect": bool,  # 设置为 true 以允许多个选择
        }
    ],
    "answers": dict[str, str] | None,
    # 由权限系统填充的用户答案。多选
    # 答案是所选标签的逗号连接字符串；
    # 输入时接受标签列表并强制转换为该形式
    "annotations": dict[str, dict] | None,
    # 来自用户的按问题文本键入的每个问题注释。
    # 每个值可以携带"preview"（所选选项的预览
    # 内容）和"notes"（关于选择的自由文本注释）
    "metadata": dict | None,  # 分析元数据，例如 {"source": "remember"}；不向用户显示
}

输出：

{
    "questions": [  # 被提出的问题
        {
            "question": str,
            "header": str,
            "options": [{"label": str, "description": str, "preview": str | None}],
            "multiSelect": bool,
        }
    ],
    "answers": dict[str, str],  # 将问题文本映射到答案字符串
    # 多选答案以逗号分隔
    "response": str | None,
    # 用户输入的自由形式回复而不是回答问题；当设置时，
    # Claude 收到"用户回复：..."而不是答案列表
    "annotations": dict[str, dict] | None,  # 来自用户选择的每个问题"preview"和"notes"
    "afkTimeoutMs": int | None,  # 在用户不活动这么多毫秒后对话自动解决时设置；用户回答时不存在
}

Bash

工具名称： Bash

输入：

{
    "command": str,  # 要执行的命令
    "timeout": int | None,  # 可选的超时时间（毫秒）（最大 600000；更高的值被限制为最大值）
    "description": str | None,  # 清晰、简洁的描述（5-10 个单词）
    "run_in_background": bool | None,  # 设置为 true 以在后台运行
}

输出：

{
    "stdout": str,  # 命令的输出；stdout 和 stderr 合并到这个一个交错流中
    "stderr": str,  # 工具本身添加的通知，不是命令的 stderr
    "interrupted": bool,  # 命令是否被中断
    "isImage": bool | None,  # stdout 是否包含图像数据
    "backgroundTaskId": str | None,  # 如果命令在后台运行，后台任务的 ID
}

Monitor

工具名称： Monitor

运行后台源并将每个事件传递给 Claude，以便它可以做出反应而无需轮询：command 运行脚本并每个 stdout 行发出一个事件，ws 打开 WebSocket 并每个文本帧发出一个事件。恰好提供 command 或 ws 中的一个。

当 Monitor 运行命令时，它遵循与 Bash 相同的权限规则；WebSocket 监视会单独提示批准。ws 源需要 Claude Code v2.1.195 或更高版本。见 Monitor 工具参考 了解行为和提供商可用性。

输入：

{
    "command": str | None,  # Shell 脚本；每个 stdout 行是一个事件，退出结束监视
    "ws": dict | None,  # WebSocket 源：{"url": str, "protocols": list[str] | None}；每个文本帧是一个事件
    "description": str,  # 在通知中显示的简短描述
    "timeout_ms": int | None,  # 截止时间（毫秒）（默认 300000，最大 3600000；有效截止时间最多为 1800000）
}

输出：

{
    "taskId": str,  # 后台监视任务的 ID
    "timeoutMs": int,  # 监视的有效截止时间（毫秒）
    "persistent": bool | None,  # False：每个监视都有一个截止时间
}

Edit

工具名称： Edit

输入：

{
    "file_path": str,  # 要修改的文件的绝对路径
    "old_string": str,  # 要替换的文本
    "new_string": str,  # 替换为的文本
    "replace_all": bool | None,  # 替换所有出现（默认 False）
}

输出：

{
    "message": str,  # 确认消息
    "replacements": int,  # 进行的替换次数
    "file_path": str,  # 被编辑的文件路径
}

Read

工具名称： Read

输入：

{
    "file_path": str,  # 要读取的文件的绝对路径
    "offset": int | None,  # 开始读取的行号
    "limit": int | None,  # 要读取的行数
}

输出（文本文件）：

{
    "content": str,  # 带行号的文件内容
    "total_lines": int,  # 文件中的总行数
    "lines_returned": int,  # 实际返回的行数
}

输出（图像）：

{
    "image": str,  # Base64 编码的图像数据
    "mime_type": str,  # 图像 MIME 类型
    "file_size": int,  # 文件大小（字节）
}

Write

工具名称： Write

输入：

{
    "file_path": str,  # 要写入的文件的绝对路径
    "content": str,  # 要写入文件的内容
}

输出：

{
    "message": str,  # 成功消息
    "bytes_written": int,  # 写入的字节数
    "file_path": str,  # 被写入的文件路径
}

Glob

工具名称： Glob

输入：

{
    "pattern": str,  # 用于匹配文件的 glob 模式
    "path": str | None,  # 要搜索的目录（默认为 cwd）
}

输出：

{
    "matches": list[str],  # 匹配的文件路径数组
    "count": int,  # 找到的匹配数
    "search_path": str,  # 使用的搜索目录
}

Grep

工具名称： Grep

输入：

{
    "pattern": str,  # 正则表达式模式
    "path": str | None,  # 要搜索的文件或目录
    "glob": str | None,  # 用于过滤文件的 glob 模式
    "type": str | None,  # 要搜索的文件类型
    "output_mode": str | None,  # "content"、"files_with_matches" 或 "count"
    "-i": bool | None,  # 不区分大小写的搜索
    "-n": bool | None,  # 显示行号
    "-B": int | None,  # 每个匹配前显示的行数
    "-A": int | None,  # 每个匹配后显示的行数
    "-C": int | None,  # 每个匹配前后显示的行数
    "head_limit": int | None,  # 将输出限制为前 N 行/条目
    "multiline": bool | None,  # 启用多行模式
}

输出（content 模式）：

{
    "matches": [
        {
            "file": str,
            "line_number": int | None,
            "line": str,
            "before_context": list[str] | None,
            "after_context": list[str] | None,
        }
    ],
    "total_matches": int,
}

输出（files_with_matches 模式）：

{
    "files": list[str],  # 包含匹配的文件
    "count": int,  # 包含匹配的文件数
}

NotebookEdit

工具名称： NotebookEdit

输入：

{
    "notebook_path": str,  # Jupyter 笔记本的绝对路径
    "cell_id": str | None,  # 要编辑的单元格的 ID
    "new_source": str,  # 单元格的新源代码
    "cell_type": "code" | "markdown" | None,  # 单元格的类型
    "edit_mode": "replace" | "insert" | "delete" | None,  # 编辑操作类型
}

输出：

{
    "message": str,  # 成功消息
    "edit_type": "replaced" | "inserted" | "deleted",  # 执行的编辑类型
    "cell_id": str | None,  # 受影响的单元格 ID
    "total_cells": int,  # 编辑后笔记本中的总单元格数
}

WebFetch

工具名称： WebFetch

输入：

{
    "url": str,  # 要从中获取内容的 URL
    "prompt": str,  # 在获取的内容上运行的提示
}

输出：

{
    "bytes": int,  # 获取的内容大小（字节）
    "code": int,  # HTTP 响应代码
    "codeText": str,  # HTTP 响应代码文本
    "result": str,  # 通过将提示应用于内容得到的处理结果
    "durationMs": int,  # 获取和处理内容的时间（毫秒）
    "url": str,  # 被获取的 URL
}

WebSearch

工具名称： WebSearch

输入：

{
    "query": str,  # 要使用的搜索查询
    "allowed_domains": list[str] | None,  # 仅包含来自这些域的结果
    "blocked_domains": list[str] | None,  # 永远不包含来自这些域的结果
}

输出：

{
    "query": str,  # 搜索查询
    "results": list[str | {"tool_use_id": str, "content": list[{"title": str, "url": str}]}],
    "durationSeconds": float,  # 搜索持续时间（秒）
}

TodoWrite

工具名称： TodoWrite

输入：

{
    "todos": [
        {
            "content": str,  # 任务描述
            "status": "pending" | "in_progress" | "completed",  # 任务状态
            "activeForm": str,  # 描述的活跃形式
        }
    ]
}

输出：

{
    "message": str,  # 成功消息
    "stats": {"total": int, "pending": int, "in_progress": int, "completed": int},
}

TaskCreate

工具名称： TaskCreate

输入：

{
    "subject": str,  # 简短的任务标题
    "description": str,  # 详细的任务正文
    "activeForm": str | None,  # 进行中时显示的现在时标签
    "metadata": dict | None,  # 任意调用者元数据
}

输出：

{
    "task": {"id": str, "subject": str},  # 创建的任务及其分配的 ID
}

TaskUpdate

工具名称： TaskUpdate

输入：

{
    "taskId": str,  # 要修补的任务的 ID
    "status": Literal["pending", "in_progress", "completed", "deleted"] | None,
    "subject": str | None,
    "description": str | None,
    "activeForm": str | None,
    "addBlocks": list[str] | None,  # 此任务现在阻止的任务 ID
    "addBlockedBy": list[str] | None,  # 现在阻止此任务的任务 ID
    "owner": str | None,
    "metadata": dict | None,
}

输出：

{
    "success": bool,
    "taskId": str,
    "updatedFields": list[str],  # 更改的字段名称
    "error": str | None,
    "statusChange": {"from": str, "to": str} | None,
}

TaskGet

工具名称： TaskGet

输入：

{
    "taskId": str,  # 要读取的任务的 ID
}

输出：

{
    "task": {
        "id": str,
        "subject": str,
        "description": str,
        "status": Literal["pending", "in_progress", "completed"],
        "blocks": list[str],
        "blockedBy": list[str],
    } | None,  # 当 ID 未找到时为 None
}

TaskList

工具名称： TaskList

输入：

{}

输出：

{
    "tasks": [
        {
            "id": str,
            "subject": str,
            "status": Literal["pending", "in_progress", "completed"],
            "owner": str | None,
            "blockedBy": list[str],
        }
    ],
}

TaskOutput

在 Claude Code v2.1.277 中移除。之前检索来自运行中或已完成的后台任务的输出，BashOutput 被接受作为别名；Claude 改用 Read 在后台任务的输出文件上读取。

disallowed_tools 条目或仍然命名任一名称的拒绝规则被忽略而不发出警告。

TaskStop

工具名称： TaskStop。之前的名称 KillShell 和 KillBash 仍然被接受作为别名。

输入：

{
    "task_id": str | None,  # 要停止的后台任务的 ID
    "shell_id": str | None,  # 已弃用：改用 task_id
}

输出：

{
    "message": str,  # 关于操作的状态消息
    "task_id": str,  # 被停止的任务的 ID
    "task_type": str,  # 被停止的任务的类型
    "command": str | None,  # 被停止的任务的命令或描述
}

ExitPlanMode

工具名称： ExitPlanMode

输入：

{
    "plan": str  # 用户要运行以获得批准的计划
}

输出：

{
    "message": str,  # 确认消息
    "approved": bool | None,  # 用户是否批准了计划
}

ListMcpResources

工具名称： ListMcpResourcesTool

输入：

{
    "server": str | None  # 可选的服务器名称以按其过滤资源
}

输出：

{
    "resources": [
        {
            "uri": str,
            "name": str,
            "description": str | None,
            "mimeType": str | None,
            "server": str,
        }
    ],
    "total": int,
}

ReadMcpResource

工具名称： ReadMcpResourceTool

输入：

{
    "server": str,  # MCP 服务器名称
    "uri": str,  # 要读取的资源 URI
}

输出：

{
    "contents": [
        {"uri": str, "mimeType": str | None, "text": str | None, "blob": str | None}
    ],
    "server": str,
}

构建持续对话界面

以下示例保持一个 ClaudeSDKClient 在多个回合中保持连接，以便 Claude 记住之前的消息。输入 new 可以断开连接并重新连接以获得新的会话，或输入 exit 结束对话。

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

错误处理

以下示例将 query() 调用包装在四种 错误类型 的处理程序中，这些是 SDK 会抛出的错误。

此示例捕获 ResultError，这需要 Python Agent SDK 0.2.140 或更高版本。

import asyncio

from claude_agent_sdk import (
    query,
    CLINotFoundError,
    ProcessError,
    ResultError,
    CLIJSONDecodeError,
)


async def main():
    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"
        )
    # Catch ResultError before ProcessError, which it subclasses. Its message
    # carries the error text. A failed final request, such as an API error,
    # arrives with subtype "success", so branch on terminal_reason first.
    except ResultError as e:
        if e.terminal_reason == "api_error":
            print(f"API request failed: {e}")
        else:
            print(f"Query ended with an error result ({e.terminal_reason or e.subtype}): {e}")
    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}")


asyncio.run(main())

沙箱配置

`SandboxSettings`

沙箱行为的配置。使用此来启用命令沙箱和以编程方式配置网络限制。

class SandboxSettings(TypedDict, total=False):
    enabled: bool
    autoAllowBashIfSandboxed: bool
    excludedCommands: list[str]
    allowUnsandboxedCommands: bool
    network: SandboxNetworkConfig
    ignoreViolations: SandboxIgnoreViolations
    enableWeakerNestedSandbox: bool

示例用法

import asyncio

from claude_agent_sdk import query, ClaudeAgentOptions

sandbox_settings = {
    "enabled": True,
    "autoAllowBashIfSandboxed": True,
    "failIfUnavailable": True,
    "network": {"allowLocalBinding": True},
}


async def main():
    try:
        async for message in query(
            prompt="Build and test my project",
            options=ClaudeAgentOptions(sandbox=sandbox_settings),
        ):
            print(message)
    except Exception as error:
        # A single-shot query() raises after yielding an error result,
        # such as when failIfUnavailable is set and the sandbox can't start.
        print(f"Session ended with an error: {error}")


asyncio.run(main())

`SandboxNetworkConfig`

沙箱模式的网络特定配置。这些设置适用于当父 SandboxSettings 中的 enabled 为 True 时的沙箱化 Bash 命令。它们不限制 WebFetch 工具，该工具改用 权限规则。

class SandboxNetworkConfig(TypedDict, total=False):
    allowedDomains: list[str]
    deniedDomains: list[str]
    allowManagedDomainsOnly: bool
    allowUnixSockets: list[str]
    allowAllUnixSockets: bool
    allowLocalBinding: bool
    allowMachLookup: list[str]
    httpProxyPort: int
    socksProxyPort: int

`SandboxIgnoreViolations`

用于忽略特定沙箱违规的配置。

class SandboxIgnoreViolations(TypedDict, total=False):
    file: list[str]
    network: list[str]

沙箱外命令的权限回退

当 allowUnsandboxedCommands 启用时，模型可以通过在工具输入中设置 dangerouslyDisableSandbox: True 来请求在沙箱外运行命令。这些请求回退到现有权限系统，意味着你的 can_use_tool 处理程序将被调用，允许你实现自定义授权逻辑。

你的 excludedCommands 条目改为自动绕过沙箱，无需模型参与；sandbox.excludedCommands 涵盖何时应用条目。

以下示例记录每个沙箱外请求并拒绝它，除非你自己的授权逻辑允许它：

import asyncio
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    HookMatcher,
    PermissionResultAllow,
    PermissionResultDeny,
    ToolPermissionContext,
)


def is_command_authorized(command: str | None) -> bool:
    # Replace with your own authorization logic
    return False



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)


asyncio.run(main())

另见

• SDK 概述 - 一般 SDK 概念
• TypeScript SDK 参考 - TypeScript SDK 文档
• 自定义工具 - 为 Claude 定义可调用的进程内 MCP 工具
• CLI 参考 - 命令行界面
• 常见工作流 - 分步指南