Skip to content

Python Agent SDK 提供两种与 Claude Code 交互的方式:query() 适合一次性任务,每次调用默认创建新会话;ClaudeSDKClient 维护多轮对话上下文,支持中断、文件回退和 MCP 服务状态查询。使用 pip install claude-agent-sdk 安装后,通过 @tool 装饰器定义自定义工具,用 create_sdk_mcp_server() 集成到会话中。需要注意 TypedDict 配置项(如 ThinkingConfig)在运行时是普通字典,必须用键访问而非属性。

Claude Code Agent SDK Python 参考:query 与 ClaudeSDKClient

Python Agent SDK 的完整 API 参考,包括函数、类型和类的详细说明。

安装

bash
pip install claude-agent-sdk

选择 query() 还是 ClaudeSDKClient

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

快速对比

特性query()ClaudeSDKClient
会话 (Session)默认创建新会话复用同一会话
对话 (Conversation)单次交换同一上下文内多次交换
连接 (Connection)自动管理手动控制
流式输入✅ 支持✅ 支持
中断 (Interrupts)❌ 不支持✅ 支持
钩子 (Hooks)✅ 支持✅ 支持
自定义工具✅ 支持✅ 支持
继续聊天通过 continue_conversationresume 手动✅ 自动
适用场景一次性任务连续对话

什么时候用 query()(一次性任务)

适合场景:

  • 不需要对话历史的一次性问题
  • 不依赖之前交换上下文的独立任务
  • 简单的自动化脚本
  • 每次希望从头开始的情况

什么时候用 ClaudeSDKClient(连续对话)

适合场景:

  • 继续对话 - 需要 Claude 记住上下文
  • 追问 - 基于之前的回答继续提问
  • 交互式应用 - 聊天界面、REPL
  • 响应驱动逻辑 - 下一步动作取决于 Claude 的回答
  • 会话控制 - 明确管理对话生命周期

函数 Functions

query()

默认情况下,每次与 Claude Code 交互时都会创建一个新会话。返回一个异步迭代器,按到达顺序产生消息。每次调用 query() 都是全新开始,不记得之前的交互,除非在 ClaudeAgentOptions 中传递 continue_conversation=Trueresume。参见会话

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

参数 Parameters

参数类型说明
promptstr | AsyncIterable[dict]输入提示,可以是字符串或用于流式模式的异步迭代器
optionsClaudeAgentOptions | None可选配置对象(默认为 ClaudeAgentOptions()
transportTransport | None与 CLI 进程通信的可选自定义传输层

返回值 Returns

返回 AsyncIterator[Message],从对话中产生消息。

示例 - 带选项

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

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

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

参数 Parameters

参数类型说明
namestr工具的唯一标识符
descriptionstr工具功能的人类可读描述
input_schematype | dict[str, Any]定义工具输入参数的 schema(见下方)
annotationsToolAnnotations | None可选 MCP 工具注解,向客户端提供行为提示

输入 schema 选项 Input schema options

  1. 简单类型映射(推荐):

    python
    {"text": str, "count": int, "enabled": bool}
  2. JSON Schema 格式(复杂校验):

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

返回值 Returns

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

示例

python
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

mcp.types 重新导出(也可用 from claude_agent_sdk import ToolAnnotations)。所有字段都是可选的提示;客户端不应依赖它们做安全决策。

字段类型默认值说明
titlestr | NoneNone工具的人类可读标题
readOnlyHintbool | NoneFalse若为 True,工具不修改其环境
destructiveHintbool | NoneTrue若为 True,工具可能执行破坏性更新(仅在 readOnlyHintFalse 时有意义)
idempotentHintbool | NoneFalse若为 True,相同参数的重复调用无额外效果(仅在 readOnlyHintFalse 时有意义)
openWorldHintbool | NoneTrue若为 True,工具与外部实体交互(如网页搜索)。若为 False,工具域是封闭的(如记忆工具)
python
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 服务器。

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

参数 Parameters

参数类型默认值说明
namestr-服务器的唯一标识符
versionstr"1.0.0"服务器版本字符串
toolslist[SdkMcpTool[Any]] | NoneNone使用 @tool 装饰器创建的工具函数列表

返回值 Returns

返回 McpSdkServerConfig 对象,可以传递给 ClaudeAgentOptions.mcp_servers

示例

python
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],  # 传递装饰后的函数
)

# 与 Claude 一起使用
options = ClaudeAgentOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=["mcp__calc__add", "mcp__calc__multiply"],
)

list_sessions()

列出过去的会话及其元数据。按项目目录过滤或列出所有项目中的会话。同步调用,立即返回。

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

参数 Parameters

参数类型默认值说明
directorystr | NoneNone要列出会话的目录。省略时,返回所有项目中的会话
limitint | NoneNone最多返回的会话数
include_worktreesboolTruedirectory 位于 git 仓库内时,包含所有 worktree 路径中的会话

返回类型: SDKSessionInfo

属性类型说明
session_idstr唯一会话标识
summarystr显示标题:自定义标题、自动生成的摘要或第一条提示
last_modifiedint最后修改时间(毫秒级 Unix 时间戳)
file_sizeint | None会话文件大小(字节),远程存储后端可能为 None
custom_titlestr | None用户设置的会话标题
first_promptstr | None会话中的第一条有意义的用户提示
git_branchstr | None会话结束时的 Git 分支
cwdstr | None会话的工作目录
tagstr | None用户设置的会话标签(参见 tag_session()
created_atint | None会话创建时间(毫秒级 Unix 时间戳)

示例

打印某个项目最近 10 个会话。结果按 last_modified 降序排列,因此第一个是最新的。省略 directory 则搜索所有项目。

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

检索过去会话中的消息。同步调用,立即返回。

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

参数 Parameters

参数类型默认值说明
session_idstr必需要检索消息的会话 ID
directorystr | NoneNone要查找的项目目录。省略时搜索所有项目
limitint | NoneNone最多返回的消息数
offsetint0从开头跳过的消息数

返回类型: SessionMessage

属性类型说明
typeLiteral["user", "assistant"]消息角色
uuidstr唯一消息标识符
session_idstr会话标识符
messageAny原始消息内容
parent_tool_use_idNone保留供将来使用

示例

python
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 读取单个会话的元数据,无需扫描整个项目目录。同步调用,立即返回。

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

参数 Parameters

参数类型默认值说明
session_idstr必需要查找的会话 UUID
directorystr | NoneNone项目目录路径。省略时搜索所有项目目录

返回 SDKSessionInfo,或 None(未找到会话时)。

示例

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

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

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

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

参数 Parameters

参数类型默认值说明
session_idstr必需要重命名的会话 UUID
titlestr必需新标题。去除首尾空格后不能为空
directorystr | NoneNone项目目录路径。省略时搜索所有项目目录

如果 session_id 不是有效 UUID 或 title 为空,则抛出 ValueError;如果找不到会话,则抛出 FileNotFoundError

示例

重命名最新会话以便后续更容易找到。新标题将出现在后续读取的 SDKSessionInfo.custom_title 中。

python
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 可清除标签。重复调用是安全的;最新的标签生效。同步。

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

参数 Parameters

参数类型默认值说明
session_idstr必需要打标签的会话 UUID
tagstr | None必需标签字符串,或 None 以清除。存储前会进行 Unicode 清理
directorystr | NoneNone项目目录路径。省略时搜索所有项目目录

如果 session_id 不是有效 UUID 或 tag 清理后为空,则抛出 ValueError;如果找不到会话,则抛出 FileNotFoundError

示例

给会话打标签,然后在后续读取时按标签过滤。传递 None 可清除现有标签。

python
from claude_agent_sdk import list_sessions, tag_session

# 给会话打标签
tag_session("550e8400-e29b-41d4-a716-446655440000", "needs-review")

# 稍后:查找所有带有该标签的会话
for session in list_sessions(directory="/path/to/project"):
    if session.tag == "needs-review":
        print(session.summary)

类 Classes

ClaudeSDKClient

在多次消息交换之间维持对话会话。 这是 TypeScript SDK 中 query() 函数内部工作的 Python 等价物——它创建一个可以继续对话的客户端对象。

主要特性 Key Features

  • 会话连续性:在多次 query() 调用之间维持对话上下文
  • 同一对话:会话保留之前的消息
  • 中断支持:可以在任务执行中途停止
  • 显式生命周期:你可以控制会话的开始和结束
  • 响应驱动流程:可以对响应做出反应并发送后续问题
  • 自定义工具和钩子:支持自定义工具(用 @tool 装饰器创建)和钩子
python
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: str) -> None
    async def set_model(self, model: str | None = None) -> None
    async def rewind_files(self, user_message_id: str) -> None
    async def get_mcp_status(self) -> 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

