第 21 章 Agent SDK:把 Claude Code 内核嵌入你的程序

核对日期:2026-07-08。Agent SDK 前身 Claude Code SDK,2025-09-29 更名独立发布。本章代码示例为说明性骨架,非来自真实仓库。

21.1 结论

Agent SDK 把 Claude Code 的内核——agent loop、工具系统、上下文管理——抽成 Python/TypeScript 库,让你在自己的进程里构建自定义 agent。一句话定位:Claude Code 是产品,Agent SDK 是内核。CC 解决"人在终端里与 Claude 协作",SDK 解决"把这套协作能力嵌入你自己的程序"。

需要先纠正一个常见误解:Agent SDK 的核心入口不是 client.beta.messages.tool_runner(那是 Anthropic Client SDK 的 beta 功能,层次在 Messages API),而是 query() 函数。query() 内置完整的 agent loop、内置工具执行、错误处理与类型化消息流。理解这个区分,才能选对工具(21.5 节展开)。

21.2 SDK 是什么

发布背景:Agent SDK 前身是 Claude Code SDK,2025-09-29 正式更名并独立发布。定位是"把 Claude Code 作为库"——同一套 agent loop、同一套内置工具(Read/Edit/Bash/Glob/Grep 等)、同一套上下文管理(自动压缩、prompt caching、subagents)。

三层定位要分清:

层次	是什么	你写什么
Anthropic Client SDK	原始 Messages API 客户端	自己写 while stop_reason == "tool_use" 循环、自己实现每个工具
Client SDK + tool_runner(beta)	Client SDK 之上的半自动 loop	工具仍自己实现,loop 样板减少
Agent SDK	完整 agent 内核(工具+loop+上下文)	只写 prompt 和配置,工具开箱即用

对比 Client SDK 手写 loop:

# Client SDK: 你自己实现 tool loop
response = client.messages.create(...)
while response.stop_reason == "tool_use":
    result = your_tool_executor(response.tool_use)
    response = client.messages.create(tool_result=result, **params)

# Agent SDK: loop 和工具都内置
async for message in query(prompt="Fix the bug in auth.py"):
    print(message)

关键差别:Client SDK 里你要实现 your_tool_executor——读文件、跑命令、解析结果、处理错误;Agent SDK 这些都内置,Read/Edit/Bash 等工具开箱即用,Claude 自主调用、SDK 自主执行。

21.3 语言与安装

官方支持两种语言:

• Python:pip install claude-agent-sdk(需 Python 3.10+)
• TypeScript:npm install @anthropic-ai/claude-agent-sdk(需 Node.js 18+)

TypeScript SDK 捆绑一个原生 Claude Code 二进制作为可选依赖,无需单独安装 Claude Code CLI。Python SDK 纯库形式。

认证:默认读 ANTHROPIC_API_KEY 环境变量。也支持 Bedrock(CLAUDE_CODE_USE_BEDROCK=1)、Vertex AI(CLAUDE_CODE_USE_VERTEX=1)、Azure Foundry(CLAUDE_CODE_USE_FOUNDRY=1)等第三方 provider。注意:除非 Anthropic 事先批准,第三方开发者不能在自己的产品里提供 claude.ai 登录——只能用 API key 认证。

