Agent SDK实战：编程式AI代理开发

graph TD
    A[Agent SDK 核心架构] --> B[Agent Loop 代理循环]
    A --> C[Tool System 工具系统]
    A --> D[Hook System 钩子系统]
    A --> E[Subagent 子代理]

    B --> B1[消息发送]
    B --> B2[工具调用决策]
    B --> B3[结果处理]
    B --> B4[循环终止判断]

    C --> C1[内置工具: Read/Write/Bash]
    C --> C2[自定义工具: MCP In-Process]
    C --> C3[外部MCP服务器]

    D --> D1[PreToolUse: 工具调用前]
    D --> D2[PostToolUse: 工具调用后]
    D --> D3[Notification: 通知]

    E --> E1[Task Dispatch]
    E --> E2[Parallel Execution]
    E --> E3[Result Aggregation]

mkdir my-claude-agent && cd my-claude-agent
npm init -y
npm install @anthropic-ai/claude-agent-sdk
npm install -D typescript @types/node tsx
npx tsc --init

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "outDir": "./dist",
    "strict": true,
    "esModuleInterop": true
  }
}

export ANTHROPIC_API_KEY="sk-ant-xxxxx"

ANTHROPIC_API_KEY=sk-ant-xxxxx

// verify.ts
import { query } from "@anthropic-ai/claude-agent-sdk";

const result = await query({
  prompt: "回复'Agent SDK 连接成功'，不要执行其他操作",
});

console.log(result);

npx tsx verify.ts

import { query } from "@anthropic-ai/claude-agent-sdk";

const result = await query({
  prompt: "读取当前目录的 package.json，列出所有依赖项",
  options: {
    model: "claude-sonnet-4-20250514",
    maxTurns: 10,
    allowedTools: ["Read", "Bash"],
    cwd: process.cwd(),
  },
});

import { query } from "@anthropic-ai/claude-agent-sdk";

const stream = query({
  prompt: "分析 src/ 目录下所有 TypeScript 文件的代码质量",
  options: {
    outputFormat: "stream-json",
    allowedTools: ["Read", "Glob", "Grep"],
  },
});

for await (const event of stream) {
  switch (event.type) {
    case "assistant_message":
      process.stdout.write(event.content);
      break;
    case "tool_use":
      console.log(`\n[工具调用] ${event.toolName}: ${JSON.stringify(event.input)}`);
      break;
    case "tool_result":
      console.log(`[工具结果] 长度: ${event.output.length} 字符`);
      break;
    case "result":
      console.log("\n--- 代理执行完成 ---");
      console.log(`总轮次: ${event.totalTurns}`);
      console.log(`总Token: ${event.totalTokenUsage}`);
      break;
  }
}

sequenceDiagram
    participant Dev as 开发者代码
    participant SDK as Agent SDK
    participant Claude as Claude API
    participant Tools as 工具层

    Dev->>SDK: query(prompt, options)
    SDK->>Claude: 发送消息 + 工具定义
    Claude->>SDK: 返回响应(含工具调用)

    loop Agent Loop
        SDK->>Tools: 执行工具调用
        Tools->>SDK: 返回工具结果
        SDK->>Claude: 发送工具结果
        Claude->>SDK: 继续响应或完成
    end

    SDK->>Dev: 返回最终结果 / 流式事件

// tools/database-query.ts
import { defineTool } from "@anthropic-ai/claude-agent-sdk";

export const databaseQueryTool = defineTool({
  name: "database_query",
  description: "查询 PostgreSQL 数据库，执行只读 SQL 查询。仅支持 SELECT 语句。",
  inputSchema: {
    type: "object" as const,
    properties: {
      sql: {
        type: "string",
        description: "要执行的 SQL SELECT 查询语句",
      },
      database: {
        type: "string",
        description: "数据库名称，默认为 'main'",
        enum: ["main", "analytics", "logs"],
      },
      limit: {
        type: "number",
        description: "返回结果的最大行数，默认 100",
      },
    },
    required: ["sql"],
  },
  handler: async (input) => {
    // 安全校验：只允许 SELECT
    const normalizedSql = input.sql.trim().toUpperCase();
    if (!normalizedSql.startsWith("SELECT")) {
      return {
        isError: true,
        content: "安全限制：仅支持 SELECT 查询",
      };
    }

    const limit = input.limit ?? 100;
    const sqlWithLimit = normalizedSql.includes("LIMIT")
      ? input.sql
      : `${input.sql.replace(/;$/, "")} LIMIT ${limit}`;

    // 这里接入实际的数据库客户端
    const results = await executeQuery(input.database ?? "main", sqlWithLimit);

    return {
      content: JSON.stringify(results, null, 2),
    };
  },
});