方法 Methods

方法说明
__init__(options)使用可选配置初始化客户端
connect(prompt)连接到 Claude,可选初始提示或消息流
query(prompt, session_id)以流式模式发送新请求
receive_messages()以异步迭代器方式接收所有来自 Claude 的消息
receive_response()接收消息,直到并包括一个 ResultMessage
interrupt()发送中断信号(仅在流式模式下有效)
set_permission_mode(mode)更改当前会话的权限模式
set_model(model)更改当前会话使用的模型。传递 None 恢复为默认值
rewind_files(user_message_id)将文件恢复到指定用户消息时的状态。需要 enable_file_checkpointing=True。参见文件检查点
get_mcp_status()获取所有已配置 MCP 服务器的状态。返回 McpStatusResponse
reconnect_mcp_server(server_name)重试连接到连接失败或断开的 MCP 服务器
toggle_mcp_server(server_name, enabled)在会话中期启用或禁用 MCP 服务器。禁用会移除其工具
stop_task(task_id)停止正在运行的后台任务。消息流中随后会出现状态为 "stopped"TaskNotificationMessage
get_server_info()获取服务器信息,包括会话 ID 和功能
disconnect()断开与 Claude 的连接

上下文管理器支持 Context Manager Support

客户端可以用作异步上下文管理器,实现自动连接管理:

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

重要: 在遍历消息时,避免使用 break 提前退出,这可能导致 asyncio 清理问题。相反,让迭代自然完成,或使用标志来跟踪已找到所需内容。

示例 - 继续对话

python
import asyncio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage


async def main():
    async with ClaudeSDKClient() as client:
        # 第一个问题
        await client.query("What's the capital of France?")

        # 处理响应
        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}")

        # 后续问题 - 会话保留了之前的上下文
        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}")

        # 继续追问 - 仍在同一对话中
        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 流式输入

python
import asyncio
from claude_agent_sdk import ClaudeSDKClient


async def message_stream():
    """动态生成消息。"""
    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:
        # 向 Claude 流式输入
        await client.query(message_stream())

        # 处理响应
        async for message in client.receive_response():
            print(message)

        # 同一会话中继续提问
        await client.query("Should we be concerned about these readings?")

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


asyncio.run(main())

示例 - 使用中断

python
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:
        # 启动一个长时间运行的任务
        await client.query("Count from 1 to 100 slowly, using the bash sleep command")

        # 让它运行一会儿
        await asyncio.sleep(2)

        # 中断任务
        await client.interrupt()
        print("Task interrupted!")

        # 清空中断任务的残留消息(包括它的 ResultMessage)
        async for message in client.receive_response():
            if isinstance(message, ResultMessage):
                print(f"Interrupted task finished with subtype={message.subtype!r}")
                # 对于被中断的任务,subtype 是 "error_during_execution"

        # 发送新命令
        await client.query("Just say hello instead")

        # 现在接收新的响应
        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())

中断后的缓冲区行为: interrupt() 发送停止信号,但不清除消息缓冲区。中断任务已产生的消息(包括 ResultMessage,其 subtype="error_during_execution")仍然留在流中。在读取新查询的响应之前,必须用 receive_response() 将它们排空。如果在 interrupt() 后立即发送新查询,且只调用一次 receive_response(),你将收到中断任务的消息,而不是新查询的响应。

示例 - 高级权限控制

python
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:
    """自定义工具权限逻辑。"""

    # 阻止写入系统目录
    if tool_name == "Write" and input_data.get("file_path", "").startswith("/system/"):
        return PermissionResultDeny(
            message="System directory write not allowed", interrupt=True
        )

    # 重定向敏感文件操作
    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}
        )

    # 其他所有情况都允许
    return PermissionResultAllow(updated_input=input_data)


