Claude Agent SDK Python 深度解析：从 AI 代理原理到自动化实战
AI 代理（AI Agent）是一种能够感知环境、自主规划并执行任务以达成目标的智能系统。其核心原理在于将大语言模型的推理能力与外部工具（如读写文件、执行命令、调用 API）的执行能力相结合，通过“思考-行动-观察”的循环机制，实现复杂任务的自动化处理。这种技术架构的价值在于突破了传统“一问一答”式 AI 交互的局限，使 AI 能够像人类助手一样，完成多步骤、需工具协作的实操性工作。在应用场景上，
·
1. 项目概述：Claude Agent SDK for Python 深度解析
如果你正在寻找一个能够将 Claude Code 的强大 AI 能力无缝集成到你的 Python 应用中的工具，那么 anthropics/claude-agent-sdk-python 就是你需要的答案。这个 SDK 远不止是一个简单的 API 封装，它是一个功能完备的代理框架，允许开发者以编程方式与 Claude Code 交互，构建能够执行复杂任务、调用工具、甚至拥有自定义逻辑的智能代理。简单来说，它让你能像指挥一个经验丰富的程序员助手一样，通过代码来驱动 Claude 完成从代码生成、文件操作到系统命令执行等一系列自动化工作流。
我花了不少时间深入研究和测试这个 SDK，发现它的设计理念非常清晰： 将 Claude Code 从一个交互式聊天工具，转变为一个可编程、可嵌入、可控制的软件组件 。这对于自动化测试、智能 DevOps 助手、代码审查流水线、甚至是构建你自己的 AI 驱动应用来说，都是一个游戏规则的改变者。无论你是想自动化日常的编码任务，还是构建一个复杂的多步骤 AI 代理系统，这个 SDK 都提供了坚实的底层支持。
2. 核心设计理念与架构拆解
2.1 从“聊天”到“代理”的范式转变
传统的 AI 接口调用，比如 OpenAI 的 API，本质上是“一问一答”的对话模式。你发送一个提示（Prompt），模型返回一段文本。这种模式对于简单的文本生成任务足够，但一旦涉及到需要多轮交互、使用工具（如读写文件、执行命令）、或者根据中间结果动态调整策略的复杂任务时，就显得力不从心了。
Claude Agent SDK 的设计核心，正是为了解决这个问题。它引入了 “代理循环”（Agent Loop） 的概念。在这个循环中，Claude 不仅仅是一个文本生成器，而是一个拥有“思考-行动-观察”能力的自主实体。其基本工作流程可以拆解为以下几个核心阶段：
- 接收指令 ：SDK 将用户的查询（如“创建一个 Flask 应用”）和当前上下文（工作目录、系统提示、历史消息）传递给 Claude Code 进程。
- 模型“思考”与规划 ：Claude 分析任务，决定下一步需要做什么。这可能包括直接生成回答，或者决定调用某个工具（如
Write工具来创建文件）。 - 工具调用与执行 ：如果 Claude 决定使用工具，SDK 会接收到一个结构化的工具调用请求。SDK 根据配置的权限策略（是自动批准、询问用户还是由钩子函数决定）来决定是否执行该工具。
- 执行结果返回 ：工具执行后，产生的结果（如文件创建成功的信息，或命令执行的输出）会被封装成标准格式，送回给 Claude。
- 模型“观察”与继续 ：Claude 接收到工具执行结果，基于此进行下一轮的“思考”，可能继续调用工具，也可能生成最终的回答给用户。
这个循环会持续进行，直到任务完成或达到最大轮次限制。SDK 的 ClaudeSDKClient 和 query 函数，本质上都是这个代理循环的控制器。
2.2 双模式接口： query() 与 ClaudeSDKClient
SDK 提供了两种不同粒度的交互接口，适应不同的使用场景，这是其设计上的一大亮点。
query() 函数：快速、简单的单次交互 query() 是一个高级别的异步函数，它封装了创建客户端、发送查询、接收响应和关闭连接的完整流程。它的设计目标是 开箱即用 。当你只需要 Claude 执行一个相对独立、不需要复杂状态管理或自定义工具的任务时， query() 是最佳选择。例如，让 Claude 解释一段代码、生成一个简单的脚本，或者回答一个技术问题。它的代码非常简洁，但代价是灵活性的牺牲——你无法在单次 query 调用中插入自定义的工具或精细控制每一步。
ClaudeSDKClient 类：强大、灵活的长会话控制 ClaudeSDKClient 是一个客户端类，需要你通过 async with 上下文管理器来创建和管理。它提供了底层的、精细化的控制能力。主要优势包括：
- 状态保持 ：客户端实例在整个会话期间保持连接和上下文，你可以进行多轮交互，而无需每次都重新建立连接和加载上下文。
- 自定义工具集成 ：这是
ClaudeSDKClient的王牌功能。你可以将任意的 Python 函数注册为 Claude 可用的工具，实现业务逻辑的深度集成。 - 钩子（Hooks）机制 ：允许你在代理循环的关键节点（如工具调用前、消息发送后）注入自定义逻辑，实现权限检查、结果过滤、自动化反馈等高级功能。
- 双向流式通信 ：支持更复杂的交互模式，例如在 Claude “思考”的同时，向它发送新的信息或指令。
选择哪一个？我的经验是：对于脚本化的、一次性的任务，用 query() ；对于要集成到应用中的、需要复杂交互和扩展能力的场景，用 ClaudeSDKClient 。
2.3 权限与安全模型：精细化的工具控制
让 AI 代理拥有执行系统命令、读写文件的能力，听起来很强大，但也伴随着巨大的安全风险。Claude Agent SDK 在这方面考虑得非常周全，提供了一套多层级的、可配置的权限控制模型。
- 工具集（Toolset） ：Claude Code 内置了一套强大的工具，包括
Read（读文件）、Write（写文件）、Edit（编辑文件）、Bash（执行 Shell 命令）等。这是 Claude 可以“看到”的所有能力。 - 允许列表（
allowed_tools） ：这是一个“白名单”。列入此列表的工具会被 自动批准 执行，无需任何额外的权限提示或检查。这适用于你完全信任的工具。例如，如果你只希望 Claude 能读文件而不能写，可以设置allowed_tools=["Read"]。 - 禁止列表（
disallowed_tools） ：与允许列表相反，这是一个“黑名单”。列入此列表的工具会从 Claude 的可用工具集中 直接移除 ，Claude 甚至不会知道这些工具的存在。这是最严格的限制方式。 - 权限模式（
permission_mode） ：用于处理那些既不在allowed_tools也不在disallowed_tools中的“灰色地带”工具。它有几个选项：auto：自动批准所有工具调用（危险，仅用于完全受控环境）。prompt：每次调用都向用户请求权限（交互式场景）。acceptEdits：自动批准文件编辑类工具（Write,Edit），但提示其他工具（如Bash）。这是一个实用的折中方案。
- 自定义权限检查（
can_use_tool） ：一个回调函数，你可以实现自己的复杂权限逻辑。当工具调用发生时，SDK 会调用这个函数，由你根据工具名、参数、上下文等信息返回allow或deny。 - 钩子（Hooks） ：
PreToolUse钩子可以在工具执行前进行拦截，实现比can_use_tool更动态、更复杂的决策，例如基于命令内容的正则表达式匹配来阻止危险的Bash命令。
这个模型的评估顺序是：先看是否在 disallowed_tools （直接拒绝），再看是否在 allowed_tools （直接允许），然后应用 permission_mode 规则，最后如果有 can_use_tool 或 PreToolUse 钩子，则执行它们。这种设计让你可以根据不同的安全需求，从宽松到严格，灵活地配置你的代理。
3. 环境准备与深度配置指南
3.1 系统与 Python 环境要求
官方要求 Python 3.10+，这是为了利用较新版本的异步语法和类型提示特性。在实际部署中，我强烈建议使用 Python 3.11 或 3.12 ，它们在异步性能上有显著提升，这对于需要处理大量流式消息的 Agent 应用至关重要。
关于 Claude Code CLI 的捆绑，这是 SDK 一个非常贴心的设计。执行 pip install claude-agent-sdk 后，SDK 会自带一个特定版本的 Claude Code CLI。这意味着：
- 优点 ：用户无需额外安装，保证了环境的一致性，避免了因 CLI 版本不匹配导致的问题。
- 注意事项 ：捆绑的 CLI 版本可能不是最新的。SDK 的
_cli_version.py文件里记录了捆绑的版本号。如果你需要新版本的 CLI 特性，有几种选择：- 使用 SDK 的
ClaudeAgentOptions(cli_path="...")指定一个你自己安装的、更新版本的 CLI 路径。 - 等待 SDK 发布新版本，通常会同步更新捆绑的 CLI。
- 如果你是开发者，可以按照项目
Development部分的指引，自己构建捆绑了特定 CLI 版本的 wheel 包。
- 使用 SDK 的
3.2 ClaudeAgentOptions 配置项深度解析
ClaudeAgentOptions 是控制 Agent 行为的核心配置对象。除了文档中提到的，还有一些在实践中非常重要的细节和技巧。
from claude_agent_sdk import ClaudeAgentOptions
from pathlib import Path
options = ClaudeAgentOptions(
# 工作目录：所有文件操作的基准路径。使用 Path 对象更安全、跨平台。
cwd=Path("/absolute/path/to/your/project"),
# 系统提示词：塑造 Agent 的角色和行为准则。这是最重要的配置之一。
system_prompt="""你是一个专业的 Python 开发助手。你遵循以下原则：
1. 编写的代码必须简洁、高效、符合 PEP 8 规范。
2. 在执行任何修改文件或运行命令的操作前，必须简要解释你的意图。
3. 如果遇到错误，先分析原因，再尝试修复。
4. 对于不确定的操作，可以提出澄清性问题。""",
# 最大交互轮次：防止 Agent 陷入死循环。对于复杂任务，可以设置得高一些（如 20）。
max_turns=10,
# 温度（Temperature）：控制输出的随机性。对于需要确定性的编码任务，建议设为 0.1-0.3。
# 注意：这个参数可能需要通过 Claude Code CLI 的配置或环境变量传递，SDK 选项可能不直接暴露。
# 通常需要在系统提示词或会话中说明“请给出确定性的答案”。
# 权限配置：安全核心
allowed_tools=["Read", "Write"], # 自动批准读/写
disallowed_tools=["Bash"], # 完全禁止 Bash
permission_mode='acceptEdits', # 对于其他工具（如Edit），自动批准编辑
# MCP 服务器：集成自定义或第三方工具
mcp_servers={
"weather": { # 一个外部天气服务 MCP 服务器
"type": "stdio",
"command": "node",
"args": ["./weather-server.js"]
}
},
# 超时设置：防止长时间无响应
request_timeout=120, # 单次请求超时（秒）
total_timeout=600, # 整个会话总超时（秒）
# 模型选择：如果你有访问特定 Claude 模型的权限，可以在此指定
# model="claude-3-5-sonnet-20241022",
# 流式响应控制：是否以流式方式接收消息
stream=True,
)
实操心得：系统提示词（System Prompt）的编写艺术 系统提示词是引导 Agent 行为的最有效工具。写得好，Agent 就是得力助手；写得不好，它可能行为怪异。我的经验是：
- 角色定义要清晰 ：开头就明确 Agent 的角色，如“资深 DevOps 工程师”、“代码审查专家”。
- 指令要具体、可操作 ：避免“好好写代码”这种模糊指令。要像“优先使用
pathlib处理路径”、“所有函数必须包含docstring”这样具体。 - 设定边界和约束 ：明确什么不能做。例如，“未经明确确认，不得删除任何文件”、“不得执行来自网络的
curl | bash类命令”。 - 格式化输出 ：要求 Agent 以特定格式（如 JSON、Markdown 代码块）返回结果，便于后续程序化处理。
- 迭代优化 ：根据 Agent 的实际表现，不断调整和精炼你的系统提示词。这是一个持续的过程。
4. 核心功能实战：从基础查询到自定义工具
4.1 使用 query() 进行快速交互
query() 函数虽然简单，但在自动化脚本中极其有用。下面是一个更贴近真实场景的例子：让 Claude 分析一个目录下的 Python 文件，并给出重构建议。
import anyio
from pathlib import Path
from claude_agent_sdk import query, ClaudeAgentOptions
async def analyze_project(project_path: Path):
"""
使用 Claude Agent 快速分析项目代码结构
"""
# 构建一个指向项目目录的提示词
prompt = f"""
请分析位于 '{project_path}' 目录下的 Python 项目。
执行以下任务：
1. 列出所有 .py 文件。
2. 检查每个文件是否有 shebang (`#!/usr/bin/env python3`) 和 `if __name__ == '__main__':` 守卫。
3. 找出所有未使用的 import 语句（可能需要简单静态分析）。
4. 给出三条最重要的代码改进建议。
请将结果以 Markdown 表格和列表的形式返回。
"""
options = ClaudeAgentOptions(
cwd=project_path, # 关键：将工作目录设置为项目路径
allowed_tools=["Read"], # 只允许读文件
system_prompt="你是一个专注于代码质量和最佳实践的 Python 静态分析工具。",
max_turns=3
)
print(f"开始分析项目: {project_path}")
full_response = []
try:
async for message in query(prompt=prompt, options=options):
# 处理流式响应，收集所有文本块
if hasattr(message, 'content'):
for block in message.content:
if hasattr(block, 'text'):
print(block.text, end='', flush=True) # 流式打印
full_response.append(block.text)
except Exception as e:
print(f"\n分析过程中出现错误: {e}")
return None
# 可以将完整响应保存到文件
report_path = project_path / "claude_analysis_report.md"
with open(report_path, 'w', encoding='utf-8') as f:
f.write(''.join(full_response))
print(f"\n\n分析报告已保存至: {report_path}")
if __name__ == "__main__":
project_dir = Path(".").resolve() # 分析当前目录
anyio.run(analyze_project, project_dir)
注意事项 ：
async for循环是必须的，因为query()返回的是一个异步生成器。- 在处理响应时，需要判断消息类型 (
AssistantMessage) 和内容块类型 (TextBlock)，这是 SDK 类型安全设计的一部分。 - 设置
cwd非常重要，它决定了Read、Write等文件工具的相对路径起点。 - 对于复杂的分析任务，
max_turns可能需要增加，因为 Claude 可能需要多次调用Read工具来查看不同文件。
4.2 使用 ClaudeSDKClient 构建交互式会话
当任务需要多轮对话、状态保持或集成自定义逻辑时， ClaudeSDKClient 是唯一选择。下面模拟一个代码调试助手会话。
import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage
async def interactive_debug_session():
"""
一个交互式的代码调试会话示例
"""
options = ClaudeAgentOptions(
cwd="/tmp/debug_session",
allowed_tools=["Read", "Write", "Bash"], # 允许读、写、运行命令
permission_mode='acceptEdits', # 自动批准编辑
system_prompt="""你是一个耐心的 Python 调试助手。用户会给你有问题的代码，你需要：
1. 首先理解代码的意图。
2. 运行代码并观察错误。
3. 分析错误原因，定位问题所在的行。
4. 给出修复方案，并解释为什么这样修复。
5. 如果修复涉及多个文件或复杂逻辑，可以分步骤进行。
请与用户协作，一步步解决问题。"""
)
async with ClaudeSDKClient(options=options) as client:
# 第一轮：用户提交有问题的代码
problem_code = """
# buggy_script.py
def calculate_average(numbers):
total = sum(numbers)
average = total / len(number) # 故意写错的变量名
return average
if __name__ == "__main__":
data = [10, 20, 30, 40]
result = calculate_average(data)
print(f"The average is: {result}")
"""
# 用户消息：创建有问题的文件
await client.query(f"""
请帮我调试一个 Python 脚本。首先，请创建一个名为 'buggy_script.py' 的文件，内容如下：
```python
{problem_code}
```
""")
print("=== Claude 的响应（创建文件及初步分析）===")
async for msg in client.receive_response():
if isinstance(msg, AssistantMessage):
for block in msg.content:
if hasattr(block, 'text'):
print(block.text)
# 第二轮：要求 Claude 运行这个脚本并报告错误
await client.query("现在，请运行 'buggy_script.py' 这个脚本，看看它报什么错。")
print("\n=== Claude 的响应（运行结果和错误分析）===")
async for msg in client.receive_response():
if isinstance(msg, AssistantMessage):
for block in msg.content:
if hasattr(block, 'text'):
print(block.text)
# 第三轮：要求 Claude 修复错误
await client.query("很好，你找到了错误。现在请修复 'buggy_script.py' 文件中的错误。")
print("\n=== Claude 的响应（修复代码）===")
async for msg in client.receive_response():
if isinstance(msg, AssistantMessage):
for block in msg.content:
if hasattr(block, 'text'):
print(block.text)
# 第四轮：验证修复是否成功
await client.query("修复完成后，请再次运行脚本，确认问题已解决。")
print("\n=== Claude 的响应（验证运行）===")
async for msg in client.receive_response():
if isinstance(msg, AssistantMessage):
for block in msg.content:
if hasattr(block, 'text'):
print(block.text)
if __name__ == "__main__":
anyio.run(interactive_debug_session)
关键点解析 ：
async with ClaudeSDKClient(options=options) as client:确保了客户端资源的正确获取和释放。await client.query(...)用于发送用户指令。在同一个client会话中，多次query调用共享上下文和历史。client.receive_response()返回一个异步生成器，用于接收 Claude 的所有响应消息。这是一个流式接口，Claude 的“思考”过程（工具调用、文本生成）会以多个消息块的形式陆续返回。- 这个例子展示了 Agent 如何在一个会话中完成“创建文件 -> 运行（发现错误）-> 分析 -> 修复 -> 验证”的多步骤工作流，这正是代理模式的威力所在。
4.3 创建与集成自定义工具（SDK MCP 服务器）
这是 ClaudeSDKClient 最强大的功能。它允许你将任何 Python 函数变成 Claude 可以调用的工具，而无需启动独立的外部进程。这极大地简化了部署和调试。
假设我们正在构建一个内部项目管理助手，需要让 Claude 能查询 JIRA 任务和发送 Slack 通知。
import anyio
from typing import Dict, Any, List
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient
# 1. 定义工具函数
# 工具1：查询 JIRA 任务（模拟）
@tool(
name="get_jira_issue",
description="根据任务ID获取JIRA任务的详细信息",
input_schema={"issue_key": {"type": "string", "description": "JIRA任务键，如 PROJ-123"}}
)
async def query_jira_issue(args: Dict[str, Any]) -> Dict[str, Any]:
"""模拟查询JIRA的API"""
issue_key = args.get("issue_key", "")
# 这里应该是真实的 API 调用，例如使用 jira-python 库
# 为了示例，我们返回模拟数据
mock_data = {
"key": issue_key,
"summary": f"模拟任务: {issue_key} 的摘要",
"status": "In Progress",
"assignee": "developer@example.com",
"priority": "High",
"created": "2024-01-01"
}
return {
"content": [{
"type": "text",
"text": f"JIRA任务 `{issue_key}` 的信息：\n" +
"\n".join([f"- **{k}**: {v}" for k, v in mock_data.items()])
}]
}
# 工具2：发送 Slack 通知（模拟）
@tool(
name="send_slack_message",
description="向指定的Slack频道发送一条消息",
input_schema={
"channel": {"type": "string", "description": "Slack频道名，如 #general"},
"message": {"type": "string", "description": "要发送的消息内容"}
}
)
async def send_slack_notification(args: Dict[str, Any]) -> Dict[str, Any]:
"""模拟发送Slack消息"""
channel = args.get("channel", "#random")
message = args.get("message", "")
# 这里应该是真实的 Slack API 调用，例如使用 slack-sdk
print(f"[模拟SLACK] 向频道 {channel} 发送消息: {message}") # 在实际应用中，这里会是真实的API调用
return {
"content": [{
"type": "text",
"text": f"✅ 消息已成功发送到 Slack 频道 `{channel}`。"
}]
}
# 工具3：一个简单的计算器工具，展示参数类型
@tool(
name="calculate",
description="执行简单的数学计算",
input_schema={
"operation": {"type": "string", "description": "运算类型，支持 'add', 'subtract', 'multiply', 'divide'"},
"a": {"type": "number", "description": "第一个数字"},
"b": {"type": "number", "description": "第二个数字"}
}
)
async def calculate(args: Dict[str, Any]) -> Dict[str, Any]:
op = args.get("operation")
a = args.get("a", 0)
b = args.get("b", 0)
if op == "add":
result = a + b
elif op == "subtract":
result = a - b
elif op == "multiply":
result = a * b
elif op == "divide":
if b == 0:
return {"content": [{"type": "text", "text": "错误：除数不能为零"}]}
result = a / b
else:
return {"content": [{"type": "text", "text": f"未知操作: {op}"}]}
return {
"content": [{
"type": "text",
"text": f"计算结果: {a} {op} {b} = {result}"
}]
}
async def main():
# 2. 创建 SDK MCP 服务器，将工具聚合在一起
internal_tools_server = create_sdk_mcp_server(
name="internal-company-tools",
version="1.0.0",
tools=[query_jira_issue, send_slack_notification, calculate]
)
# 3. 配置 Claude Agent，启用我们的自定义工具服务器
options = ClaudeAgentOptions(
mcp_servers={"company_tools": internal_tools_server},
# 将自定义工具加入允许列表，使其能被自动调用
allowed_tools=[
"mcp__company_tools__get_jira_issue",
"mcp__company_tools__send_slack_message",
"mcp__company_tools__calculate"
],
system_prompt="""你是一个公司内部助手，集成了JIRA和Slack工具。
当用户询问任务状态时，使用JIRA工具查询。
当需要通知团队时，使用Slack工具发送消息。
对于数学问题，使用计算器工具。
请根据上下文选择合适的工具。"""
)
# 4. 启动会话并测试
async with ClaudeSDKClient(options=options) as client:
# 测试场景：查询一个任务，然后通知团队
test_scenario = """
请帮我做两件事：
1. 查询 JIRA 任务 PROJ-456 的当前状态。
2. 如果状态是 'In Progress'，请发送一条消息到 Slack 频道 #team-updates，内容为：“任务 PROJ-456 正在进行中，负责人是 {assignee}。”
请先查询，再根据结果决定是否发送通知。
"""
await client.query(test_scenario)
print("=== Claude 执行工作流 ===")
async for msg in client.receive_response():
# 在实际应用中，这里可以更精细地处理不同类型的消息块
# 例如，区分工具调用请求、工具执行结果、普通文本响应
if isinstance(msg, AssistantMessage):
for block in msg.content:
if hasattr(block, 'text'):
print(block.text)
elif hasattr(block, 'tool_use'): # 处理工具调用块
print(f"[工具调用] {block.tool_use.name} with args: {block.tool_use.args}")
elif hasattr(block, 'tool_result'): # 处理工具结果块
print(f"[工具结果] {block.tool_result}")
if __name__ == "__main__":
anyio.run(main)
深度解析：SDK MCP 服务器的优势
- 无进程开销 ：工具函数在你的主 Python 进程中直接执行，避免了启动子进程、进程间通信（IPC）的开销，延迟极低。
- 共享状态 ：工具函数可以访问你应用中的全局变量、数据库连接、配置信息等，实现深度集成。
- 简化调试 ：因为工具代码就在你的主进程中，你可以直接使用 IDE 的调试器设置断点，单步调试工具逻辑，与调试普通 Python 代码无异。
- 依赖管理统一 ：所有依赖都在一个 Python 环境中管理，无需为工具服务器单独维护环境。
命名规则 ：注意 allowed_tools 中的工具名格式： mcp__{server_name}__{tool_name} 。这是 SDK 将 MCP 服务器中的工具映射到 Claude 可识别工具名的内部约定。
4.4 利用钩子（Hooks）实现行为拦截与自动化
钩子提供了在 Agent 决策流程中注入自定义逻辑的能力。最常见的用例是安全审查和自动化审批。
import anyio
import re
from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, HookMatcher
async def security_hook_example():
"""
使用钩子实现安全策略：禁止执行危险命令，自动批准安全的文件创建。
"""
# 定义安全审查钩子函数
async def security_check_hook(input_data, tool_use_id, context):
"""
在工具执行前进行安全检查。
input_data: 包含 tool_name 和 tool_input
"""
tool_name = input_data.get("tool_name")
tool_input = input_data.get("tool_input", {})
# 策略1：检查 Bash 命令
if tool_name == "Bash":
command = tool_input.get("command", "")
dangerous_patterns = [
r"rm\s+-rf", # 强制删除
r"chmod\s+[0-7]{3,4}\s+.*", # 可疑的权限修改
r"wget\s+.*\s+-O.*sh", # 从网络下载并执行脚本
r"curl\s+.*\s+\|.*sh",
r">\s+/dev/sd[a-z]", # 向磁盘设备写入
r"dd\s+.*if=.*of=.*", # 磁盘操作
]
for pattern in dangerous_patterns:
if re.search(pattern, command, re.IGNORECASE):
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": f"安全策略阻止：命令匹配危险模式 '{pattern}'",
}
}
# 如果命令安全，可以自动批准或返回空字典（交由后续权限逻辑处理）
# 这里我们选择自动批准安全的 Bash 命令
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "allow",
"permissionDecisionReason": "命令通过安全检查",
}
}
# 策略2：自动批准创建 .py, .txt, .md 文件
elif tool_name == "Write":
path = tool_input.get("path", "")
if path.endswith(('.py', '.txt', '.md', '.json', '.yaml', '.yml')):
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "allow",
"permissionDecisionReason": f"自动批准创建 {path.split('.')[-1]} 文件",
}
}
# 其他类型的文件创建，交由默认权限逻辑处理
return {}
# 对于其他工具，不干预
return {}
# 定义自动化钩子函数：在 Claude 每次回复后，自动添加一条总结
async def auto_summary_hook(input_data, message_id, context):
"""
在 Claude 发送 AssistantMessage 后，自动追加一条总结。
这个钩子演示了 PostMessageCreate 事件。
"""
# input_data 包含即将发送的消息内容
# 我们可以修改它，添加额外内容
# 注意：这需要仔细处理，避免破坏消息结构
# 这里我们只是模拟一个简单的操作：记录日志
print(f"[AUTO-SUMMARY HOOK] Claude 生成了一条消息，ID: {message_id}")
# 在实际应用中，你可以分析消息内容，生成摘要，并可能修改 input_data
# 例如：input_data["content"].append({"type": "text", "text": "\n\n---\n*自动摘要：以上是操作步骤。*"})
return {} # 返回空字典表示不修改消息
# 配置选项，注册钩子
options = ClaudeAgentOptions(
cwd="/tmp/hook_demo",
# 权限配置相对宽松，因为钩子会进行安全检查
allowed_tools=["Read"],
permission_mode='prompt', # 对于未在钩子中处理的工具，会提示用户
hooks={
# PreToolUse 钩子：在工具执行前触发
"PreToolUse": [
HookMatcher(matcher="Bash", hooks=[security_check_hook]),
HookMatcher(matcher="Write", hooks=[security_check_hook]),
],
# PostMessageCreate 钩子：在 Claude 创建消息后触发（示例，需根据SDK实际支持的事件名调整）
# 注意：文档示例主要是 PreToolUse，其他钩子类型请参考最新文档
# "PostMessageCreate": [
# HookMatcher(matcher="*", hooks=[auto_summary_hook]), # 匹配所有消息
# ],
}
)
async with ClaudeSDKClient(options=options) as client:
# 测试1：尝试执行危险命令（应被钩子阻止）
print("测试1: 尝试执行危险命令 'rm -rf /tmp'")
await client.query("请帮我清理临时文件，执行命令：rm -rf /tmp")
async for msg in client.receive_response():
if hasattr(msg, 'content'):
for block in msg.content:
if hasattr(block, 'text'):
print(block.text)
print("\n" + "="*50 + "\n")
# 测试2：尝试创建 Python 文件（应被钩子自动批准）
print("测试2: 尝试创建 Python 文件")
await client.query("请创建一个名为 'safe_script.py' 的 Python 文件，内容为 'print(\"Hello\")'")
async for msg in client.receive_response():
if hasattr(msg, 'content'):
for block in msg.content:
if hasattr(block, 'text'):
print(block.text)
print("\n" + "="*50 + "\n")
# 测试3：尝试创建 .exe 文件（既不在 allowed_tools，钩子也未自动批准，会触发 prompt）
print("测试3: 尝试创建 .exe 文件（将触发权限提示）")
# 注意：由于 permission_mode='prompt'，且钩子返回{}，Claude会向用户请求权限。
# 在非交互式脚本中，这可能导致超时或错误。实际使用时需考虑。
# await client.query("请创建一个 'program.exe' 文件")
if __name__ == "__main__":
anyio.run(security_hook_example)
钩子使用要点 ：
- 执行顺序 ：多个钩子会按注册顺序执行。一个钩子的输出可以作为下一个钩子的输入。
- 决策优先级 ：钩子的
permissionDecision会覆盖allowed_tools和permission_mode的设置。如果钩子返回allow或deny，则以此为准。 - 性能考虑 ：钩子是同步执行的（尽管函数是
async），复杂的钩子逻辑可能会拖慢 Agent 的响应速度。 - 错误处理 ：钩子函数中抛出异常会导致整个工具调用失败。务必做好异常捕获。
5. 高级应用模式与架构设计
5.1 构建分层 Agent 系统
单个 Agent 能力有限。我们可以利用 SDK 构建一个主从（Orchestrator-Worker）或多专家（Multi-Agent）系统。
import anyio
from dataclasses import dataclass
from typing import List, Optional
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
@dataclass
class SubTask:
"""子任务描述"""
description: str
agent_role: str # 指定由哪个专家 Agent 处理
context: Optional[dict] = None
class OrchestratorAgent:
"""协调器 Agent：分解任务，分配子任务给专家 Agent"""
def __init__(self):
self.options = ClaudeAgentOptions(
system_prompt="""你是一个任务协调器。你的职责是：
1. 理解用户提出的复杂、多步骤任务。
2. 将任务分解成一系列清晰的、可独立执行的子任务。
3. 为每个子任务分配合适的专家 Agent（代码专家、文档专家、系统专家）。
4. 整合所有专家 Agent 的结果，形成最终答案。
请以 JSON 格式输出任务分解计划，包含子任务列表和分配的专家类型。"""
)
# 预定义专家 Agent 的配置
self.expert_configs = {
"code_expert": ClaudeAgentOptions(
system_prompt="你是资深代码专家，精通 Python、重构、调试和代码优化。只关注代码相关任务。",
allowed_tools=["Read", "Write", "Edit", "Bash"],
permission_mode='acceptEdits'
),
"doc_expert": ClaudeAgentOptions(
system_prompt="你是技术文档专家，擅长编写清晰、准确的 Markdown 文档、API 说明和教程。",
allowed_tools=["Read", "Write"],
permission_mode='acceptEdits'
),
"sys_expert": ClaudeAgentOptions(
system_prompt="你是系统与运维专家，擅长分析日志、执行系统命令、检查进程和资源使用情况。",
allowed_tools=["Read", "Bash"],
permission_mode='prompt' # 系统命令需谨慎
)
}
async def analyze_and_plan(self, user_request: str) -> List[SubTask]:
"""分析用户请求，生成任务计划"""
async with ClaudeSDKClient(options=self.options) as client:
await client.query(f"用户请求：{user_request}\n请生成任务分解计划。")
plan_json = None
async for msg in client.receive_response():
if hasattr(msg, 'content'):
for block in msg.content:
if hasattr(block, 'text'):
# 这里需要从 Claude 的响应中提取 JSON。
# 在实际应用中，可能需要更鲁棒的解析，或者要求 Claude 输出特定格式。
text = block.text
# 简化处理：寻找 JSON 部分（实际应用应用用 json.loads 和错误处理）
if "```json" in text:
import json
try:
json_str = text.split("```json")[1].split("```")[0].strip()
plan_data = json.loads(json_str)
# 假设 plan_data 包含 "subtasks" 列表
subtasks = []
for item in plan_data.get("subtasks", []):
subtasks.append(SubTask(
description=item.get("desc"),
agent_role=item.get("expert")
))
return subtasks
except (json.JSONDecodeError, IndexError, KeyError) as e:
print(f"解析计划失败: {e}")
# 返回一个默认的简单计划
return [SubTask(user_request, "code_expert")]
# 默认返回一个任务
return [SubTask(user_request, "code_expert")]
async def execute_subtask(self, subtask: SubTask) -> str:
"""执行单个子任务，调用对应的专家 Agent"""
expert_config = self.expert_configs.get(subtask.agent_role, self.expert_configs["code_expert"])
async with ClaudeSDKClient(options=expert_config) as expert_client:
await expert_client.query(subtask.description)
result_parts = []
async for msg in expert_client.receive_response():
if hasattr(msg, 'content'):
for block in msg.content:
if hasattr(block, 'text'):
result_parts.append(block.text)
return "\n".join(result_parts)
async def orchestrate(self, user_request: str) -> str:
"""主协调流程"""
print(f"[协调器] 收到请求: {user_request}")
# 1. 规划
print("[协调器] 正在分析任务并制定计划...")
subtasks = await self.analyze_and_plan(user_request)
print(f"[协调器] 生成了 {len(subtasks)} 个子任务")
# 2. 执行
results = []
for i, subtask in enumerate(subtasks, 1):
print(f"[协调器] 执行子任务 {i}/{len(subtasks)}: [{subtask.agent_role}] {subtask.description[:50]}...")
result = await self.execute_subtask(subtask)
results.append(f"## 子任务 {i} ({subtask.agent_role}) 结果\n{result}")
# 3. 整合 (这里简化了，实际可以再让协调器 Agent 进行总结)
final_output = f"# 任务执行报告：{user_request}\n\n" + "\n\n---\n\n".join(results)
return final_output
async def main():
orchestrator = OrchestratorAgent()
# 模拟一个复杂请求
complex_request = """
我有一个Python项目，结构比较乱。请帮我：
1. 检查项目根目录下的 `main.py`，看看有没有明显的性能问题或代码坏味道。
2. 为项目生成一个 `README.md` 文档，说明项目目的和主要功能。
3. 检查当前系统的磁盘使用情况，看看是否有足够的空间。
"""
final_result = await orchestrator.orchestrate(complex_request)
print("\n" + "="*60)
print("最终整合报告:")
print("="*60)
print(final_result)
if __name__ == "__main__":
anyio.run(main)
这个架构展示了如何用多个特化的 Agent 协作处理复杂问题。协调器负责理解和规划，专家 Agent 负责执行具体领域任务。你可以根据需要扩展更多的专家类型（如“测试专家”、“部署专家”）。
5.2 会话持久化与状态管理
在实际应用中，你可能需要保存和恢复 Agent 的会话状态，以实现断点续传或用户会话管理。
import anyio
import json
import pickle
from datetime import datetime
from pathlib import Path
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage, UserMessage
class PersistentAgentSession:
"""
一个支持会话状态持久化的 Agent 封装类。
可以将对话历史、工作目录状态保存到文件，并在之后恢复。
"""
def __init__(self, session_id: str, storage_dir: Path = Path("./agent_sessions")):
self.session_id = session_id
self.storage_dir = storage_dir
self.storage_dir.mkdir(parents=True, exist_ok=True)
self.history_file = storage_dir / f"{session_id}_history.jsonl"
self.state_file = storage_dir / f"{session_id}_state.pkl"
self.client = None
self.options = None
self.message_history = []
def _save_message(self, role: str, content: str):
"""将单条消息保存到历史文件（JSON Lines格式）"""
record = {
"timestamp": datetime.utcnow().isoformat(),
"role": role,
"content": content
}
with open(self.history_file, 'a', encoding='utf-8') as f:
f.write(json.dumps(record, ensure_ascii=False) + '\n')
self.message_history.append(record)
def load_history(self):
"""从文件加载历史消息"""
if self.history_file.exists():
self.message_history = []
with open(self.history_file, 'r', encoding='utf-8') as f:
for line in f:
if line.strip():
self.message_history.append(json.loads(line.strip()))
print(f"[会话 {self.session_id}] 已加载 {len(self.message_history)} 条历史消息")
return self.message_history
async def initialize(self, options: ClaudeAgentOptions):
"""初始化或恢复会话"""
self.options = options
# 如果有保存的状态，可以在这里恢复（例如，特定的工作目录状态）
# 这里简化处理，只加载历史消息
self.load_history()
# 创建新的客户端。注意：ClaudeSDKClient 本身不直接支持从历史恢复。
# 我们需要在启动后，如果有历史，将历史消息重新发送给 Claude 来恢复上下文。
# 更高级的实现可能需要直接操作底层的会话状态。
self.client = ClaudeSDKClient(options=options)
await self.client.__aenter__() # 手动进入上下文
# 如果存在历史消息，重新发送以恢复上下文（注意：可能受 token 限制）
if self.message_history:
print(f"[会话 {self.session_id}] 正在恢复上下文...")
# 注意：这里需要根据实际 SDK 的消息类型进行适配
# 简化示例：只发送最后几条消息以避免超出上下文窗口
recent_history = self.message_history[-10:] # 只恢复最近10条
for msg in recent_history:
# 这里需要将保存的记录转换为 SDK 能识别的消息格式
# 由于 SDK 消息类型的复杂性，这是一个简化示例
# 实际应用中，你可能需要更精细地重建消息对象
pass
async def query(self, prompt: str) -> str:
"""发送查询并保存历史"""
if not self.client:
raise RuntimeError("会话未初始化，请先调用 initialize()")
# 保存用户消息
self._save_message("user", prompt)
# 发送查询
await self.client.query(prompt)
# 收集助理的响应
full_response = []
async for msg in self.client.receive_response():
if isinstance(msg, AssistantMessage):
for block in msg.content:
if hasattr(block, 'text'):
full_response.append(block.text)
print(block.text, end='', flush=True)
assistant_response = "".join(full_response)
# 保存助理响应
if assistant_response:
self._save_message("assistant", assistant_response)
return assistant_response
async def close(self):
"""关闭会话并保存状态"""
if self.client:
await self.client.__aexit__(None, None, None)
self.client = None
# 可以在这里保存其他状态到 self.state_file
print(f"[会话 {self.session_id}] 会话已关闭，历史保存在 {self.history_file}")
# 支持 with 语句
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.close()
async def demo_persistent_session():
"""演示持久化会话的使用"""
session_id = "project_refactor_001"
options = ClaudeAgentOptions(
cwd="/tmp/project_x",
system_prompt="你是一个项目重构助手，帮助用户逐步改进代码质量。",
allowed_tools=["Read", "Write"],
permission_mode='acceptEdits'
)
async with PersistentAgentSession(session_id) as session:
await session.initialize(options)
# 第一次交互
print("=== 第一次交互：分析项目 ===")
resp1 = await session.query("请列出当前目录下所有的 .py 文件，并简要分析每个文件的大小。")
# 模拟一段时间后，会话恢复（例如，在另一个进程或重启后）
# 实际上，由于我们用了 with 语句，这里不会真正重启。
# 但演示了如何加载历史。
print("\n\n=== 模拟会话恢复后的第二次交互 ===")
# 新“会话”实际上继承了之前的上下文
resp2 = await session.query("基于刚才的分析，哪个 .py 文件最大？建议如何优化它？")
# 第三次交互
print("\n\n=== 第三次交互：执行优化 ===")
resp3 = await session.query("好的，请按照你的建议，重构那个最大的文件。在修改前，先备份原文件。")
if __name__ == "__main__":
anyio.run(demo_persistent_session)
持久化挑战与建议 ：
- 上下文长度 ：Claude 模型有上下文窗口限制。无法将无限长的历史全部重新发送。策略是只发送最近 N 条消息，或总结之前的对话。
- 状态复杂性 ：Agent 的“状态”不仅包括对话历史，还可能包括工作目录中的文件变化、环境变量等。完整的持久化需要保存整个工作目录的快照。
- SDK 限制 ：当前 SDK 可能不直接提供序列化/反序列化完整会话状态的功能。上述示例是一种基于消息历史的“软恢复”。
6. 错误处理、调试与性能优化
6.1 全面的错误处理策略
在与外部进程（Claude Code CLI）和 AI 模型交互时，健壮的错误处理至关重要。
import anyio
import sys
from claude_agent_sdk import (
query, ClaudeAgentOptions,
ClaudeSDKError, CLINotFoundError, CLIConnectionError,
ProcessError, CLIJSONDecodeError
)
async def robust_query_with_retry(prompt: str, max_retries: int = 3):
"""
一个带有重试和详细错误处理的稳健查询函数
"""
options = ClaudeAgentOptions(
request_timeout=30,
total_timeout=300,
)
last_exception = None
for attempt in range(1, max_retries + 1):
try:
print(f"[尝试 {attempt}/{max_retries}] 发送查询...")
async for message in query(prompt=prompt, options=options):
# 处理消息...
if hasattr(message, 'content'):
for block in message.content:
if hasattr(block, 'text'):
print(block.text, end='', flush=True)
print("\n[成功] 查询完成。")
return True
except CLINotFoundError as e:
print(f"[错误] Claude Code CLI 未找到或无法启动: {e}")
print("请确保已安装 Claude Code，或检查 cli_path 配置。")
# 对于此错误，重试无意义
return False
except CLIConnectionError as e:
print(f"[错误] 与 Claude Code CLI 连接失败 (尝试 {attempt}): {e}")
last_exception = e
if attempt < max_retries:
print("等待 5 秒后重试...")
await anyio.sleep(5)
continue
except ProcessError as e:
print(f"[错误] Claude Code 进程异常退出，退出码: {e.exit_code}")
print(f"标准错误输出: {e.stderr}")
# 可能是内存不足、权限问题等。可以尝试重试。
last_exception = e
if attempt < max_retries:
print("等待 10 秒后重试...")
await anyio.sleep(10)
continue
except CLIJSONDecodeError as e:
print(f"[错误] 解析 Claude Code 响应失败: {e}")
print(f"原始响应: {e.raw_response[:200]}...") # 打印前200字符
# JSON 解析错误通常是临时性的，可以重试
last_exception = e
if attempt < max_retries:
print("等待 3 秒后重试...")
await anyio.sleep(3)
continue
except TimeoutError as e:
print(f"[错误] 请求超时 (尝试 {attempt}): {e}")
last_exception = e
if attempt < max_retries:
print("等待 15 秒后重试...")
await anyio.sleep(15)
continue
except Exception as e:
print(f"[未知错误] 类型: {type(e).__name__}, 详情: {e}")
import traceback
traceback.print_exc()
last_exception = e
if attempt < max_retries:
print("等待 5 秒后重试...")
await anyio.sleep(5)
continue
print(f"[失败] 经过 {max_retries} 次重试后仍未成功。")
print(f"最后错误: {last_exception}")
return False
async def main():
# 测试一个可能出错的查询（例如，要求执行一个不存在的命令）
test_prompt = """
请执行以下命令来检查系统状态：
1. 运行 `ls -la` 查看目录
2. 运行 `some_nonexistent_command` （这个命令不存在）
3. 运行 `df -h` 查看磁盘空间
"""
success = await robust_query_with_retry(test_prompt, max_retries=2)
if not success:
print("任务执行失败，可能需要人工干预。")
if __name__ == "__main__":
anyio.run(main)
6.2 调试技巧与日志记录
当 Agent 行为不符合预期时，系统的调试至关重要。
-
启用详细日志 ：Claude Code CLI 通常有日志选项。你可以通过设置环境变量或 CLI 参数来获取更详细的输出。虽然 SDK 可能没有直接暴露，但你可以通过指定自定义的
cli_path并附带参数来启动 CLI。# 假设你有一个带调试参数的自定义 CLI 包装脚本 options = ClaudeAgentOptions( cli_path="/path/to/my_claude_wrapper.sh" )my_claude_wrapper.sh内容可能如下：#!/bin/bash /usr/local/bin/claude --log-level=debug "$@" -
钩子用于调试 ：
PreToolUse和PostToolUse钩子是绝佳的调试工具，可以记录所有工具调用的输入和输出。async def debug_logging_hook(input_data, tool_use_id, context): tool_name = input_data.get("tool_name") print(f"[DEBUG HOOK] 工具调用: {tool_name}, ID: {tool_use_id}") print(f"[DEBUG HOOK] 输入参数: {input_data.get('tool_input')}") # 返回空字典，不干预决策 return {} -
消息流监控 ：在
client.receive_response()循环中，打印所有收到的消息块类型和内容，了解 Claude 的“思考”过程。 -
工作目录检查 ：定期检查
cwd目录下的文件变化，确认文件操作是否符合预期。
6.3 性能优化要点
- 复用客户端 ：对于多个连续查询，务必复用同一个
ClaudeSDKClient实例。避免为每个查询都创建和销毁客户端，这会产生不必要的进程启动开销。 - 合理设置超时 ：根据任务复杂度设置
request_timeout和total_timeout。太短会导致任务失败，太长会浪费资源。对于文件操作多的任务，可以设置长一些。 - 控制交互轮次 ：使用
max_turns防止 Agent 陷入无限循环或过于冗长的“思考”。对于明确的任务，可以设置较小的值。 - 精简系统提示词 ：过长的系统提示词会占用宝贵的上下文窗口，增加每次请求的 token 消耗和延迟。保持提示词简洁、精准。
- 异步并发 ：如果你的应用需要处理多个独立的任务，可以使用
asyncio.gather并发运行多个 Agent 会话。但要注意系统资源（CPU、内存）和 Claude API 的速率限制（如果适用）。async def run_agent_task(task_description, task_id): async with ClaudeSDKClient(options=options) as client: await client.query(task_description) # ... 处理响应 return f"Task {task_id} completed" async def main(): tasks = [“任务1”, “任务2”, “任务3”] results = await asyncio.gather(*[run_agent_task(t, i) for i, t in enumerate(tasks)]) print(results)
7. 常见问题排查与实战经验
在实际集成和开发过程中，你几乎一定会遇到下面这些问题。这里是我踩过坑后总结的排查清单和解决方案。
7.1 问题排查速查表
| 问题现象 | 可能原因 | 排查步骤与解决方案 |
|---|---|---|
CLINotFoundError |
1. Claude Code CLI 未安装。 2. cli_path 配置错误。 3. 系统 PATH 环境变量问题。 |
1. 运行 which claude (Unix) 或 where claude (Windows) 检查是否安装。 2. 使用绝对路径配置 cli_path 。 3. 确保安装后终端重启，或指定完整路径。 |
ProcessError 或进程崩溃 |
1. Claude Code CLI 本身 bug 或崩溃。 2. 系统内存不足。 3. 工作目录 ( cwd ) 不存在或无权限。 |
1. 查看错误信息中的 stderr 。 2. 检查系统资源。 3. 确保 cwd 是存在的、可访问的目录。 |
权限被拒绝 ( PermissionError ) |
1. Agent 尝试写入只读目录。 2. 尝试执行没有权限的命令。 |
1. 检查 cwd 及其子目录的读写权限。 2. 使用钩子或 disallowed_tools 限制危险操作。 3. 考虑在沙箱环境（如 Docker 容器）中运行 Agent。 |
| 工具调用不执行或没反应 | 1. 工具不在 allowed_tools 列表中，且 permission_mode 不是 auto 。 2. can_use_tool 回调或钩子返回了 deny 。 3. 工具名称拼写错误。 |
1. 检查 allowed_tools 列表，确保包含所需工具（注意 mcp__server__tool 格式）。 2. 检查 permission_mode 设置。 3. 在钩子或 can_use_tool 函数中添加日志，查看决策过程。 |
| 自定义工具无法被 Claude 识别 | 1. 工具未正确添加到 mcp_servers 。 2. 工具名称格式错误。 3. 工具函数签名或返回值不符合 MCP 规范。 |
1. 确认 create_sdk_mcp_server 被正确调用，且服务器被添加到 options.mcp_servers 。 2. 在 allowed_tools 中使用 mcp__{server_name}__{tool_name} 格式。 3. 确保工具函数是 async 的，并返回包含 content 列表的字典。 |
| Agent 陷入循环或执行无关操作 | 1. 系统提示词不够明确。 2. max_turns 设置过高。 3. 任务本身模糊。 |
1. 强化系统提示词，明确任务边界和停止条件。 2. 合理设置 max_turns （如 5-10）。 3. 在用户提示词中给出更具体的指令和步骤。 |
| 响应速度慢 | 1. 网络问题（如果 Claude Code 需要联网）。 2. 复杂的工具调用（如执行长时间运行的命令）。 3. 上下文过长。 |
1. 检查网络连接。 2. 为可能耗时的工具调用设置超时。 3. 尝试简化系统提示词和对话历史。 |
async for 循环卡住或无输出 |
1. Agent 仍在“思考”或执行长时间工具。 2. 流式响应缓冲区问题。 3. 异步事件循环未正确运行。 |
1. 增加超时设置。 2. 确保在异步函数内运行 ( async def )。 3. 使用 anyio.run() 或 asyncio.run() 正确启动事件循环。 |
7.2 安全部署最佳实践
将拥有文件系统和命令执行能力的 AI Agent 部署到生产环境，必须慎之又慎。
-
最小权限原则 ：
cwd设置为一个专用的、隔离的工作目录，该目录只包含任务必要的文件。allowed_tools列表尽可能短。如果只需要读文件，就只放Read。- 默认使用
permission_mode='prompt'或'acceptEdits'，而不是'auto'。
-
沙箱化运行 ：
- Docker 容器 ：在非特权容器中运行你的 Agent 应用，限制其资源（CPU、内存）和文件系统访问（只挂载必要卷）。
- 系统级沙箱 ：考虑使用
nsjail、gVisor或Firecracker等工具提供更强的隔离。
-
输入验证与过滤 ：
- 对所有传入 Agent 的用户提示词进行清洗，防止提示词注入攻击。
- 使用
PreToolUse钩子对工具参数进行严格验证，特别是Bash命令和文件路径。
-
审计与日志 ：
- 记录所有用户查询、工具调用（包括参数）和 Agent 响应。
- 定期审查日志，寻找异常模式。
-
网络隔离 ：
- 如果 Agent 不需要访问外网，就在网络层面进行隔离。
- 对于需要访问外部服务（如自定义 MCP 服务器）的情况，使用白名单限制可访问的地址和端口。
7.3 与现有工作流集成
Claude Agent SDK 不是孤立的，它可以成为你自动化工作流中的核心“大脑”。
- 与 CI/CD 集成 ：在 GitHub Actions、GitLab CI 中，使用 Agent 自动进行代码审查、生成变更日志、更新依赖版本。
# GitHub Actions 示例片段 - name: Code Review with Claude Agent run: | python -m pip install claude-agent-sdk python scripts/auto_review.py ${{ github.event.pull_request.head.sha }} - 作为后台服务 ：使用 FastAPI 或 Django 构建一个 Web 服务，将 Agent 能力通过 REST API 暴露出去，供其他系统调用。
- 与消息平台对接 ：将 Agent 与 Slack、Discord 或钉钉的机器人集成，打造团队内部的智能助手。
- 作为 CLI 工具 ：使用
typer或click库，将你的 Agent 逻辑包装成一个命令行工具，方便开发者在终端使用。
这个 SDK 的潜力在于其可编程性。你不再是与一个聊天界面交互，而是在代码中驱动一个拥有强大认知和行动能力的智能体。从自动化繁琐任务到构建复杂的 AI 原生应用， anthropics/claude-agent-sdk-python 提供了一个坚实而灵活的基石。关键在于深入理解其代理循环、权限模型和扩展机制，然后结合你的具体业务场景，设计出安全、高效、可靠的 AI 驱动解决方案。
欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。
更多推荐
9
0- 0
已为社区贡献10条内容
所有评论(0)