Claude Agent SDK Python 完整 API 参考
Python Agent SDK 的完整 API 参考,包括所有函数、类型和类。
Agent SDK 参考 - Python
Python Agent SDK 的完整 API 参考,包括所有函数、类型和类。
安装
pip install claude-agent-sdk
在 query() 和 ClaudeSDKClient 之间选择
Python SDK 提供了两种与 Claude Code 交互的方式:
快速比较
| 功能 | query() | ClaudeSDKClient |
|---|---|---|
| 会话 | 每次创建新会话 | 重用同一会话 |
| 对话 | 单次交换 | 同一上下文中的多次交换 |
| 连接 | 自动管理 | 手动控制 |
| 流式输入 | 支持 | 支持 |
| 中断 | 不支持 | 支持 |
| hooks | 支持 | 支持 |
| 自定义工具 | 支持 | 支持 |
| 继续聊天 | 每次新会话 | 保持对话 |
| 用例 | 一次性任务 | 持续对话 |
何时使用 query()(每次新会话)
最适合:
- 不需要对话历史的一次性问题
- 不需要来自之前交换的上下文的独立任务
- 简单的自动化脚本
- 当你想每次都重新开始时
何时使用 ClaudeSDKClient(持续对话)
最适合:
- 继续对话 - 当你需要 Claude 记住上下文时
- 后续问题 - 基于之前的响应进行构建
- 交互式应用程序 - 聊天界面、REPL
- 响应驱动的逻辑 - 当下一步操作取决于 Claude 的响应时
- 会话控制 - 显式管理对话生命周期
函数
query()
为每次与 Claude Code 的交互创建一个新会话。返回一个异步迭代器,当消息到达时产生消息。每次调用 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",
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]]
参数
| 参数 | 类型 | 描述 |
|---|---|---|
name | str | 工具的唯一标识符 |
description | str | 工具功能的人类可读描述 |
input_schema | type | dict[str, Any] | 定义工具输入参数的模式(见下文) |
annotations | ToolAnnotations | None | 可选的 MCP 工具注解,为客户端提供行为提示 |
输入模式选项
-
简单类型映射(推荐):
{"text": str, "count": int, "enabled": bool} -
JSON Schema 格式(用于复杂验证):
{ "type": "object", "properties": { "text": {"type": "string"}, "count": {"type": "integer", "minimum": 0}, }, "required": ["text"], }
示例
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 重新导出(也可以从 claude_agent_sdk 导入)。所有字段都是可选的提示;客户端不应依赖它们做出安全决策。
| 字段 | 类型 | 默认值 | 描述 |
|---|---|---|---|
title | str | None | None | 工具的人类可读标题 |
readOnlyHint | bool | None | False | 如果为 True,工具不修改其环境 |
destructiveHint | bool | None | True | 如果为 True,工具可能执行破坏性更新(仅当 readOnlyHint 为 False 时有意义) |
idempotentHint | bool | None | False | 如果为 True,使用相同参数的重复调用没有额外效果(仅当 readOnlyHint 为 False 时有意义) |
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
参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
name | str | - | 服务器的唯一标识符 |
version | str | "1.0.0" | 服务器版本字符串 |
tools | list[SdkMcpTool[Any]] | None | None | 使用 @tool 装饰器创建的工具函数列表 |
示例
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],
)
# Use with Claude
options = ClaudeAgentOptions(
mcp_servers={"calc": calculator},
allowed_tools=["mcp__calc__add", "mcp__calc__multiply"],
)
list_sessions()
列出带有元数据的过去会话。按项目目录过滤或列出所有项目中的会话。同步;立即返回。
def list_sessions(
directory: str | None = None,
limit: int | None = None,
include_worktrees: bool = True
) -> list[SDKSessionInfo]
参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
directory | str | None | None | 列出会话的目录。省略时,返回所有项目中的会话 |
limit | int | None | None | 返回的最大会话数 |
include_worktrees | bool | True | 当 directory 在 git 仓库内时,包括所有 worktrees 路径中的会话 |
返回类型:SDKSessionInfo
| 属性 | 类型 | 描述 |
|---|---|---|
session_id | str | 唯一会话标识符 |
summary | str | 显示标题:自定义标题、自动生成的摘要或第一个提示 |
last_modified | int | 上次修改时间(自纪元以来的毫秒数) |
file_size | int | None | 会话文件大小(字节)(远程存储后端为 None) |
custom_title | str | None | 用户设置的会话标题 |
first_prompt | str | None | 会话中的第一个有意义的用户提示 |
git_branch | str | None | 会话结束时的 Git 分支 |
cwd | str | None | 会话的工作目录 |
tag | str | None | 用户设置的会话标签 |
created_at | int | None | 会话创建时间(自纪元以来的毫秒数) |
示例
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]
示例
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
rename_session()
通过追加自定义标题条目来重命名会话。重复调用是安全的;最新的标题获胜。同步。
def rename_session(
session_id: str,
title: str,
directory: str | None = None,
) -> None
tag_session()
标记会话。传递 None 以清除标签。重复调用是安全的;最新的标签获胜。同步。
def tag_session(
session_id: str,
tag: str | None,
directory: str | None = None,
) -> None
示例
from claude_agent_sdk import list_sessions, tag_session
# Tag a session
tag_session("550e8400-e29b-41d4-a716-446655440000", "needs-review")
# Later: find all sessions with that tag
for session in list_sessions(directory="/path/to/project"):
if session.tag == "needs-review":
print(session.summary)
类
ClaudeSDKClient
在多次交换中维持对话会话。 这是 TypeScript SDK 的 query() 函数内部工作方式的 Python 等价物 - 它创建一个可以继续对话的客户端对象。
关键特性
- 会话连续性:在多个
query()调用中维持对话上下文 - 同一对话:会话保留之前的消息
- 中断支持:可以在任务中途停止执行
- 显式生命周期:你控制会话何时开始和结束
- 响应驱动的流程:可以对响应做出反应并发送后续消息
- 自定义工具和 hooks:支持自定义工具(使用
@tool装饰器创建)和 hooks
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
上下文管理器支持
客户端可以用作异步上下文管理器以自动管理连接:
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:
# First question
await client.query("What's the capital of France?")
# Process response
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
# Follow-up question - the session retains the previous context
await client.query("What's the population of that city?")
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
asyncio.run(main())
示例 - 使用中断
import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, ResultMessage
async def interruptible_task():
options = ClaudeAgentOptions(allowed_tools=["Bash"], permission_mode="acceptEdits")
async with ClaudeSDKClient(options=options) as client:
# Start a long-running task
await client.query("Count from 1 to 100 slowly, using the bash sleep command")
# Let it run for a bit
await asyncio.sleep(2)
# Interrupt the task
await client.interrupt()
print("Task interrupted!")
# Drain the interrupted task's messages (including its ResultMessage)
async for message in client.receive_response():
if isinstance(message, ResultMessage):
print(f"Interrupted task finished with subtype={message.subtype!r}")
# Send a new command
await client.query("Just say hello instead")
# Now receive the new response
async for message in client.receive_response():
if isinstance(message, ResultMessage) and message.subtype == "success":
print(f"New result: {message.result}")
asyncio.run(interruptible_task())
中断后的缓冲行为: interrupt() 发送停止信号但不清除消息缓冲区。被中断任务已产生的消息,包括其 ResultMessage(带 subtype="error_during_execution"),保留在流中。你必须在读取新查询的响应之前用 receive_response() 清空它们。
类型
ℹ️
@dataclassvsTypedDict: 此 SDK 使用两种类型。用@dataclass装饰的类(如ResultMessage、AgentDefinition、TextBlock)在运行时是对象实例,支持属性访问:msg.result。用TypedDict定义的类(如ThinkingConfigEnabled、McpStdioServerConfig、SyncHookJSONOutput)在运行时是普通字典,需要键访问: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
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) |
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 # 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
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 # Deprecated: use thinking instead
thinking: ThinkingConfig | None = None
effort: Literal["low", "medium", "high", "xhigh", "max"] | None = None
enable_file_checkpointing: bool = False
session_store: SessionStore | None = None
session_store_flush: SessionStoreFlushMode = "batched"
处理缓慢或停滞的 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窗口。CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS:使用run_in_background启动的子代理的停滞监视器。默认600000。CLAUDE_ENABLE_STREAM_WATCHDOG=1与CLAUDE_STREAM_IDLE_TIMEOUT_MS:当响应体停止流式传输时中止请求。
OutputFormat
结构化输出验证的配置。将其作为 dict 传递给 ClaudeAgentOptions 上的 output_format 字段:
{
"type": "json_schema",
"schema": {...}, # Your JSON Schema definition
}
SystemPromptPreset
使用 Claude Code 的预设系统提示和可选添加的配置。
class SystemPromptPreset(TypedDict):
type: Literal["preset"]
preset: Literal["claude_code"]
append: NotRequired[str]
exclude_dynamic_sections: NotRequired[bool]
SettingSource
控制 SDK 从哪些基于文件系统的配置源加载设置。
SettingSource = Literal["user", "project", "local"]
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: Literal["low", "medium", "high", "xhigh", "max"] | int | None = None
permissionMode: PermissionMode | None = None
PermissionMode
用于控制工具执行的权限模式。
PermissionMode = Literal[
"default", # Standard permission behavior
"acceptEdits", # Auto-accept file edits
"plan", # Planning mode - read-only tools only
"dontAsk", # Deny anything not pre-approved instead of prompting
"bypassPermissions", # Bypass all permission checks (use with caution)
]
CanUseTool
工具权限回调函数的类型别名。
CanUseTool = Callable[
[str, dict[str, Any], ToolPermissionContext], Awaitable[PermissionResult]
]
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
PermissionResult
权限回调结果的联合类型。
PermissionResult = PermissionResultAllow | PermissionResultDeny
PermissionResultAllow
指示应允许工具调用的结果。
@dataclass
class PermissionResultAllow:
behavior: Literal["allow"] = "allow"
updated_input: dict[str, Any] | None = None
updated_permissions: list[PermissionUpdate] | None = None
PermissionResultDeny
指示应拒绝工具调用的结果。
@dataclass
class PermissionResultDeny:
behavior: Literal["deny"] = "deny"
message: str = ""
interrupt: bool = False
ToolsPreset
使用 Claude Code 的默认工具集的预设工具配置。
class ToolsPreset(TypedDict):
type: Literal["preset"]
preset: Literal["claude_code"]
ThinkingConfig
控制扩展思考行为。三种配置的联合:
class ThinkingConfigAdaptive(TypedDict):
type: Literal["adaptive"]
class ThinkingConfigEnabled(TypedDict):
type: Literal["enabled"]
budget_tokens: int
class ThinkingConfigDisabled(TypedDict):
type: Literal["disabled"]
ThinkingConfig = ThinkingConfigAdaptive | ThinkingConfigEnabled | ThinkingConfigDisabled
SdkBeta
SDK 测试功能的字面类型。
SdkBeta = Literal["context-1m-2025-08-07"]
McpSdkServerConfig
使用 create_sdk_mcp_server() 创建的 SDK MCP 服务器的配置。
class McpSdkServerConfig(TypedDict):
type: Literal["sdk"]
name: str
instance: Any
McpServerConfig
MCP 服务器配置的联合类型。
McpServerConfig = (
McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig
)
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
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
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
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
错误类型
ClaudeSDKError
所有 SDK 错误的基础异常类。
class ClaudeSDKError(Exception):
"""Base error for Claude SDK."""
CLINotFoundError
当 Claude Code CLI 未安装或找不到时引发。
class CLINotFoundError(CLIConnectionError):
def __init__(
self, message: str = "Claude Code not found", cli_path: str | None = None
):
...
CLIConnectionError
当连接到 Claude Code 失败时引发。
class CLIConnectionError(ClaudeSDKError):
"""Failed to connect to Claude Code."""
ProcessError
当 Claude Code 进程失败时引发。
class ProcessError(ClaudeSDKError):
def __init__(
self, message: str, exit_code: int | None = None, stderr: str | None = None
):
self.exit_code = exit_code
self.stderr = stderr
CLIJSONDecodeError
当 JSON 解析失败时引发。
class CLIJSONDecodeError(ClaudeSDKError):
def __init__(self, line: str, original_error: Exception):
self.line = line
self.original_error = original_error
Hook 类型
有关 Hook 类型的详细文档,包括所有 Hook 输入/输出类型,请参阅完整的 Python SDK 源代码。
工具输入/输出类型
所有内置 Claude Code 工具的完整输入/输出模式文档已包含在 SDK 源代码中。
本文翻译自 Anthropic Claude Code 官方文档,最近一次同步:2025-05-01。