<!-- title: Agent SDK编程式集成与多仓库批量自动化 -->

## 概述

本教程聚焦 Claude Code 从"命令行工具"升级为"可编程运行时"的两大能力：**Agent SDK**（TypeScript/Python 编程式调用）与 **Routines**（保存的配置+多仓库工作集）。读者将学会用 30 行代码驱动 Claude 自主读文件、跑命令、改代码，并用 `xargs`/`parallel` 把同一套指令批量投放到几十个仓库并行执行。适用人群：已熟悉交互式 Claude Code、希望把它嵌入脚本/服务/批处理流水线的后端工程师、平台/SRE、研发效能团队。

## 前置条件

- Claude Code CLI 已安装且可正常交互运行（`claude --version` 有输出）
- 一个可用的认证方式：OAuth 登录（`claude`）或 `ANTHROPIC_API_KEY` 环境变量
- Node.js ≥ 20（TypeScript SDK）或 Python ≥ 3.10（Python SDK）
- 基本的命令行与 shell 脚本经验；了解 JSON 流式解析更佳
- 准备 3~5 个本地的 git 仓库目录用于批量实战（可以是任意示例项目）
## 一、为什么需要编程式集成

交互式的 `claude` 适合人在回路（human-in-the-loop）的场景：你打字、它思考、你审核。但真实工程里有一类需求它力不从心——**把 Claude 当成一个被调用的函数**：

- 每天凌晨 3 点，对 50 个微服务仓库跑一遍"依赖漏洞扫描+自动提 PR"
- 在 CI 流水线的某个 stage 里，让 Claude 根据失败的日志生成修复补丁，再交给人工 review
- 给团队内部写一个"代码体检"HTTP 服务：传入仓库 URL，返回结构化诊断报告
- 构建一个编排层，先调 Claude 分析、再调 Codex 改、最后调 Gemini 复核

这类场景的共同点是：**没有人在屏幕前敲回车**。Claude 必须以子进程或库的形式被唤起、被喂参数、被收集输出、被关闭。Claude Code 提供了三层能力来满足它，复杂度递增：

| 层次 | 形态 | 适用 | 本章 |
|---|---|---|---|
| L1 | `claude -p` 无头模式 | 简单脚本、一次性任务 | 第一节 |
| L2 | Agent SDK（TS/Python） | 服务化、流式、复杂编排 | 第二、三节 |
| L3 | Routines + 批处理 | 多仓库、可复用配置 | 第四、五节 |

下面逐层展开。

## 二、L1：`-p` 无头模式——最轻量的集成

`-p`（`--print`）让 Claude 不启动 TUI，而是接收一个 prompt、自主执行若干轮工具调用、把最终答案打印到 stdout 后退出。这是把 Claude 嵌入 shell 脚本最直接的方式。

### 2.1 最小可用调用

```bash
cd ~/repos/my-service
claude -p "用三句话总结这个仓库的架构，并指出最可能的技术债。"
```

预期输出：一段纯文本落到 stdout，进程退出码为 0。把输出重定向到文件即可留档：

```bash
claude -p "..." > review.txt
```

### 2.2 关键参数

```bash
claude -p \
  --output-format json \          # 结构化输出，便于 jq 解析
  --verbose \                     # json 模式下附带每一步的会话信息
  --max-turns 15 \                # 限制自主轮数，防止失控
  --allowedTools "Read,Grep,Glob,Bash"\  # 收紧权限白名单
  --model claude-sonnet-4-6 \     # 指定模型
  "为 src/ 下所有导出的函数补上 JSDoc 注释，只改注释不改逻辑"
```

参数说明：

- `--output-format json`：返回一个 JSON 对象，包含 `result`（最终文本）、`cost_usd`（本次花费）、`duration_ms`、`num_turns`、`session_id` 等。**做成本核算和任务追踪时务必用它**。
- `--output-format stream-json`：以换行分隔的 JSON 流输出每个事件（工具调用、文本块、结果），适合实时展示进度条或事件驱动处理。
- `--max-turns`：限制 Agent 自主循环次数。生产环境强烈建议设置，避免 prompt 不清晰时模型反复试错烧 token。
- `--allowedTools`：在 CLI 已授予权限的基础上再做一次白名单收紧。例如只想让它"读和分析"，就只给 `Read,Grep,Glob`，不给 `Edit,Write`。
- `--permission-mode`：可选 `acceptEdits`、`bypassPermissions`、`plan` 等，控制它能否自动执行写操作。批量无人值守场景常用 `acceptEdits`。

