Claude Agent SDK for Python 架构设计

📄️Claude Agent SDK for Python 架构设计

Public API、Internal 层、Transport 层、CLI 子进程协作全解

目录

1. 执行摘要
2. 系统概述
3. 系统架构
4. 核心组件详解
5. 数据流分析
6. 类型系统
7. 使用示例

---

执行摘要

Claude Agent SDK for Python 是一个用于与 Claude Code 交互的高级 Python 软件开发工具包。该 SDK 提供了两种主要交互模式：

• 一次性查询模式 (query() 函数)：适用于简单的、无状态的问答场景
• 交互式会话模式 (ClaudeSDKClient 类)：适用于需要双向通信、状态管理和实时交互的场景

关键特性

• MCP (Model Context Protocol) 服务器支持：支持进程内和外部 MCP 服务器
• 钩子 (Hooks) 系统：提供 11 种不同的钩子事件用于生命周期管理
• 子代理 (Subagents) 支持：可定义和使用具有特定工具集的自定义代理
• 权限控制：灵活的工具权限管理机制
• 错误恢复：完善的错误处理和重试机制

技术栈

• 异步框架：anyio (支持 asyncio 和 trio)
• 子进程管理：通过 Claude Code CLI 进行通信
• 类型安全：完整的类型注解和数据类定义
• MCP 集成：兼容 MCP (Model Context Protocol) 规范

---

系统概述

系统定位

Claude Agent SDK 作为 Python 应用与 Claude Code CLI 之间的桥梁，提供了高层抽象接口，使开发者能够轻松集成 Claude 的 AI 能力到自己的应用中。

核心设计理念

1. 分层抽象：提供从简单到复杂的多层次 API
2. 类型安全：完整的类型注解支持
3. 异步优先：所有 I/O 操作都是异步的
4. 可扩展性：通过钩子和 MCP 服务器支持自定义扩展

---

系统架构

整体架构图

包结构

src/claude_agent_sdk/
├── __init__.py              # 公共 API 导出
├── _cli_version.py          # CLI 版本信息
├── _errors.py               # 错误类定义
├── _version.py              # SDK 版本信息
├── client.py                # ClaudeSDKClient 类
├── query.py                 # query() 函数
├── types.py                 # 类型定义
└── _internal/
    ├── __init__.py
    ├── client.py            # InternalClient 实现
    ├── message_parser.py    # 消息解析器
    ├── query.py             # Query 控制协议处理
    ├── session_mutations.py # 会话变更操作
    ├── sessions.py          # 会话列表和查询
    └── transport/
        ├── __init__.py
        └── subprocess_cli.py # 子进程 CLI 传输实现

组件关系图

---

核心组件详解

1. 公共 API 层

1.1 query() 函数 - 一次性查询

位置: src/claude_agent_sdk/query.py

用途: 用于简单的、无状态的一次性查询场景。

特点:

• 单向通信：发送所有输入，接收所有输出
• 自动管理连接生命周期
• 适合批量处理和自动化脚本

工作流程:

1.2 ClaudeSDKClient 类 - 交互式会话

位置: src/claude_agent_sdk/client.py

用途: 用于需要双向通信、状态管理和实时交互的场景。

核心方法:

方法	用途
connect()	建立连接并初始化会话
query()	发送新的查询
receive_messages()	接收所有消息
receive_response()	接收直到 ResultMessage
interrupt()	中断当前执行
set_permission_mode()	动态更改权限模式
set_model()	动态更改模型
get_mcp_status()	获取 MCP 服务器状态
get_context_usage()	获取上下文使用情况
disconnect()	关闭连接

生命周期:

2. 内部实现层

2.1 Query 类 - 控制协议处理

位置: src/claude_agent_sdk/_internal/query.py

职责:

• 处理与 CLI 的控制协议通信
• 管理初始化流程
• 处理钩子回调
• 路由 SDK MCP 服务器消息

控制协议消息类型:

2.2 MessageParser - 消息解析

位置: src/claude_agent_sdk/_internal/message_parser.py