最小可运行示例(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"]),
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

验证:运行 python agent.py,应输出当前目录文件列表。失败边界:若 pip 报 No matching distribution found,说明 Python 版本低于 3.10,用 python3 --version 核对。

21.4 核心:agent loop、工具、上下文管理

这三件与 Claude Code 完全一致——这就是"内核库化"的含义。

Agent loop 的五个消息类型:

类型	何时产生	用途
SystemMessage	会话生命周期事件	subtype="init" 携带 session_id;compact_boundary 标记压缩发生
AssistantMessage	每次 Claude 响应	含文本和工具调用请求
UserMessage	工具执行后	携带工具结果反馈给 Claude
StreamEvent	启用 partial messages 时	实时文本 delta、工具输入 chunk
ResultMessage	loop 结束	最终文本、token 用量、成本、session_id

loop 的工作方式:Claude 收到 prompt → 评估 → 请求工具调用 → SDK 执行工具 → 结果反馈 → Claude 再评估 → ... 直到 Claude 产出不含工具调用的纯文本响应。一个 turn 是"Claude 产出工具调用到结果反馈"的一个完整往返。

内置工具集(与 CC 一致):

• 文件:Read Edit Write
• 搜索:Glob Grep
• 执行:Bash
• Web:WebSearch WebFetch
• 发现:ToolSearch(按需加载工具 schema,避免预载全部)
• 编排:Agent(子代理)Skill AskUserQuestion TaskCreate TaskUpdate

上下文管理:

• Prompt caching:系统提示、工具定义、CLAUDE.md 内容跨 turn 自动缓存,只有首次请求付全成本
• 自动压缩:上下文接近上限时,SDK 自动 summarize 旧历史,保留近期对话与关键决策;触发时 yield SystemMessage(subtype="compact_boundary")
• Subagents:每个子代理开全新上下文(看不到父代理历史),只把最终结果作为 tool result 返回——主上下文只增长摘要,不增长子任务全量 transcript

权限三件套控制工具执行:

• allowed_tools:预批准列表(只读 agent 用 ["Read", "Glob", "Grep"])
• disallowed_tools:禁止列表
• permission_mode:default(需回调批准)/acceptEdits(自动批准文件编辑)/plan(只读)/dontAsk(未预批准一律拒绝)/auto(模型分类器决定)/bypassPermissions(全开,仅隔离环境用)

预算控制:max_turns(工具调用往返上限)、max_budget_usd(成本上限)。命中任一上限,SDK 返回 ResultMessage 且 subtype 为 error_max_turns 或 error_max_budget_usd。

21.5 Tool Runner 与 query():概念厘清

官方文档与社区材料里,"Tool Runner" 这个词容易混淆。需要厘清三个层次:

1. Anthropic Client SDK 的 client.beta.messages.tool_runner:Messages API 客户端的 beta 功能,帮你跑 while stop_reason == "tool_use" 的循环,但工具实现仍由你提供。它减少样板代码,不提供内置工具。

2. Agent SDK 的 query():Agent SDK 的核心入口,内置完整 agent loop 且内置工具(Read/Edit/Bash 等)。你不需要实现任何工具。

3. 关系:tool_runner 是从 Client SDK 通往 agent 化的中间台阶;Agent SDK 是更上层的封装,把"工具也内置"这件事做完了。

如果只用 Client SDK + tool_runner,你写的是:

# Client SDK + tool_runner: 工具你自己实现,loop 自动跑
def my_read_tool(path):
    return open(path).read()

response = client.beta.messages.tool_runner.run(
    tools=[{"name": "read", "implementation": my_read_tool}],
    # ...
)

如果用 Agent SDK,同样的"读文件"需求:

# Agent SDK: 工具内置,只配 allowed_tools
async for msg in query(
    prompt="Read auth.py and summarize",
    options=ClaudeAgentOptions(allowed_tools=["Read"]),
):
    print(msg)

选型规则:需要 Claude Code 同款工具(文件/命令/搜索/Web)用 Agent SDK;只想要自己的业务工具(查数据库、调内部 API)用 Client SDK + tool_runner。本章聚焦 Agent SDK;Client SDK 的 tool_runner 属于 API 层话题。

Agent SDK query() 自带的 agent loop 已包含三项工程化能力:

• 错误包装:工具失败时把错误作为 tool result 反馈给 Claude,让它自我修正或换路径,而非直接崩溃
• 类型安全:Python 用 isinstance(msg, ResultMessage) 判消息类型;TypeScript 用 msg.type === "result" 字符串字段
• human-in-the-loop:通过 canUseTool 回调(权限批准)或 AskUserQuestion 工具(向用户提问)实现

21.6 MCP in-process:在 SDK 内挂载 MCP 工具

Agent SDK 可以直接在进程内挂载 MCP 服务器,给 agent 加外部能力(浏览器、数据库、API)。配置通过 mcp_servers 选项:

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    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"]
                }
            }
        ),
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

验证:运行后应看到 Claude 调用 Playwright MCP 打开页面并描述内容。失败边界:首次运行 npx 会拉取 @playwright/mcp 包,网络不通会卡住或超时;MCP server crash 表现为工具调用返回错误结果,Claude 通常会换路径重试。

上下文成本警告:MCP server 的每个工具 schema 都会进入每次请求的上下文。工具多时,几个 MCP server 可能在上千 token 的 schema 还没用工具前就吃掉可观上下文。解决方案是启用 ToolSearch(按需加载工具 schema,而非全量预载)。在 Google Cloud Agent Platform 或非第一方 ANTHROPIC_BASE_URL 下,tool search 默认关闭,schema 全量加载——这种环境更要控制 MCP server 数量。

21.7 会话管理:多轮、持久化、流式

多轮(resume):从 SystemMessage(subtype="init")或 ResultMessage 捕获 session_id,下次 query() 传 resume=session_id 即可恢复完整上下文(已读文件、已完成分析、已执行动作)。

from claude_agent_sdk import (
    query, ClaudeAgentOptions, SystemMessage, ResultMessage
)