### 2.3 流式输出解析（stream-json）

```bash
claude -p --output-format stream-json --verbose "列出所有 TODO 注释的位置" | \
  jq -r 'select(.type=="tool_use" and .name=="Grep") | .input.pattern'
```

每个流事件是一个独立 JSON 对象，按类型（`system`、`assistant`、`tool_use`、`tool_result`、`result`）过滤即可。这在构建"实时进度看板"时非常有用。

### 2.4 常见问题

| 现象 | 原因 | 解决 |
|---|---|---|
| 提示需要授权，脚本卡住 | 默认权限模式会等待用户确认 | 加 `--permission-mode acceptEdits`，或预先在 `settings.json` 配 `allow` 列表 |
| 退出码非 0 但看不出错 | stderr 被吞了 | 用 `2>err.log` 单独捕获 stderr |
| JSON 解析失败 | 默认输出是纯文本 | 必须显式加 `--output-format json` |
| 重复登录态丢失 | 每个子进程独立认证 | 用 `ANTHROPIC_API_KEY` 而非 OAuth，或确保共享同一 `~/.claude` |

## 三、L2：Agent SDK——把 Claude 当函数调用

当 `-p` 不够用时（比如需要把 Claude 嵌进一个长期运行的服务、需要在同一会话里多轮交互、需要拿到结构化对象而非解析文本），就轮到 Agent SDK 上场。它把 Claude Code 的整套运行时（工具系统、权限、会话、流式事件）封装成一个库函数 `query()`。

### 3.1 安装

```bash
# TypeScript / Node
npm install @anthropic-ai/claude-agent-sdk

# Python
pip install claude-agent-sdk
```

### 3.2 TypeScript 最小示例

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

async function main() {
  const messages = [{ role: "user", content: "列出 src/ 下所有 TypeScript 文件并统计每个文件的行数" }];

  // query() 是一个异步生成器，逐个 yield 事件
  for await (const message of query({
    prompt: "分析本仓库的测试覆盖率现状，给出提升到 80% 的优先动作清单",
    options: {
      cwd: "/home/me/repos/my-service",  // 工作目录，决定 Claude 能操作哪个仓库
      allowedTools: ["Read", "Grep", "Glob", "Bash"],  // 工具白名单
      maxTurns: 15,
      model: "claude-sonnet-4-6",
      permissionMode: "acceptEdits",
    },
  })) {
    // 每条 message 是一个流事件：assistant 文本、tool_use、tool_result、result...
    if (message.type === "assistant") {
      for (const block of message.message.content) {
        if (block.type === "text") process.stdout.write(block.text);
      }
    }
  }
}

main();
```

### 3.3 Python 最小示例

```python
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        cwd="/home/me/repos/my-service",
        allowed_tools=["Read", "Grep", "Glob", "Bash"],
        max_turns=15,
        model="claude-sonnet-4-6",
        permission_mode="acceptEdits",
    )

    async for message in query(
        prompt="找出所有未处理的 Promise rejection 和 bare except，列出文件:行号",
        options=options,
    ):
        if getattr(message, "type", None) == "assistant":
            for block in message.message.content:
                if block.type == "text":
                    print(block.text, end="")

anyio.run(main)
```

### 3.4 三种使用形态

SDK 并非只有"一次性 prompt"一种用法，理解这三种形态决定了你如何编排：

**形态 A：单次查询（one-shot）**。一个 prompt，自主跑完，返回结果。对应上面的示例，最适合"分析类""报告类"任务。

**形态 B：多轮对话（multi-turn）**。保留 `session_id`，下一次 `query()` 时传入 `resume`，在同一上下文里继续追问。适合需要人工中途介入审核、再让它继续的场景：

```python
# 第一轮：让它调研
result = await query(prompt="调研迁移到 ESM 的可行性", options=opts)
session_id = result....  # 从结果事件中取出

