Claude Agent SDK Python - 专业教学文档

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 核心特性

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 环境变量

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 asyncdefmain():asyncfor message in query(prompt="What is 2 + 2?"):print(message) anyio.run(main)

带选项的查询

from claude_agent_sdk import( query, ClaudeAgentOptions, AssistantMessage, TextBlock )asyncdefmain(): options = ClaudeAgentOptions( system_prompt="You are a helpful assistant", max_turns=1)asyncfor message in query( prompt="Tell me a joke", options=options ):ifisinstance(message, AssistantMessage):for block in message.content:ifisinstance(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 )asyncdefparse_messages(): options = ClaudeAgentOptions( allowed_tools=["Read","Write"], system_prompt="You are a file assistant")asyncfor 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 可用工具

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 asyncdefinteractive_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")asyncfor message in client.receive_response():print(message)await client.disconnect()# 方式 2: 使用上下文管理器（推荐）asyncwith ClaudeSDKClient(options=options)as client:await client.query("List files in current directory")asyncfor message in client.receive_response():print(message)

5.1.3 消息接收模式

# 模式 1: 接收所有消息（无限）asyncfor message in client.receive_messages():print(message)# 需要手动中断或等待会话结束# 模式 2: 接收单个响应（到 ResultMessage 为止）asyncfor 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 asyncdefstreaming_prompts():# 定义异步生成器asyncdefprompt_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"}}# 使用异步迭代器作为 promptasyncfor 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 )asyncdefcustom_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 模式)asyncwith ClaudeSDKClient(options=options)as client:await client.query("Create a test file")# 工具使用将经过自定义权限检查

5.3.2 权限结果类型

