Claude Agent SDK Python 技术文档

Claude Agent SDK Python 技术文档 | 极客日志

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

# Python 版本要求 (推荐 3.10+) python --version# >= 3.10# 操作系统支持# - macOS (Intel & Apple Silicon)# - Linux (x86_64, ARM64)# - Windows (WSL2 推荐)

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

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

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

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

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

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

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)

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

from claude_agent_sdk import( TextBlock,# 文本内容 ThinkingBlock,# 思考过程 ToolUseBlock,# 工具使用 ToolResultBlock # 工具结果)# 内容块层次结构 AssistantMessage.content → List[ContentBlock] ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

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

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

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=[]# 空列表表示禁用所有工具)

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

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

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)

# 模式 1: 接收所有消息（无限）asyncfor message in client.receive_messages():print(message)# 需要手动中断或等待会话结束
# 模式 2: 接收单个响应（到 ResultMessage 为止）asyncfor message in client.receive_response():print(message)# 自动在 ResultMessage 后停止

# 中断当前操作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',[]))}")

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)

{
"type":"user",
"message":{
"role":"user",
"content":"Hello, Claude!"
},
"parent_tool_use_id":None,
"session_id":"default"
}

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")# 工具使用将经过自定义权限检查

@dataclass
classPermissionResultAllow:
 behavior: Literal["allow"]="allow"
 updated_input:dict[str, Any]|None=None# 修改后的输入 updated_permissions:list[PermissionUpdate]|None=None# 权限更新
@dataclass
classPermissionResultDeny:
 behavior: Literal["deny"]="deny"
 message:str=""# 拒绝原因 interrupt:bool=False# 是否中断执行

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

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

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)

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

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}"}]}

# 简单字典模式@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

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"])

# 创建多个服务器 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"])

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

#!/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())

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

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

# 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

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]# 应用到此匹配的钩子)]})

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

# 基本控制{"continue_":True,# 是否继续执行（默认 True）"suppressOutput":False,# 是否抑制输出（默认 False）"stopReason":"..."# 停止原因（当 continue_ 为 False 时）}
# 决策字段{"decision":"block",# 阻止执行（仅对特定 Hook 有效）"systemMessage":"...",# 显示给用户的消息"reason":"..."# 给 Claude 的反馈}

# PreToolUse 特定输出{"hookSpecificOutput":{"hookEventName":"PreToolUse","permissionDecision":"allow"|"deny"|"ask","permissionDecisionReason":"...","updatedInput":{...}# 修改后的输入}}
# PostToolUse 特定输出{"hookSpecificOutput":{"hookEventName":"PostToolUse","additionalContext":"..."# 额外上下文}}

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{}
# 其他工具不处理

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{}

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']}")}}

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}

# 异步 Hook 允许延迟执行asyncdefasync_validation(input_data, tool_use_id, context):
"""异步验证 - 可用于外部 API 调用"""
return{"async_":True,# 启用异步"asyncTimeout":5000,# 5 秒超时# 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())

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

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

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

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

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

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)

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

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)

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

# 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",[])))

# ❌ 不好：每次查询都创建新连接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

# ❌ 不好：串行处理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)

# ❌ 不好：累积所有消息 messages =[msg asyncfor msg in query(prompt)]
# ✅ 好：流式处理asyncdefprocess_large_response():asyncfor message in query(prompt):await process_message(message)# 立即处理# 不累积在内存中

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

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

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

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"

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'))

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

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}

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

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!")

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

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

export ANTHROPIC_API_KEY="your_key_here"

# 1. 确认工具在 allowed_tools 中 options = ClaudeAgentOptions( allowed_tools=["Read","Write","Bash"],# 明确指定 permission_mode="acceptEdits"# 或 "bypassPermissions" 测试)
# 2. 检查工具名称拼写
# 3. 确认系统提示引导 Claude 使用工具

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

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)

# 0.1.0 之前版本from claude_agent_sdk import ClaudeCodeOptions # ❌ 已弃用
# 0.1.0+ 版本from claude_agent_sdk import ClaudeAgentOptions # ✅ 新名称

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

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

Claude Agent SDK Python 技术文档

1. 概述

1.1 什么是 Claude Agent SDK?

1.2 核心特性

1.3 应用场景

2. 安装与环境配置

2.1 系统要求

2.2 安装方式

通过 pip 安装（推荐）

开发环境安装

2.3 Claude Code CLI

2.4 环境变量

3. 核心架构

3.1 架构概览

3.2 核心组件

3.2.1 Transport 层

3.2.2 Query 引擎

3.2.3 Message Parser

3.3 数据流

4. 基础使用

4.1 快速开始

最简示例

带选项的查询

4.2 消息类型系统

4.2.1 消息类型概览

4.2.2 内容块类型

4.2.3 完整示例：消息解析

4.3 工具使用

4.3.1 可用工具

4.3.2 工具配置

4.3.3 权限模式

4.4 工作目录配置

5. 高级功能

5.1 ClaudeSDKClient - 交互式客户端

5.1.1 何时使用 ClaudeSDKClient

5.1.2 基本使用模式

5.1.3 消息接收模式

5.1.4 会话控制

5.2 流式模式

5.2.1 异步迭代器模式

5.2.2 消息格式

5.3 工具权限回调

5.3.1 自定义权限逻辑

5.3.2 权限结果类型

5.4 文件检查点与回滚

5.4.1 启用文件检查点

5.5 结构化输出

5.5.1 JSON Schema 验证

5.6 沙箱配置

5.6.1 启用沙箱

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

6.1 什么是 MCP?

6.2 SDK MCP 服务器的优势

6.3 创建自定义工具

6.3.1 使用 @tool 装饰器

6.3.2 输入模式类型

6.4 创建 MCP 服务器

6.4.1 基本服务器创建

6.4.2 多个服务器

6.5 混合服务器配置

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

7. Hooks 系统

7.1 Hooks 概览

7.1.1 支持的 Hook 事件

7.1.2 不支持的事件（Python SDK）

7.2 Hook 回调函数

7.2.1 函数签名

7.2.2 Hook 输入类型

7.3 Hook 配置

7.3.1 使用 HookMatcher

7.3.2 匹配器模式

7.4 Hook 输出格式

7.4.1 控制字段

7.4.2 Hook 特定输出

7.5 常见 Hook 模式

7.5.1 权限控制（PreToolUse）

7.5.2 结果验证（PostToolUse）

7.5.3 上下文注入（UserPromptSubmit）

7.5.4 执行控制（Stop Hook）

7.6 异步 Hook