本页目录36
- 一次性任务用 query(),需要多轮对话上下文、中断、权限回调等能力时用 ClaudeSDKClient
- 自定义工具通过 @tool 装饰器 + create_sdk_mcp_server() 注册为 SDK 内置 MCP 服务器,再通过 allowed_tools 授权
- ClaudeAgentOptions 是最核心的配置对象,涵盖工具、权限模式、系统提示词、MCP 服务器、模型、思考预算、子代理定义等全部行为开关
- can_use_tool 权限回调只在权限判定“落到需要询问”时才会被调用,已被 allowed_tools/规则/权限模式自动放行的调用不会触发它
- TypedDict 类型(如 ThinkingConfigEnabled、McpStdioServerConfig)在运行时是普通字典,需用 config["key"] 取值,不能用属性访问
- 会话相关函数(list_sessions/get_session_messages/rename_session/tag_session 等)用于脱离对话流程单独查询、重命名、打标签历史会话
本文是对 Claude Agent SDK 官方文档「Python SDK reference」页面的中文整理,完整与最新内容以原文为准:https://code.claude.com/docs/en/agent-sdk/python 说明:受抓取工具长度限制,原文页面末尾关于 Hook 类型完整字段、消息内容块(TextBlock/ThinkingBlock/ToolUseBlock/ToolResultBlock 等)完整定义、错误类(ClaudeSDKError 及子类)、SandboxSettings、SdkPluginConfig、SessionStore 协议的详细字段表未能完整抓取,本文对这些部分仅做提及、不列具体字段表,避免编造,请以原文链接为准。
安装
python3 -m venv .venv
source .venv/bin/activate
pip install claude-agent-sdk
使用 uv、Windows PowerShell 或配置 API key,参见 Agent SDK 快速上手文档的 Setup 部分。
query() 与 ClaudeSDKClient 对比
| 特性 | query() | ClaudeSDKClient |
|---|---|---|
| 会话 | 默认创建新会话 | 复用同一会话 |
| 对话 | 单次往返 | 多次往返 |
| 连接 | 自动管理 | 手动控制 |
| 流式输入 | ✅ | ✅ |
| 中断 | ❌ | ✅ |
| Hooks | ✅ | ✅ |
| 自定义工具 | ✅ | ✅ |
| 继续对话 | 需手动 | ✅ 自动 |
| 适用场景 | 一次性任务 | 持续对话 |
交互式应用、或下一步动作依赖 Claude 上一步响应时,使用 ClaudeSDKClient。
函数
query()
async def query(
*,
prompt: str | AsyncIterable[dict[str, Any]],
options: ClaudeAgentOptions | None = None,
transport: Transport | None = None
) -> AsyncIterator[Message]
参数:
| 参数 | 类型 | 说明 |
|---|---|---|
prompt | str | AsyncIterable[dict] | 字符串或用于流式输入的异步可迭代对象 |
options | ClaudeAgentOptions | None | 可选配置(None 时默认为 ClaudeAgentOptions()) |
transport | Transport | None | 可选的自定义 CLI 通信传输层 |
返回值: AsyncIterator[Message],产出对话中的消息。
示例:
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
options = ClaudeAgentOptions(
system_prompt="You are an expert Python developer",
permission_mode="acceptEdits",
)
async for message in query(prompt="Create a Python web server", options=options):
print(message)
asyncio.run(main())
tool()
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]]
参数:
| 参数 | 类型 | 说明 |
|---|---|---|
name | str | 工具的唯一标识符 |
description | str | 人类可读的描述 |
input_schema | type | dict[str, Any] | 定义输入参数的 schema |
annotations | ToolAnnotations | None | 可选的 MCP 工具注解 |
input_schema 的两种写法:
- 简单类型映射(推荐):
{"text": str, "count": int, "enabled": bool}
- JSON Schema 格式:
{
"type": "object",
"properties": {
"text": {"type": "string"},
"count": {"type": "integer", "minimum": 0},
},
"required": ["text"],
}
返回值: 装饰器函数,返回 SdkMcpTool 实例。
示例:
from claude_agent_sdk import tool
from typing import Any
@tool("greet", "Greet a user", {"name": str})
async def greet(args: dict[str, Any]) -> dict[str, Any]:
return {"content": [{"type": "text", "text": f"Hello, {args['name']}!"}]}
ToolAnnotations
从 mcp.types 重新导出,所有字段均可选。
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
title | str | None | None | 人类可读标题 |
readOnlyHint | bool | None | False | 该工具不修改环境 |
destructiveHint | bool | None | True | 该工具可能执行破坏性更新 |
idempotentHint | bool | None | False | 重复调用无额外效果 |
openWorldHint | bool | None | True | 该工具与外部实体交互 |
带注解的示例:
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()
def create_sdk_mcp_server(
name: str,
version: str = "1.0.0",
tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig
参数:
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
name | str | - | 服务器唯一标识符 |
version | str | "1.0.0" | 服务器版本号字符串 |
tools | list[SdkMcpTool[Any]] | None | None | 工具函数列表 |
返回值: McpSdkServerConfig,用于 ClaudeAgentOptions.mcp_servers。
示例:
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions
@tool("add", "Add two numbers", {"a": float, "b": float})
async def add(args):
return {"content": [{"type": "text", "text": f"Sum: {args['a'] + args['b']}"}]}
@tool("multiply", "Multiply two numbers", {"a": float, "b": float})
async def multiply(args):
return {"content": [{"type": "text", "text": f"Product: {args['a'] * args['b']}"}]}
calculator = create_sdk_mcp_server(
name="calculator",
version="2.0.0",
tools=[add, multiply],
)
options = ClaudeAgentOptions(
mcp_servers={"calc": calculator},
allowed_tools=["mcp__calc__add", "mcp__calc__multiply"],
)
会话管理函数
list_sessions()
def list_sessions(
directory: str | None = None,
limit: int | None = None,
offset: int = 0,
include_worktrees: bool = True
) -> list[SDKSessionInfo]
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
directory | str | None | None | 要列出会话的目录,省略则列出所有项目 |
limit | int | None | None | 返回的最大会话数 |
offset | int | 0 | 从起始位置跳过的会话数 |
include_worktrees | bool | True | 若处于 git 仓库中,是否包含 worktree 路径 |
返回类型 SDKSessionInfo:
| 属性 | 类型 | 说明 |
|---|---|---|
session_id | str | 会话唯一标识符 |
summary | str | 展示标题 |
last_modified | int | 最后修改时间(自 epoch 起的毫秒数) |
file_size | int | None | 会话文件大小(字节) |
custom_title | str | None | 用户设置的会话标题 |
first_prompt | str | None | 首条有意义的用户提示词 |
git_branch | str | None | 会话结束时所在的 git 分支 |
cwd | str | None | 工作目录 |
tag | str | None | 用户设置的会话标签 |
created_at | int | None | 创建时间(自 epoch 起的毫秒数) |
示例:
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]
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
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 | str | None | 触发该消息的 Agent 工具调用块 ID |
parent_agent_id | str | None | 父子代理 ID |
示例:
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()
def get_session_info(
session_id: str,
directory: str | None = None,
) -> SDKSessionInfo | None
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
session_id | str | 必填 | 要查询的会话 UUID |
directory | str | None | None | 项目目录,省略则搜索所有项目 |
返回值: SDKSessionInfo,未找到时为 None。
from claude_agent_sdk import get_session_info
info = get_session_info("550e8400-e29b-41d4-a716-446655440000")
if info:
print(f"{info.summary} (branch: {info.git_branch}, tag: {info.tag})")
rename_session()
def rename_session(
session_id: str,
title: str,
directory: str | None = None,
) -> None
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
session_id | str | 必填 | 要重命名的会话 UUID |
title | str | 必填 | 新标题(去除首尾空白后不能为空) |
directory | str | None | None | 项目目录,省略则搜索所有项目 |
异常: session_id 无效或 title 为空时抛出 ValueError;未找到会话时抛出 FileNotFoundError。
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()
def tag_session(
session_id: str,
tag: str | None,
directory: str | None = None,
) -> None
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
session_id | str | 必填 | 要打标签的会话 UUID |
tag | str | None | 必填 | 标签字符串,传 None 清除标签 |
directory | str | None | None | 项目目录,省略则搜索所有项目 |
异常: session_id 无效,或标签清洗后为空时抛出 ValueError;未找到会话时抛出 FileNotFoundError。
from claude_agent_sdk import list_sessions, tag_session
sessions = list_sessions(directory="/path/to/project", limit=1)
if sessions:
tag_session(sessions[0].session_id, "needs-review")
for session in list_sessions(directory="/path/to/project"):
if session.tag == "needs-review":
print(session.summary)
类:ClaudeSDKClient
跨多次往返维护同一会话状态。
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
方法:
| 方法 | 说明 |
|---|---|
__init__(options) | 用可选配置初始化 |
connect(prompt) | 连接,可带初始提示词/消息流 |
query(prompt, session_id) | 在流式模式下发送新请求 |
receive_messages() | 以异步迭代器形式接收所有消息 |
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) | 重试连接失败/已断开的服务器 |
toggle_mcp_server(server_name, enabled) | 在会话中途启用/禁用服务器 |
stop_task(task_id) | 停止正在运行的后台任务,随后会收到状态为 stopped 的 TaskNotificationMessage |
get_server_info() | 获取服务器信息,包括会话 ID 与能力 |
disconnect() | 断开与 Claude 的连接 |
上下文管理器用法:
import asyncio
from claude_agent_sdk import ClaudeSDKClient
async def main():
async with ClaudeSDKClient() as client:
await client.query("Hello Claude")
async for message in client.receive_response():
print(message)
asyncio.run(main())
⚠️ 注意: 迭代消息时避免用 break 提前退出,否则可能导致 asyncio 清理问题。让迭代自然结束,或使用标志变量控制。
示例——延续对话:
import asyncio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage
async def main():
async with ClaudeSDKClient() as client:
# First question
await client.query("What's the capital of France?")
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
# Follow-up - session retains context
await client.query("What's the population of that city?")
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
# Another follow-up - still in same conversation
await client.query("What are some famous landmarks there?")
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
asyncio.run(main())
示例——流式输入:
import asyncio
from claude_agent_sdk import ClaudeSDKClient
async def message_stream():
"""Generate messages dynamically."""
yield {
"type": "user",
"message": {"role": "user", "content": "Analyze the following data:"},
}
await asyncio.sleep(0.5)
yield {
"type": "user",
"message": {"role": "user", "content": "Temperature: 25°C, Humidity: 60%"},
}
await asyncio.sleep(0.5)
yield {
"type": "user",
"message": {"role": "user", "content": "What patterns do you see?"},
}
async def main():
async with ClaudeSDKClient() as client:
# Stream input to Claude
await client.query(message_stream())
async for message in client.receive_response():
print(message)
# Follow-up in same session
await client.query("Should we be concerned about these readings?")
async for message in client.receive_response():
print(message)
asyncio.run(main())
示例——使用中断:
import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, ResultMessage
async def interruptible_task():
options = ClaudeAgentOptions(allowed_tools=["Bash"], permission_mode="acceptEdits")
async with ClaudeSDKClient(options=options) as client:
# Start long-running task
await client.query("Count from 1 to 100 slowly, using the bash sleep command")
# Let it run
await asyncio.sleep(2)
# Interrupt
await client.interrupt()
print("Task interrupted!")
# Drain interrupted task's messages
async for message in client.receive_response():
if isinstance(message, ResultMessage):
print(f"Interrupted task: terminal_reason={message.terminal_reason!r}")
# Send new command
await client.query("Just say hello instead")
# Receive new response
async for message in client.receive_response():
if isinstance(message, ResultMessage) and message.subtype == "success":
print(f"New result: {message.result}")
asyncio.run(interruptible_task())
⚠️ 中断后的缓冲区行为: interrupt() 发送停止信号,但不会清空消息缓冲区。被中断任务已产生的消息仍留在消息流里。读取新查询的响应前,必须先用 receive_response() 排空这些残留消息。
示例——高级权限控制:
import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
from claude_agent_sdk.types import (
PermissionResultAllow,
PermissionResultDeny,
ToolPermissionContext,
)
async def custom_permission_handler(
tool_name: str, input_data: dict, context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
"""Custom logic for tool permissions."""
# Block writes to system directories
if tool_name == "Write" and input_data.get("file_path", "").startswith("/system/"):
return PermissionResultDeny(
message="System directory write not allowed", interrupt=True
)
# Redirect sensitive file operations
if tool_name in ["Write", "Edit"] and "config" in input_data.get("file_path", ""):
safe_path = f"./sandbox/{input_data['file_path']}"
return PermissionResultAllow(
updated_input={**input_data, "file_path": safe_path}
)
# Allow everything else
return PermissionResultAllow(updated_input=input_data)
async def main():
options = ClaudeAgentOptions(can_use_tool=custom_permission_handler)
async with ClaudeSDKClient(options=options) as client:
await client.query("Update the system config file")
async for message in client.receive_response():
print(message)
asyncio.run(main())
类型(Types)
提示:@dataclass 与 TypedDict 的区别
- dataclass(如
ResultMessage、TextBlock):运行时是对象实例,用属性访问,如msg.result - TypedDict(如
ThinkingConfigEnabled、McpStdioServerConfig):运行时是普通字典,需用键访问,如config["budget_tokens"],不能写config.budget_tokens
SdkMcpTool
@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 注解 |
Transport
自定义传输层实现的抽象基类。
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() | 关闭输入流 |
导入方式: from claude_agent_sdk import Transport
ClaudeAgentOptions
Claude Code 查询的配置 dataclass。
@dataclass
class ClaudeAgentOptions:
tools: list[str] | ToolsPreset | None = None
allowed_tools: list[str] = field(default_factory=list)
system_prompt: str | SystemPromptPreset | SystemPromptFile | None = None
mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
strict_mcp_config: bool = False
permission_mode: PermissionMode | None = None
continue_conversation: bool = False
resume: str | None = None
session_id: str | None = None
max_turns: int | None = None
max_budget_usd: float | None = None
disallowed_tools: list[str] = field(default_factory=list)
model: str | None = None
fallback_model: str | None = None
betas: list[SdkBeta] = field(default_factory=list)
output_format: dict[str, Any] | None = None
permission_prompt_tool_name: str | None = None
cwd: str | Path | None = None
cli_path: str | Path | None = None
settings: str | None = None
add_dirs: list[str | Path] = field(default_factory=list)
env: dict[str, str] = field(default_factory=dict)
extra_args: dict[str, str | None] = field(default_factory=dict)
max_buffer_size: int | None = None
debug_stderr: Any = sys.stderr # Deprecated
stderr: Callable[[str], None] | None = None
can_use_tool: CanUseTool | None = None
hooks: dict[HookEvent, list[HookMatcher]] | None = None
user: str | None = None
include_partial_messages: bool = False
include_hook_events: bool = False
forward_subagent_text: bool = False
fork_session: bool = False
resume_session_at: str | None = None
resume_drops_turn: str | None = None
agents: dict[str, AgentDefinition] | None = None
setting_sources: list[SettingSource] | None = None
skills: list[str] | Literal["all"] | None = None
sandbox: SandboxSettings | None = None
plugins: list[SdkPluginConfig] = field(default_factory=list)
max_thinking_tokens: int | None = None # Deprecated: use thinking
thinking: ThinkingConfig | None = None
effort: EffortLevel | None = None
enable_file_checkpointing: bool = False
session_store: SessionStore | None = None
session_store_flush: SessionStoreFlushMode = "batched"
load_timeout_ms: int = 60_000
task_budget: TaskBudget | None = None
全部字段:
| 属性 | 类型 | 默认值 | 说明 |
|---|---|---|---|
tools | list[str] | ToolsPreset | None | None | 工具配置。用 {"type": "preset", "preset": "claude_code"} 使用 Claude Code 默认工具集 |
allowed_tools | list[str] | [] | 自动批准的工具列表。不代表限制 Claude 只能用这些工具。列出任务跟踪类工具会使会话启用该功能;未列出的工具会走 permission_mode 与 can_use_tool 的判定流程。要屏蔽工具用 disallowed_tools |
system_prompt | str | SystemPromptPreset | SystemPromptFile | None | None | 系统提示词配置。传字符串为自定义提示词;传 {"type": "preset", "preset": "claude_code"} 使用 Claude Code 预设(可带 "append");传 {"type": "file", "path": "..."} 从文件加载 |
mcp_servers | dict[str, McpServerConfig] | str | Path | {} | MCP 服务器配置,或指向配置文件的路径 |
strict_mcp_config | bool | False | 为 True 时仅使用传入的服务器,忽略项目 .mcp.json、用户设置、插件与 claude.ai 连接器 |
permission_mode | PermissionMode | None | None | 工具使用的权限模式 |
continue_conversation | bool | False | 继续最近一次对话 |
resume | str | None | None | 要恢复的会话 ID |
session_id | str | None | None | 使用指定会话 ID 而非自动生成,必须是 UUID。不能与 continue_conversation 或 resume 同时使用,除非同时设置了 fork_session |
max_turns | int | None | None | 最大代理轮次(工具调用往返次数) |
max_budget_usd | float | None | None | 当客户端估算的费用达到该美元值时停止 |
disallowed_tools | list[str] | [] | 要拒绝的工具。裸名如 "Bash" 会将其从上下文中移除;带作用域如 "Bash(rm *)" 会保留该工具可见,但拒绝匹配的调用,在所有权限模式下均生效 |
model | str | None | None | Claude 模型别名或完整名称 |
fallback_model | str | None | None | 主模型失败时的备用模型 |
betas | list[SdkBeta] | [] | 要启用的 beta 特性 |
output_format | dict[str, Any] | None | None | 结构化响应的输出格式 |
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] | [] | 额外可访问的目录 |
env | dict[str, str] | {} | 合并到进程环境变量之上的环境变量 |
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 | 用于拦截事件的 hook 配置 |
user | str | None | None | 用户标识符 |
include_partial_messages | bool | False | 是否包含部分流式事件。开启后会产出 StreamEvent 消息 |
include_hook_events | bool | False | 是否将 hook 生命周期事件作为 HookEventMessage 包含在内 |
forward_subagent_text | bool | False | 转发子代理的文本/思考块。默认只发出 tool_use 与 tool_result |
fork_session | bool | False | 恢复会话时分叉到新会话 ID,而非延续原会话 |
resume_session_at | str | None | None | 仅加载到指定 UUID 消息为止的对话内容。通常与 resume、fork_session 搭配使用 |
resume_drops_turn | str | None | None | 截断轮次时要丢弃的用户提示词 UUID。若被丢弃的范围内存在未归因条目,CLI 会拒绝执行。需要 v2.1.223+ |
agents | dict[str, AgentDefinition] | None | None | 以编程方式定义的子代理 |
plugins | list[SdkPluginConfig] | [] | 从指定路径加载自定义插件 |
sandbox | SandboxSettings | None | None | 配置沙箱行为 |
setting_sources | list[SettingSource] | None | None(默认加载全部) | 控制从文件系统加载哪些设置。传 [] 可禁用 user/project/local;策略(policy)设置始终会加载 |
skills | list[str] | Literal["all"] | None | None | 可用技能。传 "all" 或名称列表。SDK 会自动把 Skill 加入 allowed_tools。名称格式非法或使用通配符会抛出 ValueError |
max_thinking_tokens | int | None | None | 已弃用 —— 最大思考 token 数,改用 thinking |
thinking | ThinkingConfig | None | None | 扩展思考行为配置,优先级高于 max_thinking_tokens |
effort | EffortLevel | None | None | 思考深度的 effort 等级 |
enable_file_checkpointing | bool | False | 启用文件变更跟踪,以支持回退(rewind) |
session_store | SessionStore | None | None | 镜像写入外部后端,便于从另一台主机恢复会话 |
session_store_flush | Literal["batched", "eager"] | "batched" | 何时刷新条目:"batched" 按轮次或缓冲区满时刷新;"eager" 每帧刷新 |
load_timeout_ms | int | 60000 | session_store.load() 与 list_subkeys() 单次调用的超时(毫秒) |
task_budget | TaskBudget | None | None | 配合 task-budgets-2026-03-13 beta 使用的 API 侧 token 预算,传 {"total": <int>} |
用于 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:单次请求超时(毫秒),默认600000CLAUDE_CODE_MAX_RETRIES:最大重试次数,默认10,上限15;每次重试拥有各自独立的超时窗口CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS:run_in_background子代理的停滞看门狗超时(毫秒),默认600000;每次流事件会重置该计时CLAUDE_ENABLE_STREAM_WATCHDOG(配合CLAUDE_STREAM_IDLE_TIMEOUT_MS):在响应头已到达但响应体停滞时中止请求。默认开启,设为0可关闭;超时默认值/最小值均为300000毫秒
OutputFormat
结构化输出校验配置,作为字典传给 output_format 字段:
{
"type": "json_schema",
"schema": {...}, # JSON Schema definition
}
| 字段 | 是否必填 | 说明 |
|---|---|---|
type | 是 | 必须为 "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" |
append | 否 | 追加的额外指令 |
exclude_dynamic_sections | 否 | 将随会话变化的上下文(cwd、git 标记、memory 路径)移到首条消息中,以提升 prompt cache 命中率 |
SystemPromptFile
从文件加载自定义系统提示词。
class SystemPromptFile(TypedDict):
type: Literal["file"]
path: str
| 字段 | 是否必填 | 说明 |
|---|---|---|
type | 是 | 必须为 "file" |
path | 是 | 提示词文件路径 |
提示词较大时建议用文件形式:SDK 会通过 argv 传字符串,受操作系统限制(Linux 约 128 KB,Windows 整条命令约 32 KB)。
SettingSource
控制加载哪些文件系统配置来源。
SettingSource = Literal["user", "project", "local"]
| 取值 | 说明 | 位置 |
|---|---|---|
"user" | 全局用户设置 | ~/.claude/settings.json |
"project" | 团队共享的项目设置 | .claude/settings.json |
"local" | 本地项目设置(已 gitignore) | .claude/settings.local.json |
默认值: 省略或为 None 时,会像 CLI 一样加载 user、project、local 三者;端点策略(endpoint policy)始终会加载;服务端管理的配置会在符合条件时获取。
禁用文件系统设置:
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
async for message in query(
prompt="Analyze this code",
options=ClaudeAgentOptions(setting_sources=[]),
):
print(message)
asyncio.run(main())
仅加载项目设置:
async for message in query(
prompt="Run CI checks",
options=ClaudeAgentOptions(setting_sources=["project"]),
):
print(message)
仅用 SDK 配置、不读文件系统:
import asyncio
from claude_agent_sdk import AgentDefinition, ClaudeAgentOptions, query
async def main():
async for message in query(
prompt="Review this PR",
options=ClaudeAgentOptions(
setting_sources=[],
agents={
"code-reviewer": AgentDefinition(
description="Reviews code changes",
prompt="You are a code reviewer. Report issues in the diff.",
),
},
allowed_tools=["Read", "Grep", "Glob"],
),
):
print(message)
asyncio.run(main())
要加载 CLAUDE.md 指令,需在 setting_sources 中包含 "project"。
设置优先级(从高到低):
- Local(
.claude/settings.local.json) - Project(
.claude/settings.json) - User(
~/.claude/settings.json)
程序化传入的选项会覆盖文件系统设置;受管理的策略(managed policy)会覆盖程序化选项。
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
字段命名: 使用 camelCase(如 disallowedTools、permissionMode、maxTurns),而非 snake_case。
| 字段 | 是否必填 | 说明 |
|---|---|---|
description | 是 | 何时应使用该代理 |
prompt | 是 | 系统提示词 |
tools | 否 | 允许使用的工具名列表,省略则可用全部子代理工具 |
disallowedTools | 否 | 要移除的工具。支持 MCP 模式:mcp__server、mcp__server__*、mcp__* |
model | 否 | 模型覆盖:别名、完整 ID,或 "inherit" |
skills | 否 | 预加载的技能名称,未列出的技能仍可通过 Skill 工具调用 |
memory | 否 | 来源:"user"、"project" 或 "local" |
mcpServers | 否 | 服务器名称,或内联的 {name: config} 字典 |
initialPrompt | 否 | 作为首轮自动提交的内容 |
maxTurns | 否 | 停止前的最大轮次 |
background | 否 | 是否为非阻塞的后台任务 |
effort | 否 | 推理 effort 等级(名称或整数) |
permissionMode | 否 | 工具执行的权限模式 |
PermissionMode
PermissionMode = Literal[
"default", # Standard permission behavior
"acceptEdits", # Auto-accept file edits
"plan", # Explore without editing
"dontAsk", # Deny unlisted instead of prompting
"bypassPermissions", # Bypass checks; ask rules still prompt
"auto", # Model classifier approves/denies
]
译注:default(标准权限行为)、acceptEdits(自动接受文件编辑)、plan(只探索不编辑)、dontAsk(对未列出项直接拒绝而非询问)、bypassPermissions(绕过检查,但 ask 类规则仍会询问)、auto(由模型分类器自动批准/拒绝)。
EffortLevel
EffortLevel = Literal[
"low", # Minimal thinking, fastest
"medium", # Moderate thinking
"high", # Deep reasoning
"xhigh", # Extended; falls back to "high" if unsupported
"max", # Maximum effort
]
译注:low(最少思考,最快)、medium(中等思考)、high(深度推理)、xhigh(扩展模式,不支持时降级为 high)、max(最大 effort)。
CanUseTool
工具权限回调的类型别名。
CanUseTool = Callable[
[str, dict[str, Any], ToolPermissionContext], Awaitable[PermissionResult]
]
回调接收:
tool_name:被调用的工具input_data:工具输入参数context:ToolPermissionContext,包含附加信息
返回值: PermissionResult(PermissionResultAllow 或 PermissionResultDeny)
说明: 仅在权限判定流程落到需要提示时才会被调用。通过 allowed_tools、设置中的 allow 规则、权限模式已预先批准的调用不会触发它。若需对每次调用都进行拦截判断,应使用 PreToolUse hook。
ToolPermissionContext
传给权限回调的上下文信息。
@dataclass
class ToolPermissionContext:
signal: Any | None = None
suggestions: list[PermissionUpdate] = field(default_factory=list)
tool_use_id: str | None = None
agent_id: str | None = None
blocked_path: str | None = None
decision_reason: str | None = None
title: str | None = None
display_name: str | None = None
description: str | None = None
| 字段 | 类型 | 说明 |
|---|---|---|
signal | Any | None | 预留给未来的中止信号支持 |
suggestions | list[PermissionUpdate] | CLI 给出的建议。Bash 类调用会附带目标为 localSettings 的建议 |
tool_use_id | str | None | 工具调用 ID,对 can_use_tool 始终有值 |
agent_id | str | None | 子代理 ID,主代理时为 None |
blocked_path | str | None | 触发该请求的文件路径 |
decision_reason | str | None | 触发请求的原因,来自 PreToolUse hook 的 permissionDecisionReason |
title | str | None | 完整的提示句子,应作为主要展示文本 |
display_name | str | None | 用于按钮的简短动作短语 |
description | str | None | 副标题 |
PermissionResult
权限回调结果的联合类型。
PermissionResult = PermissionResultAllow | PermissionResultDeny
PermissionResultAllow
@dataclass
class PermissionResultAllow:
behavior: Literal["allow"] = "allow"
updated_input: dict[str, Any] | None = None
updated_permissions: list[PermissionUpdate] | None = None
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
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 | 用于 add/replace/remove 的规则 |
behavior | Literal["allow", "deny", "ask"] | None | 行为 |
mode | PermissionMode | None | setMode 时的模式 |
directories | list[str] | None | 用于 add/remove 的目录 |
destination | Literal[...] | None | 应用的目标位置 |
PermissionRuleValue
权限更新用的规则。
@dataclass
class PermissionRuleValue:
tool_name: str
rule_content: str | None = None
ToolsPreset
预设工具配置。
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 | 启用思考并指定 token 预算 |
disabled | type | 关闭思考 |
可选字段 display 控制是否返回思考文本:"summarized" 或 "omitted"。在 Opus 4.7+ 上,API 默认值为 "omitted";设为 "summarized" 才会收到 ThinkingBlock 输出。Claude Code 不会向 Bedrock/GCP Agent Platform 发送该参数,因此即使设为 "summarized",这些平台上仍返回空内容。
TypedDict —— 运行时是普通字典:
from claude_agent_sdk import ClaudeAgentOptions, ThinkingConfigEnabled
# Option 1: dict literal (recommended)
options = ClaudeAgentOptions(thinking={"type": "enabled", "budget_tokens": 20000})
# Option 2: constructor-style (returns dict)
config = ThinkingConfigEnabled(type="enabled", budget_tokens=20000)
print(config["budget_tokens"]) # 20000
# config.budget_tokens raises AttributeError
TaskBudget
API 侧的 token 任务预算。
class TaskBudget(TypedDict):
total: int
| 字段 | 类型 | 说明 |
|---|---|---|
total | int | 总 token 预算 |
以普通字典形式传入:ClaudeAgentOptions(task_budget={"total": 50000})
SdkBeta
SDK 的 beta 特性。
SdkBeta = Literal["context-1m-2025-08-07"]
配合 ClaudeAgentOptions 的 betas 字段使用。
⚠️ 警告: context-1m-2025-08-07 将于 2026 年 4 月 30 日退役。该 beta 对 Sonnet 4.5/4 无效,超出 200k 上下文窗口会返回错误。请迁移至 Claude Opus 5、Sonnet 5、Sonnet 4.6、Opus 4.6/4.7/4.8(这些模型以标准价格提供 1M 上下文,无需该 header)。
McpSdkServerConfig
通过 create_sdk_mcp_server() 创建的 SDK MCP 服务器。
class McpSdkServerConfig(TypedDict):
type: Literal["sdk"]
name: str
instance: Any # MCP Server instance
McpServerConfig
MCP 服务器配置的联合类型:
McpServerConfig = (
McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig
)
McpStdioServerConfig
class McpStdioServerConfig(TypedDict):
type: NotRequired[Literal["stdio"]] # Optional for backwards compatibility
command: str
args: NotRequired[list[str]]
env: NotRequired[dict[str, str]]
McpSSEServerConfig
class McpSSEServerConfig(TypedDict):
type: Literal["sse"]
url: str
headers: NotRequired[dict[str, str]]
McpHttpServerConfig
class McpHttpServerConfig(TypedDict):
type: Literal["http"]
url: str
headers: NotRequired[dict[str, str]]
McpServerStatusConfig
get_mcp_status() 返回的 MCP 服务器配置形式,是所有配置变体加上仅输出用的 claudeai-proxy 的联合类型:
McpServerStatusConfig = (
McpStdioServerConfig
| McpSSEServerConfig
| McpHttpServerConfig
| McpSdkServerConfigStatus
| McpClaudeAIProxyServerConfig
)
McpSdkServerConfigStatus:可序列化形式,含 type("sdk")与 name(str),不含 instance。
McpClaudeAIProxyServerConfig:字段为 type("claudeai-proxy")、url(str)、id(str)。
McpStatusResponse
ClaudeSDKClient.get_mcp_status() 的返回值。
class McpStatusResponse(TypedDict):
mcpServers: list[McpServerStatus]
McpServerStatus
已连接 MCP 服务器的状态。
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 | 服务器配置 |
scope(可选) | str | 作用域信息 |
tools(可选) | list[McpToolInfo] | 可用工具列表 |
消息类型(简述)
原文页面列出了异步迭代器中常见返回的消息类型名称,但本次抓取未能取得其完整字段定义(详见文首说明)。已知的类型名称包括:
TextBlock:响应中的文本内容ThinkingBlock:扩展思考输出(display: "summarized"时出现)ToolUseBlock:工具调用ToolResultBlock:工具执行结果AssistantMessage:包含若干内容块ResultMessage:最终响应,含结果/错误信息(示例中出现过.result、.subtype、.terminal_reason等属性)StreamEvent:部分流式更新(include_partial_messages=True时出现)HookEventMessage:hook 生命周期事件(include_hook_events=True时出现)TaskNotificationMessage:后台任务状态更新
此外,原文还包含 Hook 相关类型(HookEvent、HookMatcher 等,见 ClaudeAgentOptions.hooks 字段)、SandboxSettings、SdkPluginConfig、SessionStore 接口,以及 ClaudeSDKError 及其子类(CLIConnectionError、CLINotFoundError、ProcessError、CLIJSONDecodeError)等错误类型的完整定义。这些内容未能在本次整理中完整抓取,如需精确字段列表,请查阅原文:https://code.claude.com/docs/en/agent-sdk/python