本教程教你用 Claude Agent SDK(原 Claude Code SDK)以编程方式把 Claude Code 的引擎嵌入自己的脚本,构建一个能自主读码、定位、修复 Bug 并闭环验证的代理。区别于命令行 -p 一次性调用,SDK 让你拿到完整的 Agent 循环、工具执行、上下文管理,可在 Python/TypeScript 里精细编排权限、流式消息、自定义工具门禁与会话续跑。适合已有命令行使用经验、想构建 CI 自动化、内部修复机器人或批处理流水线的开发者,预计 2–3 小时读完并跑通实战项目。
npx(TypeScript SDK 自带平台对应的 Claude Code 原生二进制,无需单独装 Claude Code);若用 Python,则需 Python 3.10+async/await、for await)git、npm/pip很多人对 Claude Code 自动化的认知停留在第一层,把它们搞混会选错工具:
| 层次 | 形态 | 典型场景 | 局限 |
|---|---|---|---|
| L1 命令行 | claude -p "修复bug" --output-format json |
一次性脚本、cron | 无中间状态、无法编程式拦截工具 |
| L2 Hooks/Subagent | settings.json 钩子 + 子代理委派 | 编辑器内护栏、任务分流 | 配置驱动,灵活性受限于规则 |
| L3 编程式 SDK | query() 嵌入你的程序 |
CI 机器人、多轮对话、自定义门禁 | 需写代码,但可控性最强 |
本教程聚焦 L3。它的核心价值是:工具执行、上下文管理、重试编排都由 SDK 处理,你只消费消息流——但你可以在任意位置插手。
创建项目目录并安装:
mkdir my-fixbot && cd my-fixbot
npm init -y
npm install @anthropic-ai/claude-agent-sdk
认证用环境变量,SDK 不会自动读取 .env 文件,必须显式导出:
export ANTHROPIC_API_KEY=sk-ant-xxxxxxxx
若走云厂商渠道,按需设置:
# Amazon Bedrock
export CLAUDE_CODE_USE_BEDROCK=1
# Google Vertex AI
export CLAUDE_CODE_USE_VERTEX=1
# Azure AI Foundry
export CLAUDE_CODE_USE_FOUNDRY=1
# Claude Platform on AWS
export CLAUDE_CODE_USE_ANTHROPIC_AWS=1
export ANTHROPIC_AWS_WORKSPACE_ID=your-workspace-id
预期结果:node -e "require('@anthropic-ai/claude-agent-sdk')" 不报错即安装成功。常见问题:若提示找不到原生二进制,确认 Node 版本 ≥ 18,并检查 node_modules/@anthropic-ai/claude-agent-sdk/ 下是否存在平台子包。
query 是唯一的主入口,返回一个异步迭代器,你用 for await 逐条消费 Claude 产生的消息:
// fixbot.ts
import { query } from "@anthropic-ai/claude-agent-sdk";
for await (const message of query({
prompt: "列出当前目录下的 .py 文件",
options: { allowedTools: ["Bash", "Glob"] },
})) {
if (message.type === "result" && message.subtype === "success") {
console.log(message.result);
}
}
运行:npx tsx fixbot.ts。这三部分是 SDK 的骨架:
prompt:可以是字符串,也可以是异步生成器(见第五步)。options:代理配置,如 allowedTools(预批准工具)、permissionMode(权限模式)、systemPrompt、mcpServers、maxTurns、cwd。为什么和 -p 不同:循环本身持续运转,Claude 自动「思考 → 调工具 → 观察结果 → 决定下一步」,直到任务完成或报错。你不必实现工具执行逻辑——Read/Edit/Bash/Glob 等内置工具由 SDK 直接跑。
无人值守场景下,最危险的是 Agent 执行了你没预期的破坏性命令。SDK 提供两道闸门:
allowedTools:白名单,只允许 Claude 用列出的工具。permissionMode:全局策略,取值 "default"(每次问人)、"acceptEdits"(自动批准文件编辑)、"bypassPermissions"(全自动,慎用)、"plan"(只规划不执行)。for await (const message of query({
prompt: "审查 utils.ts 里会导致崩溃的 Bug 并修复",
options: {
cwd: "./repo", // 限定工作目录,防止越界
allowedTools: ["Read", "Edit", "Glob", "Grep"],
permissionMode: "acceptEdits", // 自动批准文件编辑
maxTurns: 15, // 防止失控,硬上限
},
})) {
// ...处理消息
}
铁律:批处理/CI 永远不要用 bypassPermissions 配合无限制工具集。最小权限 + maxTurns 上限是底线。allowedTools 里不放 Bash 或只放受限的命令,能避免 rm -rf 之类的灾难。
不做过滤时,迭代器会吐出原始系统消息(很吵)。实战中你要识别三类消息:
for await (const message of query({ prompt: "...", options: { allowedTools: ["Read", "Edit"] } })) {
if (message.type === "assistant" && message.message?.content) {
for (const block of message.message.content) {
if ("text" in block) {
console.log("💭 推理:", block.text);
} else if ("name" in block) {
console.log("🔧 调用工具:", block.name);
}
}
} else if (message.type === "result") {
console.log(`✅ 结束: ${message.subtype}, 耗时 ${message.duration_ms}ms`);
console.log("最终结果:", message.result);
}
}
assistant 消息:Claude 的输出,content 是若干 block——text 是推理文本,name 是要调用的工具。result 消息:循环终态,subtype 为 success/error_max_turns/error_during_execution 等,带 duration_ms、total_cost_usd、num_turns 等元数据。这套结构让你能把 Agent 嵌进可观测性系统:每个工具调用落日志、每次成本入库、失败时触发告警。
当 prompt 是一个异步生成器时,进入「流式输入模式」——这是 SDK 最强的能力:Agent 作为长驻进程,接收多轮输入、处理中断、跨轮维持上下文。
import { query } from "@anthropic-ai/claude-agent-sdk";
import { readFile } from "fs/promises";
async function* messages() {
yield { type: "user", message: { role: "user", content: "分析 auth.ts 的安全问题" } };
await new Promise((r) => setTimeout(r, 2000)); // 等外部信号
yield {
type: "user",
message: {
role: "user",
content: [
{ type: "text", text: "再结合这张架构图" },
{ type: "image", source: { type: "base64", media_type: "image/png", data: await readFile("arch.png", "base64") } },
],
},
};
}
for await (const m of query({ prompt: messages(), options: { allowedTools: ["Read", "Grep"], maxTurns: 10 } })) {
if (m.type === "result" && m.subtype === "success") console.log(m.result);
}
要点:可附加多模态(图片)做视觉分析;可在生成器里 await 任意异步事件(用户输入、定时器、消息队列),实现「人在回路」交互;生成器抛错时流以 Claude Code process aborted 错误结束。Python 对应的是 ClaudeSDKClient + receive_response()。
L3 区别于 L1 的关键:你能在工具真正执行前编程式拦截。canUseTool 回调对每次工具调用做审批决策——这是构建安全护栏的程序化手段:
const options = {
allowedTools: ["Bash", "Edit"],
permissionMode: "default",
// 伪代码:实际签名以官方 TS reference 为准,回调接收 toolName 与 input,返回 boolean 或 { behavior, updatedInput }
hooks: {
PreToolUse: [
{
matcher: "Bash",
hooks: [
{
type: "command",
command: 'node ./guard.js "$TOOL_INPUT"', // 命中危险命令以 exit 2 阻断
},
],
},
],
},
};
实战策略:
- 命令白名单:Bash 调用前解析 input.command,正则匹配 rm -rf|curl|wget|> 等危险模式,命中则拒绝。
- 范围限定:Edit/Write 调用前校验 file_path 必须在 cwd 子树内。
- 审计落库:无论批准与否,把 {tool, input, decision, ts} 写入审计日志。
flowchart TD
A[query 入口] --> B{Agent 决策}
B -->|选择工具| C[PreToolUse 钩子/canUseTool]
C -->|拒绝| D[阻断并反馈给 Agent]
C -->|批准| E[执行工具]
E --> F[工具结果回灌 Agent]
F --> B
B -->|任务完成| G[result 消息]
D --> B
G --> H[消费方:日志/告警/续跑]
长任务可能因超时或错误中断,resume 让你从上次断点继续,而非从头来过。配合单轮模式(不实时打印、一次性收集全部消息),非常适合 CI/批处理:
import { query } from "@anthropic-ai/claude-agent-sdk";
import { writeFileSync } from "fs";
// 单轮模式:收集全部消息,适合后台/CI
const collected: any[] = [];
for await (const m of query({
prompt: "跑 npm test,修复所有失败用例,不要改测试本身",
options: { allowedTools: ["Read", "Edit", "Bash"], permissionMode: "acceptEdits", maxTurns: 30 },
})) {
collected.push(m);
}
const result = collected.find((m) => m.type === "result");
writeFileSync("fix-report.json", JSON.stringify(result, null, 2));
console.log(`cost=$${result.total_cost_usd?.toFixed(4)} turns=${result.num_turns}`);
// 续跑:用上一轮返回的 session_id 接着干
if (result.subtype === "error_max_turns") {
for await (const m of query({
prompt: "继续刚才未完成的修复",
options: { resume: result.session_id, allowedTools: ["Read", "Edit", "Bash"], permissionMode: "acceptEdits" },
})) {
/* ... */
}
}
记住两个生产要点:总是设 maxTurns(防失控烧钱);检查 subtype 区分成功与各类失败,据此决定是续跑还是告警。
下面是一个完整可运行的修复机器人:它接收一个目标文件,让 Agent 自主审查并修复 Bug,然后用独立的验证 Agent 跑测试确认修复有效,最后产出结构化报告。
先准备有 Bug 的样本:
// utils.ts(故意留两个 Bug)
export function calculateAverage(numbers: number[]): number {
let total = 0;
for (const n of numbers) total += n;
return total / numbers.length; // 空数组会除以零
}
export function getUserName(user: any): string {
return user.name.toUpperCase(); // user 为 null 会抛 TypeError
}
主程序 fixbot.ts:
import { query } from "@anthropic-ai/claude-agent-sdk";
import { writeFileSync } from "fs";
const TARGET = "./utils.ts";
interface Report {
target: string;
fixStatus: string;
turns: number;
costUsd: number;
verifyStatus: string;
summary: string;
}
// 阶段一:修复
async function fix(): Promise<any> {
console.log("🔧 阶段一:自主修复");
let final: any = null;
for await (const m of query({
prompt: `审查 ${TARGET} 中会导致崩溃的边界 Bug(如空数组、null 入参)并修复,保持函数签名不变。`,
options: {
cwd: ".",
allowedTools: ["Read", "Edit", "Grep"],
permissionMode: "acceptEdits",
maxTurns: 12,
systemPrompt: "你是严谨的修复代理,只改必要逻辑,加最小防御性代码,不引入新依赖。",
},
})) {
if (m.type === "assistant" && m.message?.content) {
for (const b of m.message.content) {
if ("name" in b) console.log(" ->", b.name);
}
}
if (m.type === "result") final = m;
}
return final;
}
// 阶段二:验证(独立 Agent,只读 + 跑测试,不改代码)
async function verify(): Promise<any> {
console.log("🧪 阶段二:闭环验证");
let final: any = null;
for await (const m of query({
prompt: `读 ${TARGET},确认 calculateAverage([]) 不再崩溃、getUserName(null) 不再抛错。用 node -e 实跑断言,报告通过或失败。`,
options: {
cwd: ".",
allowedTools: ["Read", "Bash"],
permissionMode: "bypassPermissions",
maxTurns: 8,
},
})) {
if (m.type === "result") final = m;
}
return final;
}
(async () => {
const fixRes = await fix();
const verifyRes = await verify();
const report: Report = {
target: TARGET,
fixStatus: fixRes?.subtype ?? "unknown",
turns: fixRes?.num_turns ?? 0,
costUsd: (fixRes?.total_cost_usd ?? 0) + (verifyRes?.total_cost_usd ?? 0),
verifyStatus: verifyRes?.subtype ?? "unknown",
summary: verifyRes?.result ?? "(无摘要)",
};
writeFileSync("fix-report.json", JSON.stringify(report, null, 2));
console.log("\n📋 报告:\n", JSON.stringify(report, null, 2));
})();
运行:npx tsx fixbot.ts。预期输出:阶段一打印 -> Read -> Edit 序列,阶段二打印 -> Read -> Bash,最后产出 fix-report.json,fixStatus 与 verifyStatus 均为 success,costUsd 给出真实花费。
这个案例体现了 L3 的精髓:两个独立 Agent 分工(修复者可写、验证者只读+执行),职责隔离,避免「自己改自己验」的盲区,正是企业 CI 修复机器人需要的形态。
flowchart LR
subgraph 你的程序
A[业务编排:Node/Python]
end
subgraph "Claude Agent SDK"
B[query:Agent循环]
C[工具执行:Read/Edit/Bash]
D[上下文/重试/会话]
end
E[PreToolUse门禁]
F[文件系统/CI环境]
G[第三方模型:Bedrock/Vertex/Azure]
A -->|prompt+options| B
B --> C
C --> E
E -->|批准| F
E -->|拒绝| B
F --> C
B --> D
D --> B
B -->|消息流| A
A -.->|认证/渠道| G
.env,必须在运行 shell 里 export ANTHROPIC_API_KEY。maxTurns,并对高成本工具做白名单;监控 total_cost_usd。allowedTools,systemPrompt 明确「只改必要逻辑」,并用只读验证 Agent 复核 diff。error_max_turns 频繁:任务太大,拆成多阶段(定位 → 修复 → 验证),用 resume 续跑,而非提高单轮上限。options.mcpServers 里声明 MCP 服务器配置,Agent 即可调用其工具,与内置工具一视同仁。code.claude.com/docs/en/agent-sdk/overview —— Python/TS 双语言入门与 bug-fixing agent 示例。github.com/anthropics/claude-agent-sdk-typescript —— 源码、issue 与从 Claude Code SDK 迁移指南。-p 与 hooks(L1/L2),否则难以理解 Agent 循环与消息流。[!tip] 51学通信知识星球 星主:爱卫生(微信:gprshome201101)——公众号「51学通信」创办者,19 年通信行业运维与讲师经验,著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书(各大电商平台有售)。 星球定位:深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域,主打运维岗位与信令分析。 当前规模:5200 篇主题 / 990 个回答,198 元/年。 加入福利:加入星球后请加爱卫生微信,拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题,另有群内分享。 AI 加持:51学通信知识星球已作为 skill 接入 WorkBuddy,使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例,做总结、找文档和 PPT 等场景全覆盖。