// tools/slack-notify.ts
import { defineTool } from "@anthropic-ai/claude-agent-sdk";

export const slackNotifyTool = defineTool({
  name: "slack_notify",
  description: "向指定 Slack 频道发送通知消息。用于在关键操作完成后通知团队。",
  inputSchema: {
    type: "object" as const,
    properties: {
      channel: {
        type: "string",
        description: "Slack 频道名称，如 '#deployments'",
      },
      message: {
        type: "string",
        description: "要发送的消息内容",
      },
      severity: {
        type: "string",
        description: "消息严重级别",
        enum: ["info", "warning", "critical"],
      },
    },
    required: ["channel", "message"],
  },
  handler: async (input) => {
    const colorMap = { info: "#36a64f", warning: "#f2c744", critical: "#e01e5a" };

    const payload = {
      channel: input.channel,
      attachments: [{
        color: colorMap[input.severity ?? "info"],
        text: input.message,
        ts: Math.floor(Date.now() / 1000).toString(),
      }],
    };

    const response = await fetch(process.env.SLACK_WEBHOOK_URL!, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify(payload),
    });

    return {
      content: response.ok
        ? `消息已发送到 ${input.channel}`
        : `发送失败: ${response.status} ${response.statusText}`,
    };
  },
});

// agent.ts
import { query } from "@anthropic-ai/claude-agent-sdk";
import { databaseQueryTool } from "./tools/database-query";
import { slackNotifyTool } from "./tools/slack-notify";

const result = await query({
  prompt: "检查过去 24 小时的错误日志数量，如果超过 100 条则通知 #ops 频道",
  options: {
    model: "claude-sonnet-4-20250514",
    maxTurns: 15,
    customTools: [databaseQueryTool, slackNotifyTool],
    allowedTools: ["Read", "Bash", "database_query", "slack_notify"],
    systemPrompt: "你是一个运维监控助手。先查询数据，再根据结果决定是否需要告警。",
  },
});

console.log(result);

flowchart LR
    subgraph 自定义工具层
        A[database_query<br/>SQL只读查询]
        B[slack_notify<br/>Slack消息推送]
        C[jira_create<br/>Jira工单创建]
        D[metric_fetch<br/>监控指标拉取]
    end

    subgraph Agent SDK 核心
        E[Agent Loop<br/>推理-决策-执行]
    end

    subgraph 内置工具层
        F[Read/Write/Edit]
        G[Bash]
        H[Glob/Grep]
    end

    I[用户 Prompt] --> E
    E --> A & B & C & D & F & G & H
    A & B & C & D & F & G & H --> E

    style A fill:#4a9eff,color:#fff
    style B fill:#e01e5a,color:#fff
    style C fill:#0052cc,color:#fff
    style D fill:#f4a261,color:#000

// hooks/pre-tool-security.ts
import { Hook } from "@anthropic-ai/claude-agent-sdk";

