2.9 Claude Agent SDK:构建自主 AI 代理
什么是 Agent SDK
Claude Agent SDK 是 Anthropic 官方提供的开发工具包,允许开发者以编程方式构建自主 AI 代理。它将驱动 Claude Code 的核心能力(工具调用、代理循环、上下文管理)封装为 Python 和 TypeScript 库。
命名变更:原名 "Claude Code SDK" 已更名为 "Claude Agent SDK",以更好地反映其构建各类 AI 代理的广泛能力。
核心理念
SDK 的核心设计原则是"给你的代理一台计算机"——通过提供文件操作、命令执行等工具,Claude 可以像人类开发者一样工作:
读取文件 → 搜索代码 → 编辑内容 → 执行命令 → 验证结果核心价值
| 特性 | 说明 |
|---|---|
| 内置工具 | 文件读写、Bash 命令、代码搜索等开箱即用 |
| 代理循环 | 自动处理工具调用链,无需手动实现 |
| 上下文管理 | 自动压缩长对话,防止超出上下文限制 |
| 权限控制 | 细粒度控制代理可使用的工具 |
| 会话管理 | 跨多轮对话保持上下文,支持会话恢复 |
| MCP 集成 | 通过 Model Context Protocol 连接外部服务 |
与其他工具的对比
Agent SDK vs Client SDK
| 维度 | Client SDK | Agent SDK |
|---|---|---|
| 工具执行 | 需要自己实现 | 内置自动执行 |
| 代理循环 | 手动编写 | 自动处理 |
| 复杂度 | 更底层,更灵活 | 更高层,更便捷 |
| 适用场景 | 简单 API 调用 | 复杂自主任务 |
# Client SDK:需要自己实现工具循环
response = client.messages.create(...)
while response.stop_reason == "tool_use":
result = your_tool_executor(response.tool_use)
response = client.messages.create(tool_result=result, ...)
# Agent SDK:Claude 自动处理工具调用
async for message in query(prompt="Fix the bug in auth.py"):
print(message)Agent SDK vs Claude Code CLI
| 使用场景 | 最佳选择 |
|---|---|
| 交互式开发 | CLI |
| CI/CD 流水线 | SDK |
| 自定义应用 | SDK |
| 一次性任务 | CLI |
| 生产自动化 | SDK |
快速开始
1. 安装 Claude Code(运行时)
# macOS/Linux/WSL
curl -fsSL https://claude.ai/install.sh | bash
# Homebrew
brew install --cask claude-code
# npm
npm install -g @anthropic-ai/claude-code2. 安装 SDK
TypeScript/JavaScript:
npm install @anthropic-ai/claude-agent-sdkPython:
pip install claude-agent-sdk3. 设置 API Key
export ANTHROPIC_API_KEY=your-api-key4. 运行第一个代理
Python:
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
async for message in query(
prompt="What files are in this directory?",
options=ClaudeAgentOptions(allowed_tools=["Bash", "Glob"])
):
print(message)
asyncio.run(main())TypeScript:
import { query } from "@anthropic-ai/claude-agent-sdk";
for await (const message of query({
prompt: "What files are in this directory?",
options: { allowedTools: ["Bash", "Glob"] },
})) {
console.log(message);
}内置工具
SDK 提供以下开箱即用的工具:
| 工具 | 功能 |
|---|---|
| Read | 读取工作目录中的任意文件 |
| Write | 创建新文件 |
| Edit | 精确编辑现有文件 |
| Bash | 运行终端命令、脚本、git 操作 |
| Glob | 按模式查找文件(**/*.ts,src/**/*.py) |
| Grep | 使用正则搜索文件内容 |
| WebSearch | 搜索网络获取最新信息 |
| WebFetch | 获取并解析网页内容 |
| Task | 启动子代理处理子任务 |
使用示例
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
# 创建一个搜索 TODO 注释的代理
async for message in query(
prompt="Find all TODO comments and create a summary",
options=ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep"])
):
print(message)
asyncio.run(main())核心功能详解
1. Hooks(钩子)
在代理生命周期的关键点运行自定义代码:
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
async for message in query(
prompt="Suggest improvements to utils.py",
options=ClaudeAgentOptions(
permission_mode="acceptEdits",
hooks={
"PostToolUse": [{
"matcher": "Edit|Write",
"hooks": [{
"type": "command",
"command": "echo \"$(date): file modified\" >> ./audit.log"
}]
}]
}
)
):
print(message)
asyncio.run(main())可用 Hooks:PreToolUse、PostToolUse、Stop、SessionStart、SessionEnd、UserPromptSubmit 等。
2. Subagents(子代理)
启动专门的子代理处理聚焦的子任务:
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
# 启用 Task 工具让 Claude 可以创建子代理
async for message in query(
prompt="Analyze this codebase for security vulnerabilities",
options=ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep", "Task"])
):
print(message)
asyncio.run(main())3. MCP 集成
通过 Model Context Protocol 连接外部系统(数据库、浏览器、API 等):
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
# 连接 Playwright MCP 服务器进行浏览器自动化
async for message in query(
prompt="Open example.com and describe what you see",
options=ClaudeAgentOptions(
mcp_servers={
"playwright": {"command": "npx", "args": ["@playwright/mcp@latest"]}
}
)
):
print(message)
asyncio.run(main())4. 权限控制
精确控制代理可以使用的工具:
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
# 创建只读代理,只能分析不能修改
async for message in query(
prompt="Review this code for best practices",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Glob", "Grep"],
permission_mode="bypassPermissions"
)
):
print(message)
asyncio.run(main())权限模式:
| 模式 | 说明 |
|---|---|
default | 默认权限,需要用户确认危险操作 |
acceptEdits | 自动接受文件编辑 |
bypassPermissions | 跳过所有权限确认 |
plan | 计划模式,只读不写 |
5. 会话管理
跨多轮对话保持上下文,支持会话恢复:
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
session_id = None
# 第一次查询:获取 session_id
async for message in query(
prompt="Read the authentication module",
options=ClaudeAgentOptions(allowed_tools=["Read", "Glob"])
):
if hasattr(message, 'subtype') and message.subtype == 'init':
session_id = message.data.get('session_id')
# 恢复会话,保持完整上下文
async for message in query(
prompt="Now find all places that call it", # "it" 指代认证模块
options=ClaudeAgentOptions(resume=session_id)
):
pass
asyncio.run(main())自定义工具
使用 createSdkMcpServer 和 tool 辅助函数创建类型安全的自定义工具:
TypeScript 示例
import { query, tool, createSdkMcpServer } from "@anthropic-ai/claude-agent-sdk";
import { z } from "zod";
// 创建自定义 MCP 服务器
const customServer = createSdkMcpServer({
name: "my-custom-tools",
version: "1.0.0",
tools: [
tool(
"get_weather",
"Get current temperature for a location",
{
latitude: z.number().describe("Latitude coordinate"),
longitude: z.number().describe("Longitude coordinate")
},
async (args) => {
const response = await fetch(
`https://api.open-meteo.com/v1/forecast?latitude=${args.latitude}&longitude=${args.longitude}¤t=temperature_2m`
);
const data = await response.json();
return {
content: [{
type: "text",
text: `Temperature: ${data.current.temperature_2m}°C`
}]
};
}
)
]
});
// 使用自定义工具
async function* generateMessages() {
yield {
type: "user" as const,
message: {
role: "user" as const,
content: "What's the weather in San Francisco?"
}
};
}
for await (const message of query({
prompt: generateMessages(),
options: {
mcpServers: { "my-custom-tools": customServer },
allowedTools: ["mcp__my-custom-tools__get_weather"]
}
})) {
console.log(message);
}Python 示例
from claude_agent_sdk import tool, create_sdk_mcp_server, query, ClaudeAgentOptions
from typing import Any
import aiohttp
# 使用 @tool 装饰器定义自定义工具
@tool("get_weather", "Get current temperature for a location", {"latitude": float, "longitude": float})
async def get_weather(args: dict[str, Any]) -> dict[str, Any]:
async with aiohttp.ClientSession() as session:
url = f"https://api.open-meteo.com/v1/forecast?latitude={args['latitude']}&longitude={args['longitude']}¤t=temperature_2m"
async with session.get(url) as response:
data = await response.json()
return {
"content": [{
"type": "text",
"text": f"Temperature: {data['current']['temperature_2m']}°C"
}]
}
# 创建 MCP 服务器
custom_server = create_sdk_mcp_server(
name="my-custom-tools",
version="1.0.0",
tools=[get_weather]
)
# 使用自定义工具
async def message_generator():
yield {
"type": "user",
"message": {
"role": "user",
"content": "What's the weather in San Francisco?"
}
}
async def main():
async for message in query(
prompt=message_generator(),
options=ClaudeAgentOptions(
mcp_servers={"my-custom-tools": custom_server},
allowed_tools=["mcp__my-custom-tools__get_weather"]
)
):
print(message)工具命名规则
MCP 工具的名称格式:mcp__{server_name}__{tool_name}
例如:服务器 my-custom-tools 中的工具 get_weather → mcp__my-custom-tools__get_weather
代理设计模式
四阶段代理循环
Claude Agent SDK 推荐的代理工作模式:
┌─────────────────────────────────────────────────────────────┐
│ 代理循环(Agent Loop) │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 1.收集上下文 │───▶│ 2.采取行动 │───▶│ 3.验证工作 │ │
│ │ Gather │ │ Take Action │ │ Verify │ │
│ │ Context │ │ │ │ │ │
│ └──────────────┘ └──────────────┘ └──────┬───────┘ │
│ ▲ │ │
│ │ │ │
│ └────────────── 4.重复 ──────────────────┘ │
│ Repeat │
└─────────────────────────────────────────────────────────────┘1. 收集上下文(Gather Context)
- 智能搜索:使用
grep、glob等工具定位相关文件 - 语义搜索:速度快但准确性较低
- 子代理:并行处理,各自维护隔离上下文
- 自动压缩:长对话自动总结,防止超出限制
2. 采取行动(Take Action)
- 自定义工具:定义高频操作如
fetchInbox、searchEmails - Bash 脚本:灵活执行任意计算机操作
- 代码生成:SDK 擅长生成精确、可复用的代码
- MCP 集成:预建集成 Slack、GitHub、Asana 等服务
3. 验证工作(Verify Work)
- 规则校验:使用代码 linting 反馈错误
- 视觉反馈:对 HTML 等输出截图验证
- LLM 评判:用另一个模型评估输出质量
4. 重复(Repeat)
直到任务完成或达到停止条件。
迁移指南
从 Claude Code SDK 迁移
包名变更:
| 语言 | 旧包名 | 新包名 |
|---|---|---|
| TypeScript | @anthropic-ai/claude-code | @anthropic-ai/claude-agent-sdk |
| Python | claude-code-sdk | claude-agent-sdk |
TypeScript 迁移步骤
# 1. 卸载旧包
npm uninstall @anthropic-ai/claude-code
# 2. 安装新包
npm install @anthropic-ai/claude-agent-sdk// 3. 更新导入
// 之前
import { query, tool } from "@anthropic-ai/claude-code";
// 之后
import { query, tool } from "@anthropic-ai/claude-agent-sdk";Python 迁移步骤
# 1. 卸载旧包
pip uninstall claude-code-sdk
# 2. 安装新包
pip install claude-agent-sdk# 3. 更新导入和类型名
# 之前
from claude_code_sdk import query, ClaudeCodeOptions
# 之后
from claude_agent_sdk import query, ClaudeAgentOptions破坏性变更
1. 系统提示不再默认
# v0.0.x - 默认使用 Claude Code 系统提示
async for message in query(prompt="Hello"):
print(message)
# v0.1.0 - 默认空系统提示
# 恢复旧行为:
async for message in query(
prompt="Hello",
options=ClaudeAgentOptions(
system_prompt={"type": "preset", "preset": "claude_code"}
)
):
print(message)
# 或使用自定义系统提示:
async for message in query(
prompt="Hello",
options=ClaudeAgentOptions(
system_prompt="You are a helpful coding assistant"
)
):
print(message)2. 设置源不再默认加载
# v0.0.x - 自动加载所有设置
async for message in query(prompt="Hello"):
print(message)
# v0.1.0 - 默认不加载设置
# 恢复旧行为:
async for message in query(
prompt="Hello",
options=ClaudeAgentOptions(
setting_sources=["user", "project", "local"]
)
):
print(message)实战案例
案例 1:Bug 修复代理
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def fix_bug():
async for message in query(
prompt="Find and fix the bug in auth.py that causes login failures",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Edit", "Bash", "Grep"],
permission_mode="acceptEdits"
)
):
print(message)
asyncio.run(fix_bug())案例 2:代码审查代理
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def review_code():
async for message in query(
prompt="Review the latest git diff for security issues and code quality",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Glob", "Grep", "Bash"],
permission_mode="bypassPermissions", # 只读,无需权限确认
system_prompt="""You are a senior code reviewer.
Focus on:
1. Security vulnerabilities
2. Performance issues
3. Code style and best practices
Provide specific, actionable feedback."""
)
):
print(message)
asyncio.run(review_code())案例 3:文档生成代理
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def generate_docs():
async for message in query(
prompt="Generate API documentation for all functions in src/api/",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Write", "Glob", "Grep"],
permission_mode="acceptEdits"
)
):
print(message)
asyncio.run(generate_docs())案例 4:多工具数据库代理
from claude_agent_sdk import tool, create_sdk_mcp_server, query, ClaudeAgentOptions
from typing import Any
import json
@tool("query_database", "Execute SQL query", {"query": str})
async def query_database(args: dict[str, Any]) -> dict[str, Any]:
# 实际项目中连接真实数据库
results = [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
return {
"content": [{
"type": "text",
"text": f"Found {len(results)} rows:\n{json.dumps(results, indent=2)}"
}]
}
@tool("insert_record", "Insert a new record", {"table": str, "data": dict})
async def insert_record(args: dict[str, Any]) -> dict[str, Any]:
return {
"content": [{
"type": "text",
"text": f"Inserted record into {args['table']}"
}]
}
database_server = create_sdk_mcp_server(
name="database",
version="1.0.0",
tools=[query_database, insert_record]
)
async def main():
async def message_generator():
yield {
"type": "user",
"message": {
"role": "user",
"content": "List all users in the database"
}
}
async for message in query(
prompt=message_generator(),
options=ClaudeAgentOptions(
mcp_servers={"database": database_server},
allowed_tools=["mcp__database__query_database", "mcp__database__insert_record"]
)
):
print(message)最佳实践
设计原则
- 明确工具边界:只授予代理完成任务所需的最小工具集
- 验证输出:使用 Hooks 或后处理验证代理的输出
- 错误处理:自定义工具要优雅处理错误,返回有意义的消息
- 日志审计:使用 Hooks 记录所有重要操作
性能优化
- 使用子代理:将复杂任务分解给专门的子代理
- 限制工具:减少可用工具数量可提高响应速度
- 会话复用:使用
resume参数复用会话避免重复初始化
安全建议
- 权限最小化:使用
allowed_tools限制可用工具 - 环境隔离:在沙箱环境运行不可信代理
- 输入验证:对用户输入进行验证和清理
- 敏感信息保护:不要在提示中包含 API Key 等敏感信息
调试技巧
查看详细输出
遍历消息流查看所有事件:
async for message in query(prompt="...", options=options):
print(f"Type: {message.type}")
if hasattr(message, 'subtype'):
print(f"Subtype: {message.subtype}")
print(f"Content: {message}")
print("---")常见问题
| 问题 | 解决方案 |
|---|---|
| MCP 工具不工作 | 确保使用异步生成器作为 prompt |
| 权限错误 | 检查 permission_mode 设置 |
| 工具未找到 | 验证 allowed_tools 中的工具名格式 |
| 会话恢复失败 | 确保 session_id 正确保存和传递 |
应用场景
Claude Agent SDK 可用于构建多种类型的 AI 代理:
| 类型 | 示例 |
|---|---|
| 开发工具 | SRE 机器人、安全审计员、代码审查代理 |
| 业务助手 | 法律助理、财务顾问、客户支持 |
| 自动化 | CI/CD 集成、文档生成、测试自动化 |
| 研究工具 | 深度研究代理、数据分析助手 |
本节小结
- Agent SDK = 编程化的 Claude Code — 相同能力,SDK 形式
- 内置工具丰富 — 文件操作、命令执行、网络搜索等开箱即用
- 自定义工具 — 通过 MCP 服务器扩展任意功能
- 权限控制 — 细粒度控制代理能力
- 会话管理 — 跨对话保持上下文
- 四阶段循环 — 收集上下文 → 采取行动 → 验证工作 → 重复
参考资料
- Claude Agent SDK 官方概述
- 迁移指南
- 自定义工具文档
- Anthropic 工程博客 - Building Agents
- SDK 示例仓库
- TypeScript SDK GitHub
- Python SDK GitHub
下一节: 2.20 本章小结 — 回顾 Commands、Skills、Plugins、Subagent、Hooks 和 Agent SDK 的完整知识体系