Claude Code Agent SDK 开发指南

什么是 Agent SDK

Agent SDK 是 Anthropic 官方提供的开发工具包,将 Claude Code 的 AI Agent 能力封装为可编程接口,让开发者能够在自己的应用中集成 AI 自动化能力。

核心价值

类比理解: 如果 Claude Code 是一个完整的 IDE,那么 Agent SDK 就是它的"引擎"

| Claude Code | Agent SDK | |-------------|-----------| | 完整的桌面应用 | 可嵌入的库 | | 交互式界面 | 编程接口 | | 手动触发任务 | 自动化工作流 | | 单一用户 | 多用户/批量处理 |

适用场景:

CI/CD 集成 - 自动代码审查、测试生成
批量处理 - 同时处理多个代码库
自定义工作流 - 构建专属的 AI 编程助手
企业集成 - 嵌入到内部开发平台

| 维度 | Claude API | Agent SDK | |------|-----------|-----------| | 定位 | 通用对话 API | 编程专用 Agent | | 工具能力 | 需自己实现 | 内置文件/命令工具 | | Agent Loop | 需手动编排 | 自动多轮执行 | | 子代理 | 不支持 | 支持并行子任务 | | 适用场景 | 聊天、问答 | 代码生成、自动化 |

关键区别: Agent SDK = Claude API + 内置工具 + Agent Loop + 子代理系统

快速开始

环境准备

系统要求:

Python 3.10+ 或 Node.js 18+
Claude Code CLI (SDK 会自动下载)
Anthropic API Key

获取 API Key:

访问 Anthropic Console
注册/登录账号
创建 API Key
设置环境变量:

export ANTHROPIC_API_KEY="sk-ant-api03-your-key-here"

Python 安装

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/macOS

# 安装 SDK
pip install claude-agent-sdk

# 验证安装
python -c "from claude_agent_sdk import query; print('安装成功')"

TypeScript 安装

# 初始化项目
npm init -y

# 安装 SDK
npm install @anthropic-ai/agent-sdk

# 验证安装
node -e "require('@anthropic-ai/agent-sdk'); console.log('安装成功')"

Hello World 示例

Python 版本:

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        model='claude-sonnet-4-6'
    )
    
    async for message in query(
        prompt="创建一个 hello.txt 文件,内容为 'Hello, Agent SDK!'",
        options=options
    ):
        print(message)

asyncio.run(main())

TypeScript 版本:

import { query, ClaudeAgentOptions } from '@anthropic-ai/agent-sdk';

async function main() {
  const options: ClaudeAgentOptions = {
    model: 'claude-sonnet-4-6'
  };
  
  for await (const message of query({
    prompt: "创建一个 hello.txt 文件,内容为 'Hello, Agent SDK!'",
    options
  })) {
    console.log(message);
  }
}

main();

核心概念

query 函数

query 是 SDK 的核心 API,启动一个 Agent 会话:

async for message in query(prompt, options):
    # 处理消息流

参数说明:

| 参数 | 类型 | 说明 | |------|------|------| | prompt | str | 任务描述 | | options | ClaudeAgentOptions | 配置选项 |

返回值: AsyncIterator,流式返回消息

ClaudeAgentOptions 配置

options = ClaudeAgentOptions(
    # 模型选择
    model='claude-sonnet-4-6',  # 推荐
    
    # 权限控制
    allowed_tools=['Read', 'Write', 'Bash'],
    permission_mode='default',  # default/acceptEdits/bypassPermissions
    
    # 系统提示
    system_prompt='你是代码审查专家',
    
    # 执行限制
    max_turns=20,  # 最大对话轮数
    
    # 工作目录
    cwd='/path/to/project',
    
    # 自定义工具
    mcp_servers={'tools': my_tools}
)

消息类型

Agent 返回多种消息类型:

from claude_agent_sdk.types import (
    AssistantMessage,  # AI 的回复
    ToolUseBlock,      # 工具调用
    ToolResultBlock,   # 工具结果
    TextBlock          # 文本内容
)

async for message in query(prompt, options):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock):
                print(f"AI: {block.text}")
            elif isinstance(block, ToolUseBlock):
                print(f"调用工具: {block.name}")
    
    elif isinstance(message, ToolResultBlock):
        print(f"工具结果: {message.content}")

Agent Loop 执行流程

1. 接收任务 (prompt)
   ↓
2. AI 分析并决策
   ↓
3. 调用工具 (Read/Write/Bash/Agent)
   ↓
4. 获取工具结果
   ↓