export const securityHook: Hook = {
  name: "security-gate",
  events: ["PreToolUse"],
  handler: async (event) => {
    // 阻止对生产环境文件的写入
    if (event.toolName === "Write" || event.toolName === "Edit") {
      const filePath = event.input.file_path ?? "";
      const blockedPaths = [
        "/etc/",
        "/var/www/production/",
        "/opt/deploy/",
      ];

      if (blockedPaths.some((p) => filePath.startsWith(p))) {
        return {
          decision: "block" as const,
          reason: `安全策略禁止修改生产路径: ${filePath}`,
        };
      }
    }

    // 审计所有 Bash 命令
    if (event.toolName === "Bash") {
      const command = event.input.command ?? "";
      const dangerousPatterns = [
        /rm\s+-rf\s+\//,
        /DROP\s+TABLE/i,
        /DELETE\s+FROM/i,
        />\s*\/dev\//,
        /curl.*\|\s*sh/,
      ];

      for (const pattern of dangerousPatterns) {
        if (pattern.test(command)) {
          return {
            decision: "block" as const,
            reason: `危险命令被安全策略拦截: ${command.substring(0, 80)}...`,
          };
        }
      }

      // 记录审计日志
      await logAudit({
        tool: event.toolName,
        input: event.input,
        timestamp: new Date().toISOString(),
      });
    }

    return { decision: "allow" as const };
  },
};

// hooks/post-tool-metrics.ts
import { Hook } from "@anthropic-ai/claude-agent-sdk";

export const metricsHook: Hook = {
  name: "execution-metrics",
  events: ["PostToolUse"],
  handler: async (event) => {
    const metrics = {
      tool: event.toolName,
      duration: event.duration,
      success: !event.isError,
      tokenCost: event.tokenUsage,
      timestamp: new Date().toISOString(),
    };

    // 推送到监控系统
    await pushToMetrics(metrics);

    // 如果执行失败超过阈值，自动告警
    if (event.isError) {
      console.error(`[工具失败] ${event.toolName}: ${event.error}`);
    }

    return { decision: "allow" as const };
  },
};

import { query } from "@anthropic-ai/claude-agent-sdk";
import { securityHook } from "./hooks/pre-tool-security";
import { metricsHook } from "./hooks/post-tool-metrics";

const result = await query({
  prompt: "重构 src/utils/ 下的所有工具函数，添加 JSDoc 注释",
  options: {
    hooks: [securityHook, metricsHook],
    maxTurns: 30,
  },
});

flowchart TD
    A[Claude 决定调用工具] --> B{PreToolUse Hook}
    B -->|block| C[返回拒绝原因<br/>代理重新规划]
    B -->|allow| D[执行工具]
    D --> E{PostToolUse Hook}
    E -->|block| F[丢弃结果<br/>代理重新规划]
    E -->|allow| G[结果返回代理<br/>继续推理]

    H[代理启动] --> I{Notification Hook}
    I --> J[记录启动事件]

    style B fill:#ff6b6b,color:#fff
    style E fill:#4ecdc4,color:#fff
    style C fill:#ffe66d,color:#000
    style F fill:#ffe66d,color:#000

// multi-file-review.ts
import { query } from "@anthropic-ai/claude-agent-sdk";
import * as path from "path";
import * as fs from "fs/promises";

async function reviewFile(filePath: string) {
  return query({
    prompt: `审查以下文件的代码质量，关注：安全性、性能、可维护性。

文件路径: ${filePath}

请输出结构化的审查报告，包含问题列表和改进建议。`,
    options: {
      model: "claude-sonnet-4-20250514",
      maxTurns: 5,
      allowedTools: ["Read"],
      outputFormat: "json",
      systemPrompt: "你是一位高级代码审查工程师。输出 JSON 格式的审查报告。",
    },
  });
}

async function parallelReview(directory: string) {
  // 1. 扫描目标文件
  const files = await fs.readdir(directory, { recursive: true });
  const tsFiles = files
    .filter((f) => f.toString().endsWith(".ts"))
    .map((f) => path.join(directory, f.toString()));

  console.log(`发现 ${tsFiles.length} 个 TypeScript 文件`);

  // 2. 并行启动子代理（限制并发数）
  const CONCURRENCY = 5;
  const results: Map<string, any> = new Map();

  for (let i = 0; i < tsFiles.length; i += CONCURRENCY) {
    const batch = tsFiles.slice(i, i + CONCURRENCY);
    const batchResults = await Promise.all(
      batch.map(async (file) => {
        const result = await reviewFile(file);
        return { file, result };
      })
    );
    batchResults.forEach((r) => results.set(r.file, r.result));
    console.log(`已完成 ${Math.min(i + CONCURRENCY, tsFiles.length)}/${tsFiles.length}`);
  }

  // 3. 汇总报告
  return Object.fromEntries(results);
}