职责: 将 CLI 的原始 JSON 输出解析为类型安全的 Python 对象。

支持的消息类型:

• UserMessage - 用户消息
• AssistantMessage - 助手消息（包含内容块）
• SystemMessage - 系统消息
• ResultMessage - 结果消息（包含成本和使用统计）
• StreamEvent - 流事件
• RateLimitEvent - 速率限制事件

消息类型:

内容块类型:

3. 传输层

3.1 Transport 抽象接口

位置: src/claude_agent_sdk/_internal/transport/__init__.py

定义了传输层必须实现的核心方法：

• connect() - 建立连接
• write(data) - 写入数据
• read_messages() - 读取消息流
• close() - 关闭连接

3.2 SubprocessCLITransport - 子进程传输实现

位置: src/claude_agent_sdk/_internal/transport/subprocess_cli.py

核心职责:

• 管理 Claude Code CLI 子进程的生命周期
• 处理 stdin/stdout/stderr 流
• 缓冲和解析 JSON 消息
• 版本检查和验证

CLI 查找策略:

CLI 查找位置 (按优先级):

1. 捆绑的 CLI (_bundled/ 目录)
2. shutil.which("claude")
3. ~/.npm-global/bin/claude
4. /usr/local/bin/claude
5. ~/.local/bin/claude
6. ~/node_modules/.bin/claude
7. ~/.yarn/bin/claude
8. ~/.claude/local/claude

---

数据流分析

1. 一次性查询数据流

2. 交互式会话数据流

3. MCP 工具调用数据流

4. 钩子执行数据流

---

类型系统

1. ClaudeAgentOptions - 配置选项

位置: src/claude_agent_sdk/types.py (第 1175 行)

这是 SDK 的核心配置类，包含所有可配置选项。

@dataclass
class ClaudeAgentOptions:
    # 工具配置
    tools: list[str] | ToolsPreset | None = None
    allowed_tools: list[str] = field(default_factory=list)
    disallowed_tools: list[str] = field(default_factory=list)

    # 系统提示
    system_prompt: str | SystemPromptPreset | SystemPromptFile | None = None

    # MCP 服务器
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)

    # 权限控制
    permission_mode: PermissionMode | None = None
    can_use_tool: CanUseTool | None = None

    # 会话管理
    continue_conversation: bool = False
    resume: str | None = None
    session_id: str | None = None
    fork_session: bool = False

    # 限制配置
    max_turns: int | None = None
    max_budget_usd: float | None = None
    task_budget: TaskBudget | None = None

    # 模型配置
    model: str | None = None
    fallback_model: str | None = None
    betas: list[SdkBeta] = field(default_factory=list)

    # 钩子配置
    hooks: dict[HookEvent, list[HookMatcher]] | None = None

    # 代理配置
    agents: dict[str, AgentDefinition] | None = None

    # 环境配置
    cwd: str | Path | None = None
    cli_path: str | Path | None = None
    env: dict[str, str] = field(default_factory=dict)

    # 高级特性
    thinking: ThinkingConfig | None = None
    effort: Literal["low", "medium", "high", "max"] | None = None
    output_format: dict[str, Any] | None = None
    enable_file_checkpointing: bool = False
    sandbox: SandboxSettings | None = None

2. 权限模式 (PermissionMode)

3. 钩子事件类型

事件名称	触发时机	用途
PreToolUse	工具使用前	权限控制、参数验证、修改输入
PostToolUse	工具使用后	结果审核、添加上下文
PostToolUseFailure	工具使用失败后	错误处理、重试逻辑
UserPromptSubmit	用户提示提交时	添加上下文、内容过滤
Stop	会话停止时	清理资源
SubagentStart	子代理启动时	子代理初始化
SubagentStop	子代理停止时	子代理清理
PreCompact	上下文压缩前	自定义压缩逻辑
Notification	通知事件	状态通知处理
PermissionRequest	权限请求时	动态权限决策

4. 钩子回调签名

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

HookJSONOutput 类型:

5. MCP 服务器配置类型

---

使用示例

1. 基本使用

1.1 使用 query() 进行一次性查询

