<!-- title: 无头模式输出契约三法实测 -->

## 概述

当你用 `claude -p` 把 Claude Code 塞进 CI 流水线、定时巡检或批量脚本时，最先撞墙的从来不是"能不能跑通"，而是"返回的东西能不能被机器稳定解析"。本教程聚焦一个被严重低估的核心命题——**输出契约（output contract）**：如何让无头 Agent 吐出结构化、可校验、可被下游管道逐字段消费的结果。我们将实测三种递进的方法，并用一个多仓库代码质量巡检管道把它们串成可运行的生产级编排。

**适用人群**：已读过 `-p` 基础与一篇 hooks 实战、想把 Claude Code 真正织入自动化流水线的工程师、SRE、平台开发者。

## 前置条件

- Claude Code CLI 已安装且 `claude -p "ping"` 能正常返回
- Python 3.10+（用于编排器示例），或熟悉 `jq`
- 已读过官方 headless 文档，了解 `-p` 与 `--output-format` 的存在
- 一个被测代码仓库（任意语言均可）

## Step 0：先认清"信封"与"内容"的分层

无头模式的输出分为两层，这是理解后面所有方法的前提：

```bash
claude -p "用一句话总结这个目录" \
  --output-format json \
  --allowedTools "Read,Glob"
```

返回的**单个 JSON 信封**长这样（字段稳定、官方文档明确）：

```json
{
  "type": "result",
  "subtype": "success",
  "session_id": "abc123...",
  "result": "这是一个面向初学者的 Python 示例仓库……",
  "is_error": false,
  "duration_ms": 8421,
  "num_turns": 2,
  "total_cost_usd": 0.0142,
  "usage": { "input_tokens": 1024, "output_tokens": 128, "cache_read_input_tokens": 0 }
}
```

**关键认知**：信封本身是结构化的（你可以放心 `jq .total_cost_usd`、`jq .is_error`），但信封里的 `.result` 仍然是一段**自由文本字符串**。所谓"输出契约"，本质上是在约束 `.result` 这段文本的内容，或者绕开它。三种方法的分野，全在这里。

**预期结果**：你能用 `jq -r .result` 取出文本，用 `jq .total_cost_usd` 取出成本，但取出的文本是否为合法 JSON、字段是否符合预期，信封不负责保证。

**常见问题**：

- *报 `--output-format` 找不到*：CLI 版本过旧，运行 `claude update` 或 `npm i -g @anthropic-ai/claude-code@latest`。
- *返回里没有 `usage` 字段*：检查是否误加了 `--verbose` 的旧写法；新版默认在 json 模式下携带 usage。

## 方法 A：Prompt 约束 + 文本解析（脆弱基线，先看反面）

最直觉的做法：在系统提示里"命令"模型只输出 JSON，然后用 `jq` 解析 `.result`。

```bash
claude -p "分析 README.md 的可读性。只输出 JSON，不要任何解释、不要 markdown 代码块。格式严格为：
{\"grade\": \"A|B|C\", \"issues\": [\"...\"], \"word_count\": 123}" \
  --output-format json \
  --allowedTools "Read" \
  --append-system-prompt "You are a JSON-only API. Never output prose. Never wrap output in backticks."
```

然后解析：

```bash
claude -p "$PROMPT" --output-format json --allowedTools Read \
  | jq -r '.result' \
  | jq '.grade'
```

**为什么这是反面教材**：模型偶尔会做三件破坏契约的事——

1. 在 JSON 前后加上"好的，结果如下："之类的引导语；
2. 用 ` ```json ` 代码块包裹；
3. 在字段里混入未转义的换行或引号。

任何一种都会让第二个 `jq` 直接报 `parse error`，整条管道崩掉。在生产里你会看到每几百次就失败一次的"玄学错误"，根因全在这里。

**何时仍可用**：一次性脚本、对失败零容忍度低的场景、或你能容忍"失败就重试到成功"。否则请往下看。

**缓解小技巧（治标）**：解析前做一次正则净化，把首个 `{` 到末个 `}` 之间的内容截出来，再喂给 `jq`：

```bash
jq -r '.result' | grep -oP '\{.*\}' | tail -1 | jq '.grade'
```

但这仍是赌模型行为，不构成契约。

## 方法 B：JSON 信封 + 强约束（半结构化，够用但不稳）

方法 B 不再要求模型"输出 JSON"，而是让模型输出**带固定标记字段的的结构化文本**，再由编排器解析。它比 A 稳，但仍依赖模型纪律。

```bash
claude -p "审查 git diff（已暂存）。按下面格式逐行输出，每行一个字段：
GRADE: <A|B|C|D>
SEVERITY: <low|med|high>
SUMMARY: <不超过50字>
TOP_ISSUE: <最主要的一条问题>" \
  --output-format json \
  --allowedTools "Bash(git diff:*)" \
  --permission-mode acceptEdits \
  --append-system-prompt "Output exactly 4 lines, each starting with the exact field name. No extra lines."