5. 检查任务是否完成
   ├─ 未完成 → 回到步骤 2
   └─ 已完成 → 返回最终结果

示例流程:

任务: "分析 main.py 并添加类型注解"

轮次1: AI → Read(main.py)
轮次2: 工具返回文件内容
轮次3: AI → Edit(main.py, 添加类型注解)
轮次4: 工具返回编辑成功
轮次5: AI → 任务完成

内置工具系统

可用工具列表

| 工具名 | 功能 | 示例 | |--------|------|------| | Read | 读取文件 | 读取配置文件 | | Write | 写入文件 | 创建新文件 | | Edit | 编辑文件 | 修改代码 | | Bash | 执行命令 | 运行测试 | | Glob | 文件搜索 | 查找所有 .py 文件 | | Grep | 内容搜索 | 搜索函数定义 | | Agent | 子代理 | 并行处理任务 |

工具权限控制

# 只读 Agent
readonly_options = ClaudeAgentOptions(
    allowed_tools=['Read', 'Glob', 'Grep']
)

# 代码审查 Agent
review_options = ClaudeAgentOptions(
    allowed_tools=['Read', 'Glob', 'Grep', 'Bash'],
    system_prompt="只能读取和运行测试,不能修改文件"
)

# 完全权限 Agent
full_options = ClaudeAgentOptions(
    allowed_tools=['Read', 'Write', 'Edit', 'Bash', 'Glob', 'Grep', 'Agent'],
    permission_mode='acceptEdits'
)

工具使用示例

文件操作:

# 读取文件
prompt = "读取 config.json 并解析配置"

# 写入文件
prompt = "创建 README.md,包含项目说明"

# 编辑文件
prompt = "在 main.py 的第10行后添加日志语句"

命令执行:

# 运行测试
prompt = "执行 pytest 并报告结果"

# 安装依赖
prompt = "运行 pip install -r requirements.txt"

# Git 操作
prompt = "提交当前更改,commit 信息为 'feat: add logging'"

搜索操作:

# 文件搜索
prompt = "找出所有包含 TODO 注释的 Python 文件"

# 内容搜索
prompt = "搜索所有定义了 async def 的函数"

子代理系统

什么是子代理

子代理 (Subagent) 是独立的 Agent 实例,可以并行处理子任务。

使用场景:

并行处理 - 同时分析多个模块
任务隔离 - 每个子任务独立上下文
大规模任务 - 拆分复杂项目

基础用法

prompt = """
请使用子代理并行处理:
1. 子代理1: 分析 src/ 目录的代码质量
2. 子代理2: 分析 tests/ 目录的测试覆盖率
3. 汇总两个子代理的结果
"""

async for message in query(prompt, options):
    print(message)

高级配置

定义子代理:

from claude_agent_sdk.types import AgentDefinition

# 定义专用子代理
code_reviewer = AgentDefinition(
    name="code-reviewer",
    system_prompt="你是代码审查专家,关注安全和性能",
    allowed_tools=["Read", "Grep"],
    max_turns=10
)

test_generator = AgentDefinition(
    name="test-generator",
    system_prompt="你是测试专家,生成完整的单元测试",
    allowed_tools=["Read", "Write"],
    max_turns=15
)

# 使用子代理
options = ClaudeAgentOptions(
    agents=[code_reviewer, test_generator]
)

prompt = """
使用 code-reviewer 审查 main.py
使用 test-generator 为 main.py 生成测试
"""

并行限制

最多同时运行 10 个子代理
超过限制会自动排队
子代理有独立的上下文和工具权限

自定义工具开发

MCP 工具集成

Agent SDK 支持通过 MCP (Model Context Protocol) 添加自定义工具:

from claude_agent_sdk.mcp import MCPServer, Tool

# 定义工具
@Tool(
    name="calculate",
    description="执行数学计算",
    input_schema={
        "type": "object",
        "properties": {
            "expression": {"type": "string", "description": "数学表达式"}
        },
        "required": ["expression"]
    }
)
async def calculate(args):
    expression = args.get('expression')
    try:
        result = eval(expression)
        return {
            "content": [{
                "type": "text",
                "text": f"{expression} = {result}"
            }]
        }
    except Exception as e:
        return {
            "content": [{
                "type": "text",
                "text": f"计算错误: {str(e)}"
            }],
            "isError": True
        }

# 创建 MCP 服务器
my_tools = MCPServer(
    name="math-tools",
    tools=[calculate]
)

# 使用自定义工具
options = ClaudeAgentOptions(
    mcp_servers={"math": my_tools},
    allowed_tools=["mcp__math__calculate"]
)