# （人工审核中间结论后）

# 第二轮：基于同一会话继续
await query(prompt="好，现在按你建议的方案动手改 package.json 和 tsconfig", 
            options={**opts, "resume": session_id})
```

**形态 C：自定义工具（custom tools）**。SDK 允许你注册宿主进程里的函数作为 MCP 工具，让 Claude 在执行过程中回调你的代码。例如把"查公司内部监控平台""查 Jira 工单"封装成工具，Claude 就能在分析代码时顺手拉取线上指标。这是把 Claude 与企业内部系统打通的关键闸门。

```typescript
// 注册一个自定义工具：查内部 SLO 面板
const options = {
  mcpServers: {
    "slo-panel": {
      type: "stdio",
      command: "node",
      args: ["./slo-mcp-server.js"],  // 你自己实现的 MCP server
    },
  },
  // Claude 会像调用 Read/Bash 一样调用 slo-panel 暴露的工具
};
```

### 3.5 常见问题

- **`query()` 一直不返回**：多半是 `maxTurns` 没设、模型在死循环。务必设上限，并在外层加超时（如 Python 的 `anyio.fail_after`）。
- **工具调用被拒**：SDK 默认沿用 `settings.json` 的权限；若想完全程序化，用 `permissionMode: "bypassPermissions"`（仅在可信沙箱内），或注册 `canUseTool` 回调自定义审批逻辑。
- **跨进程会话丢失**：`session_id` 保存在 `cwd` 下的会话存储里，宿主进程要有读写权限，且 `cwd` 要稳定。

## 四、L3：Routines——可复用的"配置+工作集"

当同一套指令要在多个仓库上反复跑时，把 prompt、工具白名单、模型、目标仓库列表打包成一个**Routine**，是最干净的复用方式。Routines 是 Claude Code 官方提供的"保存的配置"机制：一个 Routine = 一个 prompt + 一个或多个仓库 + 一组配置约束。

### 4.1 为什么需要 Routine

裸用 `claude -p` 的问题在于：参数散落在各处脚本里，改一次要改 N 个地方；不同人跑出的结果不一致（有人忘了加 `--max-turns`，有人忘了限制工具）。Routine 把这些固化下来，让"团队统一的代码体检流程"成为一个可分发、可版本化的对象。

### 4.2 定义与运行

Routines 通常通过 Claude Code 的配置文件或 `claude` 交互式命令创建/保存（具体语法以你本机版本为准）。一个 Routine 的逻辑结构如下：

```yaml
# 概念性结构（实际以本机版本配置格式为准）
name: weekly-security-scan
prompt: |
  扫描本仓库的依赖清单与代码，找出已知 CVE、硬编码密钥、
  不安全的反序列化点。输出一份 Markdown 报告，按严重程度排序。
repositories:
  - /home/me/repos/billing-service
  - /home/me/repos/auth-gateway
  - /home/me/repos/notification-worker
options:
  allowedTools: [Read, Grep, Glob, Bash]
  maxTurns: 20
  model: claude-sonnet-4-6
  permissionMode: acceptEdits
  outputFormat: json
```

运行一个 Routine 等价于：对其中**每一个仓库**，用相同的 prompt 和 options 跑一次独立的 Claude 会话，并汇总结果。仓库之间的执行天然隔离——各自的工作目录、各自的上下文，互不污染。

### 4.3 Routine vs `-p` vs SDK 的取舍

| 需求 | 推荐方案 |
|---|---|
| 一次性、临时任务 | `claude -p` |
| 团队共享、重复执行的固定流程 | Routine |
| 需要嵌入自有服务、复杂事件处理 | Agent SDK |
| 需要自定义工具、与企业系统打通 | Agent SDK + MCP |

## 五、多仓库并行批处理实战

这是本教程的重头戏。目标：对一组仓库**并行**执行同一套 Claude 指令，收集结构化结果，控制并发与成本。

### 5.1 并发模型：每仓库一个独立进程

关键认知：**Claude Code 的每次 `-p` 调用或 SDK `query()` 都是一个独立的 OS 进程/会话**。这意味着多仓库天然适合用进程级并行——不存在共享状态、不需要锁、不会互相干扰。并发度只受两个约束：

1. **API 速率限制**（RPM/TPM）：并发太高会被 429 拒绝
2. **本机 CPU/内存**：每个 Claude 进程会持有一定上下文

经验值：中小仓库（< 10 万行）的"分析类"任务，**并发 4~8** 是一个安全的起点；写操作类任务降到 2~4。

### 5.2 方案 A：xargs 并行（最简单）

```bash
#!/usr/bin/env bash
set -euo pipefail