```

解析（键值对天然健壮，行序错乱也能兜住）：

```python
import json, re, subprocess

raw = subprocess.run(["claude", "-p", PROMPT, "--output-format", "json",
                      "--allowedTools", "Bash(git diff:*)"], capture_output=True, text=True).stdout
envelope = json.loads(raw)
fields = {}
for line in envelope["result"].splitlines():
    m = re.match(r"^(GRADE|SEVERITY|SUMMARY|TOP_ISSUE):\s*(.+)$", line.strip())
    if m:
        fields[m.group(1)] = m.group(2).strip()
assert fields.get("GRADE") in {"A","B","C","D"}, f"契约违反: {fields}"
```

**优点**：解析逻辑可容错（缺行能检测、多行能截断）；`GRADE` 的枚举校验能当场拦下"玄学输出"并触发重试。

**缺点**：多行自由文本里如果 `SUMMARY` 被模型折行，仍会漏。本质问题没解决——**我们仍在解析模型自由生成的文本**。要真正解决，得换赛道。

## 方法 C：工具调用契约（生产级，把 schema 变成强约束）

这是本教程的核心。Anthropic 的模型对**工具参数**的 schema 遵守度，远高于对"请输出 JSON"这类软指令。原因很朴素：工具参数走的是结构化生成通道，参数必须匹配 JSON Schema，否则工具调用无法发起。

所以契约设计变成：**定义一个"提交结果"工具，它的 input schema 就是你想要的输出结构；让 Agent 把答案通过调用这个工具交上来**。然后从 `stream-json` 的事件流里精准取出 `tool_use.input`，得到一个 100% 符合 schema 的结构化对象。

### C-1：定义提交工具（一个极简 MCP 服务器）

新建 `submit_tool.py`，用 FastMCP 暴露一个 `record_review` 工具，参数即契约：

```python
# submit_tool.py — 仅用于"接收"结构化结果，本身几乎不做事
from fastmcp import FastMCP

mcp = FastMCP("review-submitter")

@mcp.tool()
def record_review(grade: str, severity: str, summary: str, top_issue: str, file_count: int) -> dict:
    """提交一次代码审查的结构化结果。调用此工具即视为审查完成。"""
    # 校验是契约的一部分：枚举不合法时抛错，模型会被迫修正
    assert grade in {"A", "B", "C", "D"}, f"grade 非法: {grade}"
    assert severity in {"low", "med", "high"}, f"severity 非法: {severity}"
    return {"ok": True, "received": {"grade": grade, "severity": severity,
            "summary": summary, "top_issue": top_issue, "file_count": file_count}}

if __name__ == "__main__":
    mcp.run()  # stdio 传输
```

用 `mcp.json` 注册（或直接命令行 `--mcp-config`）：

```json
{
  "mcpServers": {
    "review": { "command": "python", "args": ["submit_tool.py"] }
  }
}
```

### C-2：提示词只要求"调用工具"，不要求"输出格式"

```bash
claude -p "审查当前仓库已暂存的改动。完成后必须调用 record_review 工具提交结果，不要在对话里复述 JSON。" \
  --output-format stream-json \
  --allowedTools "Bash(git diff:*)" "mcp__review__record_review" \
  --permission-mode acceptEdits \
  --max-turns 6 \
  --mcp-config mcp.json
```

注意两个要点：`record_review` 被显式列入 `--allowedTools`（否则无头模式下不会被授权）；用 `stream-json` 是因为我们要在事件流里抓取工具调用，而 `json` 信封只在最后给 `result` 文本，丢失了 `tool_use` 结构。

### C-3：从事件流精准提取结构化结果

`stream-json` 是换行分隔的 NDJSON，每行一个事件，`type` 字段区分种类。我们要的是 `assistant` 事件里 `content` 数组中 `type=="tool_use"` 且 `name=="record_review"` 的那个块，其 `.input` 就是契约对象：

```python
# orchestrator.py — 抽取 + 重试 + 聚合的核心
import json, subprocess, time

