-p流式协议配钩子造批量引擎

首页 › tutorials › -p流式协议配钩子造批量引擎

概述

本教程教你把 claude -p 从"问一句答一句"升级为可编程、可观测的批量任务引擎：用 --bare 保证可复现、用 --output-format 与 --json-schema 把自然语言变成机器可读的结构化数据、用 stream-json 全程追踪事件流，再把 2026 年大幅扩展的钩子（Hook）生命周期装成自动化质量门禁。最终产出一个读任务队列、自动拦危险命令、按任务计成本、可恢复的批量调度器。适合已熟悉交互模式、想把 Claude Code 嵌入脚本/CI/流水线的工程师。

前置条件

Claude Code ≥ v2.1.205：该版本起 --json-schema 会对非法 schema 报错退出（更早版本会静默忽略并退回纯文本），且 system/init 事件带 capabilities 能力探测字段。
Node.js + npm：Claude Code 的安装运行环境。
jq：解析 JSON 输出的必备工具（brew install jq 或 apt install jq）。
一个测试仓库：本教程会在其根目录写 .claude/settings.json 与脚本。
API Key 或已登录：ANTHROPIC_API_KEY 环境变量，或交互式 claude 登录过。
基础 Shell 与 JSON 知识。

第 1 步：认清 `-p` 模式的三重身份

给任何 claude 命令加上 -p（或 --print）即进入非交互模式：发一个提示、收一个结果、退出，没有对话循环。但它远不止"问一次"，实际上能扮演三种角色：

# 身份一：一次性问答
claude -p "auth 模块负责做什么？"

# 身份二：管道工具（读 stdin，写 stdout）
cat build-error.txt | claude -p '用一句话说出这个构建错误的根因' > cause.txt

# 身份三：可编程子进程（结构化输出 + 事件流）
claude -p "提取 auth.py 的函数名" --output-format json

关键开关：`--bare`（脚本推荐模式，将成为 `-p` 默认）

默认情况下，-p 会像交互会话一样加载当前目录与 ~/.claude 里的全部上下文——队友机器上的某个 hook、项目 .mcp.json 里的 MCP 服务器、本地 CLAUDE.md 都会生效，导致同一脚本在不同机器上结果不一致。加上 --bare 会跳过 hooks、skills、plugins、MCP servers、auto memory、CLAUDE.md 的自动发现，只认你显式传入的 flag：

claude --bare -p "总结这个文件" --allowedTools "Read"

官方明确：--bare 是脚本和 SDK 调用的推荐模式，未来会成为 -p 的默认行为。CI 流水线一律加 --bare。

常见问题

现象	原因	解决
脚本卡住不动	`-p` 仍可能弹权限提示	加 `--allowedTools "Read,Edit,Bash"` 或 `--permission-mode acceptEdits`
stdin 报 `exceeds 10MB`	v2.1.128 起 stdin 上限 10MB	把大文件写盘，在 prompt 里用路径引用，别用管道灌
后台进程（dev server）把进程挂住	v2.1.163 前永不退出	升级；或设 `CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS`，默认 10 分钟封顶

第 2 步：三种输出格式与 `jq` 解析

--output-format 决定结果形态：

text（默认）：纯文本。
json：结构化对象，含 result（文本结果）、session_id、total_cost_usd、按模型拆分的费用、usage。
stream-json：换行分隔的 NDJSON，逐事件实时吐出（下一步详解）。

json 模式让脚本化调用无需查用量面板就能追踪每次开销：

claude -p "总结这个项目" --output-format json | jq '{result: .result, cost: .total_cost_usd, sid: .session_id}'

预期输出形如：

{
  "result": "一个基于 FastAPI 的认证服务……",
  "cost": 0.0123,
  "sid": "abc123..."
}

第 3 步：`--json-schema` 锁定结构化输出（引擎可机器读的关键）

批量引擎要可靠解析每个任务的结果，就不能靠正则去抠自然语言。用 --json-schema 给一个 JSON Schema，结果会落到 structured_output 字段：