import anyio
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
    ResultMessage
)

async def basic_example():
    """简单的一次性查询示例"""

    # 基本查询
    async for message in query(prompt="2 + 2 等于多少？"):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")
        elif isinstance(message, ResultMessage):
            print(f"成本: ${message.total_cost_usd:.4f}")

async def with_options_example():
    """带自定义选项的查询"""

    options = ClaudeAgentOptions(
        system_prompt="你是一个乐于助人的助手，用简单的方式解释事情。",
        max_turns=1,
        model="claude-sonnet-4-5-20250929"
    )

    async for message in query(
        prompt="用一句话解释什么是 Python。",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")

async def with_tools_example():
    """使用特定工具的查询"""

    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write"],
        system_prompt="你是一个乐于助人的文件助手。",
        permission_mode="acceptEdits"  # 自动接受文件编辑
    )

    async for message in query(
        prompt="创建一个叫 hello.txt 的文件，内容是 'Hello, World!'",
        options=options
    ):
        # 处理消息...
        pass

if __name__ == "__main__":
    anyio.run(basic_example)

1.2 使用 ClaudeSDKClient 进行交互式会话

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

async def interactive_session():
    """交互式会话示例"""

    options = ClaudeAgentOptions(
        system_prompt="你是一个编程助手。",
        allowed_tools=["Read", "Grep", "Bash"]
    )

    # 使用 async with 自动管理连接
    async with ClaudeSDKClient(options=options) as client:
        # 第一个问题
        print("用户: 帮我分析这个目录的结构")
        await client.query("帮我分析当前目录的结构")

        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")
            elif isinstance(msg, ResultMessage):
                print("--- 第一轮结束 ---")

        # 第二个问题（基于上下文）
        print("\n用户: 这里有 Python 代码吗？")
        await client.query("这里有 Python 代码吗？分析一下主要文件")

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

async def dynamic_configuration():
    """动态配置示例"""

    async with ClaudeSDKClient() as client:
        # 发送第一个查询
        await client.query("用默认模型分析...")
        async for _ in client.receive_response():
            pass

        # 动态切换模型
        await client.set_model("claude-opus-4-7")

        # 用新模型继续
        await client.query("现在用 Opus 进行更深入的分析...")
        async for _ in client.receive_response():
            pass

        # 动态切换权限模式
        await client.set_permission_mode("acceptEdits")

        # 获取 MCP 状态
        mcp_status = await client.get_mcp_status()
        for server in mcp_status["mcpServers"]:
            print(f"{server['name']}: {server['status']}")

        # 获取上下文使用情况
        usage = await client.get_context_usage()
        print(f"上下文使用: {usage['percentage']:.1f}%")

if __name__ == "__main__":
    asyncio.run(interactive_session)

2. MCP (Model Context Protocol) 使用

2.1 创建 SDK MCP 服务器（进程内）

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

# 使用 @tool 装饰器定义工具
@tool("add", "将两个数字相加", {"a": float, "b": float})
async def add_numbers(args: dict[str, Any]) -> dict[str, Any]:
    """加法工具"""
    result = args["a"] + args["b"]
    return {
        "content": [{"type": "text", "text": f"{args['a']} + {args['b']} = {result}"}]
    }

@tool("multiply", "将两个数字相乘", {"a": float, "b": float})
async def multiply_numbers(args: dict[str, Any]) -> dict[str, Any]:
    """乘法工具"""
    result = args["a"] * args["b"]
    return {
        "content": [{"type": "text", "text": f"{args['a']} × {args['b']} = {result}"}]
    }

@tool("divide", "将两个数字相除", {"a": float, "b": float})
async def divide_numbers(args: dict[str, Any]) -> dict[str, Any]:
    """除法工具，带错误处理"""
    if args["b"] == 0:
        return {
            "content": [{"type": "text", "text": "错误：不能除以零"}],
            "is_error": True
        }
    result = args["a"] / args["b"]
    return {
        "content": [{"type": "text", "text": f"{args['a']} ÷ {args['b']} = {result}"}]
    }