async for message in query(
    prompt="计算 (123 + 456) * 2",
    options=options
):
    print(message)

工具最佳实践

| 原则 | 说明 | |------|------| | 清晰命名 | 工具名应描述功能 | | 详细描述 | 帮助 AI 理解何时使用 | | 输入验证 | 检查参数有效性 | | 错误处理 | 返回清晰的错误信息 | | 幂等性 | 多次调用结果一致 |

Hooks 系统

什么是 Hooks

Hooks 允许在工具执行前后插入自定义逻辑,实现监控、过滤、日志等功能。

可用 Hook 类型:

| Hook | 触发时机 | 用途 | |------|---------|------| | PreToolUse | 工具执行前 | 权限检查、参数验证 | | PostToolUse | 工具执行后 | 日志记录、结果处理 |

基础示例

from claude_agent_sdk.types import HookMatcher

async def log_tool_use(input_data, tool_use_id, context):
    """记录工具调用"""
    tool_name = input_data.get("tool_name")
    print(f"[日志] 调用工具: {tool_name}")
    return {}

async def block_dangerous_commands(input_data, tool_use_id, context):
    """阻止危险命令"""
    if input_data.get("tool_name") == "Bash":
        command = input_data.get("tool_input", {}).get("command", "")
        dangerous = ["rm -rf", "sudo", "format"]
        
        for pattern in dangerous:
            if pattern in command:
                return {
                    "permission": {
                        "permissionDecision": "deny",
                        "permissionDecisionReason": f"危险命令被阻止: {pattern}"
                    }
                }
    return {}

# 配置 Hooks
options = ClaudeAgentOptions(
    allowed_tools=["Bash", "Read", "Write"],
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Bash", hooks=[block_dangerous_commands]),
        ],
        "PostToolUse": [
            HookMatcher(matcher=".*", hooks=[log_tool_use]),
        ],
    }
)

权限控制模式

| 模式 | 说明 | 适用场景 | |------|------|---------| | default | 每次文件操作需确认 | 生产环境 | | acceptEdits | 自动接受文件编辑 | 个人项目 | | bypassPermissions | 跳过所有权限检查 | 完全受控环境 |

# 生产环境配置
production_options = ClaudeAgentOptions(
    model='claude-sonnet-4-6',
    allowed_tools=['Read', 'Glob', 'Grep'],
    permission_mode='default',
    max_turns=10,
    sandbox={"enabled": True}
)

实战案例

案例1: 自动代码审查

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def code_review(file_path):
    """自动代码审查"""
    options = ClaudeAgentOptions(
        model='claude-sonnet-4-6',
        allowed_tools=['Read', 'Grep'],
        system_prompt="""你是代码审查专家。审查代码时关注:
        1. 安全漏洞
        2. 性能问题
        3. 代码规范
        4. 错误处理
        
        输出格式:
        🔴 严重: 必须修复
        🟡 警告: 建议修复
        🟢 提示: 可优化
        """
    )
    
    prompt = f"审查文件 {file_path},提供详细分析"
    
    async for message in query(prompt, options):
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text'):
                    print(block.text)

# 使用
asyncio.run(code_review("src/main.py"))

案例2: 批量测试生成

async def generate_tests_batch(files):
    """批量生成测试"""
    options = ClaudeAgentOptions(
        model='claude-sonnet-4-6',
        allowed_tools=['Read', 'Write', 'Agent'],
        permission_mode='acceptEdits'
    )
    
    # 使用子代理并行处理
    tasks = "\n".join([f"- 为 {f} 生成单元测试" for f in files])
    prompt = f"""
    使用子代理并行处理以下任务:
    {tasks}
    
    每个测试文件应包含:
    1. 导入语句
    2. 测试类
    3. 至少3个测试用例
    4. 边界条件测试
    """
    
    async for message in query(prompt, options):
        print(message)

# 使用
files = ["src/auth.py", "src/database.py", "src/api.py"]
asyncio.run(generate_tests_batch(files))

案例3: CI/CD 集成

async def ci_pipeline():
    """CI/CD 流水线"""
    options = ClaudeAgentOptions(
        model='claude-sonnet-4-6',
        allowed_tools=['Read', 'Bash', 'Grep'],
        max_turns=15
    )
    
    prompt = """
    执行 CI 检查流程:
    1. 运行 pytest 检查测试
    2. 运行 flake8 检查代码规范
    3. 运行 mypy 检查类型
    4. 汇总所有问题并生成报告
    
    如果发现问题,按严重程度分类列出。
    """
    
    report = []
    async for message in query(prompt, options):
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text'):
                    report.append(block.text)
    
    # 保存报告
    with open("ci_report.md", "w") as f:
        f.write("\n".join(report))
    
    print("CI 报告已生成: ci_report.md")