claude -p "提取 auth.py 的主要函数名" \
  --output-format json \
  --json-schema '{"type":"object","properties":{"functions":{"type":"array","items":{"type":"string"}}},"required":["functions"]}' \
  | jq '.structured_output'

预期输出：

{ "functions": ["login", "verify_token", "refresh", "logout"] }

注意事项

v2.1.205 起：schema 非法会报 Error: --json-schema is not a valid JSON Schema 并以非零状态退出。更早版本会静默忽略并退回纯文本——批量场景务必升级，否则脚本会把纯文本当成结构化数据解析而崩溃。
format（如 "format":"email"）只作为注解，不强制校验。
复杂 schema 可存文件复用：先把 schema 写进 schema.json，脚本里 --json-schema "$(cat schema.json)"。

第 4 步：`stream-json` 全程事件流（引擎可观测的关键）

claude -p "解释递归" --output-format stream-json --verbose --include-partial-messages

每一行是一个事件对象。核心事件四类：

system/init：流的第一个事件，报告 model、tools、加载的 MCP 服务器与 plugins，以及可选的 capabilities 数组（用于能力探测，比比较版本号更稳）。
stream_event：逐 token 增量，event.delta.type == "text_delta" 即文本片段。
system/api_retry：可重试的 API 错误重试前发出，可用于自定义退避/熔断。字段：attempt、max_retries、retry_delay_ms、error_status、error（取值如 rate_limit/overloaded/billing_error）。
result：最终汇总，含 result 文本、total_cost_usd、duration_ms、num_turns、is_error。

用 jq 实现"实时打字 + 末尾汇总"：

claude -p "写一首关于版本控制的短诗" --output-format stream-json --verbose --include-partial-messages \
  | jq -rj 'select(.type=="stream_event" and .event.delta.type?=="text_delta") | .event.delta.text'

实战技巧：批量引擎里把整条流写进 run.jsonl，事后用 jq 'select(.type=="system/api_retry")' 统计重试，用 jq 'select(.type=="result")' 聚合成本与耗时。

第 5 步：2026 钩子全景与"三档节奏"

钩子是在生命周期固定点自动执行的 shell 命令、HTTP 端点、LLM 提示、MCP 工具或 agent。理解它先抓住三档节奏：

flowchart LR
    A[会话级 一次] --> B[SessionStart]
    A --> C[Setup *-p/--init*]
    A --> D[SessionEnd]
    E[轮级 每个提示] --> F[UserPromptSubmit]
    E --> G[Stop / StopFailure]
    H[工具级 每次调用] --> I[PreToolUse]
    H --> J[PostToolUse / PostToolBatch]

2026 版新增/值得关注的事件：

Setup：用 --init-only，或 -p 模式下用 --init/--maintenance 时触发，专为 CI 一次性准备设计。
PostToolBatch：一整批并行工具调用解析完毕、下一次模型调用前触发——批量并行场景的对账点。
TaskCreated / TaskCompleted：通过 TaskCreate 建任务/标完成时触发，可做进度看板。
FileChanged：磁盘文件变更触发，matcher 指定监听文件名。

配置是三层嵌套：事件 → matcher 组 → handler。matcher 求值规则：

matcher 取值	求值方式	例子
`"*"` / `""` / 省略	全匹配	每次该事件都触发
仅含字母、数字、`_`、`-`、空格、`,`、`\\|`	精确串（或用 `\\|`/`,` 列举）	`Bash`、`Edit\\|Write`
含其他字符	JavaScript 正则（非锚定）	`^Notebook`、`mcp__memory__.*`

第 6 步：把钩子装成自动化质量门禁

