English Documentation | GitHub Repository

---

📚 目录

• 1. 概述
• 2. 安装与环境配置
• 3. 核心架构
• 4. 基础使用
• 5. 高级功能
• 6. MCP服务器与自定义工具
• 7. Hooks系统
• 8. 错误处理
• 9. 最佳实践
• 10. 常见问题

---

1. 概述

1.1 什么是 Claude Agent SDK?

Claude Agent SDK for Python 是 Anthropic 官方推出的 Python 软件开发工具包，用于与 Claude Code 进行程序化交互。该 SDK 提供了两种核心交互模式：

• 查询模式 (query()): 简单的一次性交互，适用于无状态、单向的对话场景
• 客户端模式 (ClaudeSDKClient): 双向交互式会话，支持状态保持、中断控制和实时消息流

1.2 核心特性

特性	说明
🚀 简单易用	直观的 API 设计，支持 async/await 异步编程
🔧 工具集成	内置支持 Read、Write、Bash、Edit 等文件系统工具
🛡️ 权限控制	多级别权限模式，支持自定义工具使用策略
🔌 MCP 服务器	支持进程内 MCP 服务器，性能优于外部进程方案
🪝 Hooks 系统	6 种事件钩子，支持细粒度控制对话流程
📊 结构化输出	支持 JSON Schema 验证的结构化响应
💰 成本追踪	实时监控 API 调用成本
🏖️ 沙箱支持	可选的命令执行沙箱环境

1.3 应用场景

• 自动化脚本: CI/CD 流水线、代码生成、批量处理
• 开发工具: IDE 插件、代码审查工具、智能重构
• 交互式应用: 聊天机器人、REPL 环境、调试助手
• 数据处理: 文件转换、数据提取、报告生成

---

2. 安装与环境配置

2.1 系统要求

# Python 版本要求 (推荐 3.10+)
python --version  # >= 3.10

# 操作系统支持
# - macOS (Intel & Apple Silicon)
# - Linux (x86_64, ARM64)
# - Windows (WSL2 推荐)

2.2 安装方式

通过 pip 安装（推荐）

pip install claude-agent-sdk

开发环境安装

# 克隆仓库
git clone https://github.com/anthropics/claude-agent-sdk-python.git
cd claude-agent-sdk-python

# 安装依赖
pip install -e ".[dev]"

# 运行测试
pytest

2.3 Claude Code CLI

从 0.1.8 版本开始，Claude Code CLI 已自动捆绑，无需单独安装：

# SDK 会自动使用捆绑的 CLI
# 如需使用系统级安装，可指定路径：
ClaudeAgentOptions(cli_path="/path/to/claude")

如需手动安装 CLI：

curl -fsSL https://claude.ai/install.sh | bash

2.4 环境变量

环境变量	说明	默认值
CLAUDE_CODE_ENTRYPOINT	SDK 入口点标识	sdk-py / sdk-py-client
CLAUDE_CODE_STREAM_CLOSE_TIMEOUT	流关闭超时时间（毫秒）	60000
ANTHROPIC_API_KEY	API 密钥	-

---

3. 核心架构

3.1 架构概览

┌─────────────────────────────────────────────────────────────┐
│                     Your Application                          │
├─────────────────────────────────────────────────────────────┤
│  ┌──────────────┐                    ┌───────────────────┐   │
│  │   query()    │                    │ ClaudeSDKClient   │   │
│  │   Function   │                    │    Class        │   │
│  └──────┬───────┘                    └────────┬──────────┘   │
├─────────┼───────────────────────────────────────┼──────────────┤
│         │  Internal Implementation              │              │
│  ┌──────▼───────┐                    ┌────────▼──────────┐   │
│  │ InternalClient │                    │     Query         │   │
│  └──────┬───────┘                    └────────┬──────────┘   │
│         │                                    │              │
│  ┌──────▼───────────────────────────────────▼──────────┐   │
│  │              Transport Layer                          │   │
│  │  ┌────────────────────────────────────────────────┐  │   │
│  │  │         SubprocessCLITransport                 │  │   │
│  │  │    ( communicates with Claude CLI )            │  │   │
│  │  └────────────────────────────────────────────────┘  │   │
│  └───────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────┘

3.2 核心组件

3.2.1 Transport 层

Transport 层负责与 Claude Code CLI 的底层通信：

• SubprocessCLITransport: 默认实现，通过子进程与 CLI 交互
• Transport 接口: 可扩展接口，支持自定义传输实现

3.2.2 Query 引擎

Query 类处理控制协议和消息路由：

• 管理异步消息流
• 处理工具权限回调
• 协调 MCP 服务器通信
• 支持钩子的注册和执行

3.2.3 Message Parser

负责将 CLI 的原始输出解析为类型安全的 Python 对象：

• parse_message(): 解析 JSON 消息
• 支持 5 种消息类型
• 提供内容块级别的解析

3.3 数据流

1. 应用层调用 query() 或 ClaudeSDKClient
   ↓
2. InternalClient 处理配置验证
   ↓
3. Transport 建立与 CLI 的连接
   ↓
4. Query 引擎初始化控制协议
   ↓
5. 消息流开始双向通信
   ↓
6. Message Parser 实时解析响应
   ↓
7. 应用层处理解析后的消息对象

---

4. 基础使用

4.1 快速开始

最简示例

import anyio
from claude_agent_sdk import query

async def main():
    async for message in query(prompt="What is 2 + 2?"):
        print(message)

anyio.run(main)

带选项的查询

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