def extract_contract(lines):
    """从 stream-json 事件流里提取 record_review 的 input 参数。"""
    for line in lines:
        evt = json.loads(line)
        if evt.get("type") != "assistant":
            continue
        for block in evt.get("message", {}).get("content", []):
            if block.get("type") == "tool_use" and block.get("name") == "record_review":
                return block["input"]      # ← 100% 符合 schema 的结构化对象
    return None

def run_agent_once(prompt, retries=3):
    for attempt in range(retries):
        proc = subprocess.run(
            ["claude", "-p", prompt, "--output-format", "stream-json",
             "--allowedTools", "Bash(git diff:*)", "mcp__review__record_review",
             "--permission-mode", "acceptEdits", "--max-turns", "6",
             "--mcp-config", "mcp.json"],
            capture_output=True, text=True, timeout=300)
        contract = extract_contract(proc.stdout.splitlines())
        if contract:
            return contract                 # 契约达成
        time.sleep(2 ** attempt)            # 指数退避：1s, 2s, 4s
    raise RuntimeError(f"agent 重试 {retries} 次仍未达成契约")
```

**为什么这是真正的契约**：`record_review` 的参数由 schema 强约束，模型生成的就是合法 JSON；枚举非法会在工具内 `assert` 报错，错误经 `tool_result` 回流，模型会自我修正后再次调用。你拿到的 `block["input"]` 是一个干净的 dict，可以直接喂给 `json.dumps`、写库、或拼进聚合报告——**不存在解析失败这一说**。

### 三法对比

| 维度 | A：Prompt 约束 | B：键值文本 | C：工具调用契约 |
|---|---|---|---|
| 解析可靠性 | 低（玄学崩溃） | 中（行序敏感） | 高（schema 强约束） |
| 字段类型保证 | 无 | 无 | 有（int/enum/必填） |
| 失败可检测 | 难（静默错） | 可（断言） | 可（工具内校验） |
| 实现成本 | 最低 | 低 | 中（需一个 MCP 工具） |
| 适合场景 | 一次性脚本 | 简单 CI 门禁 | 生产批量管道 |

```mermaid
flowchart TD
    Q["需要机器消费结果?"] -->|否| A["方法A: Prompt约束<br/>文本输出即可"]
    Q -->|是| R["能容忍偶发失败?"]
    R -->|能| B["方法B: 键值文本<br/>编排器容错解析"]
    R -->|不能| C["方法C: 工具调用契约<br/>schema强约束"]
    C --> C1["定义 record_* MCP工具"]
    C1 --> C2["提示词只要求调用工具"]
    C2 --> C3["stream-json 抓 tool_use.input"]
    C3 --> C4["枚举/类型校验 + 重试"]
    style C fill:#e8f5e9,stroke:#2e7d32
```

## 进阶：错误重试、超时与部分失败隔离

生产管道的成败不在"快乐路径"，而在异常路径。三件事必须处理：

**1. 契约未达成的重试**：上面 `run_agent_once` 已演示指数退避。判断"未达成"的标准是 `extract_contract` 返回 `None`——模型跑完了却没调用提交工具。

**2. 超时熔断**：无头 Agent 偶尔会陷入工具调用死循环（反复 `git diff`）。`--max-turns 6` 给出硬上限，`subprocess(timeout=300)` 给出壁钟兜底，二者缺一不可。

**3. 部分失败隔离**：批量场景下，一个仓库的失败绝不能拖垮整批。把每个仓库跑在独立子进程里（天然隔离），失败只记录、不抛出：

```python
results = []
for repo in repos:
    try:
        contract = run_agent_once(build_prompt(repo))
        results.append({"repo": repo, "ok": True, **contract})
    except Exception as e:
        results.append({"repo": repo, "ok": False, "error": str(e)})
        continue   # ← 隔离：单点失败不影响其他仓库
```

**4. 成本护栏**：从 `stream-json` 的 `result` 事件提取 `total_cost_usd` 累加，超过预算阈值就停止派发新任务。这是无人值守场景避免"一觉醒来账单爆炸"的关键：

```python
if sum(r.get("cost", 0) for r in results) > BUDGET_USD:
    log("预算达上限，剩余仓库跳过"); break