@dataclassclassPermissionResultAllow: behavior: Literal["allow"]="allow" updated_input:dict[str, Any]|None=None# 修改后的输入 updated_permissions:list[PermissionUpdate]|None=None# 权限更新@dataclassclassPermissionResultDeny: 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)asyncwith ClaudeSDKClient(options=options)as client:# 执行文件操作await client.query("Create file1.txt with content A")# 获取检查点 ID checkpoint_id =Noneasyncfor msg in client.receive_response():ifhasattr(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 中获取验证后的 JSONasyncfor message in query( prompt="Extract person info from: John Doe, 30 years old, skills: Python, AI", options=options ):ifisinstance(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 服务器的优势

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})asyncdefadd_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})asyncdefdivide_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}"}]}# 定义工具 - 访问应用状态classDataStore:def__init__(self): self.items =[] self.counter =0 store = DataStore()@tool("add_item","Add item to store",{"item":str})asyncdefadd_item(args:dict[str, Any])->dict[str, Any]:"""Add item to application state.""" store.items.append(args["item"]) store.counter +=1return{"content":[{"type":"text","text":f"Added: {args['item']} (Total: {store.counter})"}]}@tool("list_items","List all items in store",{})asyncdeflist_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})asyncdefgreet(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"]})asyncdefprocess_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})asyncdefcalculate(args):"""安全地计算数学表达式""" expression = args["expression"]# 安全验证：只允许数字、运算符和括号 allowed_chars =set("0123456789+-*/.() ")ifnotall(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",{})asyncdefget_history(args):"""获取计算历史"""# 这里可以连接到持久化存储 history =getattr(calculate,"history",[])ifnot history:return{"content":[{"type":"text","text":"No calculations yet"}]} history_text ="\\n".join(f"{i+1}. {expr}"for i, expr inenumerate(history))return{"content":[{"type":"text","text":f"History:\\n{history_text}"}]}# 创建服务器asyncdefmain(): 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.")asyncwith ClaudeSDKClient(options=options)as client:# 测试 1: 简单计算await client.query("Calculate 15 * 7 + 3")asyncfor msg in client.receive_response():print(f"Response: {msg}")# 测试 2: 获取历史await client.query("Show me the calculation history")asyncfor 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 事件

7.1.2 不支持的事件（Python SDK）

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

7.2 Hook 回调函数

7.2.1 函数签名

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

7.2.2 Hook 输入类型

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

# PreToolUse 输入classPreToolUseHookInput(BaseHookInput): hook_event_name: Literal["PreToolUse"] tool_name:str tool_input:dict[str, Any]# PostToolUse 输入 classPostToolUseHookInput(BaseHookInput): hook_event_name: Literal["PostToolUse"] tool_name:str tool_input:dict[str, Any] tool_response: Any # UserPromptSubmit 输入classUserPromptSubmitHookInput(BaseHookInput): hook_event_name: Literal["UserPromptSubmit"] prompt:str

7.3 Hook 配置

7.3.1 使用 HookMatcher

from claude_agent_sdk import ClaudeAgentOptions, HookMatcher # 定义 Hook 回调asyncdefcheck_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）

asyncdefstrict_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）

asyncdefvalidate_command_output(input_data, tool_use_id, context):"""验证命令输出""" tool_response = input_data.get("tool_response","")# 检查错误if"error"instr(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"instr(tool_response).lower():return{"hookSpecificOutput":{"hookEventName":"PostToolUse","additionalContext":"✅ Command executed successfully"}}return{}

7.5.3 上下文注入（UserPromptSubmit）

asyncdefadd_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）

asyncdefstop_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 允许延迟执行asyncdefasync_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: 文件权限控制asyncdeffile_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","")# 阻止系统文件ifany(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: 命令审计asyncdefcommand_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: 错误处理asyncdeferror_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","")ifisinstance(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: 上下文增强asyncdefcontext_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.")# 使用配置asyncdefmain():asyncwith ClaudeSDKClient(options=options)as client:await client.query("Create a test.txt file with 'Hello World'")asyncfor 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:asyncfor message in query(prompt="Hello"):passexcept 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:asyncfor message in query(prompt="Hello"):passexcept 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:asyncfor message in query(prompt="Hello"):passexcept 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 )asyncdefrobust_query_with_error_handling():"""带有完整错误处理的查询示例""" options = ClaudeAgentOptions( system_prompt="You are a helpful assistant", allowed_tools=["Read","Write"], max_turns=3)try:asyncfor 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}")raiseasyncdefrobust_client_with_error_handling():"""带有完整错误处理的客户端示例""" options = ClaudeAgentOptions( system_prompt="You are a coding assistant", allowed_tools=["Read","Write","Bash"]) client =Nonetry: client = ClaudeSDKClient(options=options)await client.connect()await client.query("Help me with Python")asyncfor 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 asyncdefquery_with_retry(prompt:str, max_retries:int=3, delay:float=1.0):"""带重试机制的查询""" options = ClaudeAgentOptions(max_turns=1)for attempt inrange(max_retries):try:asyncfor 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")raiseexcept Exception:# 非连接错误，不重试raise# 使用示例asyncdefmain():asyncfor message in query_with_retry("What is Python?"):print(message)

8.3.3 超时处理

import asyncio from claude_agent_sdk import query, ClaudeAgentOptions asyncdefquery_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")raiseasyncdefcollect_messages(message_iter):"""收集所有消息""" messages =[]asyncfor 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.pyimport json from pathlib import Path from claude_agent_sdk import ClaudeAgentOptions classConfig:def__init__(self, config_path:str| Path):withopen(config_path,'r')as f: self.config = json.load(f)defget_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:asyncfor msg in query(prompt):pass# ✅ 好：复用客户端连接asyncwith ClaudeSDKClient(options=options)as client:for prompt in prompts:await client.query(prompt)asyncfor msg in client.receive_response():pass

9.2.2 批量处理

# ❌ 不好：串行处理for item in items: result =await process_single(item)# ✅ 好：并行处理asyncdefprocess_batch(items, concurrency=5): semaphore = asyncio.Semaphore(concurrency)asyncdefprocess_with_semaphore(item):asyncwith semaphore:returnawait process_single(item) tasks =[process_with_semaphore(item)for item in items]returnawait asyncio.gather(*tasks)

9.2.3 内存管理

# ❌ 不好：累积所有消息 messages =[msg asyncfor msg in query(prompt)]# ✅ 好：流式处理asyncdefprocess_large_response():asyncfor 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 输入验证

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

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.asyncioasyncdeftest_calculator_tool():"""测试计算器工具"""@tool("add","Add numbers",{"a":int,"b":int})asyncdefadd(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.asyncioasyncdeftest_query_integration():"""测试完整查询流程""" options = ClaudeAgentOptions( max_turns=1, system_prompt="Return only 'OK'") messages =[]asyncfor message in query(prompt="Say OK", options=options): messages.append(message)assertlen(messages)>0assertany(msg.content for msg in messages ifhasattr(msg,'content'))

9.5 监控和可观测性

9.5.1 成本监控

from claude_agent_sdk import ResultMessage classCostTracker:def__init__(self): self.total_cost =0.0 self.query_count =0deftrack_query(self, messages):for message in messages:ifisinstance(message, ResultMessage):if message.total_cost_usd: self.total_cost += message.total_cost_usd self.query_count +=1print(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 classPerformanceMonitor:def__init__(self): self.metrics ={}asyncdefmonitor_client(self, client: ClaudeSDKClient, session_name:str): start_time = time.time() message_count =0# 包装 receive_response original_receive_response = client.receive_response asyncdefmonitored_receive_response():nonlocal message_count asyncfor message in original_receive_response(): message_count +=1yield 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 >0else0}

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__}")# 运行简单查询验证asyncdeftest_installation():asyncfor message in query(prompt="Say 'Installation successful'"):ifhasattr(message,'content'):print("✅ Installation verified!")

10.2 连接问题

Q3: 连接超时或失败？

A: 检查以下几点：

增加超时

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

网络代理

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

环境变量

exportANTHROPIC_API_KEY="your_key_here"

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()}"))

检查消息流

asyncdefdebug_message_flow():asyncfor 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. 术语表