async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are a helpful assistant",
        max_turns=1
    )
    
    async for message in query(
        prompt="Tell me a joke", 
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

4.2 消息类型系统

4.2.1 消息类型概览

# 5 种核心消息类型
from claude_agent_sdk import (
    UserMessage,        # 用户消息
    AssistantMessage,   # 助手消息
    SystemMessage,      # 系统消息
    ResultMessage,      # 结果消息（含成本信息）
    StreamEvent         # 流事件（部分消息）
)

4.2.2 内容块类型

from claude_agent_sdk import (
    TextBlock,         # 文本内容
    ThinkingBlock,     # 思考过程
    ToolUseBlock,      # 工具使用
    ToolResultBlock    # 工具结果
)

# 内容块层次结构
AssistantMessage.content → List[ContentBlock]
ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

4.2.3 完整示例：消息解析

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import (
    AssistantMessage,
    ResultMessage,
    TextBlock,
    ToolUseBlock,
    ToolResultBlock
)

async def parse_messages():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write"],
        system_prompt="You are a file assistant"
    )
    
    async for message in query(
        prompt="Create a hello.txt file with 'Hello World'",
        options=options
    ):
        match message:
            case AssistantMessage():
                print(f"Model: {message.model}")
                for block in message.content:
                    match block:
                        case TextBlock(text=text):
                            print(f"Text: {text}")
                        case ToolUseBlock(id=tool_id, name=name, input=input_data):
                            print(f"Tool: {name} (ID: {tool_id})")
                            print(f"Input: {input_data}")
                            
            case ResultMessage():
                print(f"Cost: ${message.total_cost_usd}")
                print(f"Duration: {message.duration_ms}ms")
                print(f"Turns: {message.num_turns}")

4.3 工具使用

4.3.1 可用工具

工具名	说明	权限级别
Read	读取文件内容	低
Write	写入文件内容	中
Edit	编辑文件（多行）	中
Bash	执行 shell 命令	高
WebFetch	获取网页内容	中
StrReplace	字符串替换	低

4.3.2 工具配置

from claude_agent_sdk import ClaudeAgentOptions

# 方式 1: allowed_tools 列表
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Bash"],
    permission_mode='acceptEdits'  # 自动接受文件编辑
)

# 方式 2: 使用 tools 预设
options = ClaudeAgentOptions(
    tools={"type": "preset", "preset": "claude_code"},
    disallowed_tools=["Bash"]  # 排除特定工具
)

# 方式 3: 禁用所有工具
options = ClaudeAgentOptions(
    tools=[]  # 空列表表示禁用所有工具
)

4.3.3 权限模式

PermissionMode = Literal[
    "default",          # 默认：危险操作需要确认
    "acceptEdits",      # 自动接受文件编辑
    "plan",             # 计划模式
    "bypassPermissions" # 绕过所有权限（谨慎使用）
]

4.4 工作目录配置

from pathlib import Path
from claude_agent_sdk import ClaudeAgentOptions

# 方式 1: 字符串路径
options = ClaudeAgentOptions(cwd="/path/to/project")

# 方式 2: Path 对象
options = ClaudeAgentOptions(cwd=Path("/path/to/project"))

# 方式 3: 相对路径
options = ClaudeAgentOptions(cwd="./my-project")

---

5. 高级功能

5.1 ClaudeSDKClient - 交互式客户端

5.1.1 何时使用 ClaudeSDKClient

使用场景：

• ✅ 交互式对话（聊天机器人、REPL）
• ✅ 需要根据响应动态发送消息
• ✅ 需要中断控制能力
• ✅ 长时间运行的会话
• ✅ 实时用户输入

不使用场景：

• ❌ 一次性简单查询
• ❌ 批量处理独立提示
• ❌ 所有输入已预先知道

5.1.2 基本使用模式

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def interactive_example():
    options = ClaudeAgentOptions(
        system_prompt="You are a helpful coding assistant",
        allowed_tools=["Read", "Write", "Bash"]
    )
    
    # 方式 1: 显式连接/断开
    client = ClaudeSDKClient(options=options)
    await client.connect()
    
    await client.query("Help me create a Python script")
    async for message in client.receive_response():
        print(message)
    
    await client.disconnect()
    
    # 方式 2: 使用上下文管理器（推荐）
    async with ClaudeSDKClient(options=options) as client:
        await client.query("List files in current directory")
        async for message in client.receive_response():
            print(message)

5.1.3 消息接收模式

# 模式 1: 接收所有消息（无限）
async for message in client.receive_messages():
    print(message)
    # 需要手动中断或等待会话结束

# 模式 2: 接收单个响应（到 ResultMessage 为止）
async for message in client.receive_response():
    print(message)
    # 自动在 ResultMessage 后停止

5.1.4 会话控制

# 中断当前操作
await client.interrupt()

# 动态更改权限模式
await client.set_permission_mode('acceptEdits')

# 切换模型
await client.set_model('claude-sonnet-4-5')

# 获取服务器信息
info = await client.get_server_info()
print(f"可用命令: {len(info.get('commands', []))}")

5.2 流式模式

5.2.1 异步迭代器模式

from claude_agent_sdk import query, ClaudeAgentOptions

async def streaming_prompts():
    # 定义异步生成器
    async def prompt_generator():
        yield {"type": "user", "message": {"role": "user", "content": "Hello"}}
        yield {"type": "user", "message": {"role": "user", "content": "How are you?"}}
        yield {"type": "user", "message": {"role": "user", "content": "Tell me a joke"}}
    
    # 使用异步迭代器作为 prompt
    async for message in query(prompt=prompt_generator()):
        print(message)

5.2.2 消息格式

{
    "type": "user",                           # 消息类型
    "message": {
        "role": "user",                       # 角色
        "content": "Hello, Claude!"           # 内容
    },
    "parent_tool_use_id": None,               # 父工具使用 ID
    "session_id": "default"                   # 会话 ID
}

5.3 工具权限回调

5.3.1 自定义权限逻辑

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

async def custom_tool_permission(
    tool_name: str,
    tool_input: dict,
    context: ToolPermissionContext
) -> PermissionResult:
    """自定义工具使用权限检查"""
    
    # 示例：阻止特定文件写入
    if tool_name == "Write":
        file_path = tool_input.get("file_path", "")
        if "sensitive" in file_path:
            return PermissionResultDeny(
                message=f"Cannot write to sensitive file: {file_path}"
            )
    
    # 示例：修改 Bash 命令输入
    if tool_name == "Bash":
        command = tool_input.get("command", "")
        if "rm -rf /" in command:
            return PermissionResultDeny(
                message="Dangerous command blocked",
                interrupt=True  # 中断执行
            )
        
        # 修改命令，添加安全检查
        updated_input = tool_input.copy()
        updated_input["command"] = f"set -e; {command}"
        
        return PermissionResultAllow(updated_input=updated_input)
    
    # 默认允许
    return PermissionResultAllow()

# 使用自定义权限回调
options = ClaudeAgentOptions(
    can_use_tool=custom_tool_permission,
    # 需要 streaming 模式
)