REPO_LIST="repos.txt"        # 每行一个仓库绝对路径
MAX_PARALLEL=4
OUT_DIR="reports/$(date +%Y%m%d)"
mkdir -p "$OUT_DIR"

# 关键：-P 控制并发数；每个仓库一个独立 Claude 进程
cat "$REPO_LIST" | xargs -I {} -P "$MAX_PARALLEL" bash -c '
  repo="$1"
  name=$(basename "$repo")
  out="'"$OUT_DIR"'/${name}.json"
  cd "$repo"
  claude -p \
    --output-format json \
    --max-turns 15 \
    --allowedTools "Read,Grep,Glob,Bash" \
    --permission-mode acceptEdits \
    "审查本仓库的安全性，列出高/中/低危问题各不超过 5 条，给出文件:行号" \
    > "$out" 2> "${out%.json}.err"
  echo "✓ $name (exit=$?)"
' _ {}
```

要点：

- `xargs -P 4` 让最多 4 个仓库同时跑
- `_ {}` 把当前仓库名作为 `$1` 传进内联脚本
- 每个仓库的 stdout/stderr 分别落盘，互不串扰
- 输出是 JSON，后续可用 `jq` 汇总

### 5.3 方案 B：GNU parallel（更强的控流）

```bash
#!/usr/bin/env bash
set -euo pipefail

OUT_DIR="reports/$(date +%Y%m%d)"
mkdir -p "$OUT_DIR"
export OUT_DIR

# parallel 支持 --jobs、--load（按系统负载限流）、--retries（失败重试）
parallel --jobs 4 --retries 1 --joblog "$OUT_DIR/joblog.txt" '
  repo={}
  name=$(basename "$repo")
  cd "$repo"
  claude -p --output-format json --max-turns 15 \
    --allowedTools "Read,Grep,Glob" \
    "生成本仓库的架构概述（500 字内）" \
    > "'"$OUT_DIR"'/${name}.json"
' :::: repos.txt

# joblog 记录每个任务的退出码、耗时，便于事后审计
```

GNU parallel 相比 xargs 的优势：`--joblog` 自动记录每任务状态、`--retries` 自动重试被 429 打断的任务、`--bar` 显示进度条。

### 5.4 方案 C：SDK 编排（最灵活）

当 xargs/parallel 不够时——例如要按仓库大小动态分配并发、要在结果间做去重合并、要对失败仓库做差异化重试——就用 SDK 自己写编排器：

```python
import asyncio, json, os
from pathlib import Path
from claude_agent_sdk import query, ClaudeAgentOptions

REPOS = Path("repos.txt").read_text().splitlines()
MAX_CONCURRENCY = 4
PROMPT = "审查安全性，按严重程度列出问题，输出 JSON：{issues:[{severity,file,line,desc}]}"

async def scan_one(repo: str, sem: asyncio.Semaphore):
    async with sem:
        opts = ClaudeAgentOptions(
            cwd=repo,
            allowed_tools=["Read", "Grep", "Glob", "Bash"],
            max_turns=15,
            model="claude-sonnet-4-6",
            permission_mode="acceptEdits",
        )
        final_text = ""
        try:
            async for msg in query(prompt=PROMPT, options=opts):
                if getattr(msg, "type", None) == "result":
                    final_text = msg.result
            return {"repo": Path(repo).name, "ok": True, "raw": final_text}
        except Exception as e:
            return {"repo": Path(repo).name, "ok": False, "error": str(e)}

