首页 › tutorials › 无头模式输出契约三法实测

概述

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

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

前置条件

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

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

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

返回的单个 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、字段是否符合预期，信封不负责保证。

常见问题：

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

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

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."

然后解析：

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

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

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

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

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

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

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

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

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

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

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."

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

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 工具，参数即契约：

# 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）：

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

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

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 就是契约对象：

# 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 门禁 生产批量管道 
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. 部分失败隔离：批量场景下，一个仓库的失败绝不能拖垮整批。把每个仓库跑在独立子进程里（天然隔离），失败只记录、不抛出：

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 累加，超过预算阈值就停止派发新任务。这是无人值守场景避免"一觉醒来账单爆炸"的关键：

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 完整版：

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，保证幂等且可补跑：

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

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

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

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

{"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"}

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

管道全景

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

常见坑与排查

延伸阅读

训练营适用性评估

参考来源

  1. Claude Code Headless — Run Claude Code programmatically（官方文档） — --output-format/--input-format/--allowedTools 等无头模式标志的权威定义
  2. Claude Code in CI/CD: headless patterns for real pipelines（Refactix） — 生产级 headless 配置、JSON 解析与成本控制的实战经验
  3. Structured Outputs with Claude: JSON Schemas, Validation, and Retry Loops（snippets.ltd） — 结构化输出的校验与重试闭环方法论，方法 C 的理论依据
  4. Headless automation: add retry, verification, and resume（GitHub Issue #28489） — 社区对无人值守场景下重试/校验/续跑的诉求与方案讨论
  5. Headless Mode and CI/CD — Tutorial（SFEIR Institute） — 无头模式接入 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学通信知识星球海报-扫码加入

📎 本分类最新文章