async with ClaudeSDKClient(options=options) as client:
    await client.query("Create a test file")
    # 工具使用将经过自定义权限检查

5.3.2 权限结果类型

@dataclass
class PermissionResultAllow:
    behavior: Literal["allow"] = "allow"
    updated_input: dict[str, Any] | None = None    # 修改后的输入
    updated_permissions: list[PermissionUpdate] | None = None  # 权限更新

@dataclass  
class PermissionResultDeny:
    behavior: Literal["deny"] = "deny"
    message: str = ""           # 拒绝原因
    interrupt: bool = False     # 是否中断执行

5.4 文件检查点与回滚

5.4.1 启用文件检查点

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient

options = ClaudeAgentOptions(
    enable_file_checkpointing=True,           # 启用检查点
    extra_args={"replay-user-messages": None} # 接收 UserMessage UUID
)

async with ClaudeSDKClient(options=options) as client:
    # 执行文件操作
    await client.query("Create file1.txt with content A")
    
    # 获取检查点 ID
    checkpoint_id = None
    async for msg in client.receive_response():
        if hasattr(msg, 'uuid') and msg.uuid:
            checkpoint_id = msg.uuid
            break
    
    # 继续操作
    await client.query("Modify file1.txt to content B")
    
    # 回滚到检查点
    if checkpoint_id:
        await client.rewind_files(checkpoint_id)
        print(f"Files reverted to checkpoint: {checkpoint_id}")

5.5 结构化输出

5.5.1 JSON Schema 验证

from claude_agent_sdk import ClaudeAgentOptions

# 定义 JSON Schema
output_schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer", "minimum": 0},
        "skills": {
            "type": "array",
            "items": {"type": "string"}
        }
    },
    "required": ["name", "age"]
}

options = ClaudeAgentOptions(
    output_format={
        "type": "json_schema",
        "schema": output_schema
    }
)

# 在 ResultMessage 中获取验证后的 JSON
async for message in query(
    prompt="Extract person info from: John Doe, 30 years old, skills: Python, AI",
    options=options
):
    if isinstance(message, ResultMessage):
        structured_data = message.structured_output
        print(f"Validated data: {structured_data}")

5.6 沙箱配置

5.6.1 启用沙箱

from claude_agent_sdk import ClaudeAgentOptions
from claude_agent_sdk.types import SandboxSettings

sandbox_config: SandboxSettings = {
    "enabled": True,                           # 启用沙箱
    "autoAllowBashIfSandboxed": True,         # 沙箱内自动批准命令
    "excludedCommands": ["docker", "git"],    # 排除的命令
    "allowUnsandboxedCommands": False,        # 不允许非沙箱命令
    "network": {
        "allowUnixSockets": ["/var/run/docker.sock"],
        "allowLocalBinding": True
    },
    "ignoreViolations": {
        "file": ["/tmp/*"],
        "network": ["localhost"]
    }
}

options = ClaudeAgentOptions(sandbox=sandbox_config)

重要说明： 文件系统和网络限制应通过权限规则配置，而非沙箱设置：

• 文件读取限制：使用 Read 拒绝规则
• 文件写入限制：使用 Edit 允许/拒绝规则
• 网络限制：使用 WebFetch 允许/拒绝规则

---

6. MCP服务器与自定义工具

6.1 什么是 MCP?

MCP (Model Context Protocol) 是 Claude 的插件协议，允许模型调用外部工具。SDK 支持两种 MCP 服务器：

1. 外部 MCP 服务器: 独立进程，通过 stdio/SSE/HTTP 通信
2. SDK MCP 服务器: 进程内执行，性能更优

6.2 SDK MCP 服务器的优势

特性	SDK MCP	外部 MCP
性能	✅ 无 IPC 开销	❌ 进程间通信
部署	✅ 单进程	❌ 多进程管理
调试	✅ 同进程	❌ 跨进程调试
状态访问	✅ 直接访问	❌ 序列化传递
子进程管理	✅ 无需	❌ 需要管理

6.3 创建自定义工具

6.3.1 使用 @tool 装饰器

from claude_agent_sdk import tool, create_sdk_mcp_server
from typing import Any

# 定义工具 - 基本计算器
@tool("add", "Add two numbers", {"a": float, "b": float})
async def add_numbers(args: dict[str, Any]) -> dict[str, Any]:
    """Add two numbers and return the result."""
    result = args["a"] + args["b"]
    return {
        "content": [
            {"type": "text", "text": f"{args['a']} + {args['b']} = {result}"}
        ]
    }

# 定义工具 - 带错误处理
@tool("divide", "Divide two numbers", {"a": float, "b": float})
async def divide_numbers(args: dict[str, Any]) -> dict[str, Any]:
    """Divide a by b with error handling."""
    if args["b"] == 0:
        return {
            "content": [
                {"type": "text", "text": "Error: Division by zero"}
            ],
            "is_error": True  # 标记为错误
        }
    
    result = args["a"] / args["b"]
    return {
        "content": [
            {"type": "text", "text": f"{args['a']} ÷ {args['b']} = {result}"}
        ]
    }

# 定义工具 - 访问应用状态
class DataStore:
    def __init__(self):
        self.items = []
        self.counter = 0

store = DataStore()

@tool("add_item", "Add item to store", {"item": str})
async def add_item(args: dict[str, Any]) -> dict[str, Any]:
    """Add item to application state."""
    store.items.append(args["item"])
    store.counter += 1
    return {
        "content": [
            {"type": "text", "text": f"Added: {args['item']} (Total: {store.counter})"}
        ]
    }

@tool("list_items", "List all items in store", {})
async def list_items(args: dict[str, Any]) -> dict[str, Any]:
    """List all items in store."""
    items_str = ", ".join(store.items) if store.items else "No items"
    return {
        "content": [
            {"type": "text", "text": f"Items: {items_str}"}
        ]
    }

6.3.2 输入模式类型

# 简单字典模式
@tool("greet", "Greet user", {"name": str, "age": int})
async def greet(args):
    return {"content": [{"type": "text", "text": f"Hello {args['name']}, age {args['age']}"}]}