// 执行
const report = await parallelReview("./src");
console.log(JSON.stringify(report, null, 2));

// pipeline.ts — 代码迁移流水线
import { query } from "@anthropic-ai/claude-agent-sdk";

// 阶段1: 分析旧代码结构
const analysis = await query({
  prompt: "分析 src/ 目录的代码结构，识别所有需要从 JavaScript 迁移到 TypeScript 的文件",
  options: {
    allowedTools: ["Read", "Glob", "Grep"],
    maxTurns: 10,
    outputFormat: "json",
  },
});

// 阶段2: 生成迁移计划
const migrationPlan = await query({
  prompt: `基于以下分析结果，生成迁移计划。按依赖顺序排列文件。

分析结果:
${JSON.stringify(analysis)}

输出 JSON 数组，每个元素包含: file, priority, dependencies, estimated_complexity`,
  options: {
    maxTurns: 3,
    outputFormat: "json",
  },
});

// 阶段3: 逐文件执行迁移
const plan = JSON.parse(migrationPlan);
for (const item of plan) {
  console.log(`迁移: ${item.file} (优先级: ${item.priority})`);

  await query({
    prompt: `将 ${item.file} 从 JavaScript 迁移到 TypeScript。
要求：
1. 添加完整的类型注解
2. 修复所有类型错误
3. 保持原有功能不变
4. 添加必要的 interface 和 type 定义`,
    options: {
      allowedTools: ["Read", "Write", "Edit", "Bash"],
      maxTurns: 20,
      systemPrompt: "你是一个 TypeScript 迁移专家。严格保持功能等价。",
    },
  });
}

// 阶段4: 验证
const validation = await query({
  prompt: "运行 TypeScript 编译检查，确认所有文件无类型错误。如果有错误，修复它们。",
  options: {
    allowedTools: ["Bash", "Read", "Edit"],
    maxTurns: 15,
  },
});

flowchart TD
    A[主代理: 分析阶段] -->|JSON 输出| B[主代理: 规划阶段]
    B -->|迁移计划| C{文件列表}

    C --> D1[子代理: 迁移 file1.ts]
    C --> D2[子代理: 迁移 file2.ts]
    C --> D3[子代理: 迁移 file3.ts]
    C --> D4[子代理: 迁移 fileN.ts]

    D1 --> E[主代理: 验证阶段]
    D2 --> E
    D3 --> E
    D4 --> E

    E -->|编译检查| F{有错误?}
    F -->|是| G[子代理: 修复错误]
    G --> E
    F -->|否| H[迁移完成 ✓]

    style A fill:#6c5ce7,color:#fff
    style B fill:#0984e3,color:#fff
    style E fill:#00b894,color:#fff
    style H fill:#00b894,color:#fff

code-review-bot/
├── src/
│   ├── index.ts          # 入口文件
│   ├── tools/
│   │   ├── git-diff.ts   # Git 差异工具
│   │   └── comment.ts    # PR 评论工具
│   ├── hooks/
│   │   └── audit.ts      # 审计 Hook
│   └── prompts/
│       └── review.ts     # 审查提示词模板
├── package.json
└── tsconfig.json

// src/tools/git-diff.ts
import { defineTool } from "@anthropic-ai/claude-agent-sdk";

export const gitDiffTool = defineTool({
  name: "git_diff",
  description: "获取 Git 仓库中指定提交范围的代码差异。用于代码审查。",
  inputSchema: {
    type: "object" as const,
    properties: {
      base: {
        type: "string",
        description: "基准分支或提交，默认为 'main'",
      },
      head: {
        type: "string",
        description: "目标分支或提交，默认为 'HEAD'",
      },
      stat: {
        type: "boolean",
        description: "是否只返回统计信息而非完整差异",
      },
    },
    required: [],
  },
  handler: async (input) => {
    const { execSync } = await import("child_process");

    const base = input.base ?? "main";
    const head = input.head ?? "HEAD";

    try {
      let command: string;
      if (input.stat) {
        command = `git diff --stat ${base}...${head}`;
      } else {
        command = `git diff ${base}...${head} --diff-filter=ACMR`;
      }

      const output = execSync(command, {
        maxBuffer: 10 * 1024 * 1024, // 10MB
        encoding: "utf-8",
      });

      return { content: output || "没有发现差异" };
    } catch (error: any) {
      return {
        isError: true,
        content: `Git 命令执行失败: ${error.message}`,
      };
    }
  },
});