# 使用 TypedDict 定义更复杂的输入模式
from typing import TypedDict, Annotated

class WeatherArgs(TypedDict):
    city: Annotated[str, "城市名称"]
    unit: Annotated[str, "温度单位: celsius 或 fahrenheit"]

@tool("get_weather", "获取城市天气", WeatherArgs)
async def get_weather(args: dict[str, Any]) -> dict[str, Any]:
    """获取天气的工具"""
    city = args["city"]
    unit = args.get("unit", "celsius")

    # 模拟天气 API 调用
    temp = 22 if unit == "celsius" else 72
    return {
        "content": [{
            "type": "text",
            "text": f"{city} 的天气：晴朗，{temp}°{unit[0].upper()}"
        }]
    }

async def mcp_example():
    """MCP 服务器使用示例"""

    # 创建 SDK MCP 服务器
    calculator_server = create_sdk_mcp_server(
        name="calculator",
        version="1.0.0",
        tools=[add_numbers, multiply_numbers, divide_numbers]
    )

    weather_server = create_sdk_mcp_server(
        name="weather",
        version="1.0.0",
        tools=[get_weather]
    )

    # 配置使用这些服务器
    options = ClaudeAgentOptions(
        mcp_servers={
            "calc": calculator_server,
            "weather": weather_server
        },
        # 预批准这些工具，避免权限提示
        allowed_tools=[
            "mcp__calc__add",
            "mcp__calc__multiply",
            "mcp__calc__divide",
            "mcp__weather__get_weather"
        ]
    )

    # 使用客户端
    async with ClaudeSDKClient(options=options) as client:
        # 使用计算器
        await client.query("计算 15 + 27 × 2")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # 使用天气工具
        print("\n---\n")
        await client.query("北京的天气怎么样？用摄氏度")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # 获取 MCP 状态
        status = await client.get_mcp_status()
        for server in status["mcpServers"]:
            print(f"\n服务器: {server['name']}")
            print(f"状态: {server['status']}")
            if server.get("tools"):
                print(f"工具: {[t['name'] for t in server['tools']]}")

if __name__ == "__main__":
    asyncio.run(mcp_example())

2.2 使用外部 MCP 服务器

from pathlib import Path
from claude_agent_sdk import ClaudeAgentOptions, query

async def external_mcp_example():
    """使用外部 MCP 服务器示例"""

    # 方式 1: 使用字典配置
    options1 = ClaudeAgentOptions(
        mcp_servers={
            "filesystem": {
                "type": "stdio",
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allow"]
            },
            "github": {
                "type": "sse",
                "url": "http://localhost:8080/sse",
                "headers": {"Authorization": "Bearer xxx"}
            }
        }
    )

    # 方式 2: 使用 JSON 配置文件
    options2 = ClaudeAgentOptions(
        mcp_servers=Path("mcp-config.json")
    )

    # 方式 3: 使用 JSON 字符串
    mcp_config_json = '''
    {
        "mcpServers": {
            "my-server": {
                "command": "my-mcp-server",
                "args": []
            }
        }
    }
    '''
    options3 = ClaudeAgentOptions(mcp_servers=mcp_config_json)

    # 使用配置
    async for message in query(prompt="列出文件", options=options1):
        # 处理消息...
        pass

3. 钩子 (Hooks) 使用

3.1 PreToolUse - 工具使用前钩子

import asyncio
import logging
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    HookMatcher,
    HookInput,
    HookContext,
    HookJSONOutput
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def security_check_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """安全检查钩子 - 阻止危险操作"""

    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]

    # 阻止删除重要文件
    if tool_name == "DeleteFile":
        file_path = tool_input.get("file_path", "")
        if "important" in file_path.lower() or "secret" in file_path.lower():
            logger.warning(f"阻止删除文件: {file_path}")
            return {
                "systemMessage": "🚫 安全策略阻止了此操作",
                "reason": "删除重要文件被安全策略禁止",
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": "安全策略：不能删除重要文件"
                }
            }

    # 阻止危险的 bash 命令
    if tool_name == "Bash":
        command = tool_input.get("command", "")
        dangerous_patterns = ["rm -rf", "mkfs", "dd if=", ":(){ :|:& };:", "chmod 777"]

        for pattern in dangerous_patterns:
            if pattern in command:
                logger.warning(f"阻止危险命令: {command}")
                return {
                    "systemMessage": "⚠️ 检测到危险命令",
                    "hookSpecificOutput": {
                        "hookEventName": "PreToolUse",
                        "permissionDecision": "deny",
                        "permissionDecisionReason": f"命令包含危险模式: {pattern}"
                    }
                }

    # 允许其他操作
    return {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "allow"
        }
    }