# JSON Schema 模式
@tool("process_data", "Process data", {
    "type": "object",
    "properties": {
        "data": {"type": "array", "items": {"type": "string"}},
        "options": {
            "type": "object",
            "properties": {
                "case_sensitive": {"type": "boolean"},
                "max_length": {"type": "integer"}
            }
        }
    },
    "required": ["data"]
})
async def process_data(args):
    # 处理复杂输入
    pass

6.4 创建 MCP 服务器

6.4.1 基本服务器创建

from claude_agent_sdk import create_sdk_mcp_server

# 创建服务器
calculator_server = create_sdk_mcp_server(
    name="calculator",           # 服务器名称（唯一）
    version="2.0.0",            # 版本号
    tools=[
        add_numbers,
        divide_numbers,
        # ... 更多工具
    ]
)

# 使用服务器
options = ClaudeAgentOptions(
    mcp_servers={
        "calc": calculator_server  # "calc" 是服务器引用名
    },
    allowed_tools=[
        "mcp__calc__add",          # 工具命名格式: mcp__{server_name}__{tool_name}
        "mcp__calc__divide"
    ]
)

6.4.2 多个服务器

# 创建多个服务器
math_server = create_sdk_mcp_server(
    name="math",
    tools=[add_numbers, multiply_numbers]
)

file_server = create_sdk_mcp_server(
    name="file_ops", 
    tools=[read_file, write_file, list_directory]
)

# 配置多个服务器
options = ClaudeAgentOptions(
    mcp_servers={
        "math": math_server,
        "files": file_server
    },
    allowed_tools=[
        "mcp__math__add",
        "mcp__math__multiply", 
        "mcp__files__read_file",
        "mcp__files__write_file"
    ]
)

6.5 混合服务器配置

from claude_agent_sdk import ClaudeAgentOptions

# 同时使用 SDK MCP 和外部 MCP
options = ClaudeAgentOptions(
    mcp_servers={
        # SDK MCP 服务器（进程内）
        "internal": calculator_server,
        
        # 外部 MCP 服务器（独立进程）
        "external": {
            "type": "stdio",
            "command": "python",
            "args": ["-m", "external_server"]
        },
        
        # SSE 服务器
        "sse_server": {
            "type": "sse",
            "url": "http://localhost:8080/sse"
        }
    }
)

6.6 完整示例：计算器服务器

#!/usr/bin/env python3
"""完整示例：带 UI 的计算器 MCP 服务器"""

import asyncio
from claude_agent_sdk import (
    ClaudeAgentOptions,
    ClaudeSDKClient,
    create_sdk_mcp_server,
    tool
)

# 工具定义
@tool("calculate", "Perform calculation", {"expression": str})
async def calculate(args):
    """安全地计算数学表达式"""
    expression = args["expression"]
    
    # 安全验证：只允许数字、运算符和括号
    allowed_chars = set("0123456789+-*/.() ")
    if not all(c in allowed_chars for c in expression):
        return {
            "content": [{"type": "text", "text": "Error: Invalid characters in expression"}],
            "is_error": True
        }
    
    try:
        # 安全计算（使用 ast 模块更安全）
        import ast
        import operator
        
        # 简化的安全求值（实际生产应使用更严格的解析）
        result = eval(expression, {"__builtins__": {}})
        return {
            "content": [{"type": "text", "text": f"{expression} = {result}"}]
        }
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"Error: {str(e)}"}],
            "is_error": True
        }

@tool("get_history", "Get calculation history", {})
async def get_history(args):
    """获取计算历史"""
    # 这里可以连接到持久化存储
    history = getattr(calculate, "history", [])
    if not history:
        return {"content": [{"type": "text", "text": "No calculations yet"}]}
    
    history_text = "\\n".join(f"{i+1}. {expr}" for i, expr in enumerate(history))
    return {"content": [{"type": "text", "text": f"History:\\n{history_text}"}]}

# 创建服务器
async def main():
    calculator = create_sdk_mcp_server(
        name="smart_calculator",
        version="1.0.0",
        tools=[calculate, get_history]
    )
    
    options = ClaudeAgentOptions(
        mcp_servers={"calc": calculator},
        allowed_tools=["mcp__calc__calculate", "mcp__calc__get_history"],
        system_prompt="You are a helpful calculator assistant."
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # 测试 1: 简单计算
        await client.query("Calculate 15 * 7 + 3")
        async for msg in client.receive_response():
            print(f"Response: {msg}")
        
        # 测试 2: 获取历史
        await client.query("Show me the calculation history")
        async for msg in client.receive_response():
            print(f"History: {msg}")

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

---

7. Hooks系统

7.1 Hooks 概览

Hooks 允许在 Claude 代理循环的特定点插入自定义逻辑，提供对对话流程的细粒度控制。

7.1.1 支持的 Hook 事件

Hook 事件	触发时机	主要用途
PreToolUse	工具执行前	权限检查、参数验证、输入修改
PostToolUse	工具执行后	结果验证、错误处理、添加上下文
UserPromptSubmit	用户提示提交时	上下文注入、提示预处理
Stop	会话停止时	清理操作、日志记录
SubagentStop	子代理停止时	子代理生命周期管理
PreCompact	会话压缩前	压缩策略控制

7.1.2 不支持的事件（Python SDK）

• SessionStart - 会话开始
• SessionEnd - 会话结束
• Notification - 通知事件

7.2 Hook 回调函数

7.2.1 函数签名

from claude_agent_sdk.types import (
    HookInput,
    HookContext,
    HookJSONOutput
)

async def my_hook(
    input_data: HookInput,           # 事件特定的输入数据
    tool_use_id: str | None,         # 工具使用 ID（如果有）
    context: HookContext             # 上下文信息
) -> HookJSONOutput:                 # 返回值控制执行流程
    pass

7.2.2 Hook 输入类型

每个 Hook 事件有特定的输入类型：

# PreToolUse 输入
class PreToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PreToolUse"]
    tool_name: str
    tool_input: dict[str, Any]

# PostToolUse 输入  
class PostToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PostToolUse"]
    tool_name: str
    tool_input: dict[str, Any]
    tool_response: Any

# UserPromptSubmit 输入
class UserPromptSubmitHookInput(BaseHookInput):
    hook_event_name: Literal["UserPromptSubmit"]
    prompt: str

7.3 Hook 配置

7.3.1 使用 HookMatcher

from claude_agent_sdk import ClaudeAgentOptions, HookMatcher

# 定义 Hook 回调
async def check_bash_command(input_data, tool_use_id, context):
    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]
    
    if tool_name != "Bash":
        return {}  # 不处理
    
    command = tool_input.get("command", "")
    
    # 阻止危险命令
    dangerous_patterns = ["rm -rf /", "curl | sh"]
    for pattern in dangerous_patterns:
        if pattern in command:
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"Dangerous command pattern: {pattern}"
                }
            }
    
    return {}  # 允许执行

