Skip to content

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 SDKAgent SDK
工具执行需要自己实现内置自动执行
代理循环手动编写自动处理
复杂度更底层,更灵活更高层,更便捷
适用场景简单 API 调用复杂自主任务
python
# 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(运行时)

bash
# macOS/Linux/WSL
curl -fsSL https://claude.ai/install.sh | bash

# Homebrew
brew install --cask claude-code

# npm
npm install -g @anthropic-ai/claude-code

2. 安装 SDK

TypeScript/JavaScript:

bash
npm install @anthropic-ai/claude-agent-sdk

Python:

bash
pip install claude-agent-sdk

3. 设置 API Key

bash
export ANTHROPIC_API_KEY=your-api-key

4. 运行第一个代理

Python:

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:

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按模式查找文件(**/*.tssrc/**/*.py
Grep使用正则搜索文件内容
WebSearch搜索网络获取最新信息
WebFetch获取并解析网页内容
Task启动子代理处理子任务

使用示例

python
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(钩子)

在代理生命周期的关键点运行自定义代码:

python
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:PreToolUsePostToolUseStopSessionStartSessionEndUserPromptSubmit 等。

2. Subagents(子代理)

启动专门的子代理处理聚焦的子任务:

python
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 等):

python
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. 权限控制

精确控制代理可以使用的工具:

python
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. 会话管理

跨多轮对话保持上下文,支持会话恢复:

python
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())

自定义工具

使用 createSdkMcpServertool 辅助函数创建类型安全的自定义工具:

TypeScript 示例

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}&current=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 示例

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']}&current=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_weathermcp__my-custom-tools__get_weather


代理设计模式

四阶段代理循环

Claude Agent SDK 推荐的代理工作模式:

┌─────────────────────────────────────────────────────────────┐
│                    代理循环(Agent Loop)                    │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ┌──────────────┐    ┌──────────────┐    ┌──────────────┐ │
│   │ 1.收集上下文  │───▶│ 2.采取行动   │───▶│ 3.验证工作   │ │
│   │ Gather       │    │ Take Action  │    │ Verify       │ │
│   │ Context      │    │              │    │              │ │
│   └──────────────┘    └──────────────┘    └──────┬───────┘ │
│         ▲                                        │         │
│         │                                        │         │
│         └────────────── 4.重复 ──────────────────┘         │
│                        Repeat                              │
└─────────────────────────────────────────────────────────────┘

1. 收集上下文(Gather Context)

  • 智能搜索:使用 grepglob 等工具定位相关文件
  • 语义搜索:速度快但准确性较低
  • 子代理:并行处理,各自维护隔离上下文
  • 自动压缩:长对话自动总结,防止超出限制

2. 采取行动(Take Action)

  • 自定义工具:定义高频操作如 fetchInboxsearchEmails
  • 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
Pythonclaude-code-sdkclaude-agent-sdk

TypeScript 迁移步骤

bash
# 1. 卸载旧包
npm uninstall @anthropic-ai/claude-code

# 2. 安装新包
npm install @anthropic-ai/claude-agent-sdk
typescript
// 3. 更新导入
// 之前
import { query, tool } from "@anthropic-ai/claude-code";

// 之后
import { query, tool } from "@anthropic-ai/claude-agent-sdk";

Python 迁移步骤

bash
# 1. 卸载旧包
pip uninstall claude-code-sdk

# 2. 安装新包
pip install claude-agent-sdk
python
# 3. 更新导入和类型名
# 之前
from claude_code_sdk import query, ClaudeCodeOptions

# 之后
from claude_agent_sdk import query, ClaudeAgentOptions

破坏性变更

1. 系统提示不再默认

python
# 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. 设置源不再默认加载

python
# 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 修复代理

python
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:代码审查代理

python
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:文档生成代理

python
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:多工具数据库代理

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

最佳实践

设计原则

  1. 明确工具边界:只授予代理完成任务所需的最小工具集
  2. 验证输出:使用 Hooks 或后处理验证代理的输出
  3. 错误处理:自定义工具要优雅处理错误,返回有意义的消息
  4. 日志审计:使用 Hooks 记录所有重要操作

性能优化

  1. 使用子代理:将复杂任务分解给专门的子代理
  2. 限制工具:减少可用工具数量可提高响应速度
  3. 会话复用:使用 resume 参数复用会话避免重复初始化

安全建议

  1. 权限最小化:使用 allowed_tools 限制可用工具
  2. 环境隔离:在沙箱环境运行不可信代理
  3. 输入验证:对用户输入进行验证和清理
  4. 敏感信息保护:不要在提示中包含 API Key 等敏感信息

调试技巧

查看详细输出

遍历消息流查看所有事件:

python
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 集成、文档生成、测试自动化
研究工具深度研究代理、数据分析助手

本节小结

  1. Agent SDK = 编程化的 Claude Code — 相同能力,SDK 形式
  2. 内置工具丰富 — 文件操作、命令执行、网络搜索等开箱即用
  3. 自定义工具 — 通过 MCP 服务器扩展任意功能
  4. 权限控制 — 细粒度控制代理能力
  5. 会话管理 — 跨对话保持上下文
  6. 四阶段循环 — 收集上下文 → 采取行动 → 验证工作 → 重复

参考资料


下一节: 2.20 本章小结 — 回顾 Commands、Skills、Plugins、Subagent、Hooks 和 Agent SDK 的完整知识体系

基于 MIT 许可证发布。内容版权归作者所有。