async def main():
    options = ClaudeAgentOptions(
        can_use_tool=custom_permission_handler, allowed_tools=["Read", "Write", "Edit"]
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Update the system config file")

        async for message in client.receive_response():
            # 将使用沙箱路径代替
            print(message)


asyncio.run(main())

类型 Types

@dataclass vs TypedDict 此 SDK 使用两种类型。带有 @dataclass 装饰器的类(如 ResultMessageAgentDefinitionTextBlock)在运行时时是对象实例,支持属性访问:msg.result。使用 TypedDict 定义的类(如 ThinkingConfigEnabledMcpStdioServerConfigSyncHookJSONOutput)在运行时时是普通字典,需要键访问:config["budget_tokens"],而不是 config.budget_tokens。两种类型都支持 ClassName(field=value) 调用语法,但只有数据类产生具有属性的对象。

SdkMcpTool

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

python
@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
属性类型说明
namestr工具的唯一标识符
descriptionstr人类可读的描述
input_schematype[T] | dict[str, Any]输入校验 schema
handlerCallable[[T], Awaitable[dict[str, Any]]]处理工具执行的异步函数
annotationsToolAnnotations | None可选的 MCP 工具注解(如 readOnlyHintdestructiveHintopenWorldHint)。来自 mcp.types

Transport

自定义传输实现的基础抽象类。用于通过自定义通道(例如远程连接而非本地子进程)与 Claude 进程通信。

这是一个低级内部 API。接口在将来版本中可能变化。自定义实现必须更新以匹配任何接口变化。

python
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: ...
方法说明
connect()连接传输层,准备通信
write(data)向传输层写入原始数据(JSON + 换行)
read_messages()产生解析后 JSON 消息的异步迭代器
close()关闭连接并清理资源
is_ready()如果传输层可以发送和接收则返回 True
end_input()关闭输入流(例如对于子进程传输,关闭 stdin)

导入:from claude_agent_sdk import Transport

ClaudeAgentOptions

Claude Code 查询的配置数据类。

python
@dataclass
class ClaudeAgentOptions:
    tools: list[str] | ToolsPreset | None = None
    allowed_tools: list[str] = field(default_factory=list)
    system_prompt: str | SystemPromptPreset | None = None
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
    strict_mcp_config: bool = False
    permission_mode: PermissionMode | None = None
    continue_conversation: bool = False
    resume: str | None = None
    max_turns: int | None = None
    max_budget_usd: float | None = None
    disallowed_tools: list[str] = field(default_factory=list)
    model: str | None = None
    fallback_model: str | None = None
    betas: list[SdkBeta] = field(default_factory=list)
    output_format: dict[str, Any] | None = None
    permission_prompt_tool_name: str | None = None
    cwd: str | Path | None = None
    cli_path: str | Path | None = None
    settings: str | None = None
    add_dirs: list[str | Path] = field(default_factory=list)
    env: dict[str, str] = field(default_factory=dict)
    extra_args: dict[str, str | None] = field(default_factory=dict)
    max_buffer_size: int | None = None
    debug_stderr: Any = sys.stderr  # 已废弃
    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
    fork_session: bool = False
    agents: dict[str, AgentDefinition] | None = None
    setting_sources: list[SettingSource] | None = None
    sandbox: SandboxSettings | None = None
    plugins: list[SdkPluginConfig] = field(default_factory=list)
    max_thinking_tokens: int | None = None  # 已废弃,改用 thinking
    thinking: ThinkingConfig | None = None
    effort: EffortLevel | None = None
    enable_file_checkpointing: bool = False
    session_store: SessionStore | None = None
    session_store_flush: SessionStoreFlushMode = "batched"
属性类型默认值说明
toolslist[str] | ToolsPreset | NoneNone工具配置。使用 {"type": "preset", "preset": "claude_code"} 启用 Claude Code 的默认工具
allowed_toolslist[str][]自动批准的工具(无需提示)。这不会将 Claude 限制为仅这些工具;未列出的工具会回退到 permission_modecan_use_tool。使用 disallowed_tools 阻止工具。参见权限
system_promptstr | SystemPromptPreset | NoneNone系统提示配置。传入字符串作为自定义提示,或使用 {"type": "preset", "preset": "claude_code"} 使用 Claude Code 的提示。添加 "append" 可扩展预设
mcp_serversdict[str, McpServerConfig] | str | Path{}MCP 服务器配置,或指向配置文件的路径
strict_mcp_configboolFalse当为 True 时,仅使用 mcp_servers 中传递的服务器,忽略项目 .mcp.json、用户设置和插件提供的 MCP 服务器。对应 CLI 的 --strict-mcp-config 标志
permission_modePermissionMode | NoneNone工具使用的权限模式
continue_conversationboolFalse继续最近的对话
resumestr | NoneNone要恢复的会话 ID
max_turnsint | NoneNone最大智能推理回合数(工具使用往返次数)
max_budget_usdfloat | NoneNone当客户端成本估算达到此美元值时停止查询。与 total_cost_usd 使用相同估算;参见跟踪成本和用量了解准确性说明
disallowed_toolslist[str][]拒绝的工具。裸名称如 "Bash" 从 Claude 的上下文中移除工具。作用域规则如 "Bash(rm *)" 留下工具,并在所有权限模式下拒绝匹配的调用,包括 bypassPermissions。参见权限
enable_file_checkpointingboolFalse启用文件更改跟踪以支持回退。参见文件检查点
modelstr | NoneNone要使用的 Claude 模型
fallback_modelstr | NoneNone如果主要模型失败,使用的备用模型
betaslist[SdkBeta][]要启用的 Beta 功能。参见 SdkBeta 获取可用选项
output_formatdict[str, Any] | NoneNone结构化响应的输出格式(例如 {"type": "json_schema", "schema": {...}})。参见结构化输出
permission_prompt_tool_namestr | NoneNone用于权限提示的 MCP 工具名称
cwdstr | Path | NoneNone当前工作目录
cli_pathstr | Path | NoneNoneClaude Code CLI 可执行文件的自定义路径
settingsstr | NoneNone设置文件路径
add_dirslist[str | Path][]Claude 可以访问的额外目录
envdict[str, str]{}环境变量,合并到继承的进程环境之上。参见环境变量了解底层 CLI 读取的变量,以及处理慢速或停滞的 API 响应了解超时相关变量
extra_argsdict[str, str | None]{}直接传递给 CLI 的额外参数
max_buffer_sizeint | NoneNone缓冲 CLI stdout 时的最大字节数
debug_stderrAnysys.stderr已废弃 - 调试输出的文件类对象。改用 stderr 回调
stderrCallable[[str], None] | NoneNone用于 CLI 的 stderr 输出的回调函数
can_use_toolCanUseTool | NoneNone工具权限回调函数。参见权限类型了解详情
hooksdict[HookEvent, list[HookMatcher]] | NoneNone用于拦截事件的钩子配置
userstr | NoneNone用户标识符
include_partial_messagesboolFalse包含部分消息流式事件。启用后,会生成 StreamEvent 消息
include_hook_eventsboolFalse在消息流中包含钩子生命周期事件,作为 HookEventMessage 对象
fork_sessionboolFalse当使用 resume 恢复时,衍生出一个新的会话 ID,而不是继续原始会话
agentsdict[str, AgentDefinition] | NoneNone编程方式定义的子代理
pluginslist[SdkPluginConfig][]从本地路径加载的自定义插件。参见插件了解详情
sandboxSandboxSettings | NoneNone编程方式配置沙箱行为。参见沙箱设置了解详情
setting_sourceslist[SettingSource] | NoneNone(CLI 默认:所有来源)控制要加载哪些文件系统设置。传递 [] 禁用用户、项目和本地设置。托管策略设置始终加载。参见使用 Claude Code 功能
skillslist[str] | Literal["all"] | NoneNone会话可用的技能。传递 "all" 启用所有已发现的技能,或传递技能名称列表。设置后,SDK 自动启用 Skill 工具,无需在 allowed_tools 中列出。参见技能
max_thinking_tokensint | NoneNone已废弃 - 思维块的最大令牌数。改用 thinking
thinkingThinkingConfig | NoneNone控制扩展思维行为。优先级高于 max_thinking_tokens
effortEffortLevel | NoneNone思维深度的努力程度
session_storeSessionStore | NoneNone将会话记录镜像到外部后端,使任何主机都能恢复。参见将会话持久化到外部存储
session_store_flushLiteral["batched", "eager"]"batched"何时将镜像的记录刷新到 session_store"batched" 每回合或缓冲区满时刷新;"eager" 每帧后触发后台刷新。当 session_storeNone 时忽略

处理慢速或停滞的 API 响应

CLI 子进程读取几个控制 API 超时和停滞检测的环境变量。通过 ClaudeAgentOptions.env 传递它们:

python
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。每次重试都有自己的 API_TIMEOUT_MS 窗口,因此最坏情况下的墙上时间大约是 API_TIMEOUT_MS × (CLAUDE_CODE_MAX_RETRIES + 1) 加上退避时间。
  • CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS:对于使用 run_in_background 启动的子代理的停滞看门狗。默认 600000。每次流式事件时重置;如果停滞,则中止子代理,标记任务失败,并将错误(含任何部分结果)传递给父代理。不适用于同步子代理。
  • CLAUDE_ENABLE_STREAM_WATCHDOG=1 配合 CLAUDE_STREAM_IDLE_TIMEOUT_MS:当标头已到达但响应体停止流式时中止请求。默认关闭。CLAUDE_STREAM_IDLE_TIMEOUT_MS 默认为 300000,并钳制到该最小值。中止的请求通过正常的重试路径进行。

OutputFormat

结构化输出校验的配置。将其作为字典传递给 ClaudeAgentOptionsoutput_format 字段:

python
# output_format 的预期字典形状
{
    "type": "json_schema",
    "schema": {...},  # 你的 JSON Schema 定义
}
字段必需说明
type必须是 "json_schema",表示 JSON Schema 校验
schema用于输出校验的 JSON Schema 定义

SystemPromptPreset

用于使用 Claude Code 的预设系统提示并可选择添加内容的配置。

python
class SystemPromptPreset(TypedDict):
    type: Literal["preset"]
    preset: Literal["claude_code"]
    append: NotRequired[str]
    exclude_dynamic_sections: NotRequired[bool]
字段必需说明
type必须是 "preset" 以使用预设系统提示
preset必须是 "claude_code" 以使用 Claude Code 的系统提示
append要追加到预设系统提示后的额外指令
exclude_dynamic_sections将会话级上下文(如工作目录、git-repo 标志、自动记忆路径)从系统提示移到第一条用户消息中。提高跨用户和机器的提示缓存复用率。参见修改系统提示

SettingSource

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

python
SettingSource = Literal["user", "project", "local"]
说明位置
"user"全局用户设置~/.claude/settings.json
"project"共享项目设置(版本控制).claude/settings.json
"local"本地项目设置(gitignored).claude/settings.local.json

默认行为

当省略 setting_sources 或传递 None 时,query() 加载与 Claude Code CLI 相同的文件系统设置:用户、项目和本地。托管策略设置始终加载。参见settingSources 不控制的内容了解哪些输入无论此选项如何都会读取,以及如何禁用它们。

为什么要使用 setting_sources

禁用文件系统设置:

python
# 不从磁盘加载用户、项目或本地设置
from claude_agent_sdk import query, ClaudeAgentOptions

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

在 Python SDK 0.1.59 及更早版本中,空列表被视为与省略选项相同,因此 setting_sources=[] 不会禁用文件系统设置。如果你需要空列表生效,请升级到更新版本。TypeScript SDK 不受影响。

显式加载所有文件系统设置:

python
from claude_agent_sdk import query, ClaudeAgentOptions

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

仅加载特定设置来源:

python
# 仅加载项目设置,忽略用户和本地设置
async for message in query(
    prompt="Run CI checks",
    options=ClaudeAgentOptions(
        setting_sources=["project"]  # 仅 .claude/settings.json
    ),
):
    print(message)

测试和 CI 环境:

python
# 确保 CI 行为一致,排除本地设置
async for message in query(
    prompt="Run tests",
    options=ClaudeAgentOptions(
        setting_sources=["project"],  # 仅团队共享设置
        permission_mode="bypassPermissions",
    ),
):
    print(message)

仅 SDK 应用:

python
# 全部通过编程方式定义。
# 传递 [] 以不加载文件系统设置来源。
async for message in query(
    prompt="Review this PR",
    options=ClaudeAgentOptions(
        setting_sources=[],
        agents={...},
        mcp_servers={...},
        allowed_tools=["Read", "Grep", "Glob"],
    ),
):
    print(message)

加载 CLAUDE.md 项目指令:

python
# 加载项目设置以包含 CLAUDE.md 文件
async for message in query(
    prompt="Add a new feature following project conventions",
    options=ClaudeAgentOptions(
        system_prompt={
            "type": "preset",
            "preset": "claude_code",  # 使用 Claude Code 的系统提示
        },
        setting_sources=["project"],  # 从项目加载 CLAUDE.md
        allowed_tools=["Read", "Write", "Edit"],
    ),
):
    print(message)

设置优先级

当加载多个来源时,设置按以下优先级合并(从高到低):

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

编程选项(如 agentsallowed_tools)会覆盖用户、项目和本地文件系统设置。托管策略设置优先于编程选项。

AgentDefinition

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

python
@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
字段必需说明
description何时使用此子代理的自然语言描述
prompt子代理的系统提示
tools允许的工具名称数组。如果省略,继承所有工具
disallowedTools要从子代理工具集中移除的工具名称数组
model此代理的模型覆盖。接受别名如 "sonnet""opus""haiku""inherit",或完整模型 ID。如果省略,使用主模型
skills预加载到代理上下文中的技能名称列表。未列出的技能仍可通过 Skill 工具调用
memory此代理的记忆来源:"user""project""local"
mcpServers此代理可用的 MCP 服务器。每个条目是服务器名称或内联的 {name: config}字典
initialPrompt当此代理作为主线程代理运行时,自动提交作为第一条用户消息
maxTurns最大智能推理回合数,之后代理停止
background当调用时,此代理作为非阻塞后台任务运行
effort此代理的推理努力程度。接受命名级别或整数。参见 EffortLevel
permissionMode此代理内工具执行的权限模式。参见 PermissionMode

AgentDefinition 字段名称使用驼峰式(camelCase),如 disallowedToolspermissionModemaxTurns。这些名称直接映射到与 TypeScript SDK 共享的网络格式。这与 ClaudeAgentOptions 不同,后者使用 Python 的 snake_case 表示等效顶层字段,如 disallowed_toolspermission_mode。因为 AgentDefinition 是数据类,传递 snake_case 关键字会在构造时抛出 TypeError

PermissionMode

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

python
PermissionMode = Literal[
    "default",          # 标准权限行为
    "acceptEdits",      # 自动接受文件编辑
    "plan",             # 计划模式 - 仅允许只读工具
    "dontAsk",          # 拒绝任何未预先批准的操作,而不是提示
    "bypassPermissions", # 绕过所有权限检查(谨慎使用)
]

EffortLevel

指导思维深度的努力程度。

python
EffortLevel = Literal[
    "low",      # 最小思维,最快响应
    "medium",   # 中等思维
    "high",     # 深度推理
    "xhigh",    # 扩展推理(仅 Opus 4.7 支持;其他模型回退到 "high")
    "max",     # 最大努力
]

CanUseTool

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

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

回调接收:

  • tool_name:被调用的工具名称
  • input_data:工具的输入参数
  • context:包含额外信息的 ToolPermissionContext

返回 PermissionResultPermissionResultAllowPermissionResultDeny)。