# 配置选项
options = ClaudeAgentOptions(
    allowed_tools=["Bash"],
    hooks={
        "PreToolUse": [
            HookMatcher(
                matcher="Bash",           # 匹配 Bash 工具
                hooks=[check_bash_command] # 应用到此匹配的钩子
            )
        ]
    }
)

7.3.2 匹配器模式

# 匹配特定工具
HookMatcher(matcher="Bash", hooks=[...])

# 匹配多个工具（使用 | 分隔）
HookMatcher(matcher="Write|Edit|StrReplace", hooks=[...])

# 匹配所有工具
HookMatcher(matcher=None, hooks=[...])

7.4 Hook 输出格式

7.4.1 控制字段

# 基本控制
{
    "continue_": True,           # 是否继续执行（默认 True）
    "suppressOutput": False,     # 是否抑制输出（默认 False）
    "stopReason": "..."          # 停止原因（当 continue_ 为 False 时）
}

# 决策字段
{
    "decision": "block",         # 阻止执行（仅对特定 Hook 有效）
    "systemMessage": "...",      # 显示给用户的消息
    "reason": "..."              # 给 Claude 的反馈
}

7.4.2 Hook 特定输出

# PreToolUse 特定输出
{
    "hookSpecificOutput": {
        "hookEventName": "PreToolUse",
        "permissionDecision": "allow" | "deny" | "ask",
        "permissionDecisionReason": "...",
        "updatedInput": {...}  # 修改后的输入
    }
}

# PostToolUse 特定输出
{
    "hookSpecificOutput": {
        "hookEventName": "PostToolUse",
        "additionalContext": "..."  # 额外上下文
    }
}

7.5 常见 Hook 模式

7.5.1 权限控制（PreToolUse）

async def strict_file_permissions(input_data, tool_use_id, context):
    """严格的文件权限控制"""
    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]
    
    if tool_name in ["Write", "Edit"]:
        file_path = tool_input.get("file_path", "")
        
        # 阻止写入特定目录
        blocked_dirs = ["/etc", "/usr", "/system"]
        for blocked_dir in blocked_dirs:
            if file_path.startswith(blocked_dir):
                return {
                    "reason": f"Writing to system directory {blocked_dir} is not allowed",
                    "systemMessage": f"🚫 Cannot write to system directory: {blocked_dir}",
                    "hookSpecificOutput": {
                        "hookEventName": "PreToolUse",
                        "permissionDecision": "deny",
                        "permissionDecisionReason": "System directory protection"
                    }
                }
        
        # 允许并记录
        return {
            "reason": f"File write approved: {file_path}",
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "allow",
                "permissionDecisionReason": "File passed security check"
            }
        }
    
    return {}  # 其他工具不处理

7.5.2 结果验证（PostToolUse）

async def validate_command_output(input_data, tool_use_id, context):
    """验证命令输出"""
    tool_response = input_data.get("tool_response", "")
    
    # 检查错误
    if "error" in str(tool_response).lower():
        return {
            "systemMessage": "⚠️ Command produced an error",
            "reason": "Tool execution failed - check syntax",
            "hookSpecificOutput": {
                "hookEventName": "PostToolUse",
                "additionalContext": "Consider running with --help flag for usage info"
            }
        }
    
    # 检查成功
    if "success" in str(tool_response).lower():
        return {
            "hookSpecificOutput": {
                "hookEventName": "PostToolUse", 
                "additionalContext": "✅ Command executed successfully"
            }
        }
    
    return {}

7.5.3 上下文注入（UserPromptSubmit）

async def add_system_context(input_data, tool_use_id, context):
    """注入系统上下文信息"""
    return {
        "hookSpecificOutput": {
            "hookEventName": "UserPromptSubmit",
            "additionalContext": (
                f"System info: Python 3.10, Working directory: {input_data['cwd']}, "
                f"Session: {input_data['session_id']}"
            )
        }
    }

7.5.4 执行控制（Stop Hook）

async def stop_on_critical_error(input_data, tool_use_id, context):
    """在检测到严重错误时停止执行"""
    # 这里可以检查全局状态或错误计数
    error_count = getattr(stop_on_critical_error, "error_count", 0)
    
    if error_count >= 3:
        return {
            "continue_": False,
            "stopReason": "Too many errors - stopping for safety",
            "systemMessage": "🛑 Execution halted due to repeated errors"
        }
    
    return {"continue_": True}

7.6 异步 Hook

# 异步 Hook 允许延迟执行
async def async_validation(input_data, tool_use_id, context):
    """异步验证 - 可用于外部 API 调用"""
    return {
        "async_": True,                    # 启用异步
        "asyncTimeout": 5000,              # 5 秒超时
        # Hook 将在后台执行，不阻塞主流程
    }

7.7 完整示例：多 Hook 配置

#!/usr/bin/env python3
"""完整示例：综合 Hook 配置"""

import asyncio
from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, HookMatcher
from claude_agent_sdk.types import HookInput, HookContext, HookJSONOutput

# Hook 1: 文件权限控制
async def file_security_hook(input_data: HookInput, tool_use_id: str | None, context: HookContext) -> HookJSONOutput:
    if input_data["tool_name"] in ["Write", "Edit", "Delete"]:
        file_path = input_data["tool_input"].get("file_path", "")
        
        # 阻止系统文件
        if any(file_path.startswith(path) for path in ["/etc", "/usr", "/bin"]):
            return {
                "reason": "System file protection",
                "systemMessage": f"🛡️ Blocked write to system file: {file_path}",
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": "System file protection"
                }
            }
        
        # 允许并记录
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "allow",
                "permissionDecisionReason": f"File write approved: {file_path}"
            }
        }
    return {}

