Claude Code官方文档Agent SDKPythonAPI 参考

Claude Agent SDK Python 完整 API 参考

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

· 阅读约 39 分钟

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]

参数

参数类型描述
promptstr | AsyncIterable[dict]输入提示,可以是字符串或用于流式模式的异步可迭代对象
optionsClaudeAgentOptions | None可选配置对象(如果为 None,默认为 ClaudeAgentOptions()
transportTransport | 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]]

参数

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

输入模式选项

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

    {"text": str, "count": int, "enabled": bool}
    
  2. 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 导入)。所有字段都是可选的提示;客户端不应依赖它们做出安全决策。

字段类型默认值描述
titlestr | NoneNone工具的人类可读标题
readOnlyHintbool | NoneFalse如果为 True,工具不修改其环境
destructiveHintbool | NoneTrue如果为 True,工具可能执行破坏性更新(仅当 readOnlyHintFalse 时有意义)
idempotentHintbool | NoneFalse如果为 True,使用相同参数的重复调用没有额外效果(仅当 readOnlyHintFalse 时有意义)
openWorldHintbool | NoneTrue如果为 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

参数

参数类型默认值描述
namestr-服务器的唯一标识符
versionstr"1.0.0"服务器版本字符串
toolslist[SdkMcpTool[Any]] | NoneNone使用 @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]

参数

参数类型默认值描述
directorystr | NoneNone列出会话的目录。省略时,返回所有项目中的会话
limitint | NoneNone返回的最大会话数
include_worktreesboolTruedirectory 在 git 仓库内时,包括所有 worktrees 路径中的会话

返回类型:SDKSessionInfo

属性类型描述
session_idstr唯一会话标识符
summarystr显示标题:自定义标题、自动生成的摘要或第一个提示
last_modifiedint上次修改时间(自纪元以来的毫秒数)
file_sizeint | None会话文件大小(字节)(远程存储后端为 None
custom_titlestr | None用户设置的会话标题
first_promptstr | None会话中的第一个有意义的用户提示
git_branchstr | None会话结束时的 Git 分支
cwdstr | None会话的工作目录
tagstr | None用户设置的会话标签
created_atint | 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() 清空它们。

类型

ℹ️ @dataclass vs TypedDict 此 SDK 使用两种类型。用 @dataclass 装饰的类(如 ResultMessageAgentDefinitionTextBlock)在运行时是对象实例,支持属性访问:msg.result。用 TypedDict 定义的类(如 ThinkingConfigEnabledMcpStdioServerConfigSyncHookJSONOutput)在运行时是普通字典,需要键访问:config["budget_tokens"],而不是 config.budget_tokensClassName(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=1CLAUDE_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。