ToolPermissionContext

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

python
@dataclass
class ToolPermissionContext:
    signal: Any | None = None  # 未来:中止信号支持
    suggestions: list[PermissionUpdate] = field(default_factory=list)
    blocked_path: str | None = None
    decision_reason: str | None = None
    title: str | None = None
    display_name: str | None = None
    description: str | None = None
字段类型说明
signalAny | None保留供将来中止信号支持
suggestionslist[PermissionUpdate]来自 CLI 的权限更新建议。Bash 提示会包含带有 localSettings 目的地的建议,因此将其返回在 updated_permissions 中会将规则写入 .claude/settings.local.json 并在会话间持久化
blocked_pathstr | None触发权限请求的文件路径(如果适用)。例如,Bash 命令尝试访问允许目录外的路径时
decision_reasonstr | None触发此权限请求的原因。当挂钩的 permissionDecisionReason 返回 "ask" 时,从 PreToolUse 挂钩转发
titlestr | None完整权限提示句子,如 Claude wants to read foo.txt。如果存在,用作主要提示文本
display_namestr | None工具操作的简短名词短语,如 Read file,适合按钮标签
descriptionstr | None权限 UI 的人类可读副标题

PermissionResult

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

python
PermissionResult = PermissionResultAllow | PermissionResultDeny

PermissionResultAllow

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

python
@dataclass
class PermissionResultAllow:
    behavior: Literal["allow"] = "allow"
    updated_input: dict[str, Any] | None = None
    updated_permissions: list[PermissionUpdate] | None = None
字段类型默认值说明
behaviorLiteral["allow"]"allow"必须为 "allow"
updated_inputdict[str, Any] | NoneNone替代原始输入的修改后输入
updated_permissionslist[PermissionUpdate] | NoneNone要应用的权限更新

PermissionResultDeny

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

python
@dataclass
class PermissionResultDeny:
    behavior: Literal["deny"] = "deny"
    message: str = ""
    interrupt: bool = False
字段类型默认值说明
behaviorLiteral["deny"]"deny"必须为 "deny"
messagestr""解释拒绝原因的消息
interruptboolFalse是否中断当前执行

PermissionUpdate

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

python
@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
字段类型说明
typeLiteral[...]权限更新操作的类型
ruleslist[PermissionRuleValue] | None用于添加/替换/移除操作的规则
behaviorLiteral["allow", "deny", "ask"] | None基于规则的操作行为
modePermissionMode | NonesetMode 操作的模式
directorieslist[str] | None添加/移除目录操作的目录列表
destinationLiteral[...] | None应用权限更新的位置

PermissionRuleValue

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

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

ToolsPreset

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

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

ThinkingConfig

控制扩展思维行为。是三种配置的联合:

python
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
变体字段说明
adaptivetype, displayClaude 自适应决定何时思考
enabledtype, budget_tokens, display以特定令牌预算启用思考
disabledtype禁用思考

可选的 display 字段控制思考文本是以 "summarized" 还是 "omitted" 方式返回。在 Claude Opus 4.7 及更高版本上,API 默认是 "omitted",因此设置 "summarized" 可以在 ThinkingBlock 输出中接收思考内容。

因为这些是 TypedDict 类,运行时时是普通字典。可以构建为字典字面量或像构造函数一样调用类;两者都产生 dict。使用 config["budget_tokens"] 访问字段,而不是 config.budget_tokens

python
from claude_agent_sdk import ClaudeAgentOptions, ThinkingConfigEnabled

# 选项1:字典字面量(推荐,无需导入)
options = ClaudeAgentOptions(thinking={"type": "enabled", "budget_tokens": 20000})

# 选项2:构造函数式(返回普通 dict)
config = ThinkingConfigEnabled(type="enabled", budget_tokens=20000)
print(config["budget_tokens"])  # 20000
# config.budget_tokens 会引发 AttributeError

SdkBeta

SDK Beta 功能的文字类型。

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

ClaudeAgentOptionsbetas 字段一起使用以启用 Beta 功能。

context-1m-2025-08-07 Beta 已于 2026 年 4 月 30 日退役。将此标头传递给 Claude Sonnet 4.5 或 Sonnet 4 没有任何效果,超过标准 200k Token 上下文窗口的请求会返回错误。要使用 100 万 Token 上下文窗口,请迁移到 Claude Sonnet 4.6、Claude Opus 4.6 或 Claude Opus 4.7,它们以标准定价包含 100 万上下文,无需 Beta 标头。

McpSdkServerConfig

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

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

McpServerConfig

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

python
McpServerConfig = (
    McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig
)

McpStdioServerConfig

python
class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]  # 向后兼容可选
    command: str
    args: NotRequired[list[str]]
    env: NotRequired[dict[str, str]]

McpSSEServerConfig

python
class McpSSEServerConfig(TypedDict):
    type: Literal["sse"]
    url: str
    headers: NotRequired[dict[str, str]]

McpHttpServerConfig

python
class McpHttpServerConfig(TypedDict):
    type: Literal["http"]
    url: str
    headers: NotRequired[dict[str, str]]

McpServerStatusConfig

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

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

McpSdkServerConfigStatusMcpSdkServerConfig 的可序列化形式,仅包含 type ("sdk") 和 name (str) 字段;进程内的 instance 被省略。McpClaudeAIProxyServerConfig 包含 type ("claudeai-proxy")、url (str) 和 id (str) 字段。

McpStatusResponse

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

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

McpServerStatus

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

python
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]]
字段类型说明
namestr服务器名称
statusstr其中之一:"connected", "failed", "needs-auth", "pending", "disabled"
serverInfodict (可选)服务器名称和版本 ({"name": str, "version": str})
errorstr (可选)如果服务器连接失败,错误消息
configMcpServerStatusConfig (可选)服务器配置。与 McpServerConfig 形状相同(stdio、SSE、HTTP 或 SDK),外加用于通过 claude.ai 连接的 claudeai-proxy 变体
scopestr (可选)配置作用域
toolslist (可选)此服务器提供的工具,每个都有 namedescriptionannotations 字段

SdkPluginConfig

在 SDK 中加载插件的配置。

python
class SdkPluginConfig(TypedDict):
    type: Literal["local"]
    path: str
字段类型说明
typeLiteral["local"]必须为 "local"(目前仅支持本地插件)
pathstr插件目录的绝对或相对路径

