综述由AI生成档详细介绍了 Claude Agent SDK for Python 的使用方法。内容包括安装配置、核心架构（查询模式与客户端模式）、基础使用（消息类型、工具权限）、高级功能（MCP 服务器、Hooks 系统、沙箱）、错误处理及最佳实践。SDK 支持异步编程、结构化输出及自定义工具集成，适用于自动化脚本、开发工具及交互式应用开发。提供了详细的代码示例和常见问题解答。
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)
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)
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 服务器：
外部 MCP 服务器: 独立进程，通过 stdio/SSE/HTTP 通信
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 importAny# 定义工具 - 基本计算器@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}"}]}
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)
# ❌ 不好：每次查询都创建新连接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)# 立即处理# 不累积在内存中
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