async def modify_input_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """修改工具输入的钩子"""

    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]

    # 在 Write 操作中添加头部注释
    if tool_name == "Write":
        file_path = tool_input.get("file_path", "")
        if file_path.endswith(".py"):
            content = tool_input.get("content", "")
            header = "# 自动生成的文件\n# 请审查后使用\n\n"
            updated_input = {
                **tool_input,
                "content": header + content
            }
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "allow",
                    "updatedInput": updated_input
                }
            }

    return {}

async def pre_tool_use_example():
    """PreToolUse 钩子示例"""

    options = ClaudeAgentOptions(
        allowed_tools=["Write", "Bash", "DeleteFile"],
        hooks={
            "PreToolUse": [
                # 匹配所有工具，进行安全检查
                HookMatcher(matcher=None, hooks=[security_check_hook]),
                # 只匹配 Write 工具
                HookMatcher(matcher="Write", hooks=[modify_input_hook]),
                # 匹配多个工具
                HookMatcher(matcher="Bash|DeleteFile", hooks=[logging_hook])
            ]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("尝试删除 important_file.txt")
        async for msg in client.receive_response():
            # 处理消息...
            pass

3.2 PostToolUse - 工具使用后钩子

async def audit_log_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """审计日志钩子 - 记录所有工具使用"""

    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]
    tool_response = input_data.get("tool_response")

    logger.info(f"工具使用审计: {tool_name}")
    logger.info(f"输入: {tool_input}")
    logger.info(f"输出: {tool_response}")

    return {}

async def validate_output_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """验证工具输出并添加上下文"""

    tool_name = input_data["tool_name"]
    tool_response = input_data.get("tool_response", "")

    # 检测到错误时添加警告
    if "error" in str(tool_response).lower():
        return {
            "systemMessage": "⚠️ 工具执行产生了错误",
            "reason": "检测到工具执行错误，请检查输出",
            "hookSpecificOutput": {
                "hookEventName": "PostToolUse",
                "additionalContext": "上一个工具执行产生了错误，考虑使用其他方法。"
            }
        }

    return {}

async def post_tool_use_example():
    """PostToolUse 钩子示例"""

    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        hooks={
            "PostToolUse": [
                HookMatcher(matcher=None, hooks=[audit_log_hook]),
                HookMatcher(matcher="Bash", hooks=[validate_output_hook])
            ]
        }
    )

3.3 其他钩子事件

async def user_prompt_submit_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """用户提示提交钩子 - 添加上下文"""

    prompt = input_data["prompt"]

    # 可以在这里添加自定义上下文
    additional_context = "用户当前时间: 2024-01-01 12:00:00"

    return {
        "hookSpecificOutput": {
            "hookEventName": "UserPromptSubmit",
            "additionalContext": additional_context
        }
    }

async def subagent_start_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """子代理启动钩子"""

    agent_id = input_data["agent_id"]
    agent_type = input_data["agent_type"]

    logger.info(f"子代理启动: {agent_id} ({agent_type})")

    return {
        "hookSpecificOutput": {
            "hookEventName": "SubagentStart",
            "additionalContext": "子代理已启动，将在独立上下文中运行。"
        }
    }

async def notification_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """通知钩子 - 处理系统通知"""

    title = input_data.get("title", "")
    message = input_data["message"]
    notification_type = input_data["notification_type"]

    logger.info(f"通知 [{notification_type}]: {title} - {message}")

    return {}