示例:

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

关于创建和使用插件的完整信息,参见插件

消息类型 Message Types

Message

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

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

UserMessage

用户输入消息。

python
@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
字段类型说明
contentstr | list[ContentBlock]消息内容,文本或内容块
uuidstr | None唯一消息标识符
parent_tool_use_idstr | None如果此消息是工具结果响应,则这是工具使用 ID
tool_use_resultdict[str, Any] | None工具结果数据(如果适用)

AssistantMessage

包含内容块的助手响应消息。

python
@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
字段类型说明
contentlist[ContentBlock]响应中的内容块列表
modelstr生成响应的模型
parent_tool_use_idstr | None如果这是嵌套响应,则为工具使用 ID
errorAssistantMessageError | None如果响应遇到错误,则为此错误类型
usagedict[str, Any] | None每条消息的令牌使用情况(与 ResultMessage.usage 相同键)
message_idstr | NoneAPI 消息 ID。一次轮次中的多条消息共享相同 ID

AssistantMessageError

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

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

SystemMessage

带有元数据的系统消息。

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

ResultMessage

包含成本和用法信息的最终结果消息。

python
@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, Any] | 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

usage 字典包含以下键(如果存在):

类型说明
input_tokensint总输入 Token 数。
output_tokensint总输出 Token 数。
cache_creation_input_tokensint用于创建新缓存条目的 Token。
cache_read_input_tokensint从缓存条目中读取的 Token。

model_usage 字典将模型名称映射到各模型的使用情况。内部字典键使用驼峰式,因为值是从底层 CLI 进程直接传递过来的,匹配 TypeScript 的 ModelUsage 类型:

类型说明
inputTokensint此模型的输入 Token。
outputTokensint此模型的输出 Token。
cacheReadInputTokensint此模型的缓存读取 Token。
cacheCreationInputTokensint此模型的缓存创建 Token。
webSearchRequestsint此模型发出的网络搜索请求数。
costUSDfloat此模型的估计成本(美元),客户端计算。参见跟踪成本和用量了解计费说明。
contextWindowint此模型的上下文窗口大小。
maxOutputTokensint此模型的最大输出 Token 限制。

StreamEvent

流式传输期间的部分消息更新事件。仅当 ClaudeAgentOptionsinclude_partial_messages=True 时接收。通过 from claude_agent_sdk.types import StreamEvent 导入。

python
@dataclass
class StreamEvent:
    uuid: str
    session_id: str
    event: dict[str, Any]  # 原始 Claude API 流事件
    parent_tool_use_id: str | None = None
字段类型说明
uuidstr此事件的唯一标识符
session_idstr会话标识符
eventdict[str, Any]原始 Claude API 流事件数据
parent_tool_use_idstr | None如果此事件来自子代理,则是父工具使用 ID

RateLimitEvent

当速率限制状态变化时发出(例如,从 "allowed" 变为 "allowed_warning")。用于在用户达到硬限制之前发出警告,或在状态为 "rejected" 时降级。

python
@dataclass
class RateLimitEvent:
    rate_limit_info: RateLimitInfo
    uuid: str
    session_id: str
字段类型说明
rate_limit_infoRateLimitInfo当前速率限制状态
uuidstr唯一事件标识符
session_idstr会话标识符

RateLimitInfo

RateLimitEvent 携带的速率限制状态。

python
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)
字段类型说明
statusRateLimitStatus当前状态。"allowed_warning" 表示接近限制;"rejected" 表示已命中限制
resets_atint | None速率限制窗口重置的 Unix 时间戳
rate_limit_typeRateLimitType | None应用的速率限制窗口
utilizationfloat | None已消耗的速率限制比例(0.0 到 1.0)
overage_statusRateLimitStatus | None按需付费超额使用的状态(如果适用)
overage_resets_atint | None超额使用窗口重置的 Unix 时间戳
overage_disabled_reasonstr | None如果状态为 "rejected",则说明超额使用不可用的原因
rawdict[str, Any]CLI 返回的完整原始字典,包括上面未建模的字段

TaskStartedMessage

当后台任务启动时发出。后台任务是在主轮次之外跟踪的任何内容:后台运行的 Bash 命令、Monitor 监听、通过 Agent 工具产生的子代理或远程代理。task_type 字段告诉你具体是哪种。此命名与 TaskAgent 工具重命名无关。

python
@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
字段类型说明
task_idstr任务的唯一标识符
descriptionstr任务描述
uuidstr唯一消息标识符
session_idstr会话标识符
tool_use_idstr | None关联的工具使用 ID
task_typestr | None哪种后台任务:"local_bash" 用于后台 Bash 和 Monitor 监听,"local_agent",或 "remote_agent"

TaskUsage

后台任务的 Token 和时间数据。

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

TaskProgressMessage

定期发出,提供运行中的后台任务的进度更新。

python
@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
字段类型说明
task_idstr任务的唯一标识符
descriptionstr当前状态描述
usageTaskUsage此任务到目前为止的 Token 使用情况
uuidstr唯一消息标识符
session_idstr会话标识符
tool_use_idstr | None关联的工具使用 ID
last_tool_namestr | None任务最后使用的工具名称

TaskNotificationMessage

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

python
@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
字段类型说明
task_idstr任务的唯一标识符
statusTaskNotificationStatus之一:"completed""failed""stopped"
output_filestr任务输出文件的路径
summarystr任务结果的摘要
uuidstr唯一消息标识符
session_idstr会话标识符
tool_use_idstr | None关联的工具使用 ID
usageTaskUsage | None任务的最终 Token 使用情况

内容块类型 Content Block Types

ContentBlock

所有内容块的联合类型。

python
ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

TextBlock

文本内容块。

python
@dataclass
class TextBlock:
    text: str

ThinkingBlock

思维内容块(适用于具有思维能力的模型)。

python
@dataclass
class ThinkingBlock:
    thinking: str
    signature: str

ToolUseBlock

工具使用请求块。

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

ToolResultBlock

工具执行结果块。

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

错误类型 Error Types

ClaudeSDKError

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

python
class ClaudeSDKError(Exception):
    """Claude SDK 基础错误。"""

CLINotFoundError

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

python
class CLINotFoundError(CLIConnectionError):
    def __init__(
        self, message: str = "Claude Code not found", cli_path: str | None = None
    ):
        """
        参数:
            message: 错误消息(默认:"Claude Code not found")
            cli_path: 未找到的 CLI 的可选路径
        """

CLIConnectionError

当与 Claude Code 的连接失败时引发。

python
class CLIConnectionError(ClaudeSDKError):
    """连接 Claude Code 失败。"""

ProcessError

当 Claude Code 进程失败时引发。

python
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

当 JSON 解析失败时引发。

python
class CLIJSONDecodeError(ClaudeSDKError):
    def __init__(self, line: str, original_error: Exception):
        """
        参数:
            line: 解析失败的文本行
            original_error: 原始的 JSON 解码异常
        """
        self.line = line
        self.original_error = original_error

钩子类型 Hook Types

关于使用钩子的完整指南(含示例和常见模式),参见钩子指南

HookEvent

支持的钩子事件类型。

python
HookEvent = Literal[
    "PreToolUse",          # 工具执行前调用
    "PostToolUse",         # 工具执行后调用
    "PostToolUseFailure",  # 工具执行失败时调用
    "UserPromptSubmit",    # 用户提交提示时调用
    "Stop",                # 停止执行时调用
    "SubagentStop",        # 子代理停止时调用
    "PreCompact",          # 消息压缩前调用
    "Notification",        # 通知事件时调用
    "SubagentStart",       # 子代理启动时调用
    "PermissionRequest",   # 需要权限决策时调用
]

TypeScript SDK 支持更多钩子事件,Python 尚不可用:SessionStartSessionEndSetupTeammateIdleTaskCompletedConfigChangeWorktreeCreateWorktreeRemovePostToolBatch

HookCallback

钩子回调函数的类型定义。

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