```

## 实战案例：多仓库代码质量夜间巡检管道

把上面所有零件组装起来。目标：每晚对 N 个仓库各跑一次结构化审查，输出一份可入库的 JSONL 报告，供看板消费。

**目录结构**：

```
nightly-review/
├── submit_tool.py        # 方法C的提交工具
├── mcp.json              # MCP 注册
├── orchestrator.py       # 编排器：派发+抽取+重试+聚合
├── repos.txt             # 仓库清单，每行一个本地路径
└── reports/              # 每晚一份 .jsonl
```

**orchestrator.py 完整版**：

```python
import json, subprocess, time, sys
from pathlib import Path

BUDGET_USD = 5.0
RETRIES, TIMEOUT, MAX_TURNS = 3, 300, 6

def extract_contract(ndjson_lines):
    for line in ndjson_lines:
        evt = json.loads(line)
        if evt.get("type") != "assistant":
            continue
        for b in evt.get("message", {}).get("content", []):
            if b.get("type") == "tool_use" and b.get("name") == "record_review":
                return b["input"]
    return None

def run_repo(repo_path):
    prompt = (f"审查仓库 {repo_path} 的最近一次提交(git show HEAD)。"
              "完成后必须调用 record_review 工具提交结果，不要复述JSON。")
    for attempt in range(RETRIES):
        try:
            p = subprocess.run(
                ["claude", "-p", prompt, "--output-format", "stream-json",
                 "--allowedTools", "Bash(git:*)", "mcp__review__record_review",
                 "--permission-mode", "acceptEdits", "--max-turns", str(MAX_TURNS),
                 "--mcp-config", "mcp.json", "--add-dir", repo_path],
                capture_output=True, text=True, timeout=TIMEOUT, cwd=repo_path)
        except subprocess.TimeoutExpired:
            continue
        contract = extract_contract(p.stdout.splitlines())
        if contract:
            # 从 result 事件取成本
            cost = 0.0
            for line in p.stdout.splitlines():
                e = json.loads(line)
                if e.get("type") == "result":
                    cost = e.get("total_cost_usd", 0.0)
            return {"repo": repo_path, "ok": True, "cost": cost, **contract}
        time.sleep(2 ** attempt)
    return {"repo": repo_path, "ok": False, "error": "contract not reached"}

def main():
    repos = [l.strip() for l in Path("repos.txt").read_text().splitlines() if l.strip()]
    spent, out = 0.0, []
    for repo in repos:
        if spent > BUDGET_USD:
            print(f"预算 {BUDGET_USD}$ 用尽，跳过 {repo}", file=sys.stderr); break
        r = run_repo(repo)
        spent += r.get("cost", 0.0)
        out.append(r)
        print(json.dumps(r, ensure_ascii=False), flush=True)
    Path("reports").mkdir(exist_ok=True)
    # 注意：时间戳由外部 cron 注入，此处仅落盘
    Path("reports/last.jsonl").write_text("\n".join(json.dumps(x, ensure_ascii=False) for x in out))

if __name__ == "__main__":
    main()
```

**派发与定时**：用系统 cron 而非脚本内 `time`，保证幂等且可补跑：

```bash
# crontab -e  —— 每晚 02:17 执行（避开整点，错峰）
17 2 * * * cd /opt/nightly-review && /usr/bin/python orchestrator.py >> logs/$(date +\%F).log 2>&1
```

**消费报告**（任意 BI / 看板 / 告警）：

```bash
# 找出所有评级为 D 的仓库，触发告警
jq -c 'select(.ok and .grade=="D")' reports/last.jsonl
# 统计总成本
jq '[.[].cost] | add' reports/last.jsonl
```

**预期产出**（`reports/last.jsonl`，每行一条契约对象）：

```json
{"repo": "/srv/web-app", "ok": true, "cost": 0.031, "grade": "B", "severity": "med", "summary": "登录模块缺少输入校验", "top_issue": "SQL拼接", "file_count": 4}
{"repo": "/srv/billing", "ok": true, "cost": 0.028, "grade": "C", "severity": "high", "summary": "金额计算用浮点", "top_issue": "float精度", "file_count": 2}
{"repo": "/srv/legacy", "ok": false, "error": "contract not reached"}
```

注意第三条：契约未达成的仓库被**隔离记录**而非崩溃整批，这正是部分失败隔离的价值。

## 管道全景

```mermaid
flowchart LR
    CRON["cron 02:17"] --> ORC["orchestrator.py"]
    ORC --> LOOP{"遍历 repos.txt"}
    LOOP -->|每个仓库| AGENT["claude -p<br/>stream-json<br/>--mcp-config"]
    AGENT --> MCP["submit_tool.py<br/>record_review工具<br/>schema强约束"]
    AGENT -->|"NDJSON事件流"| PARSE["extract_contract<br/>抓 tool_use.input"]
    PARSE -->|None| RETRY{"重试<3?"}
    RETRY -->|是| AGENT
    RETRY -->|否| ISO["隔离记录失败"]
    PARSE -->|契约对象| BUD{"超预算?"}
    BUD -->|否| APPEND["追加到结果集"]
    BUD -->|是| STOP["停止派发"]
    APPEND --> LOOP
    LOOP -->|遍历完| JSONL["reports/last.jsonl"]
    JSONL --> BI["jq消费 / 看板 / 告警"]
    ISO --> JSONL
    style MCP fill:#fff3e0,stroke:#e65100
    style PARSE fill:#e8f5e9,stroke:#2e7d32