async def permission_request_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """权限请求钩子 - 动态权限决策"""

    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]

    # 基于业务逻辑做决策
    if tool_name == "Write":
        file_path = tool_input.get("file_path", "")
        if file_path.startswith("/safe/"):
            # 自动允许安全目录的写入
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PermissionRequest",
                    "decision": {"behavior": "allow"}
                }
            }

    # 其他情况让 CLI 提示用户
    return {}

async def complete_hooks_example():
    """完整的钩子配置示例"""

    options = ClaudeAgentOptions(
        hooks={
            "PreToolUse": [
                HookMatcher(matcher=None, hooks=[security_check_hook])
            ],
            "PostToolUse": [
                HookMatcher(matcher=None, hooks=[audit_log_hook])
            ],
            "PostToolUseFailure": [
                HookMatcher(matcher=None, hooks=[error_recovery_hook])
            ],
            "UserPromptSubmit": [
                HookMatcher(matcher=None, hooks=[user_prompt_submit_hook])
            ],
            "SubagentStart": [
                HookMatcher(matcher=None, hooks=[subagent_start_hook])
            ],
            "SubagentStop": [
                HookMatcher(matcher=None, hooks=[subagent_stop_hook])
            ],
            "Notification": [
                HookMatcher(matcher=None, hooks=[notification_hook])
            ],
            "PermissionRequest": [
                HookMatcher(matcher=None, hooks=[permission_request_hook])
            ],
            "PreCompact": [
                HookMatcher(matcher=None, hooks=[pre_compact_hook])
            ],
            "Stop": [
                HookMatcher(matcher=None, hooks=[stop_hook])
            ]
        },
        # 钩子超时设置（秒）
        hooks={
            "PreToolUse": [
                HookMatcher(matcher=None, hooks=[slow_hook], timeout=30.0)
            ]
        }
    )

4. 子代理 (Subagents) 使用

import anyio
from claude_agent_sdk import (
    AgentDefinition,
    ClaudeAgentOptions,
    query,
    ClaudeSDKClient,
    AssistantMessage,
    TextBlock
)

