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 参考,包括函数、类型和类的详细说明。

安装

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。参见会话

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

参数 Parameters

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

返回值 Returns

返回 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",
        cwd="/home/user/project",
    )

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

参数 Parameters

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

输入 schema 选项 Input schema options

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

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

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

返回值 Returns

一个装饰器函数,包装工具实现并返回 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

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

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

参数 Parameters

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

返回值 Returns

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

示例

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

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

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

参数 Parameters

参数 类型 默认值 说明
directory str | None None 要列出会话的目录。省略时,返回所有项目中的会话
limit int | None None 最多返回的会话数
include_worktrees bool True directory 位于 git 仓库内时,包含所有 worktree 路径中的会话

返回类型: SDKSessionInfo

属性 类型 说明
session_id str 唯一会话标识
summary str 显示标题:自定义标题、自动生成的摘要或第一条提示
last_modified int 最后修改时间(毫秒级 Unix 时间戳)
file_size int | None 会话文件大小(字节),远程存储后端可能为 None
custom_title str | None 用户设置的会话标题
first_prompt str | None 会话中的第一条有意义的用户提示
git_branch str | None 会话结束时的 Git 分支
cwd str | None 会话的工作目录
tag str | None 用户设置的会话标签(参见 tag_session()
created_at int | None 会话创建时间(毫秒级 Unix 时间戳)

示例

打印某个项目最近 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]

参数 Parameters

参数 类型 默认值 说明
session_id str 必需 要检索消息的会话 ID
directory str | None None 要查找的项目目录。省略时搜索所有项目
limit int | None None 最多返回的消息数
offset int 0 从开头跳过的消息数

返回类型: SessionMessage

属性 类型 说明
type Literal["user", "assistant"] 消息角色
uuid str 唯一消息标识符
session_id str 会话标识符
message Any 原始消息内容
parent_tool_use_id None 保留供将来使用

示例

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

参数 Parameters

参数 类型 默认值 说明
session_id str 必需 要查找的会话 UUID
directory str | None 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

参数 Parameters

参数 类型 默认值 说明
session_id str 必需 要重命名的会话 UUID
title str 必需 新标题。去除首尾空格后不能为空
directory str | 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

参数 Parameters

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

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

示例

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

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 装饰器创建)和钩子
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

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

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

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

示例 - 继续对话

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 流式输入

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

示例 - 使用中断

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(),你将收到中断任务的消息,而不是新查询的响应。

示例 - 高级权限控制

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 工具的定义。

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

Transport

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

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

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 查询的配置数据类。

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

处理慢速或停滞的 API 响应

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

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 字段:

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

SystemPromptPreset

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

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 从哪些基于文件系统的配置源加载设置。

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

禁用文件系统设置:

# 不从磁盘加载用户、项目或本地设置
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 不受影响。

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

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)

仅加载特定设置来源:

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

测试和 CI 环境:

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

仅 SDK 应用:

# 全部通过编程方式定义。
# 传递 [] 以不加载文件系统设置来源。
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 项目指令:

# 加载项目设置以包含 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

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

@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

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

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

EffortLevel

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

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

CanUseTool

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

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

回调接收:

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

返回 PermissionResultPermissionResultAllowPermissionResultDeny)。

ToolPermissionContext

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

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

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
字段 类型 默认值 说明
behavior Literal["allow"] "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
字段 类型 默认值 说明
behavior Literal["deny"] "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
字段 类型 说明
type Literal[...] 权限更新操作的类型
rules list[PermissionRuleValue] | None 用于添加/替换/移除操作的规则
behavior Literal["allow", "deny", "ask"] | None 基于规则的操作行为
mode PermissionMode | None setMode 操作的模式
directories list[str] | None 添加/移除目录操作的目录列表
destination Literal[...] | 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
变体 字段 说明
adaptive type, display Claude 自适应决定何时思考
enabled type, budget_tokens, display 以特定令牌预算启用思考
disabled type 禁用思考

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

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

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 功能的文字类型。

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 服务器的配置。

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

McpServerConfig

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

McpServerConfig = (
    McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig
)

McpStdioServerConfig

class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]  # 向后兼容可选
    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 传输变体的联合,加上一个仅输出类型的 claudeai-proxy 变体,用于通过 claude.ai 代理的服务器。

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 键下。

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

SdkPluginConfig

在 SDK 中加载插件的配置。

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

示例:

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

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

消息类型 Message Types

Message

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

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

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

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

AssistantMessageError

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

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

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, 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_tokens int 总输入 Token 数。
output_tokens int 总输出 Token 数。
cache_creation_input_tokens int 用于创建新缓存条目的 Token。
cache_read_input_tokens int 从缓存条目中读取的 Token。

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

类型 说明
inputTokens int 此模型的输入 Token。
outputTokens int 此模型的输出 Token。
cacheReadInputTokens int 此模型的缓存读取 Token。
cacheCreationInputTokens int 此模型的缓存创建 Token。
webSearchRequests int 此模型发出的网络搜索请求数。
costUSD float 此模型的估计成本(美元),客户端计算。参见跟踪成本和用量了解计费说明。
contextWindow int 此模型的上下文窗口大小。
maxOutputTokens int 此模型的最大输出 Token 限制。

StreamEvent

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

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

RateLimitEvent

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

@dataclass
class RateLimitEvent:
    rate_limit_info: RateLimitInfo
    uuid: str
    session_id: str