# Hook 2: 命令审计
async def command_audit_hook(input_data: HookInput, tool_use_id: str | None, context: HookContext) -> HookJSONOutput:
    if input_data["tool_name"] == "Bash":
        command = input_data["tool_input"].get("command", "")
        
        # 记录所有 bash 命令（实际应用可写入日志系统）
        print(f"[AUDIT] Bash command: {command}")
        
        # 警告危险命令
        dangerous_keywords = ["curl |", "wget -O- |", "rm -rf"]
        for keyword in dangerous_keywords:
            if keyword in command:
                return {
                    "systemMessage": f"⚠️ Potentially dangerous command detected: {keyword}",
                    "hookSpecificOutput": {
                        "hookEventName": "PreToolUse",
                        "permissionDecision": "ask"  # 询问用户
                    }
                }
    
    return {}

# Hook 3: 错误处理
async def error_handler_hook(input_data: HookInput, tool_use_id: str | None, context: HookContext) -> HookJSONOutput:
    if input_data["hook_event_name"] == "PostToolUse":
        tool_response = input_data.get("tool_response", "")
        
        if isinstance(tool_response, dict) and tool_response.get("is_error"):
            error_msg = tool_response.get("content", "Unknown error")
            return {
                "systemMessage": f"❌ Tool failed: {error_msg}",
                "reason": "Tool execution encountered an error",
                "hookSpecificOutput": {
                    "hookEventName": "PostToolUse",
                    "additionalContext": "Consider retrying with different parameters"
                }
            }
    
    return {}

# Hook 4: 上下文增强
async def context_enrichment_hook(input_data: HookInput, tool_use_id: str | None, context: HookContext) -> HookJSONOutput:
    if input_data["hook_event_name"] == "UserPromptSubmit":
        prompt = input_data["prompt"]
        
        # 添加时间戳和会话信息
        import datetime
        timestamp = datetime.datetime.now().isoformat()
        
        return {
            "hookSpecificOutput": {
                "hookEventName": "UserPromptSubmit",
                "additionalContext": f"[Timestamp: {timestamp}] [Session: {input_data['session_id']}]"
            }
        }
    
    return {}

# 配置所有 hooks
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Edit", "Bash", "Delete"],
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Write|Edit|Delete", hooks=[file_security_hook]),
            HookMatcher(matcher="Bash", hooks=[command_audit_hook])
        ],
        "PostToolUse": [
            HookMatcher(matcher=None, hooks=[error_handler_hook])
        ],
        "UserPromptSubmit": [
            HookMatcher(matcher=None, hooks=[context_enrichment_hook])
        ]
    },
    system_prompt="You are a secure file assistant with comprehensive audit logging."
)

# 使用配置
async def main():
    async with ClaudeSDKClient(options=options) as client:
        await client.query("Create a test.txt file with 'Hello World'")
        async for msg in client.receive_response():
            print(f"Message: {msg}")

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

---

8. 错误处理

8.1 错误类型层次

ClaudeSDKError (基类)
├── CLIConnectionError (连接错误)
│   └── CLINotFoundError (CLI 未找到)
├── ProcessError (进程失败)
└── CLIJSONDecodeError (JSON 解析失败)

8.2 错误详情

8.2.1 CLINotFoundError

from claude_agent_sdk import CLINotFoundError

try:
    async for message in query(prompt="Hello"):
        pass
except CLINotFoundError as e:
    print(f"Claude Code CLI not found: {e}")
    print("Please install Claude Code or check cli_path")

8.2.2 ProcessError

from claude_agent_sdk import ProcessError

try:
    async for message in query(prompt="Hello"):
        pass
except ProcessError as e:
    print(f"Process failed with exit code: {e.exit_code}")
    print(f"Error output: {e.stderr}")

8.2.3 CLIJSONDecodeError

from claude_agent_sdk import CLIJSONDecodeError

try:
    async for message in query(prompt="Hello"):
        pass
except CLIJSONDecodeError as e:
    print(f"Failed to parse JSON: {e.line[:100]}...")
    print(f"Original error: {e.original_error}")

8.3 最佳实践：错误处理模式

8.3.1 全面的错误处理

import asyncio
from claude_agent_sdk import (
    query,
    ClaudeSDKClient,
    ClaudeAgentOptions,
    # 错误类型
    ClaudeSDKError,
    CLIConnectionError,
    CLINotFoundError,
    ProcessError,
    CLIJSONDecodeError
)

async def robust_query_with_error_handling():
    """带有完整错误处理的查询示例"""
    
    options = ClaudeAgentOptions(
        system_prompt="You are a helpful assistant",
        allowed_tools=["Read", "Write"],
        max_turns=3
    )
    
    try:
        async for message in query(
            prompt="Create a hello.py file",
            options=options
        ):
            # 处理消息...
            print(f"Received: {message}")
            
    except CLINotFoundError as e:
        print(f"❌ Claude Code CLI not found")
        print(f"   Error: {e}")
        print(f"   Solution: Install Claude Code or set cli_path")
        
    except CLIConnectionError as e:
        print(f"❌ Connection failed")
        print(f"   Error: {e}")
        print(f"   Solution: Check network and Claude Code status")
        
    except ProcessError as e:
        print(f"❌ Process failed")
        print(f"   Exit code: {e.exit_code}")
        print(f"   Error output: {e.stderr}")
        print(f"   Solution: Check command syntax and permissions")
        
    except CLIJSONDecodeError as e:
        print(f"❌ Failed to parse response")
        print(f"   Line: {e.line[:100]}...")
        print(f"   Original error: {e.original_error}")
        print(f"   Solution: Report this as a bug")
        
    except ClaudeSDKError as e:
        print(f"❌ SDK error: {e}")
        
    except Exception as e:
        print(f"❌ Unexpected error: {e}")
        raise

async def robust_client_with_error_handling():
    """带有完整错误处理的客户端示例"""
    
    options = ClaudeAgentOptions(
        system_prompt="You are a coding assistant",
        allowed_tools=["Read", "Write", "Bash"]
    )
    
    client = None
    try:
        client = ClaudeSDKClient(options=options)
        await client.connect()
        
        await client.query("Help me with Python")
        async for message in client.receive_response():
            print(f"Response: {message}")
            
    except CLIConnectionError as e:
        print(f"❌ Failed to connect: {e}")
        
    except Exception as e:
        print(f"❌ Error during conversation: {e}")
        
    finally:
        if client:
            try:
                await client.disconnect()
            except Exception as e:
                print(f"⚠️ Failed to disconnect: {e}")