参数:

  • input:基于 hook_event_name 的强类型钩子输入(参见 HookInput
  • tool_use_id:可选的工具使用标识符(用于与工具相关的钩子)
  • context:带有额外信息的钩子上下文

返回 HookJSONOutput,其中可能包含:

  • decision"block" 用于阻止该操作
  • systemMessage:向用户显示的警告消息
  • hookSpecificOutput:特定于钩子的输出数据

HookContext

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

python
class HookContext(TypedDict):
    signal: Any | None  # 未来:中止信号支持

HookMatcher

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

python
@dataclass
class HookMatcher:
    matcher: str | None = (
        None  # 要匹配的工具名称或模式(如 "Bash"、"Write|Edit")
    )
    hooks: list[HookCallback] = field(
        default_factory=list
    )  # 要执行的回调列表
    timeout: float | None = (
        None  # 此匹配器中所有钩子的超时秒数(默认:60)
    )

HookInput

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

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

BaseHookInput

所有钩子输入类型中存在的基字段。

python
class BaseHookInput(TypedDict):
    session_id: str
    transcript_path: str
    cwd: str
    permission_mode: NotRequired[str]
字段类型说明
session_idstr当前会话标识符
transcript_pathstr会话记录文件的路径
cwdstr当前工作目录
permission_modestr (可选)当前权限模式

PreToolUseHookInput

PreToolUse 钩子事件的输入数据。

python
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]
字段类型说明
hook_event_nameLiteral["PreToolUse"]始终为 "PreToolUse"
tool_namestr即将执行的工具名称
tool_inputdict[str, Any]工具的输入参数
tool_use_idstr此工具使用的唯一标识符
agent_idstr (可选)当钩子在子代理内部触发时,提供子代理标识符
agent_typestr (可选)当钩子在子代理内部触发时,提供子代理类型

PostToolUseHookInput

PostToolUse 钩子事件的输入数据。

python
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]
字段类型说明
hook_event_nameLiteral["PostToolUse"]始终为 "PostToolUse"
tool_namestr已执行的工具名称
tool_inputdict[str, Any]使用的输入参数
tool_responseAny工具执行的响应
tool_use_idstr此工具使用的唯一标识符
agent_idstr (可选)当钩子在子代理内部触发时,提供子代理标识符
agent_typestr (可选)当钩子在子代理内部触发时,提供子代理类型

PostToolUseFailureHookInput

PostToolUseFailure 钩子事件的输入数据。在工具执行失败时调用。

python
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]
字段类型说明
hook_event_nameLiteral["PostToolUseFailure"]始终为 "PostToolUseFailure"
tool_namestr失败的工具名称
tool_inputdict[str, Any]使用的输入参数
tool_use_idstr此工具使用的唯一标识符
errorstr来自失败执行的错误消息
is_interruptbool (可选)失败是否由中断引起
agent_idstr (可选)当钩子在子代理内部触发时,提供子代理标识符
agent_typestr (可选)当钩子在子代理内部触发时,提供子代理类型

UserPromptSubmitHookInput

UserPromptSubmit 钩子事件的输入数据。

python
class UserPromptSubmitHookInput(BaseHookInput):
    hook_event_name: Literal["UserPromptSubmit"]
    prompt: str
字段类型说明
hook_event_nameLiteral["UserPromptSubmit"]始终为 "UserPromptSubmit"
promptstr用户提交的提示

StopHookInput

Stop 钩子事件的输入数据。

python
class StopHookInput(BaseHookInput):
    hook_event_name: Literal["Stop"]
    stop_hook_active: bool
字段类型说明
hook_event_nameLiteral["Stop"]始终为 "Stop"
stop_hook_activebool停止钩子是否激活

SubagentStopHookInput

SubagentStop 钩子事件的输入数据。

python
class SubagentStopHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStop"]
    stop_hook_active: bool
    agent_id: str
    agent_transcript_path: str
    agent_type: str
字段类型说明
hook_event_nameLiteral["SubagentStop"]始终为 "SubagentStop"
stop_hook_activebool停止钩子是否激活
agent_idstr子代理的唯一标识符
agent_transcript_pathstr子代理记录文件的路径
agent_typestr子代理类型

PreCompactHookInput

PreCompact 钩子事件的输入数据。

python
class PreCompactHookInput(BaseHookInput):
    hook_event_name: Literal["PreCompact"]
    trigger: Literal["manual", "auto"]
    custom_instructions: str | None
字段类型说明
hook_event_nameLiteral["PreCompact"]始终为 "PreCompact"
triggerLiteral["manual", "auto"]触发压缩的原因
custom_instructionsstr | None压缩的自定义指令

NotificationHookInput

Notification 钩子事件的输入数据。

python
class NotificationHookInput(BaseHookInput):
    hook_event_name: Literal["Notification"]
    message: str
    title: NotRequired[str]
    notification_type: str
字段类型说明
hook_event_nameLiteral["Notification"]始终为 "Notification"
messagestr通知消息内容
titlestr (可选)通知标题
notification_typestr通知类型

SubagentStartHookInput

SubagentStart 钩子事件的输入数据。

python
class SubagentStartHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStart"]
    agent_id: str
    agent_type: str
字段类型说明
hook_event_nameLiteral["SubagentStart"]始终为 "SubagentStart"
agent_idstr子代理的唯一标识符
agent_typestr子代理类型

PermissionRequestHookInput

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

python
class PermissionRequestHookInput(BaseHookInput):
    hook_event_name: Literal["PermissionRequest"]
    tool_name: str
    tool_input: dict[str, Any]
    permission_suggestions: NotRequired[list[Any]]
字段类型说明
hook_event_nameLiteral["PermissionRequest"]始终为 "PermissionRequest"
tool_namestr请求权限的工具名称
tool_inputdict[str, Any]工具的输入参数
permission_suggestionslist[Any] (可选)CLI 建议的权限更新

HookJSONOutput

钩子回调返回值的联合类型。

python
HookJSONOutput = AsyncHookJSONOutput | SyncHookJSONOutput

SyncHookJSONOutput

带有控制和决策字段的同步钩子输出。

python
class SyncHookJSONOutput(TypedDict):
    # 控制字段
    continue_: NotRequired[bool]  # 是否继续(默认:True)
    suppressOutput: NotRequired[bool]  # 从记录中隐藏 stdout
    stopReason: NotRequired[str]  # 当 continue_ 为 False 时的消息

    # 决策字段
    decision: NotRequired[Literal["block"]]
    systemMessage: NotRequired[str]  # 向用户显示的警告消息
    reason: NotRequired[str]  # 给 Claude 的反馈

    # 特定于钩子的输出
    hookSpecificOutput: NotRequired[HookSpecificOutput]

在 Python 代码中使用 continue_(带下划线)。发送给 CLI 时会自动转换为 continue

HookSpecificOutput

包含钩子事件名称和特定于事件的字段的 TypedDict。形状取决于 hookEventName 值。关于每个钩子事件可用字段的完整详情,参见使用钩子控制执行

事件特定输出类型的区分联合。hookEventName 字段确定哪些字段有效。

python
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]  # 已废弃:改用 updatedToolOutput,适用于所有工具


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

将钩子执行推迟的异步钩子输出。

python
class AsyncHookJSONOutput(TypedDict):
    async_: Literal[True]  # 设为 True 以推迟执行
    asyncTimeout: NotRequired[int]  # 超时毫秒数

在 Python 代码中使用 async_(带下划线)。发送给 CLI 时会自动转换为 async

钩子使用示例 Hook Usage Example

此示例注册了两个钩子:一个阻止危险的 bash 命令(如 rm -rf /),另一个记录所有工具使用以供审计。安全钩子仅对 Bash 命令运行(通过 matcher),而记录钩子对所有工具运行。

python
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]:
    """验证并可能阻止危险的 bash 命令。"""
    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]:
    """记录所有工具使用以供审计。"""
    print(f"Tool used: {input_data.get('tool_name')}")
    return {}


options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(
                matcher="Bash", hooks=[validate_bash_command], timeout=120
            ),  # 2 分钟用于验证
            HookMatcher(
                hooks=[log_tool_use]
            ),  # 适用于所有工具(默认 60s 超时)
        ],
        "PostToolUse": [HookMatcher(hooks=[log_tool_use])],
    }
)

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

工具输入/输出类型 Tool Input/Output Types

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

Agent

工具名称: Agent(之前是 Task,仍可作为别名接受)

输入:

python
{
    "description": str,  # 任务的简短描述(3-5 个词)
    "prompt": str,  # 代理执行的任务
    "subagent_type": str,  # 要使用的专业代理类型
}

输出:

python
{
    "result": str,  # 来自子代理的最终结果
    "usage": dict | None,  # Token 使用统计
    "total_cost_usd": float | None,  # 估计总成本(美元)
    "duration_ms": int | None,  # 执行时长(毫秒)
}

AskUserQuestion

工具名称: AskUserQuestion

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

输入:

python
{
    "questions": [  # 要向用户提出的问题(1-4 个问题)
        {
            "question": str,  # 向用户提出的完整问题
            "header": str,  # 显示为芯片/标签的非常短的标签(最多 12 个字符)
            "options": [  # 可用选项(2-4 个选项)
                {
                    "label": str,  # 此选项的显示文本(1-5 个词)
                    "description": str,  # 此选项含义的解释
                }
            ],
            "multiSelect": bool,  # 设为 true 允许多项选择
        }
    ],
    "answers": dict[str, str | list[str]] | None,
    # 由权限系统填充的用户答案。
    # 多选答案可以是标签列表或逗号连接的字符串
}