// src/tools/comment.ts
import { defineTool } from "@anthropic-ai/claude-agent-sdk";

interface ReviewComment {
  file: string;
  line: number;
  severity: "info" | "warning" | "critical";
  message: string;
  suggestion?: string;
}

export const postCommentTool = defineTool({
  name: "post_review_comment",
  description: "发布代码审查评论。支持对特定文件行添加审查意见。",
  inputSchema: {
    type: "object" as const,
    properties: {
      summary: {
        type: "string",
        description: "整体审查摘要",
      },
      comments: {
        type: "array",
        items: {
          type: "object",
          properties: {
            file: { type: "string" },
            line: { type: "number" },
            severity: { type: "string", enum: ["info", "warning", "critical"] },
            message: { type: "string" },
            suggestion: { type: "string" },
          },
          required: ["file", "line", "severity", "message"],
        },
        description: "具体审查评论列表",
      },
      approved: {
        type: "boolean",
        description: "是否批准此 PR",
      },
    },
    required: ["summary", "comments", "approved"],
  },
  handler: async (input) => {
    const { summary, comments, approved } = input;

    // 构建审查报告
    const emoji = { info: "ℹ️", warning: "⚠️", critical: "🔴" };
    let report = `## 代码审查报告\n\n`;
    report += `**结论**: ${approved ? "✅ 批准合并" : "❌ 需要修改"}\n\n`;
    report += `### 摘要\n${summary}\n\n`;
    report += `### 详细评论 (${comments.length} 条)\n\n`;

    for (const comment of comments as ReviewComment[]) {
      report += `${emoji[comment.severity]} **${comment.file}:${comment.line}**\n`;
      report += `> ${comment.message}\n`;
      if (comment.suggestion) {
        report += `建议: \`${comment.suggestion}\`\n`;
      }
      report += "\n";
    }

    // 如果配置了 GitHub，可以自动发布到 PR
    if (process.env.GITHUB_TOKEN && process.env.PR_NUMBER) {
      await postToGitHub(report);
    }

    // 同时输出到控制台
    console.log(report);

    return {
      content: `审查报告已生成。${comments.length} 条评论。结论: ${approved ? "批准" : "需修改"}。`,
    };
  },
});

async function postToGitHub(body: string) {
  const repo = process.env.GITHUB_REPOSITORY!;
  const prNumber = process.env.PR_NUMBER!;

  await fetch(`https://api.github.com/repos/${repo}/issues/${prNumber}/comments`, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${process.env.GITHUB_TOKEN}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ body }),
  });
}

// src/prompts/review.ts
export function buildReviewPrompt(diffStat: string): string {
  return `你是一位资深代码审查工程师。请审查以下代码变更。

## 审查维度
1. **安全性**: SQL 注入、XSS、路径遍历、硬编码密钥
2. **性能**: N+1 查询、内存泄漏、不必要的循环
3. **可维护性**: 命名规范、函数复杂度、重复代码
4. **正确性**: 逻辑错误、边界条件、错误处理

## 变更统计
${diffStat}

## 审查流程
1. 使用 git_diff 工具获取完整差异（不带 stat 参数）
2. 逐文件分析变更内容
3. 使用 post_review_comment 工具发布审查结果

## 输出要求
- summary: 用中文写一段总体评价
- comments: 每个问题一条评论，必须包含文件名和行号
- approved: 只有 0 个 critical 问题时才批准`;
}

// src/index.ts
import { query } from "@anthropic-ai/claude-agent-sdk";
import { gitDiffTool } from "./tools/git-diff";
import { postCommentTool } from "./tools/comment";
import { securityHook } from "./hooks/audit";
import { buildReviewPrompt } from "./prompts/review";