字段 类型 说明
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)
字段 类型 说明
status RateLimitStatus 当前状态。"allowed_warning" 表示接近限制;"rejected" 表示已命中限制
resets_at int | None 速率限制窗口重置的 Unix 时间戳
rate_limit_type RateLimitType | None 应用的速率限制窗口
utilization float | None 已消耗的速率限制比例(0.0 到 1.0)
overage_status RateLimitStatus | None 按需付费超额使用的状态(如果适用)
overage_resets_at int | None 超额使用窗口重置的 Unix 时间戳
overage_disabled_reason str | None 如果状态为 "rejected",则说明超额使用不可用的原因
raw dict[str, Any] CLI 返回的完整原始字典,包括上面未建模的字段

TaskStartedMessage

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

@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_id str 任务的唯一标识符
description str 任务描述
uuid str 唯一消息标识符
session_id str 会话标识符
tool_use_id str | None 关联的工具使用 ID
task_type str | None 哪种后台任务:"local_bash" 用于后台 Bash 和 Monitor 监听,"local_agent",或 "remote_agent"

TaskUsage

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

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
字段 类型 说明
task_id str 任务的唯一标识符
description str 当前状态描述
usage TaskUsage 此任务到目前为止的 Token 使用情况
uuid str 唯一消息标识符
session_id str 会话标识符
tool_use_id str | None 关联的工具使用 ID
last_tool_name str | 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
字段 类型 说明
task_id str 任务的唯一标识符
status TaskNotificationStatus 之一:"completed""failed""stopped"
output_file str 任务输出文件的路径
summary str 任务结果的摘要
uuid str 唯一消息标识符
session_id str 会话标识符
tool_use_id str | None 关联的工具使用 ID
usage TaskUsage | None 任务的最终 Token 使用情况

内容块类型 Content Block Types

ContentBlock

所有内容块的联合类型。

ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

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

错误类型 Error Types

ClaudeSDKError

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

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

CLINotFoundError

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

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 的连接失败时引发。

class CLIConnectionError(ClaudeSDKError):
    """连接 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

CLIJSONDecodeError

当 JSON 解析失败时引发。

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

支持的钩子事件类型。

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

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

HookCallback

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

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

参数:

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

返回 HookJSONOutput,其中可能包含:

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

HookContext

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

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

HookMatcher

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

@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 字段。

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

BaseHookInput

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

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

PreToolUseHookInput

PreToolUse 钩子事件的输入数据。

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_name Literal["PreToolUse"] 始终为 “PreToolUse”
tool_name str 即将执行的工具名称
tool_input dict[str, Any] 工具的输入参数
tool_use_id str 此工具使用的唯一标识符
agent_id str (可选) 当钩子在子代理内部触发时,提供子代理标识符
agent_type str (可选) 当钩子在子代理内部触发时,提供子代理类型

PostToolUseHookInput

PostToolUse 钩子事件的输入数据。

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_name Literal["PostToolUse"] 始终为 “PostToolUse”
tool_name str 已执行的工具名称
tool_input dict[str, Any] 使用的输入参数
tool_response Any 工具执行的响应
tool_use_id str 此工具使用的唯一标识符
agent_id str (可选) 当钩子在子代理内部触发时,提供子代理标识符
agent_type str (可选) 当钩子在子代理内部触发时,提供子代理类型

PostToolUseFailureHookInput

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

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_name Literal["PostToolUseFailure"] 始终为 “PostToolUseFailure”
tool_name str 失败的工具名称
tool_input dict[str, Any] 使用的输入参数
tool_use_id str 此工具使用的唯一标识符
error str 来自失败执行的错误消息
is_interrupt bool (可选) 失败是否由中断引起
agent_id str (可选) 当钩子在子代理内部触发时,提供子代理标识符
agent_type str (可选) 当钩子在子代理内部触发时,提供子代理类型

UserPromptSubmitHookInput

UserPromptSubmit 钩子事件的输入数据。

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

StopHookInput

Stop 钩子事件的输入数据。

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

SubagentStopHookInput

SubagentStop 钩子事件的输入数据。

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

PreCompactHookInput

PreCompact 钩子事件的输入数据。

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

NotificationHookInput

Notification 钩子事件的输入数据。

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

SubagentStartHookInput

SubagentStart 钩子事件的输入数据。

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

PermissionRequestHookInput

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

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

HookJSONOutput

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

HookJSONOutput = AsyncHookJSONOutput | SyncHookJSONOutput

SyncHookJSONOutput

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

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 字段确定哪些字段有效。

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

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

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

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

钩子使用示例 Hook Usage Example

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

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,仍可作为别名接受)

输入:

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

输出:

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

AskUserQuestion

工具名称: AskUserQuestion

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

输入:

{
    "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,
    # 由权限系统填充的用户答案。
    # 多选答案可以是标签列表或逗号连接的字符串
}

输出:

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

Bash

工具名称: Bash

输入:

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

输出:

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

Monitor

工具名称: Monitor

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

输入:

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

输出:

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

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 notebook 的绝对路径
    "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,  # 编辑后 notebook 中的总单元格数
}

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

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

BashOutput

工具名称: BashOutput

输入:

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

输出:

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

KillBash

工具名称: KillBash

输入:

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

输出:

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

ExitPlanMode

工具名称: ExitPlanMode

输入:

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

输出:

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

ListMcpResources

工具名称: ListMcpResources

输入:

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

输出:

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

ReadMcpResource

工具名称: ReadMcpResource

输入:

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

输出:

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

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

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

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

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

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)

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

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}")

使用客户端流式模式

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 使用自定义工具

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

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

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

使用示例

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

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

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

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

SandboxIgnoreViolations

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

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

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

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

excludedCommands vs allowUnsandboxedCommands

  • excludedCommands:始终自动绕过沙箱的静态命令列表(例如 ["docker"])。模型无法控制此事。
  • allowUnsandboxedCommands:允许模型在运行时决定是否请求非沙箱执行,通过在工具输入中设置 dangerouslyDisableSandbox: True
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() 集成。