asyncio.run(ci_pipeline())

故障排查

常见错误

错误1: 模块导入失败

# 错误信息
ModuleNotFoundError: No module named 'claude_agent_sdk'

# 解决方案
source venv/bin/activate
pip install claude-agent-sdk

错误2: 认证失败

# 错误信息
AuthenticationError: Invalid API key

# 解决方案
export ANTHROPIC_API_KEY="sk-ant-api03-your-key-here"
echo $ANTHROPIC_API_KEY  # 验证

错误3: 工具权限被拒绝

# 错误信息
PermissionDenied: Tool 'Write' is not in allowed_tools

# 解决方案
options = ClaudeAgentOptions(
    allowed_tools=['Read', 'Write', 'Edit']  # 添加 Write
)

错误4: 连接超时

# 检查网络
ping api.anthropic.com

# 国内用户可能需要代理
export HTTP_PROXY="http://127.0.0.1:7890"
export HTTPS_PROXY="http://127.0.0.1:7890"

调试技巧

启用详细日志:

import logging

logging.basicConfig(level=logging.DEBUG)

打印消息详情:

async for message in query(prompt, options):
    print(f"消息类型: {type(message).__name__}")
    print(f"消息内容: {message}")
    print("-" * 40)

常见问题

Q: Agent SDK 和 Claude API 有什么区别?

Agent SDK 有 Agent Loop、内置工具和子代理系统,适合编程任务。Claude API 适合简单问答。

Q: 哪个模型性价比最高?

| 场景 | 推荐模型 | |------|---------| | 日常开发 | claude-sonnet-4-6 | | 复杂推理 | claude-opus-4-6 | | 简单任务 | claude-haiku-4-5-20251001 |

Q: 如何控制 Token 消耗?

设置 max_turns 限制轮数
精简 system_prompt
使用更小的模型
给出明确的任务描述

Q: 子代理如何通信?

主 Agent 通过 Agent 工具调用子代理,子代理返回结果给主 Agent。子代理运行在独立上下文中。

Q: 最多可以运行多少个子代理?

最多 10 个并行,超过会自动排队。

Q: 如何在生产环境使用?

production_options = ClaudeAgentOptions(
    model='claude-sonnet-4-6',
    allowed_tools=['Read', 'Glob', 'Grep'],  # 最小权限
    permission_mode='default',  # 需要确认
    max_turns=10,
    sandbox={"enabled": True}
)

Q: SDK 支持流式输出吗?

是的,query 返回 AsyncIterator,天然支持流式处理。

Q: 自定义工具可以访问外部 API 吗?

可以,自定义工具就是普通的 Python/TypeScript 函数:

import httpx

@tool("weather", "获取天气信息", {"city": str})
async def get_weather(args):
    city = args.get('city')
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.weather.com/{city}")
        return {"content": [{"type": "text", "text": response.text}]}

Q: 如何在 Jupyter Notebook 中使用?

import nest_asyncio
nest_asyncio.apply()

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def run():
    async for message in query(prompt="Hello", options=ClaudeAgentOptions()):
        print(message)

asyncio.run(run())

# 只读 Agent
ClaudeAgentOptions(
    allowed_tools=['Read', 'Glob', 'Grep']
)

# 代码审查 Agent
ClaudeAgentOptions(
    allowed_tools=['Read', 'Grep', 'Bash'],
    system_prompt="代码审查专家"
)

# 完全权限 Agent
ClaudeAgentOptions(
    allowed_tools=['Read', 'Write', 'Edit', 'Bash', 'Glob', 'Grep', 'Agent'],
    permission_mode='acceptEdits'
)

# 生产环境 Agent
ClaudeAgentOptions(
    model='claude-sonnet-4-6',
    allowed_tools=['Read', 'Glob', 'Grep'],
    permission_mode='default',
    max_turns=10,
    sandbox={"enabled": True}
)

消息类型判断

from claude_agent_sdk.types import (
    AssistantMessage,
    ToolUseBlock,
    ToolResultBlock,
    TextBlock
)

async for message in query(prompt, options):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock):
                # 处理文本
                pass
            elif isinstance(block, ToolUseBlock):
                # 处理工具调用
                pass
    elif isinstance(message, ToolResultBlock):
        # 处理工具结果
        pass