async def main():
    session_id = None
    # 第一轮:读 auth 模块
    async for message in query(
        prompt="Read the authentication module",
        options=ClaudeAgentOptions(allowed_tools=["Read", "Glob"]),
    ):
        if isinstance(message, SystemMessage) and message.subtype == "init":
            session_id = message.data["session_id"]

    # 第二轮:在上文基础上继续,"it" 指 auth 模块
    async for message in query(
        prompt="Now find all places that call it",
        options=ClaudeAgentOptions(resume=session_id),
    ):
        if isinstance(message, ResultMessage):
            print(message.result)

asyncio.run(main())

Fork:从某个 session_id 分叉出新分支,探索不同方案而不污染原会话。

持久化:默认在本地文件系统存 JSONL 会话日志。也可把 session 存到外部存储(数据库、对象存储),官方提供 "Persist sessions to external storage" 指南。

流式:默认 query() 已是 async iterator,逐消息 yield。要拿到更细粒度的 token 级流式,启用 include_partial_messages(Python)/includePartialMessages(TypeScript),会额外 yield StreamEvent 消息,含文本 delta 和工具输入 chunk——适合做实时 UI。

自动压缩与 CLAUDE.md:压缩会把旧消息 summarize 掉,早期对话的具体指令可能丢失。持久规则应写进 CLAUDE.md(通过 setting_sources 加载),因为 CLAUDE.md 内容每次请求都重新注入,不依赖对话历史。甚至可以在 CLAUDE.md 里写"压缩时保留什么"的指令,compactor 会读。

21.8 实战:用 SDK 构建代码审查 agent

目标:构建一个只读代码审查 agent,自定义系统提示、自定义工具集、严格权限、含成本上限与审计 hook。

代码骨架(Python,含标注):

import asyncio
from datetime import datetime
from claude_agent_sdk import (
    query, ClaudeAgentOptions, HookMatcher,
    ResultMessage, AssistantMessage,
)

SYSTEM_PROMPT = """You are a code review agent.
Your job: read code, find bugs, security issues, and style problems.
Constraints:
- READ-ONLY. Never edit files.
- Output findings as a numbered list with severity (HIGH/MED/LOW) and file:line.
- If the codebase is too large, focus on the entry point and its direct dependencies.
- Do not speculate about code you have not read."""

async def audit_tool(input_data, tool_use_id, context):
    """PostToolUse hook: 把每次工具调用写入审计日志。"""
    tool_name = input_data.get("tool_name", "unknown")
    tool_input = input_data.get("tool_input", {})
    with open("./audit.log", "a") as f:
        f.write(f"{datetime.now()} | {tool_name} | {tool_input}\n")
    return {}  # 空返回=不修改工具行为

async def main():
    async for message in query(
        prompt="Review src/ for bugs and security issues. Output a prioritized list.",
        options=ClaudeAgentOptions(
            system_prompt=SYSTEM_PROMPT,
            allowed_tools=["Read", "Glob", "Grep"],       # 只读
            disallowed_tools=["Edit", "Write", "Bash"],   # 硬禁写与执行
            permission_mode="dontAsk",                    # 未预批准一律拒绝
            max_turns=50,                                 # 工具调用往返上限
            max_budget_usd=2.0,                           # 成本上限
            setting_sources=["project", "user"],          # 加载 CLAUDE.md
            hooks={
                "PostToolUse": [
                    HookMatcher(matcher="Read|Glob|Grep", hooks=[audit_tool])
                ]
            },
        ),
    ):
        if isinstance(message, AssistantMessage):
            # 进度可见:每次 Claude 响应当前在做什么
            for block in message.content:
                if hasattr(block, "text"):
                    print(f"[turn] {block.text[:200]}")
        elif isinstance(message, ResultMessage):
            print(f"\n--- DONE (subtype={message.subtype}) ---")
            if message.subtype == "success":
                print(message.result)
            print(f"cost: ${message.total_cost_usd:.4f}, "
                  f"turns: {message.num_turns}")
            break

asyncio.run(main())

关键设计点标注:

1. 系统提示硬约束 READ-ONLY:即使权限配置出错,系统提示也指示模型不编辑。这是纵深防御。
2. allowed_tools + disallowed_tools + permission_mode="dontAsk" 三层:权限模型按规则数组顺序求值;dontAsk 保证未预批准工具一律拒绝,不弹确认。
3. max_turns=50 + max_budget_usd=2.0:生产 agent 必设上限。开放任务("improve this codebase")不设上限可能跑几十美元。
4. PostToolUse hook 审计:每次工具调用落日志,不进 Claude 上下文(hook 在你的进程里跑,不消耗 context window)。
5. setting_sources:加载项目与用户级 CLAUDE.md,让 agent 遵循项目规范(代码风格、禁用 API 等)。

验证步骤:

1. 准备一个含已知 bug 的小代码库(如 src/auth.py 里有 SQL 拼接)
2. 运行 python review_agent.py
3. 预期:输出含 HIGH 严重度的 SQL 注入发现,audit.log 记录所有 Read/Glob/Grep 调用,cost 在 $2 以内,turns 在 50 以内
4. 失败边界检查:
   • 若 Claude 尝试调 Edit 修改代码 → disallowed_tools 应阻止,审计日志会显示拒绝
   • 若超过 max_budget_usd → ResultMessage.subtype == "error_max_budget_usd",result 字段为空
   • 若 audit.log 未生成 → hook 未触发,检查 HookMatcher 的 matcher 正则

扩展:把 Agent 加入 allowed_tools,定义 security-auditor 和 style-checker 两个子代理,让主 agent 并行委派——每个子代理独立上下文,主上下文只收摘要。这是处理大代码库的标准模式。

21.9 SDK vs Claude Code vs Managed Agents:选型

维度	Claude Code CLI	Agent SDK	Managed Agents
形态	终端交互产品	Python/TS 库	托管 REST API
运行位置	本地进程	你的进程、你的基础设施	Anthropic 托管
工具作用对象	本地文件系统	你进程能访问的文件与服务	托管沙箱
会话状态	本地 JSONL	本地文件系统或外部存储	Anthropic 托管事件日志
自定义工具	配置/插件/MCP	进程内函数 + MCP	Claude 触发,你执行并返回
适合场景	日常交互开发、一次性任务	CI/CD、自定义应用、嵌入式 agent	生产长时 agent、无需自建沙箱

选型规则:

• 本地交互开发 → Claude Code CLI(人在终端,需要实时反馈与审查)
• 嵌入式 agent / CI/CD / 自定义应用 → Agent SDK(要在自己程序里跑 agent loop,工具作用在自己的文件与服务上)
• 云端长时、无需自建沙箱 → Managed Agents(Anthropic 托管 sandbox 与 session,适合生产异步任务)

常见路径:先用 Agent SDK 本地原型验证,再迁到 Managed Agents 上生产。两者接口理念接近,迁移成本可控。Managed Agents 细节第 22 章展开。

Claude Code CLI 与 Agent SDK 不是二选一:很多团队 CLI 用于日常开发,SDK 用于生产自动化,工作流在两者间直接迁移(同一套工具、同一套 prompt 风格)。

21.10 失败边界

1. SDK 版本与 CC 内核不一致:Agent SDK 内部依赖 Claude Code 的 agent loop 实现(TypeScript SDK 捆绑原生二进制)。SDK 版本与 CLI 版本可能不同步——CLI 升级后某些工具行为变化,旧 SDK 可能未跟进。失败表现:同一 prompt 在 CLI 和 SDK 行为不一致。对策:生产环境固定 SDK 版本,CHANGELOG 核对工具行为变更。

2. 工具定义错误:

• allowed_tools 写了不存在的工具名(如 read 应为 Read,大小写敏感)→ 工具不会被执行,Claude 收到拒绝消息
• 自定义工具 schema 不符合 MCP 规范 → 加载时报错或工具不可见
• MCP server 命令错误(如 npx 包名拼错)→ 启动时 hang 或工具调用返回连接错误

验证:启动后先跑一个简单 prompt 确认工具可用,再跑生产任务。

3. human-in-the-loop 设计缺失:

• 用 permission_mode="default" 但没提供 canUseTool 回调 → 所有未预批准工具被拒绝,agent 卡死
• 用 bypassPermissions 跑在生产或共享环境 → 危险,Bash 可执行任意命令
• 交互式 agent 没监听 AskUserQuestion 工具 → Claude 的提问被忽略,agent 静默失败

规则:default 模式必须配 canUseTool 回调;bypassPermissions 只在容器/CI 等隔离环境用;交互式 agent 要处理 AskUserQuestion 工具调用。

4. 上下文预算失控:

• MCP server 工具多 + tool search 关闭 → schema 吃满上下文,agent 还没干活就接近压缩阈值
• 长任务不设 max_turns/max_budget_usd → 开放任务可能跑几十轮、几十美元
• 大文件 Read 输出占满上下文 → 压缩频繁,早期指令丢失

对策:MCP server 精简并启用 tool search;生产 agent 必设预算上限;大文件用 Grep 定位再 Read 特定行,而非全量读。

5. 会话持久化假设:

• 默认 JSONL 存本地文件系统,容器重启会丢 → 生产环境用外部存储
• resume 依赖 session_id 正确捕获,SystemMessage(subtype="init") 没处理就丢了 → 在 init 消息里就抓 session_id
• fork 与 resume 混用导致上下文污染 → 明确区分:resume 继续原会话,fork 开新分支

---

下一章:Managed Agents:云端托管 agent