在项目根 .claude/settings.json（可提交、团队共享）写一个"危险命令拦截 + 自动格式化 + 收尾记账"的三件套：

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [
          { "type": "command",
            "if": "Bash(rm *|git push *|git push -f*)",
            "command": "${CLAUDE_PROJECT_DIR}/.claude/hooks/guard.sh" }
        ]
      }
    ],
    "PostToolUse": [
      {
        "matcher": "Edit|Write",
        "hooks": [
          { "type": "command",
            "command": "${CLAUDE_PROJECT_DIR}/.claude/hooks/fmt.sh" }
        ]
      }
    ],
    "Stop": [
      {
        "matcher": "*",
        "hooks": [
          { "type": "command",
            "command": "${CLAUDE_PROJECT_DIR}/.claude/hooks/log-stop.sh" }
        ]
      }
    ]
  }
}

拦截脚本读 stdin 的 JSON、返回 permissionDecision：

#!/bin/bash
# .claude/hooks/guard.sh
CMD=$(jq -r '.tool_input.command')
if echo "$CMD" | grep -qE 'rm -rf|git push -f|drop database'; then
  jq -n '{ hookSpecificOutput: { hookEventName: "PreToolUse",
           permissionDecision: "deny",
           permissionDecisionReason: "批量引擎禁止的危险命令" } }'
else
  exit 0   # 静默 = 不批准也不否决，走正常权限流程
fi

易踩的坑（重要）

--bare 会跳过钩子！ 上一节我们推荐批量场景用 --bare 保证可复现，但 --bare 也跳过了 hooks/MCP/CLAUDE.md。要让批量引擎同时享有"可复现"与"门禁"，二选一： - 方案 A：不用 --bare，接受会加载本地配置（适合团队统一仓库、配置已提交的场景）。 - 方案 B：仍用 --bare，但用 --settings ./batch-settings.json 显式注入门禁钩子。脚本自带配置、跨机一致且带护栏。

现象	原因	解决
钩子从不触发	matcher 写错（如把正则当精确串）	含特殊字符就走正则；精确匹配用纯字母数字
`PostToolUse` 不跑	工具调用失败	失败走 `PostToolUseFailure`，不是 `PostToolUse`
批量引擎无门禁	用了 `--bare`	改用 `--settings` 显式注入

实战案例：批量"模块健康度巡检"调度器

目标：读一个任务队列，对每个模块跑一次结构化巡检，自动拦危险命令、记录每任务成本与状态，末尾汇总。全程无人值守、可断点续跑。

目录结构

batch-inspect/
├── queue.txt                 # 任务队列，每行一条
├── schema.json               # 巡检结果 schema
├── batch-settings.json       # 门禁钩子（注入用）
├── .claude/hooks/guard.sh    # 危险命令拦截
└── batch-runner.sh           # 调度器

queue.txt：

src/auth|审查认证模块的安全风险，给出严重度分级
src/payment|审查支付模块的事务一致性风险
src/upload|审查上传模块的资源耗尽风险

schema.json（结构化输出契约）：

{
  "type": "object",
  "properties": {
    "module": { "type": "string" },
    "risks": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "severity": { "type": "string", "enum": ["high", "medium", "low"] },
          "issue": { "type": "string" },
          "location": { "type": "string" }
        },
        "required": ["severity", "issue"]
      }
    },
    "summary": { "type": "string" }
  },
  "required": ["module", "risks", "summary"]
}

batch-runner.sh（调度器核心）：

#!/bin/bash
set -uo pipefail
QUEUE="queue.txt"
LOG="run-$(date +%Y%m%d-%H%M%S).jsonl"
SETTINGS="batch-settings.json"   # 内含 guard.sh 门禁

total_cost=0
ok=0; fail=0