8.3.2 重试机制

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, CLIConnectionError

async def query_with_retry(prompt: str, max_retries: int = 3, delay: float = 1.0):
    """带重试机制的查询"""
    
    options = ClaudeAgentOptions(max_turns=1)
    
    for attempt in range(max_retries):
        try:
            async for message in query(prompt=prompt, options=options):
                yield message
            return  # 成功，退出重试循环
            
        except CLIConnectionError as e:
            if attempt < max_retries - 1:
                print(f"⚠️ Connection failed (attempt {attempt + 1}), retrying in {delay}s...")
                await asyncio.sleep(delay)
                delay *= 2  # 指数退避
            else:
                print(f"❌ All {max_retries} attempts failed")
                raise
        
        except Exception:
            # 非连接错误，不重试
            raise

# 使用示例
async def main():
    async for message in query_with_retry("What is Python?"):
        print(message)

8.3.3 超时处理

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def query_with_timeout(prompt: str, timeout: float = 30.0):
    """带超时的查询"""
    
    options = ClaudeAgentOptions(max_turns=1)
    
    try:
        # 创建任务
        query_task = asyncio.create_task(
            collect_messages(query(prompt=prompt, options=options))
        )
        
        # 等待完成或超时
        messages = await asyncio.wait_for(query_task, timeout=timeout)
        return messages
        
    except asyncio.TimeoutError:
        query_task.cancel()
        print(f"❌ Query timed out after {timeout} seconds")
        raise

async def collect_messages(message_iter):
    """收集所有消息"""
    messages = []
    async for msg in message_iter:
        messages.append(msg)
    return messages

8.4 日志记录

import logging
from claude_agent_sdk import ClaudeAgentOptions

# 配置日志
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# 创建带 stderr 回调的选项
options = ClaudeAgentOptions(
    stderr=lambda line: logging.debug(f"CLI stderr: {line.strip()}")
)

# 或者使用文件
import tempfile
with tempfile.NamedTemporaryFile(mode='w+', delete=False) as f:
    options = ClaudeAgentOptions(stderr=f)

---

9. 最佳实践

9.1 代码组织

9.1.1 项目结构

my_claude_app/
├── src/
│   ├── __init__.py
│   ├── main.py
│   ├── agents/              # 代理配置
│   │   ├── __init__.py
│   │   ├── coding_agent.py
│   │   └── analysis_agent.py
│   ├── tools/               # 自定义工具
│   │   ├── __init__.py
│   │   ├── file_tools.py
│   │   └── api_tools.py
│   ├── hooks/               # Hook 实现
│   │   ├── __init__.py
│   │   ├── security_hooks.py
│   │   └── audit_hooks.py
│   └── utils/               # 工具函数
│       ├── __init__.py
│       └── config.py
├── tests/
│   ├── test_tools.py
│   ├── test_hooks.py
│   └── fixtures/
├── config/
│   ├── agents.json
│   └── permissions.yaml
├── requirements.txt
├── pyproject.toml
└── README.md

9.1.2 配置管理

# config.py
import json
from pathlib import Path
from claude_agent_sdk import ClaudeAgentOptions

class Config:
    def __init__(self, config_path: str | Path):
        with open(config_path, 'r') as f:
            self.config = json.load(f)
    
    def get_agent_options(self, agent_name: str) -> ClaudeAgentOptions:
        agent_config = self.config["agents"][agent_name]
        
        return ClaudeAgentOptions(
            system_prompt=agent_config["system_prompt"],
            allowed_tools=agent_config["allowed_tools"],
            permission_mode=agent_config.get("permission_mode", "default"),
            max_turns=agent_config.get("max_turns"),
            hooks=self._load_hooks(agent_config.get("hooks", []))
        )

9.2 性能优化

9.2.1 连接复用

# ❌ 不好：每次查询都创建新连接
for prompt in prompts:
    async for msg in query(prompt):
        pass

# ✅ 好：复用客户端连接
async with ClaudeSDKClient(options=options) as client:
    for prompt in prompts:
        await client.query(prompt)
        async for msg in client.receive_response():
            pass

9.2.2 批量处理

# ❌ 不好：串行处理
for item in items:
    result = await process_single(item)

# ✅ 好：并行处理
async def process_batch(items, concurrency=5):
    semaphore = asyncio.Semaphore(concurrency)
    
    async def process_with_semaphore(item):
        async with semaphore:
            return await process_single(item)
    
    tasks = [process_with_semaphore(item) for item in items]
    return await asyncio.gather(*tasks)

9.2.3 内存管理

# ❌ 不好：累积所有消息
messages = [msg async for msg in query(prompt)]

# ✅ 好：流式处理
async def process_large_response():
    async for message in query(prompt):
        await process_message(message)  # 立即处理
        # 不累积在内存中

9.3 安全最佳实践

9.3.1 工具权限最小化

# ❌ 不好：允许所有工具
options = ClaudeAgentOptions(
    allowed_tools=["*"]  # 危险！
)

# ✅ 好：只允许必要的工具
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write"],  # 最小权限
    disallowed_tools=["Bash"]         # 明确禁止危险工具
)

9.3.2 输入验证

async def validate_file_path(file_path: str) -> bool:
    """验证文件路径是否安全"""
    # 阻止绝对路径
    if file_path.startswith("/"):
        return False
    
    # 阻止父目录遍历
    if ".." in file_path:
        return False
    
    # 阻止敏感文件
    sensitive_files = [".env", "config.py", "secrets.txt"]
    if any(sensitive in file_path for sensitive in sensitive_files):
        return False
    
    return True

9.3.3 沙箱使用

# 生产环境始终启用沙箱
production_sandbox = {
    "enabled": True,
    "allowUnsandboxedCommands": False,  # 禁止非沙箱命令
    "excludedCommands": ["curl", "wget", "pip"],  # 排除网络命令
    "network": {
        "allowAllUnixSockets": False,     # 禁止所有 Unix socket
        "allowLocalBinding": False        # 禁止本地端口绑定
    }
}

9.4 测试

9.4.1 单元测试

import pytest
from claude_agent_sdk import create_sdk_mcp_server, tool