async def main():
    sem = asyncio.Semaphore(MAX_CONCURRENCY)  # 信号量控并发
    results = await asyncio.gather(*[scan_one(r, sem) for r in REPOS])

    # 按严重程度排序输出总报告
    sev_order = {"high": 0, "medium": 1, "low": 2}
    all_issues.sort(key=lambda x: sev_order.get(x.get("severity", "low"), 3))
    Path("reports/summary.json").write_text(json.dumps(all_issues, indent=2, ensure_ascii=False))
    print(f"完成：{len(REPOS)} 个仓库，{len(all_issues)} 条问题")

asyncio.run(main())
```

这段编排器的关键设计：

- **`asyncio.Semaphore`** 把并发钉死在 `MAX_CONCURRENCY`，避免一次性把 50 个仓库全打出去触发 429
- **`asyncio.gather`** 让所有任务并行推进，先完成的先返回
- **结构化输出 + 容错**：要求模型输出 JSON，但解析失败时降级跳过而非整体崩溃
- **统一汇总**：跨仓库合并、排序、落盘成总报告——这是 shell 方案做不到的"结果后处理"

### 5.5 成本与限流控制

批量跑最容易翻车的就是账单和 429。三条铁律：

1. **永远设 `maxTurns`**。一次失控的自主循环可以烧掉几十美元。
2. **用 JSON 输出读取 `cost_usd`**，跑完后累加，超出预算阈值自动熔断：

```bash
total=$(cat reports/*.json | jq -s '[.[].cost_usd // 0] | add')
echo "本次总花费: $total USD"
```

3. **指数退避重试**。429 是批量场景的家常便饭，parallel 用 `--retries`，SDK 自己写 `tenacity` 装饰器：

```python
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=2, min=4, max=60),
       retry=lambda e: "429" in str(e) or "529" in str(e))
async def scan_one_with_retry(repo, sem):
    ...
```

## 六、Mermaid：批量自动化整体架构

```mermaid
flowchart TB
    subgraph 输入
        R[repos.txt<br/>仓库清单]
        P[Prompt / Routine<br/>统一指令]
        C[配置: maxTurns,<br/>allowedTools, model]
    end

    subgraph 编排层
        SEM{并发控制器<br/>Semaphore -P 4}
        DISP[任务分发器<br/>xargs / parallel / SDK]
    end

    subgraph 执行层 每仓库一个独立进程
        W1[Claude 进程 1<br/>repo-A 独立会话]
        W2[Claude 进程 2<br/>repo-B 独立会话]
        W3[Claude 进程 3<br/>repo-C 独立会话]
        WN[Claude 进程 N<br/>repo-N 独立会话]
    end

    subgraph 结果层
        J1[repo-A.json<br/>含 cost_usd]
        J2[repo-B.json]
        J3[repo-C.json]
        AGG[汇总器<br/>jq / Python]
        RPT[总报告<br/>summary.json]
    end

    R --> DISP
    P --> DISP
    C --> DISP
    DISP --> SEM
    SEM --> W1
    SEM --> W2
    SEM --> W3
    SEM --> WN
    W1 --> J1
    W2 --> J2
    W3 --> J3
    WN --> AGG
    J1 --> AGG
    J2 --> AGG
    J3 --> AGG
    AGG --> RPT

    style SEM fill:#ffe4b5
    style AGG fill:#e0ffe0
    style RPT fill:#e0ffe0
```

图中虚线框住的"并发控制器"是整个架构的命门：它决定了你是 4 倍速跑完还是被 429 雪崩。

## 七、实战案例：跨仓库依赖漏洞巡检机器人

把前面所有概念串起来，构建一个可投入生产的"每周依赖漏洞巡检"机器人。它每周一早上对一组仓库做依赖审计，按严重程度汇总，并把高危项提成交付物。

### 7.1 目录结构

```
vuln-patrol/
├── repos.txt                  # 待巡检仓库清单
├── routine.yaml               # 巡检指令与配置（Routine 思路）
├── run.sh                     # xargs 并行入口（方案 A）
├── orchestrator.py            # SDK 编排入口（方案 C，含重试与汇总）
└── reports/
    └── 20260615/
        ├── billing-service.json
        ├── auth-gateway.json
        └── summary.md
```

### 7.2 repos.txt

```
/home/me/repos/billing-service
/home/me/repos/auth-gateway
/home/me/repos/notification-worker
/home/me/repos/inventory-api
/home/me/repos/search-service
```

### 7.3 巡检 Prompt（保存为固定文本，确保可复现）

```
你是依赖安全审计员。请完成以下步骤：
1. 读取本仓库的依赖清单文件（package.json / requirements.txt / go.mod / pom.xml 等，按实际存在为准）
2. 用 Grep 搜索代码中是否使用了已知有漏洞的模式（eval、pickle.loads、不安全的反序列化、硬编码密钥）
3. 对每个发现，给出：severity(high/medium/low)、file、line、description、suggested_fix
4. 严格输出 JSON：{"repo":"<仓库名>","issues":[...],"scanned_files":N}
不要输出 JSON 以外的内容。
```

### 7.4 orchestrator.py（核心，含熔断与汇总）

```python
import asyncio, json, sys
from pathlib import Path
from datetime import date
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from claude_agent_sdk import query, ClaudeAgentOptions

REPOS = [l.strip() for l in Path("repos.txt").read_text().splitlines() if l.strip()]
PROMPT = Path("audit_prompt.txt").read_text()
MAX_CONCURRENCY = 4
BUDGET_USD = 5.0  # 整批预算上限

class RateLimited(Exception): pass

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=2, min=4, max=60),
    retry=retry_if_exception_type(RateLimited),
)
async def audit_one(repo: str, sem: asyncio.Semaphore):
    async with sem:
        opts = ClaudeAgentOptions(
            cwd=repo,
            allowed_tools=["Read", "Grep", "Glob", "Bash"],
            max_turns=20,
            model="claude-sonnet-4-6",
            permission_mode="acceptEdits",
        )
        result_text, cost = "", 0.0
        async for msg in query(prompt=PROMPT, options=opts):
            t = getattr(msg, "type", None)
            if t == "result":
                result_text = getattr(msg, "result", "")
                cost = getattr(msg, "cost_usd", 0.0) or 0.0
            if t == "system" and "429" in str(getattr(msg, "subtype", "")):
                raise RateLimited(repo)
        return {"repo": Path(repo).name, "cost": cost, "raw": result_text}

async def main():
    sem = asyncio.Semaphore(MAX_CONCURRENCY)
    spent = 0.0
    all_issues = []

    tasks = [audit_one(r, sem) for r in REPOS]
    for coro in asyncio.as_completed(tasks):  # 先完成先处理，便于提前熔断
        res = await coro
        spent += res["cost"]
        if spent > BUDGET_USD:
            print(f"⚠️ 预算熔断：已花费 ${spent:.2f} > ${BUDGET_USD}", file=sys.stderr)
            # 取消尚未完成的任务
            for t in tasks:
                t.cancel()
            break
        try:
            data = json.loads(res["raw"])
            for iss in data.get("issues", []):
                iss["repo"] = res["repo"]
                all_issues.append(iss)
        except json.JSONDecodeError:
            print(f"⚠️ {res['repo']} 输出非 JSON，跳过", file=sys.stderr)

    # 生成 Markdown 总报告
    sev = {"high": [], "medium": [], "low": []}
    for i in all_issues:
        sev.setdefault(i.get("severity", "low"), []).append(i)

    md = [f"# 依赖漏洞巡检报告 {date.today()}\n",
          f"扫描仓库 {len(REPOS)} 个，发现问题 {len(all_issues)} 个，花费 ${spent:.2f}\n"]
    for level in ["high", "medium", "low"]:
        md.append(f"\n## {level.upper()} ({len(sev[level])})\n")
        for i in sev[level]:
            md.append(f"- **{i['repo']}** `{i['file']}:{i.get('line','?')}` — {i['description']}")

    out = Path(f"reports/{date.today()}/summary.md")
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text("\n".join(md))
    print(f"报告已生成：{out}")

asyncio.run(main())
```

### 7.5 定时调度

用 cron 每周一早 7 点跑（注意避开整点，减少与全网定时任务的撞车）：

```cron
# 每周一 06:17 跑依赖巡检
17 6 * * 1 cd /home/me/vuln-patrol && /usr/bin/python3 orchestrator.py >> logs/cron.log 2>&1
```

### 7.6 预期产出

- `reports/2026-06-15/summary.md`：按 high/medium/low 分级的总报告，含每条问题的仓库、文件、行号、修复建议
- 控制台输出每个仓库的完成状态与累计花费
- 预算超限时自动熔断，不会失控烧钱
## 八、实战避坑清单

经过多轮批量任务验证，以下是最常踩的坑：

1. **`cwd` 必须显式指定**。SDK 不传 `cwd` 会用宿主进程目录，批量时极易把不同仓库的结果写到同一处。每个任务务必把自己的仓库路径作为 `cwd`。
2. **写操作类任务要单独限流**。"分析类"（只读）并发可以到 8，"改代码类"并发降到 2~3，并强烈建议 `--permission-mode plan` 先出方案、人工确认后再 `acceptEdits` 执行。
3. **结构化输出要兜底**。即使 prompt 里写了"只输出 JSON"，模型偶尔也会加前言。永远 try/except 解析，解析失败降级为"标记需人工复核"。
4. **会话隔离≠认证共享**。每个进程的会话独立，但认证（API Key / OAuth token）是共享的。并发越高，单个 key 的 RPM 压力越大——必要时分账户/分 key 池。
5. **`maxTurns` 是安全阀不是性能参数**。设太小任务完不成，设太大一次失控破产。先小批量试跑，观察"成功任务"的 `num_turns` 中位数，再设为它的 1.5~2 倍。
6. **结果幂等性**。批量任务可能重跑，prompt 要写成幂等的（"在文件开头添加注释"而不是"添加注释"），避免重复执行产生重复内容。

## 九、延伸阅读

- **Agent SDK 官方文档**：`code.claude.com/docs/en/agent-sdk/overview` —— TypeScript/Python 完整 API、自定义工具、流式事件参考
- **Claude Code Routines**：`code.claude.com/docs/en/routines` —— 保存配置与多仓库工作集的官方机制
- **Best Practices**：`code.claude.com/docs/en/best-practices` —— 环境配置、prompt 工程、子 Agent 编排的官方建议
- **Claude Agent SDK Complete Guide**（hidekazu-konishi）：实现层级的深度指南，覆盖宿主集成与多轮编排
- **Claude Code Batch Processing Guide**（smartscope.blog）：`-p` 无头模式与批量命令的系统化对照

## 十、训练营适用性评估

| 维度 | 评估 |
|---|---|
| 适用训练营类型 | 研发效能（DevEx）、平台工程、SRE 自动化、AI 工程化进阶营 |
| 学员基础要求 | 已掌握交互式 Claude Code，熟悉一门脚本语言（Python/Shell），理解进程与并发 |
| 不适用 | 零基础营（应先修"入门篇"）、纯前端营（cwd/并发概念较远） |
| 建议教学时长 | 4 课时：L1 `-p`（1h）+ L2 SDK（1.5h）+ L3 Routines（0.5h）+ 批量实战（1h） |
| 实操占比 | 建议 60% 以上，每层配一个可跑通的 mini 任务 |
| 考核点 | 能独立写出带并发控制、成本熔断、结果汇总的批量编排器 |

## 参考来源

1. [Agent SDK overview - Claude Code Docs](https://code.claude.com/docs/en/agent-sdk/overview) — 官方 Agent SDK 总览，TypeScript/Python 编程式调用入口
2. [Automate work with routines - Claude Code Docs](https://code.claude.com/docs/en/routines) — 官方 Routines 机制：保存配置 + 多仓库工作集
3. [Best practices for Claude Code - Claude Code Docs](https://code.claude.com/docs/en/best-practices) — 官方最佳实践：环境、prompt、子 Agent 编排
4. [Claude Code Batch Processing Complete Guide](https://smartscope.blog/en/generative-ai/claude/claude-code-batch-processing/) — `-p` 无头模式与批量命令的系统化对照指南
5. [Claude Agent SDK Complete Guide](https://hidekazu-konishi.com/entry/claude_agent_sdk_complete_guide.html) — 实现层级的 SDK 集成深度指南，含多轮编排与自定义工具

> [!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 等场景全覆盖。

![51学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)