while IFS='|' read -r mod task || [ -n "$mod" ]; do
  [ -z "$mod" ] && continue
  echo "▶ 巡检 $mod …"

  # 每个任务一个独立 session；结构化输出 + 限轮防爆
  out=$(claude -p "$task（模块路径：$mod）" \
        --bare \
        --settings "$SETTINGS" \
        --allowedTools "Read,Grep,Glob" \
        --max-turns 12 \
        --output-format json \
        --json-schema "$(cat schema.json)" 2>/dev/null) || true

  if [ -z "$out" ]; then
    echo "{\"module\":\"$mod\",\"status\":\"error\"}" >> "$LOG"
    fail=$((fail+1)); continue
  fi

  # 落明细 + 累计成本
  echo "$out" | jq -c --arg m "$mod" \
    '{module:$m, cost:.total_cost_usd, risks:.structured_output.risks, summary:.structured_output.summary}' >> "$LOG"
  c=$(echo "$out" | jq -r '.total_cost_usd // 0')
  total_cost=$(awk "BEGIN{printf \"%.4f\", $total_cost+$c}")
  ok=$((ok+1))
  echo "  ✓ 完成，本任务 \$$c"
done < "$QUEUE"

echo "============ 汇总 ============"
echo "成功 $ok / 失败 $fail / 总成本 \$$total_cost"
echo "明细见 $LOG"

预期运行输出

▶ 巡检 src/auth …
  ✓ 完成，本任务 $0.0184
▶ 巡检 src/payment …
  ✓ 完成，本任务 $0.0211
▶ 巡检 src/upload …
  ✓ 完成，本任务 $0.0157
============ 汇总 ============
成功 3 / 失败 0 / 总成本 $0.0552
明细见 run-20260710-143022.jsonl

明细文件可再聚合，例如统计全部 high 级风险：

jq -s '[.[] | .risks[]? | select(.severity=="high")] | length' run-*.jsonl

断点续跑

调度器是"逐行消费队列"的纯循环，天然可恢复：把已完成的模块从 queue.txt 删掉（或维护一个 done.txt 跳过清单）即可重跑。如需会话级恢复，把每任务的 session_id 也写进日志，重跑时对该任务改用 claude -p --resume <sid>。

工作流程总览

flowchart TD
    Q[queue.txt 任务队列] --> R[batch-runner.sh 调度器]
    R -->|每行| P{claude -p}
    P -->|--bare| B[可复现: 跳过本地配置]
    P -->|--settings| H[注入门禁钩子]
    H --> G1[PreToolUse 拦危险命令]
    P -->|--json-schema| S[结构化输出 structured_output]
    P -->|--output-format json| J[total_cost_usd 计费]
    P -->|--max-turns| L[限轮防爆]
    S --> jq[jq 解析]
    J --> jq
    jq --> LOG[run-*.jsonl 明细]
    LOG --> SUM[汇总: 成功/失败/成本]

训练营适用性评估

适合类型：DevOps / 平台工程 / SRE / 研效工具链训练营，以及"AI 辅助工程"进阶模块。不建议作为零基础第一课——需要先掌握交互模式与基本权限模型。
前置衔接：建议先完成"Claude Code 入门（安装与首工程）"与"读懂 Agent 循环"两课。
预计教学时长：约 3.5 小时——讲解 1.5h（-p 三身份 + 三档钩子 + 易踩的 --bare/hooks 冲突坑）+ 实操 1.5h（跑通批量巡检调度器）+ 排错与扩展 0.5h（断点续跑、成本聚合、迁移到 Agent SDK）。
可延伸作业：把 batch-runner.sh 重写为 Python 版并接入 system/api_retry 事件做熔断；或用 PostToolBatch 钩子做并行任务的对账看板。

参考来源

Run Claude Code programmatically — Claude Code Docs — 官方 headless 模式文档，-p、--bare、--output-format、--json-schema、stream-json 事件权威来源
Hooks reference — Claude Code Docs — 官方钩子参考，三档节奏、完整事件表、matcher 规则、permissionDecision 输出契约
Claude Code Batch Processing Complete Guide — smartscope.blog — /batch 命令与 claude -p headless 批量处理的系统实践
Claude Code Hooks: Complete Guide with 20+ Ready-to-Use Examples — aiorg.dev — 钩子从零到生产，含自动格式化、拦截等可直接复制的配置
A developer's Claude Code CLI reference (2026 guide) — eesel.ai — 2026 年 CLI 命令、flag、hooks 的完整参考

[!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学通信知识星球海报-扫码加入

概述