@pytest.mark.asyncio
async def test_calculator_tool():
    """测试计算器工具"""
    
    @tool("add", "Add numbers", {"a": int, "b": int})
    async def add(args):
        return {"content": [{"type": "text", "text": str(args["a"] + args["b"])}]}
    
    server = create_sdk_mcp_server(name="test", tools=[add])
    
    # 测试工具注册
    assert server["type"] == "sdk"
    assert server["name"] == "test"

9.4.2 集成测试

import pytest
from claude_agent_sdk import query, ClaudeAgentOptions

@pytest.mark.asyncio
async def test_query_integration():
    """测试完整查询流程"""
    
    options = ClaudeAgentOptions(
        max_turns=1,
        system_prompt="Return only 'OK'"
    )
    
    messages = []
    async for message in query(prompt="Say OK", options=options):
        messages.append(message)
    
    assert len(messages) > 0
    assert any(msg.content for msg in messages if hasattr(msg, 'content'))

9.5 监控和可观测性

9.5.1 成本监控

from claude_agent_sdk import ResultMessage

class CostTracker:
    def __init__(self):
        self.total_cost = 0.0
        self.query_count = 0
    
    def track_query(self, messages):
        for message in messages:
            if isinstance(message, ResultMessage):
                if message.total_cost_usd:
                    self.total_cost += message.total_cost_usd
                self.query_count += 1
                
                print(f"Query cost: ${message.total_cost_usd:.4f}")
                print(f"Total cost: ${self.total_cost:.4f}")
                print(f"Query count: {self.query_count}")

9.5.2 性能监控

import time
import asyncio
from claude_agent_sdk import ClaudeSDKClient

class PerformanceMonitor:
    def __init__(self):
        self.metrics = {}
    
    async def monitor_client(self, client: ClaudeSDKClient, session_name: str):
        start_time = time.time()
        message_count = 0
        
        # 包装 receive_response
        original_receive_response = client.receive_response
        
        async def monitored_receive_response():
            nonlocal message_count
            async for message in original_receive_response():
                message_count += 1
                yield message
        
        client.receive_response = monitored_receive_response
        
        # 在会话结束时记录指标
        try:
            yield client
        finally:
            duration = time.time() - start_time
            self.metrics[session_name] = {
                "duration": duration,
                "message_count": message_count,
                "messages_per_second": message_count / duration if duration > 0 else 0
            }

---

10. 常见问题

10.1 安装和配置

Q1: 安装后找不到 Claude Code CLI？

A: 从 0.1.8 版本开始，CLI 已自动捆绑。如果仍有问题：

# 显式指定 CLI 路径
options = ClaudeAgentOptions(
    cli_path="/path/to/claude"  # 或 ~/.claude/local/claude
)

Q2: 如何验证安装？

import claude_agent_sdk
print(f"SDK version: {claude_agent_sdk.__version__}")

# 运行简单查询验证
async def test_installation():
    async for message in query(prompt="Say 'Installation successful'"):
        if hasattr(message, 'content'):
            print("✅ Installation verified!")

10.2 连接问题

Q3: 连接超时或失败？

A: 检查以下几点：

1. 环境变量

   export ANTHROPIC_API_KEY="your_key_here"
   

2. 网络代理

   options = ClaudeAgentOptions(
       env={"HTTP_PROXY": "http://proxy.company.com:8080"}
   )
   

3. 增加超时

   import os
   os.environ["CLAUDE_CODE_STREAM_CLOSE_TIMEOUT"] = "120000"  # 120秒
   

Q4: 会话断开或不稳定？

A:

• 确保在同一个 async context 中使用客户端
• 避免长时间不活动（>5分钟）
• 使用心跳保持连接

10.3 工具使用

Q5: 工具未被调用？

A: 检查以下几点：

# 1. 确认工具在 allowed_tools 中
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Bash"],  # 明确指定
    permission_mode="acceptEdits"  # 或 "bypassPermissions" 测试
)

# 2. 检查工具名称拼写
# 3. 确认系统提示引导 Claude 使用工具

Q6: MCP 工具无法使用？

A:

# 检查工具命名格式
correct_format = "mcp__server_name__tool_name"

# 检查服务器配置
options = ClaudeAgentOptions(
    mcp_servers={
        "my_server": my_server  # 服务器名需匹配
    },
    allowed_tools=["mcp__my_server__my_tool"]  # 完整工具名
)

10.4 性能问题

Q7: 查询速度慢？

A:

• 使用 max_turns=1 限制轮数
• 避免在循环中重复创建客户端
• 使用流式模式处理大响应
• 检查网络连接

Q8: 内存占用高？

A:

• 流式处理消息，不要累积
• 及时关闭客户端连接
• 减少 max_buffer_size

10.5 调试技巧

启用调试输出

import logging
logging.basicConfig(level=logging.DEBUG)

# 或使用 stderr 回调
options = ClaudeAgentOptions(
    stderr=lambda line: print(f"CLI: {line.strip()}")
)

检查消息流

async def debug_message_flow():
    async for message in query(prompt="Hello"):
        print(f"Type: {type(message).__name__}")
        print(f"Content: {getattr(message, 'content', 'N/A')}")
        print("-" * 40)

10.6 版本兼容性

Q9: 如何迁移旧代码？

A: 查看 CHANGELOG.md 了解重大变更：

# 0.1.0 之前版本
from claude_agent_sdk import ClaudeCodeOptions  # ❌ 已弃用

# 0.1.0+ 版本
from claude_agent_sdk import ClaudeAgentOptions   # ✅ 新名称

Q10: 不同 Python 版本的兼容性？

A: SDK 支持 Python 3.10+，使用 typing_extensions 提供向后兼容：

# SDK 内部自动处理版本差异
# 用户代码无需特殊处理

---

附录

A. 参考资源

• 官方文档
• GitHub 仓库
• PyPI 包
• Claude Code 文档

B. 术语表

术语	说明
MCP	Model Context Protocol - Claude 的插件协议
SDK	Software Development Kit - 软件开发工具包
CLI	Command Line Interface - 命令行界面
Hook	钩子 - 在特定事件点执行的回调函数
Transport	传输层 - 与 CLI 通信的底层机制
Query	查询 - 一次性交互模式
Streaming	流式 - 支持实时双向通信的模式