async def code_reviewer_agent_example():
    """代码审查代理示例"""

    options = ClaudeAgentOptions(
        agents={
            "code-reviewer": AgentDefinition(
                description="专业的代码审查员，检查代码质量和潜在问题",
                prompt="""你是一位资深的代码审查专家。请仔细分析代码，关注：
                1. 潜在的 bug 和逻辑错误
                2. 性能问题
                3. 安全漏洞
                4. 代码风格和最佳实践
                5. 可维护性问题

                提供具体、建设性的反馈。""",
                tools=["Read", "Grep", "Glob"],  # 代理可用的工具
                disallowedTools=["Write", "Edit", "Bash"],  # 明确禁止的工具
                model="sonnet",  # 代理使用的模型
                skills=None,  # 代理使用的技能
                memory="project",  # 记忆范围
                maxTurns=10,  # 最大回合数
                permissionMode="plan"  # 代理的权限模式
            )
        }
    )

    # 使用代理
    async for message in query(
        prompt="使用 code-reviewer 代理审查 src/claude_agent_sdk/client.py",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")

async def multiple_agents_example():
    """多代理协作示例"""

    options = ClaudeAgentOptions(
        agents={
            "analyzer": AgentDefinition(
                description="代码结构分析员",
                prompt="分析代码结构和架构模式。",
                tools=["Read", "Grep", "Glob"],
                model="sonnet"
            ),
            "tester": AgentDefinition(
                description="测试专家",
                prompt="编写和运行测试，确保代码质量。",
                tools=["Read", "Write", "Bash", "Edit"],
                model="sonnet",
                permissionMode="acceptEdits"
            ),
            "documenter": AgentDefinition(
                description="文档编写专家",
                prompt="为代码编写清晰、全面的文档。",
                tools=["Read", "Write", "Edit"],
                model="sonnet"
            )
        },
        # 从哪些来源加载设置
        setting_sources=["user", "project", "local"]
    )

    async with ClaudeSDKClient(options=options) as client:
        # 使用分析代理
        await client.query("让 analyzer 代理分析项目结构")
        async for msg in client.receive_response():
            # 处理响应...
            pass

        # 使用测试代理
        await client.query("让 tester 代理为核心模块编写测试")
        async for msg in client.receive_response():
            # 处理响应...
            pass

async def background_agent_example():
    """后台代理示例"""

    options = ClaudeAgentOptions(
        agents={
            "monitor": AgentDefinition(
                description="后台监控代理",
                prompt="持续监控系统状态并报告问题。",
                tools=["Read", "Grep"],
                background=True,  # 后台运行
                effort="low"  # 低努力级别
            )
        }
    )

5. 错误处理

5.1 错误类型层级

from claude_agent_sdk import (
    ClaudeSDKError,
    CLIConnectionError,
    CLINotFoundError,
    CLIJSONDecodeError,
    ProcessError
)

# 错误层级:
# ClaudeSDKError (基类)
# ├── CLIConnectionError - 连接错误
# │   ├── CLINotFoundError - CLI 未找到
# │   └── CLIJSONDecodeError - JSON 解析错误
# └── ProcessError - 进程执行错误

5.2 常见错误处理示例

import anyio
from claude_agent_sdk import (
    query,
    ClaudeSDKClient,
    ClaudeAgentOptions,
    CLINotFoundError,
    CLIConnectionError,
    ProcessError,
    CLIJSONDecodeError,
    ClaudeSDKError
)

async def error_handling_example():
    """完整的错误处理示例"""

    # 示例 1: 处理 CLI 未找到错误
    try:
        options = ClaudeAgentOptions(
            cli_path="/invalid/path/to/claude"
        )
        async for message in query(prompt="你好", options=options):
            pass
    except CLINotFoundError as e:
        print(f"未找到 Claude Code CLI: {e}")
        print("请安装: npm install -g @anthropic-ai/claude-code")
        print("或通过 ClaudeAgentOptions(cli_path=...) 指定路径")

    # 示例 2: 处理连接错误
    try:
        async with ClaudeSDKClient() as client:
            await client.connect()
    except CLIConnectionError as e:
        print(f"连接失败: {e}")
        if hasattr(e, '__cause__'):
            print(f"原因: {e.__cause__}")

    # 示例 3: 处理进程错误
    try:
        options = ClaudeAgentOptions(
            cwd="/nonexistent/directory"
        )
        async for message in query(prompt="你好", options=options):
            pass
    except ProcessError as e:
        print(f"进程执行失败: {e}")
        print(f"退出代码: {e.exit_code}")
        print(f"错误输出: {e.stderr}")

    # 示例 4: 处理 JSON 解析错误
    try:
        async with ClaudeSDKClient() as client:
            await client.connect()
            # 解析错误可能在消息接收时发生
            async for msg in client.receive_messages():
                pass
    except CLIJSONDecodeError as e:
        print(f"JSON 解析失败: {e}")
        print(f"原始错误: {e.__cause__}")

    # 示例 5: 捕获所有 SDK 错误
    try:
        async for message in query(prompt="你好"):
            pass
    except ClaudeSDKError as e:
        print(f"SDK 错误: {e}")
        # 根据错误类型进行不同处理
        if isinstance(e, CLINotFoundError):
            print("需要安装 CLI")
        elif isinstance(e, ProcessError):
            print(f"进程失败，退出代码: {e.exit_code}")
        else:
            print("其他错误")

async def retry_pattern_example():
    """重试模式示例"""

    max_retries = 3
    retry_delay = 1.0  # 秒

    for attempt in range(max_retries):
        try:
            async for message in query(prompt="执行任务"):
                # 处理消息
                pass
            break  # 成功，退出循环
        except (CLIConnectionError, ProcessError) as e:
            if attempt < max_retries - 1:
                print(f"尝试 {attempt + 1} 失败，等待后重试...")
                await anyio.sleep(retry_delay * (attempt + 1))  # 退避策略
            else:
                print(f"所有 {max_retries} 次尝试都失败")
                raise

async def with_stderr_callback_example():
    """使用 stderr 回调进行调试"""

    def stderr_handler(line: str):
        """处理 CLI 的 stderr 输出"""
        print(f"[CLI stderr] {line}")
        # 可以在这里记录日志或分析错误

    options = ClaudeAgentOptions(
        stderr=stderr_handler  # 设置 stderr 回调
    )

    try:
        async for message in query(prompt="你好", options=options):
            pass
    except ClaudeSDKError as e:
        print(f"错误: {e}")
        # stderr 回调已经输出了详细的调试信息

async def result_message_error_check():
    """检查 ResultMessage 中的错误"""

    from claude_agent_sdk import ResultMessage

    async for message in query(prompt="执行可能失败的任务"):
        if isinstance(message, ResultMessage):
            if message.is_error:
                print("任务执行出错!")
                if message.errors:
                    for error in message.errors:
                        print(f"  - {error}")
                if message.permission_denials:
                    print(f"权限拒绝次数: {len(message.permission_denials)}")
            else:
                print(f"任务成功完成")
                print(f"总耗时: {message.duration_ms}ms")
                if message.total_cost_usd:
                    print(f"成本: ${message.total_cost_usd:.4f}")

5.3 工具权限回调错误处理

from claude_agent_sdk import (
    ClaudeAgentOptions,
    ClaudeSDKClient,
    CanUseTool,
    ToolPermissionContext,
    PermissionResultAllow,
    PermissionResultDeny,
    PermissionResult
)

async def tool_permission_callback_example():
    """工具权限回调示例"""

    async def can_use_tool(
        tool_name: str,
        tool_input: dict,
        context: ToolPermissionContext
    ) -> PermissionResult:
        """权限决策回调"""

        try:
            # 在这里进行权限检查
            if tool_name == "Bash":
                command = tool_input.get("command", "")
                if "danger" in command:
                    return PermissionResultDeny(
                        message="不允许执行危险命令",
                        interrupt=True
                    )

            # 记录权限决策建议
            if context.suggestions:
                print(f"CLI 建议: {context.suggestions}")

            # 允许执行
            return PermissionResultAllow()

        except Exception as e:
            # 回调出错时默认拒绝
            print(f"权限回调出错: {e}")
            return PermissionResultDeny(
                message="权限检查失败，操作被拒绝"
            )

    options = ClaudeAgentOptions(
        can_use_tool=can_use_tool,
        # 注意: can_use_tool 需要流式模式
        # 不能同时设置 permission_prompt_tool_name
    )

    # 使用时需要提供 AsyncIterable 作为 prompt
    from collections.abc import AsyncIterable

    async def prompt_stream() -> AsyncIterable[dict]:
        yield {
            "type": "user",
            "message": {"role": "user", "content": "你好"},
            "parent_tool_use_id": None,
            "session_id": "default"
        }

    async with ClaudeSDKClient(options=options) as client:
        await client.connect(prompt=prompt_stream())
        async for msg in client.receive_response():
            # 处理消息...
            pass

---

总结

Claude Agent SDK for Python 提供了一个功能丰富、设计良好的接口，用于与 Claude Code 交互。通过合理使用其提供的各种功能（MCP 服务器、钩子系统、子代理等），可以构建强大的 AI 辅助应用程序。

关键要点

1. 选择合适的 API: query() 适合简单场景，ClaudeSDKClient 适合复杂交互
2. 利用类型系统: 完整的类型注解提供了良好的开发体验和错误检查
3. 善用钩子系统: 11 种钩子事件可以实现精细的生命周期管理
4. MCP 服务器: 进程内 MCP 服务器提供高性能的工具扩展能力
5. 错误处理: 完善的错误类型和处理模式确保应用的健壮性

进一步资源

• 查看 examples/ 目录获取更多示例代码
• 阅读 tests/ 目录了解各组件的详细用法
• 参考 Claude Code 官方文档了解更多 CLI 功能

---

文档生成时间: 2026-04-18
SDK 版本: 查看 src/claude_agent_sdk/_version.py