async function main() {
  const targetBranch = process.env.TARGET_BRANCH ?? "main";
  const sourceBranch = process.env.SOURCE_BRANCH ?? "HEAD";

  // 获取差异统计（用 Bash 工具）
  const statResult = await query({
    prompt: `执行 git diff --stat ${targetBranch}...${sourceBranch}，返回输出`,
    options: {
      allowedTools: ["Bash"],
      maxTurns: 2,
    },
  });

  // 启动审查代理
  const result = await query({
    prompt: buildReviewPrompt(statResult),
    options: {
      model: "claude-sonnet-4-20250514",
      maxTurns: 25,
      customTools: [gitDiffTool, postCommentTool],
      allowedTools: ["Read", "Grep", "git_diff", "post_review_comment"],
      hooks: [securityHook],
      systemPrompt: `你是一个代码审查机器人。
规则：
- 必须使用工具获取实际代码，不要猜测
- 每个问题必须有具体的文件名和行号
- 安全问题标记为 critical，性能标记为 warning，其他标记为 info
- 审查完毕后必须调用 post_review_comment 发布结果`,
    },
  });

  console.log("\n=== 审查代理执行完毕 ===");
  console.log(result);
}

main().catch(console.error);

# 命令行运行
TARGET_BRANCH=main SOURCE_BRANCH=feature/login npx tsx src/index.ts

# CI/CD 中运行（GitHub Actions）
env:
  GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
  GITHUB_REPOSITORY: ${{ github.repository }}
  PR_NUMBER: ${{ github.event.pull_request.number }}
  TARGET_BRANCH: main
  SOURCE_BRANCH: ${{ github.head_ref }}

import { query, saveSession, loadSession } from "@anthropic-ai/claude-agent-sdk";

// 第一次运行，保存会话
const result = await query({
  prompt: "开始分析项目架构",
  options: { sessionId: "arch-analysis-001" },
});

// 后续运行，恢复上下文
const continued = await query({
  prompt: "继续上次的架构分析，现在重点关注数据库层",
  options: { sessionId: "arch-analysis-001" },
});

# CLI: 快速批量操作
for dir in packages/*/; do
  claude -p --output-format json \
    "检查 $dir 的 package.json 是否有过时的依赖" \
    --allowedTools "Read" \
    --max-turns 3
done

# SDK: 复杂的多步骤代理
npx tsx complex-agent.ts

const result = await query({
  prompt: "简要总结这个文件的功能",
  options: {
    model: "claude-haiku-4-5-20251001", // 简单任务用 Haiku
    maxTurns: 3,                         // 限制轮次
    allowedTools: ["Read"],              // 最小工具集
  },
});

// 复杂任务才用 Sonnet
const complexResult = await query({
  prompt: "重构整个模块的架构",
  options: {
    model: "claude-sonnet-4-20250514",
    maxTurns: 30,
    allowedTools: ["Read", "Write", "Edit", "Bash", "Glob", "Grep"],
  },
});

graph TB
    subgraph 用户应用
        A[TypeScript 代码]
    end

    subgraph "Agent SDK 层"
        B[query 入口函数]
        C[Agent Loop<br/>推理→决策→执行]
        D[Hook Engine<br/>PreToolUse / PostToolUse]
    end

    subgraph "工具层"
        E[内置工具<br/>Read Write Edit Bash<br/>Glob Grep WebSearch]
        F[自定义工具<br/>defineTool 注册<br/>MCP In-Process Server]
        G[外部 MCP<br/>第三方服务集成]
    end

    subgraph "Claude API"
        H[消息发送/接收]
        I[工具调用解析]
        J[流式响应]
    end

    A --> B
    B --> C
    C --> D
    D --> E & F & G
    E & F & G --> D
    D --> C
    C <--> H
    H <--> I
    H <--> J

    style B fill:#6c5ce7,color:#fff
    style C fill:#0984e3,color:#fff
    style D fill:#fdcb6e,color:#000
    style F fill:#00b894,color:#fff