```

## 常见坑与排查

- **`tool_use` 一直不出现**：九成是 `record_review` 没被加进 `--allowedTools`，或 MCP 服务器启动失败。先跑 `claude mcp list` 确认工具可见，再用 `--mcp-config` 指定路径。
- **模型复述了 JSON 又调用了工具**：无害。你只认 `tool_use.input`，对话文本直接忽略，这正是方法 C 相对 A 的健壮之处。
- **`--add-dir` 被忽略**：无头模式下 Agent 默认只能访问启动目录，跨仓库审查必须对每个仓库显式 `--add-dir` 或在 `cwd` 内启动。
- **成本统计为 0**：检查 `stream-json` 里是否有 `type=="result"` 事件；某些异常退出会缺失该事件，此时用 `--max-turns` 是否触顶来判定。
- **stream-json 解析偶发半行**：必须按**换行**缓冲，绝不能按 chunk。`splitlines()` 在进程结束后用是安全的；流式消费要用 `readline()`。

## 延伸阅读

- 官方 headless 文档：`https://docs.claude.com/en/docs/claude-code/headless` —— `--output-format` 与 `--input-format` 的权威字段说明
- 官方 Agent SDK：`https://docs.claude.com/en/api/agent-sdk/overview` —— 用 Python/TypeScript 取代 shell 编排，获取类型化事件
- MCP 规范：`https://modelcontextprotocol.io/` —— 工具 schema 的完整定义，方法 C 的底层支撑
- Claude Code GitHub Issues #28489：社区对"headless 重试、校验、resume"的讨论，与本教程的护栏设计直接相关

## 训练营适用性评估

- **平台工程 / DevOps 训练营**：高度匹配。本教程即"把 LLM 接入 CI"的工程化范本，可直接作为 2 课时的实战模块，学员产出可运行的巡检管道。
- **AI 应用开发训练营**：中高匹配。核心是"用工具调用约束结构化输出"，这是 Agent 工程的通用模式，可迁移到任意 LLM 编排场景。
- **预计教学时长**：讲授 60 分钟 + 动手实验 90 分钟。建议先让学员用方法 A 跑崩一次（亲历"玄学崩溃"），再引入方法 C，对比冲击最强。
- **前置铺垫**：需先完成一篇 `-p` 基础教程与一篇 MCP 服务器入门。

## 参考来源

1. [Claude Code Headless — Run Claude Code programmatically（官方文档）](https://docs.claude.com/en/docs/claude-code/headless) — `--output-format`/`--input-format`/`--allowedTools` 等无头模式标志的权威定义
2. [Claude Code in CI/CD: headless patterns for real pipelines（Refactix）](https://refactix.com/ai-development-engineering/claude-code-ci-cd-headless-patterns-production) — 生产级 headless 配置、JSON 解析与成本控制的实战经验
3. [Structured Outputs with Claude: JSON Schemas, Validation, and Retry Loops（snippets.ltd）](https://snippets.ltd/blog/structured-outputs-with-claude-json-schemas-validation-retry-loops) — 结构化输出的校验与重试闭环方法论，方法 C 的理论依据
4. [Headless automation: add retry, verification, and resume（GitHub Issue #28489）](https://github.com/anthropics/claude-code/issues/28489) — 社区对无人值守场景下重试/校验/续跑的诉求与方案讨论
5. [Headless Mode and CI/CD — Tutorial（SFEIR Institute）](https://institute.sfeir.com/en/claude-code/claude-code-headless-mode-and-ci-cd/tutorial/) — 无头模式接入 CI/CD 的渐进式教学，含 stream-json 消费示例

> [!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)