输出:

python
{
    "questions": [  # 被问的问题
        {
            "question": str,
            "header": str,
            "options": [{"label": str, "description": str}],
            "multiSelect": bool,
        }
    ],
    "answers": dict[str, str],  # 将问题文本映射到答案字符串
    # 多选答案为逗号分隔
}

Bash

工具名称: Bash

输入:

python
{
    "command": str,  # 要执行的命令
    "timeout": int | None,  # 可选超时毫秒数(最大 600000)
    "description": str | None,  # 清晰、简洁的描述(5-10 个词)
    "run_in_background": bool | None,  # 设为 true 在后台运行
}

输出:

python
{
    "output": str,  # 合并的 stdout 和 stderr 输出
    "exitCode": int,  # 命令的退出码
    "killed": bool | None,  # 命令是否因超时被终止
    "shellId": str | None,  # 后台进程的 Shell ID
}

Monitor

工具名称: Monitor

在后台运行一个脚本,并将每行 stdout 作为事件传递给 Claude,以便它可以无需轮询即可做出反应。Monitor 遵循与 Bash 相同的权限规则。参见 Monitor 工具参考了解行为和提供者可用性。

输入:

python
{
    "command": str,  # Shell 脚本;每行 stdout 是一个事件,退出结束监听
    "description": str,  # 通知中显示的简短描述
    "timeout_ms": int | None,  # 在此截止时间后终止(默认 300000,最大 3600000)
    "persistent": bool | None,  # 在会话生命周期内运行;用 TaskStop 停止
}

输出:

python
{
    "taskId": str,  # 后台 monitor 任务的 ID
    "timeoutMs": int,  # 超时截止时间(毫秒)(persistent 时为 0)
    "persistent": bool | None,  # 当为 True 时,一直运行直到 TaskStop 或会话结束
}

Edit

工具名称: Edit

输入:

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

输出:

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

Read

工具名称: Read

输入:

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

输出(文本文件):

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

输出(图片):

python
{
    "image": str,  # Base64 编码的图片数据
    "mime_type": str,  # 图片 MIME 类型
    "file_size": int,  # 文件大小(字节)
}

Write

工具名称: Write

输入:

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

输出:

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

Glob

工具名称: Glob

输入:

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

输出:

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

Grep

工具名称: Grep

输入:

python
{
    "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 模式):

python
{
    "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 模式):

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

NotebookEdit

工具名称: NotebookEdit

输入:

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

输出:

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

WebFetch

工具名称: WebFetch

输入:

python
{
    "url": str,  # 要获取内容的 URL
    "prompt": str,  # 对获取的内容执行的提示
}

输出:

python
{
    "bytes": int,  # 获取内容的大小(字节)
    "code": int,  # HTTP 响应码
    "codeText": str,  # HTTP 响应码文本
    "result": str,  # 将提示应用于内容后的处理结果
    "durationMs": int,  # 获取和处理内容的毫秒数
    "url": str,  # 获取的 URL
}

WebSearch

工具名称: WebSearch

输入:

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

输出:

python
{
    "query": str,  # 搜索查询
    "results": list[str | {"tool_use_id": str, "content": list[{"title": str, "url": str}]}],
    "durationSeconds": float,  # 搜索时长(秒)
}

TodoWrite

工具名称: TodoWrite

自 Claude Code v2.1.142 起,TodoWrite 默认禁用。改用 TaskCreateTaskGetTaskUpdateTaskList。参见迁移到 Task 工具来更新你的监控代码,或设置 CLAUDE_CODE_ENABLE_TASKS=0 回退到 TodoWrite

输入:

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

输出:

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

TaskCreate

工具名称: TaskCreate

输入:

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

输出:

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

TaskUpdate

工具名称: TaskUpdate

输入:

python
{
    "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,
}

输出:

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

TaskGet

工具名称: TaskGet

输入:

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

输出:

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

TaskList

工具名称: TaskList

输入:

python
{}

输出:

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

BashOutput

工具名称: BashOutput

输入:

python
{
    "bash_id": str,  # 后台 shell 的 ID
    "filter": str | None,  # 可选正则表达式,用于过滤输出行
}

输出:

python
{
    "output": str,  # 自上次检查以来的新输出
    "status": "running" | "completed" | "failed",  # 当前 shell 状态
    "exitCode": int | None,  # 完成时的退出码
}

KillBash

工具名称: KillBash

输入:

python
{
    "shell_id": str  # 要终止的后台 shell 的 ID
}

输出:

python
{
    "message": str,  # 成功消息
    "shell_id": str,  # 被终止的 shell 的 ID
}

ExitPlanMode

工具名称: ExitPlanMode

输入:

python
{
    "plan": str  # 用户批准的计划
}

输出:

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

ListMcpResources

工具名称: ListMcpResources

输入:

python
{
    "server": str | None  # 可选服务器名称,用于过滤资源
}

输出:

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

ReadMcpResource

工具名称: ReadMcpResource

输入:

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

输出:

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

使用 ClaudeSDKClient 的高级功能 Advanced Features with ClaudeSDKClient

构建持续对话界面 Building a Continuous Conversation Interface

python
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
)
import asyncio


class ConversationSession:
    """与 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":
                # 断开并重新连接以获得新会话
                await self.client.disconnect()
                await self.client.connect()
                self.turn_count = 0
                print("Started new conversation session (previous context cleared)")
                continue

            # 发送消息 - 会话保留所有之前消息
            await self.client.query(user_input)
            self.turn_count += 1

            # 处理响应
            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()  # 响应后换行

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


# 对话示例:
# Turn 1 - You: "创建一个文件叫 hello.py"
# Turn 1 - Claude: "我将为您创建 hello.py 文件..."
# Turn 2 - You: "那个文件里有什么?"
# Turn 2 - Claude: "我刚创建的 hello.py 文件包含..."(记得!)
# Turn 3 - You: "给它添加一个 main 函数"
# Turn 3 - Claude: "我会给 hello.py 添加 main 函数..."(知道是哪个文件!)

asyncio.run(main())

使用钩子修改行为 Using Hooks for Behavior Modification

python
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    HookMatcher,
    HookContext,
)
import asyncio
from typing import Any


async def pre_tool_logger(
    input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
    """在执行前记录所有工具使用。"""
    tool_name = input_data.get("tool_name", "unknown")
    print(f"[PRE-TOOL] About to use: {tool_name}")

    # 你可以在此修改或阻止工具执行
    if tool_name == "Bash" and "rm -rf" in str(input_data.get("tool_input", {})):
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": "Dangerous command blocked",
            }
        }
    return {}


async def post_tool_logger(
    input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
    """执行后记录工具结果。"""
    tool_name = input_data.get("tool_name", "unknown")
    print(f"[POST-TOOL] Completed: {tool_name}")
    return {}


async def user_prompt_modifier(
    input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
    """向用户提示添加上下文。"""
    original_prompt = input_data.get("prompt", "")

    # 添加时间戳作为 Claude 的额外上下文
    from datetime import datetime

    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    return {
        "hookSpecificOutput": {
            "hookEventName": "UserPromptSubmit",
            "additionalContext": f"[Submitted at {timestamp}] Original prompt: {original_prompt}",
        }
    }


async def main():
    options = ClaudeAgentOptions(
        hooks={
            "PreToolUse": [
                HookMatcher(hooks=[pre_tool_logger]),
                HookMatcher(matcher="Bash", hooks=[pre_tool_logger]),
            ],
            "PostToolUse": [HookMatcher(hooks=[post_tool_logger])],
            "UserPromptSubmit": [HookMatcher(hooks=[user_prompt_modifier])],
        },
        allowed_tools=["Read", "Write", "Bash"],
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("List files in current directory")

        async for message in client.receive_response():
            # 钩子会自动记录工具使用
            pass


asyncio.run(main())

实时进度监控 Real-time Progress Monitoring

python
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    ToolUseBlock,
    ToolResultBlock,
    TextBlock,
)
import asyncio


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

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Create 5 Python files with different sorting algorithms")

        # 实时监控进度
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, ToolUseBlock):
                        if block.name == "Write":
                            file_path = block.input.get("file_path", "")
                            print(f"Creating: {file_path}")
                    elif isinstance(block, ToolResultBlock):
                        print("Completed tool execution")
                    elif isinstance(block, TextBlock):
                        print(f"Claude says: {block.text[:100]}...")

        print("Task completed!")


asyncio.run(monitor_progress())

使用示例 Example Usage

基本文件操作(使用 query)

python
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ToolUseBlock
import asyncio


async def create_project():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="acceptEdits",
        cwd="/home/user/project",
    )

    async for message in query(
        prompt="Create a Python project structure with setup.py", options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    print(f"Using tool: {block.name}")


asyncio.run(create_project())

错误处理 Error handling

python
from claude_agent_sdk import query, CLINotFoundError, ProcessError, CLIJSONDecodeError

try:
    async for message in query(prompt="Hello"):
        print(message)
except CLINotFoundError:
    print(
        "Claude Code CLI not found. Try reinstalling: pip install --force-reinstall claude-agent-sdk"
    )
except ProcessError as e:
    print(f"Process failed with exit code: {e.exit_code}")
except CLIJSONDecodeError as e:
    print(f"Failed to parse response: {e}")

使用客户端流式模式

python
from claude_agent_sdk import ClaudeSDKClient
import asyncio


async def interactive_session():
    async with ClaudeSDKClient() as client:
        # 发送初始消息
        await client.query("What's the weather like?")

        # 处理响应
        async for msg in client.receive_response():
            print(msg)

        # 发送后续问题
        await client.query("Tell me more about that")

        # 处理后续响应
        async for msg in client.receive_response():
            print(msg)


asyncio.run(interactive_session())

使用 ClaudeSDKClient 使用自定义工具

python
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    tool,
    create_sdk_mcp_server,
    AssistantMessage,
    TextBlock,
)
import asyncio
from typing import Any


# 使用 @tool 装饰器定义自定义工具
@tool("calculate", "Perform mathematical calculations", {"expression": str})
async def calculate(args: dict[str, Any]) -> dict[str, Any]:
    try:
        result = eval(args["expression"], {"__builtins__": {}})
        return {"content": [{"type": "text", "text": f"Result: {result}"}]}
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"Error: {str(e)}"}],
            "is_error": True,
        }


@tool("get_time", "Get current time", {})
async def get_time(args: dict[str, Any]) -> dict[str, Any]:
    from datetime import datetime

    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return {"content": [{"type": "text", "text": f"Current time: {current_time}"}]}


async def main():
    # 使用自定义工具创建 SDK MCP 服务器
    my_server = create_sdk_mcp_server(
        name="utilities", version="1.0.0", tools=[calculate, get_time]
    )

    # 使用服务器配置选项
    options = ClaudeAgentOptions(
        mcp_servers={"utils": my_server},
        allowed_tools=["mcp__utils__calculate", "mcp__utils__get_time"],
    )

    # 使用 ClaudeSDKClient 进行交互式工具使用
    async with ClaudeSDKClient(options=options) as client:
        await client.query("What's 123 * 456?")

        # 处理计算响应
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Calculation: {block.text}")

        # 后续查询时间
        await client.query("What time is it now?")

        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Time: {block.text}")


asyncio.run(main())

沙箱配置 Sandbox Configuration

SandboxSettings

沙箱行为的配置。用于以编程方式启用命令沙箱并配置网络限制。

python
class SandboxSettings(TypedDict, total=False):
    enabled: bool
    autoAllowBashIfSandboxed: bool
    excludedCommands: list[str]
    allowUnsandboxedCommands: bool
    network: SandboxNetworkConfig
    ignoreViolations: SandboxIgnoreViolations
    enableWeakerNestedSandbox: bool
属性类型默认值说明
enabledboolFalse启用沙箱模式以进行命令执行
autoAllowBashIfSandboxedboolTrue启用沙箱时自动批准 bash 命令
excludedCommandslist[str][]始终绕过沙箱限制的命令(例如 ["docker"])。这些命令自动以非沙箱方式运行,无需模型参与
allowUnsandboxedCommandsboolTrue允许模型请求在沙箱外运行命令。当 True 时,模型可以在工具输入中设置 dangerouslyDisableSandbox,这回退到权限系统
networkSandboxNetworkConfigNone网络特定的沙箱配置
ignoreViolationsSandboxIgnoreViolationsNone配置要忽略的沙箱违规
enableWeakerNestedSandboxboolFalse为兼容性启用较弱的嵌套沙箱

使用示例

python
from claude_agent_sdk import query, ClaudeAgentOptions, SandboxSettings

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

async for message in query(
    prompt="Build and test my project",
    options=ClaudeAgentOptions(sandbox=sandbox_settings),
):
    print(message)

Unix 套接字安全allowUnixSockets 选项可以授予对强大系统服务的访问权限。例如,允许 /var/run/docker.sock 实际上通过 Docker API 授予完全的主机系统访问权限,绕过沙箱隔离。仅允许绝对必要的 Unix 套接字,并理解每个套接字的安全影响。

SandboxNetworkConfig

沙箱模式的网络特定配置。

python
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
属性类型默认值说明
allowedDomainslist[str][]沙箱进程可以访问的域名
deniedDomainslist[str][]沙箱进程不能访问的域名。优先级高于 allowedDomains
allowManagedDomainsOnlyboolFalse仅受管设置:当在受管设置中设置时,忽略非受管设置源中的 allowedDomains。通过 SDK 选项设置时无效
allowUnixSocketslist[str][]进程可以访问的 Unix 套接字路径(例如 Docker 套接字)
allowAllUnixSocketsboolFalse允许访问所有 Unix 套接字
allowLocalBindingboolFalse允许进程绑定到本地端口(例如用于开发服务器)
allowMachLookuplist[str][]仅 macOS:允许的 XPC/Mach 服务名称。支持尾部通配符
httpProxyPortintNoneHTTP 代理端口
socksProxyPortintNoneSOCKS 代理端口

内置沙箱代理基于请求的主机名强制网络允许列表,不终止或检查 TLS 流量,因此 domain fronting 等技术可能绕过它。参见沙箱安全限制了解详情,以及安全部署了解配置 TLS 终止代理的方法。

SandboxIgnoreViolations

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

python
class SandboxIgnoreViolations(TypedDict, total=False):
    file: list[str]
    network: list[str]
属性类型默认值说明
filelist[str][]要忽略违规的文件路径模式
networklist[str][]要忽略违规的网络模式

非沙箱命令的权限回退 Permissions Fallback for Unsandboxed Commands

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

excludedCommands vs allowUnsandboxedCommands

  • excludedCommands:始终自动绕过沙箱的静态命令列表(例如 ["docker"])。模型无法控制此事。
  • allowUnsandboxedCommands:允许模型在运行时决定是否请求非沙箱执行,通过在工具输入中设置 dangerouslyDisableSandbox: True
python
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    HookMatcher,
    PermissionResultAllow,
    PermissionResultDeny,
    ToolPermissionContext,
)


async def can_use_tool(
    tool: str, input: dict, context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
    # 检查模型是否请求绕过沙箱
    if tool == "Bash" and input.get("dangerouslyDisableSandbox"):
        # 模型请求在沙箱外运行此命令
        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()


# 必需:虚拟钩子保持流打开以允许 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,  # 模型可以请求非沙箱执行
            },
            permission_mode="default",
            can_use_tool=can_use_tool,
            hooks={"PreToolUse": [HookMatcher(matcher=None, hooks=[dummy_hook])]},
        ),
    ):
        print(message)

这种模式使你能够:

  • 审计模型请求:记录模型何时请求非沙箱执行
  • 实现允许列表:仅允许特定命令以非沙箱方式运行
  • 添加审批工作流:要求特权操作明确授权

使用 dangerouslyDisableSandbox: True 运行的命令具有完全系统访问权限。确保你的 can_use_tool 处理程序仔细验证这些请求。

如果 permission_mode 设置为 bypassPermissionsallow_unsandboxed_commands 启用,模型可以在没有任何审批提示的情况下自主在沙箱外执行命令。这种组合实际上允许模型静默地绕过沙箱隔离。

参见

常见问题

query()ClaudeSDKClient 有什么区别?

query() 默认每次调用创建新会话,适合一次性任务;ClaudeSDKClient 维护同一会话,支持多轮对话、中断和 MCP 服务状态查询。选择 ClaudeSDKClient 做连续对话,query() 做简单自动化。

如何安装 Python Agent SDK?

运行 pip install claude-agent-sdk。需要 Python 3.8+ 和已安装的 Claude Code CLI。如果遇到 CLINotFoundError,尝试 pip install --force-reinstall claude-agent-sdk

支持哪些内置工具?

内置工具包括 ReadWriteEditBashGlobGrepWebSearchWebFetchNotebookEditTaskCreateAgent 等。可以通过 @tool 装饰器创建自定义工具,并通过 create_sdk_mcp_server() 集成。