<!-- title: 流式JSON：打造命令行编排看板 -->

## 概述

当你把 Claude Code 塞进 shell 管道、定时任务或编排器时，`-p`（headless）模式给出的纯文本输出只是冰山一角。真正的生产级自动化，依赖的是 `--output-format stream-json` 这一**逐行 JSON 事件流**：它把"模型思考、工具调用、token 增量、成本、错误"全部结构化地吐到 stdout，让你能用 `jq` 实时解析、做看板、控预算、断点续跑。本教程面向已经能跑通 `claude -p` 但想把它升级成**可观测、可重试、可计费**的命令行 Agent 编排器的开发者。

## 前置条件

- Claude Code 已安装，`claude --version` 可正常输出
- 能跑通最小例子：`claude -p "说一句话"` 返回纯文本
- 安装 `jq`（事件流解析核心工具）：`jq --version`
- 一台能联网、已登录 Claude Code 的终端（CI 环境需配置 API Key 或 OAuth）
- 基本的 shell 脚本阅读能力

```bash
# 环境自检一行流
command -v claude && command -v jq && echo "✅ 就绪" || echo "❌ 缺依赖"
```

## 正文

### 步骤 1：三种输出格式，先认准哪一种

`--output-format` 决定 Claude Code 把结果怎么交给你，三者用途完全不同：

| 格式 | 形态 | 何时用 |
|---|---|---|
| `text`（默认） | 纯文本字符串 | 人类阅读、简单管道 |
| `json` | 单个 JSON 对象，含 `result`/`session_id`/`total_cost_usd` | 一次性脚本、取 session_id 续跑 |
| `stream-json` | 每行一个 JSON 事件（NDJSON） | 实时观测、看板、长任务进度 |

**操作说明**：用三种格式各跑一次同一个 prompt，对比 stdout。

```bash
echo "--- text ---"
claude -p "用一句话解释递归"

echo "--- json ---"
claude -p "用一句话解释递归" --output-format json | jq '{result, session_id, cost: .total_cost_usd, turns: .num_turns}'

echo "--- stream-json（只看事件类型） ---"
claude -p "用一句话解释递归" --output-format stream-json --verbose | jq -r '.type' | sort | uniq -c
```

**预期结果**：`json` 会给出类似 `{ "result": "递归是…", "session_id": "abc123…", "cost": 0.0021, "turns": 1 }` 的完整对象；`stream-json` 会打印一串事件 `type` 计数，例如若干个 `system`、`assistant`、一个 `result`。

**常见问题**：

- **Q：`stream-json` 没有任何输出就结束了？** A：很可能漏了 `--verbose`。官方文档明确：要用 `--output-format stream-json` 配合 `--verbose`（以及需要 token 级增量时加 `--include-partial-messages`）才能拿到完整事件流。
- **Q：CI 里跑 `json` 模式结果污染了输出？** A：加上 `--bare`，让脚本不被本地的 settings/CLAUDE.md 干扰，得到干净的 CI 语义。

---

### 步骤 2：拆解 stream-json 的事件流结构

stream-json 是**换行分隔的 NDJSON**，每行一个事件。理解事件分类是后续一切解析的基础。

**操作说明**：把一次真实调用的事件流落盘，逐行观察字段。

```bash
claude -p "读一下当前目录的 README 并总结" --output-format stream-json --verbose \
  > /tmp/events.ndjson

# 1) 看每行事件类型分布
jq -r '.type' /tmp/events.ndjson | sort | uniq -c

# 2) 看第一个事件（通常是 system/init）
head -n1 /tmp/events.ndjson | jq '.'
```

**预期结果**：你会看到事件的 `type` 至少包含这几类：

- `system`（含子类型 `init`）：**事件流的第一个**，报告本次会话的 model、可用 tools、MCP servers、已加载 plugins。它是会话的"身份证"。
- `assistant`：模型回复，content 里可能同时有 `text` 块和 `tool_use` 块。
- `user`：工具结果回填给模型的那一轮（tool_result）。
- `stream_event`：token 级增量（需 `--include-partial-messages`），`event.delta.type == "text_delta"` 就是逐字吐出的文本。
- `result`：**事件流的收尾**，包含最终文本、`total_cost_usd`、`duration_ms`、`num_turns`、`is_error`。

**常见问题**：

- **Q：第一行不是 `system/init`？** A：当设置了 `CLAUDE_CODE_SYNC_PLUGIN_INSTALL` 等同步安装逻辑时，首事件顺序可能变。脚本里别假设"第一行一定是 init"，要用 `select(.type=="system" and .subtype=="init")` 过滤。

---

### 步骤 3：用 jq 把事件流"翻译"成人话

事件流是给机器看的，但 `jq` 能随时把它转成你想要的任何视图。这是看板的灵魂。

**操作说明**：四段经典 jq 过滤器，分别取"逐字流式文本""模型说的话""工具调用清单""最终成本"。

```bash
# (a) 逐字流式文本（像 ChatGPT 打字机效果）
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'

# (b) 只看模型的文本回复（不含工具调用）
cat /tmp/events.ndjson \
| jq -r 'select(.type=="assistant") | .message.content[]? | select(.type=="text") | .text'

# (c) 提取所有工具调用（名称 + 关键参数）
cat /tmp/events.ndjson \
| jq -c 'select(.type=="assistant") | .message.content[]? | select(.type=="tool_use") | {name, input}'

# (d) 最终成本与耗时
cat /tmp/events.ndjson \
| jq '{cost: .total_cost_usd, ms: .duration_ms, turns: .num_turns, ok: (.is_error|not)} | select(.cost!=null)'
```

**预期结果**：
- (a) 终端边跑边打印诗句，无引号无换行污染（靠 `jq -rj` 的 `-r` 原始输出 + `-j` 不加换行）。
- (c) 输出形如 `{"name":"Read","input":{"file_path":"README.md"}}` 的工具调用序列。
- (d) 输出 `{"cost":0.0142,"ms":8341,"turns":4,"ok":true}`。

**常见问题**：

- **Q：`(c)` 报 `null`？** A：没加 `?` 容错。content 数组里可能混着 `text` 块，`select(.type=="tool_use")` 前的 `.message.content[]?` 那个 `?` 是关键，遇到非数组 content 时跳过而非报错。
- **Q：`-rj` 不认识？** A：旧版 jq 没有 `-j`，升级到 1.6+，或用 `-r` 后 `tr -d '\n'` 替代。

---

### 步骤 4：用 `--json-schema` 拿到结构化结果，告别正则解析

自动化最大的痛苦是"模型返回自由文本，你还得正则抠字段"。`--output-format json --json-schema` 直接让模型吐**符合 schema 的结构化对象**，落在 `structured_output` 字段。

**操作说明**：让 Claude 从一个文件里抽取函数名，强制返回字符串数组。

```bash
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_session"]}`。`result` 字段仍保留人类可读文本，而 `structured_output` 是机器消费的干净对象，配合 `session_id`、`total_cost_usd` 等元数据一起返回。

**常见问题**：

- **Q：schema 写错了直接报错？** A：schema 必须是合法 JSON Schema。建议先用 `jq` 校验你的 schema 字符串本身能解析：`echo '...' | jq .`。`required` 里的字段模型一定会给，未 required 的可能缺失，下游用 `// []` 兜底。
- **Q：能强制 enum 枚举值吗？** A：可以，`{"enum":["bug","feature","question"]}` 非常适合做分类任务的护栏。

---

### 步骤 5：成本与错误捕获——编排器的"刹车系统"

无人值守跑久了，必须能感知"花了多少钱、有没有出错"。`json` 模式的 payload 自带 `total_cost_usd` 和**按模型拆分的成本明细**，`result` 事件则带 `is_error` 及错误子类型。

**操作说明**：写一个 wrapper 函数，跑完后打印成本，并在出错时按子类型分诊。

```bash
run_with_cost () {
  claude -p "$1" --output-format json --bare > /tmp/out.json
  local rc=$?
  # 成本
  jq -r '"💰 cost=$\(.total_cost_usd) turns=\(.num_turns) ms=\(.duration_ms)"' /tmp/out.json
  # 错误分诊
  if jq -e '.is_error == true' /tmp/out.json >/dev/null; then
    jq -r '"❌ 错误子类型: \(.subtype // "unknown")"' /tmp/out.json
  fi
  # 结构化结果
  jq -r '.result' /tmp/out.json
  return $rc
}
run_with_cost "检查 package.json 里有没有未使用的依赖"
```

**预期结果**：一行成本（`cost=$0.0312 turns=3 ms=12044`），正常时跟结果文本；如果触发配额/限流，会打印 `❌ 错误子类型: rate_limit`。

官方文档列出的错误子类型包括：`rate_limit`、`overloaded`、`billing_error`、`invalid_request`、`model_not_found`、`server_error`、`max_output_tokens`、`oauth_org_not_allowed`、`unknown`。**编排器应据此决定退避策略**——`rate_limit`/`overloaded` 可指数退避重试，`billing_error`/`oauth_org_not_allowed` 应立刻熔断报警。

**常见问题**：

- **Q：`is_error` 字段在 stream-json 的 result 事件里也有吗？** A：有。无论 json 还是 stream-json，收尾的 result 事件都携带这套字段。stream-json 场景用 `jq 'select(.type=="result")'` 取它。
- **Q：piped stdin 报错？** A：自 Claude Code v2.1.128 起，**管道 stdin 上限 10MB**，超了会报错。大输入用 `@文件` 或先落盘再在 prompt 里引用。

---

### 步骤 6：`--resume` 与 `--continue`，把单次调用串成有状态流水线

headless 不等于无状态。`--continue` 续跑最近一次会话，`--resume <session_id>` 续跑指定会话。**关键限制：session_id 查找的作用域是当前项目目录及其 git worktree**，所以多次调用必须在同一目录下跑。

**操作说明**：第一步起一个 review 会话并抓 session_id，第二步带着它追问。

```bash
# 第一步：开 review，抓 session_id
session_id=$(claude -p "审查这个代码库的性能问题" \
  --output-format json | jq -r '.session_id')

# 第二步：在同一目录下，续跑同一会话
claude -p "把刚才发现的前 3 个问题各给出修复方案" \
  --resume "$session_id" --output-format json | jq -r '.result'

# 或者最省事：直接续最近一次
claude -p "生成一份所有问题的汇总报告" --continue | cat
```

**预期结果**：第二步能引用第一步的上下文（"刚才发现的问题"），无需重新喂背景。把多个 `-p` 调用串成一条**有状态的 Agent 工作流**，而每一步都能独立观测、独立计费。

**常见问题**：

- **Q：`--resume` 报"找不到会话"？** A：99% 是目录变了。session_id 与目录绑定，CI 里务必 `cd` 到仓库根再跑。worktree 内的会话对主仓库可见，反之不行。
- **Q：并发多个会话怎么互不干扰？** A：每次开新会话立即抓 `session_id` 存变量/存文件，后续一律 `--resume` 指定 ID，绝不用 `--continue`（它会抢最近一次，并发下会串台）。

---

### 步骤 7：权限模式与 `--max-turns`，给自动化装护栏

无人值守的最大风险是"模型干了自己不该干的事"。三道闸门：`--permission-mode`、`--max-turns`、`--bare`。

**操作说明**：按风险等级选用权限模式。

```bash
# 只读巡视（最安全）：default 模式在 CI 下退化为只读命令集，适合巡检
claude -p "检查 Dockerfile 有没有安全坏味道" --permission-mode default --bare

# 自动改文件但不执行危险命令：acceptEdits
# 文档原文：acceptEdits 让 Claude 无需提示写文件，并自动批准 mkdir/touch 等常见文件系统命令
claude -p "应用 eslint --fix 的所有修复" --permission-mode acceptEdits

# 硬上限：最多 5 轮工具调用，防止失控
claude -p "重构这个模块" --permission-mode acceptEdits --max-turns 5
```

`bypassPermissions`（俗称 yolo）会跳过一切确认，**只在一次性沙箱容器/VM 里用**，生产链路严禁。

**预期结果**：`acceptEdits` 下文件改动直接生效不卡审批；`--max-turns 5` 下任务在第 5 轮工具调用后被强制收尾，result 事件照常带成本。

**常见问题**：

- **Q：acceptEdits 模式还是会停下来问？** A：注意 2026-W23 的更新——在 acceptEdits 下写入"可运行代码的文件"前会提示确认。要真正无人值守，要么把这些写入放进沙箱，要么用更细粒度的 hooks/allow 规则。
- **Q：怎么既自动跑又不越权？** A：`acceptEdits` + `--max-turns` + `--bare` 组合，再配合 `settings.json` 的 deny 规则封死 `rm -rf`、`git push` 等。

---

### 步骤 8：组装生产级编排器 `cch`（命令行 Agent 看板）

把前面 7 步拼成一个可复用脚本：实时进度看板 + 成本累计 + 错误分诊 + 指数退避重试。

**操作说明**：保存为 `~/.local/bin/cch` 并 `chmod +x`。

```bash
#!/usr/bin/env bash
# cch — Claude Code headless orchestrator with live dashboard
set -uo pipefail

PROMPT="$1"; shift
MAX_TURNS="${CCH_MAX_TURNS:-12}"
BUDGET="${CCH_BUDGET:-2.0}"   # 单次预算上限（美元）
MAX_RETRY=3

run_once () {
  claude -p "$PROMPT" "$@" \
    --output-format stream-json --verbose --include-partial-messages \
    --permission-mode acceptEdits --max-turns "$MAX_TURNS" --bare \
  | tee /tmp/cch_events.ndjson \
  | jq -rj 'select(.type=="stream_event" and .event.delta.type?=="text_delta") | .event.delta.text'
}

attempt=0
while [ $attempt -lt $MAX_RETRY ]; do
  attempt=$((attempt+1))
  run_once "$@" || true
  # 从 result 事件读成本与错误
  result=$(jq -c 'select(.type=="result")' /tmp/cch_events.ndjson | tail -n1)
  cost=$(echo "$result" | jq -r '.total_cost_usd // 0')
  err=$(echo "$result" | jq -r '.is_error // false')
  subtype=$(echo "$result" | jq -r '.subtype // ""')
  echo   # 换行，结束流式文本
  printf '📊 尝试 %d  成本 $%.4f  错误=%s %s\n' "$attempt" "$cost" "$err" "$subtype"

  # 预算熔断
  awk "BEGIN{exit !($cost > $BUDGET)}" && { echo "🛑 超预算($BUDGET)，熔断"; break; }
  # 成功或不可重试错误 → 退出
  [ "$err" = "false" ] && break
  case "$subtype" in
    rate_limit|overloaded) sleep $((2**attempt));;   # 指数退避
    billing_error|oauth_org_not_allowed) break;;     # 立即报警
  esac
done
```

**预期结果**：执行 `cch "给 src 下所有函数补类型注解"`，终端实时打印模型生成的文本（打字机效果），结束后输出一行 `📊 尝试 1 成本 $0.2143 错误=false`。限流时自动退避重试，超预算自动熔断。`/tmp/cch_events.ndjson` 留存完整事件流供事后审计。

**常见问题**：

- **Q：想同时看进度条和成本？** A：把 `tee` 后的管道再分一路：`tee >(jq -c 'select(.type=="assistant")|.message.usage' >> /tmp/usage.log)`，实时累计 token。
- **Q：多任务并行会互相覆盖 `/tmp/cch_events.ndjson`？** A：并发时改用 `mktemp`：`EV=$(mktemp)`，把它传进去，结束时清理。

---

## 实战案例：自动 Issue 分类与回填机器人

把整套能力串成一个**可运行**的端到端管道：扫描一个仓的开放 issue → 用 `--json-schema` 强制分类成 `bug/feature/question` 并打分 → 用 `--resume` 续跑同一会话让模型写处置建议 → 解析成本写日志。全程 headless、可观测、有预算护栏。

```bash
#!/usr/bin/env bash
# issue-triage.sh — 用法: ./issue-triage.sh <repo>  例 myorg/myrepo
set -uo pipefail
REPO="$1"; BUDGET=3.0

# 1) 拉取未关闭 issue 的标题（gh CLI）
mapfile -t TITLES < <(gh issue list -R "$REPO" --state open --limit 20 \
                       --json number,title -q '.[] | "\(.number)\t\(.title)"')

# 2) 对每个 issue：结构化分类 + 抓 session_id
: > /tmp/triage.log
for line in "${TITLES[@]}"; do
  num="${line%%$'\t'*}"; title="${line#*$'\t'}"
  out=$(claude -p "这是一条 GitHub issue，请分类。标题: $title" \
        --output-format json --bare \
        --json-schema '{"type":"object","properties":{
            "label":{"enum":["bug","feature","question"]},
            "confidence":{"type":"number"},
            "suggestion":{"type":"string"}},
            "required":["label","confidence","suggestion"]}')
  sid=$(echo "$out" | jq -r '.session_id')
  cost=$(echo "$out" | jq -r '.total_cost_usd')
  parsed=$(echo "$out" | jq -c '.structured_output')

  # 3) 续跑同一会话，让模型基于分类写出 200 字内处置建议
  reply=$(claude -p "基于你刚才的分类，用 200 字内给出处置建议" \
          --resume "$sid" --output-format json | jq -r '.result')

  # 4) 回填标签（演示用，真实环境按需放开）
  label=$(echo "$parsed" | jq -r '.label')
  echo "gh issue edit $num -R $REPO --add-label \"$label\"   # dry-run"
  printf '#%s [%s %.2f] $%s\n  %s\n\n' "$num" "$label" \
         "$(echo $parsed|jq -r .confidence)" "$cost" "$reply" >> /tmp/triage.log

  # 预算护栏
  total=$(awk '{s+=$1}END{print s}' /tmp/triage_costs 2>/dev/null || echo 0)
  awk "BEGIN{exit !($total > $BUDGET)}" && { echo "🛑 预算 $BUDGET 用尽，停止"; break; }
done

echo "✅ 完成，详见 /tmp/triage.log"
```

**运行预期**：`./issue-triage.sh myorg/myrepo` 会逐条打印 `#42 [bug 0.91] $0.0088` 加一段处置建议，并在 `/tmp/triage.log` 汇总。每条 issue 都用 `--resume` 复用分类时的会话上下文，省 token、保连贯；成本逐条累加，超 `$3.0` 自动停。

---

## 工作流程图

```mermaid
flowchart TD
    A[Shell / CI 触发] --> B["claude -p --output-format stream-json<br/>--verbose --include-partial-messages"]
    B -->|NDJSON 事件流| C{事件 type}
    C -->|system/init| D[会话身份证:<br/>model / tools / MCP / plugins]
    C -->|stream_event text_delta| E[jq -rj 实时打字机]
    C -->|assistant tool_use| F[工具调用清单]
    C -->|result| G[收尾:<br/>total_cost_usd / num_turns / is_error / subtype]

    G --> H{is_error?}
    H -->|false| I[输出结构化结果]
    H -->|true| J{subtype 分诊}
    J -->|rate_limit / overloaded| K[指数退避重试]
    J -->|billing / oauth| L[立即熔断报警]
    K --> B

    G -. 成本累计 .-> M[(预算护栏)]
    M -. 超 BUDGET .-> L
    I --> N[--resume session_id 续跑<br/>串成有状态流水线]
    N --> B
```

## 延伸阅读

- **Agent SDK 取代纯脚本**：当 shell+jq 编排器复杂到难以维护时，迁移到 [Claude Agent SDK](https://docs.claude.com/en/agent-sdk/overview)，它提供等价的 agent loop、回调式流式（`Stream responses in real-time`）和更优雅的错误处理。
- **会话管理全景**：`--continue`/`--resume` 之外还有 `--from-pr`、交互式 `/resume` 选择器、session 命名与分支，见 [Sessions 文档](https://docs.claude.com/en/sessions)。
- **沙箱化无人值守**：`bypassPermissions`（yolo）必须配沙箱，官方推荐 Bubblewrap/Seatbelt/Docker/devcontainer，见 [Sandbox environments](https://docs.claude.com/en/sandbox-environments)。
- **成本与缓存**：`ENABLE_PROMPT_CACHING_1H`、`cache_read_input_tokens` 等可把批量调用的 token 成本再砍一截，见 [Cost tracking](https://docs.claude.com/en/agent-sdk/cost-tracking)。
- **第三方封装参考**：[peakflames/claude-print](https://github.com/peakflames/claude-print) 提供跨平台实时进度回显；[claude-code-automation](https://skills.rest/skill/claude-code-automation) 演示了 headless + streaming 双模式封装思路。

## 训练营适用性评估

| 维度 | 评估 |
|---|---|
| **适合的训练营** | DevOps/SRE 自动化营、AI 工程化营、平台工程营；不适合零基础（需 shell+jq 基础） |
| **学员前置** | 能写 50 行 bash、会用 `jq`、理解管道与退出码 |
| **建议教学时长** | 4 课时：1 课时事件流理论（步骤1-3）、1 课时结构化与成本（步骤4-5）、1 课时会话与护栏（步骤6-7）、1 课时实战案例（步骤8+案例） |
| **可裁剪点** | 时间紧可跳过步骤4（json-schema），但步骤3和步骤5是核心，不建议省 |
| **进阶作业** | 让学员把 `cch` 扩展成支持多任务并发的 worker（用 `mktemp` 隔离 + `xargs -P`），并加一个 token 实时累计看板 |

## 参考来源

1. [Run Claude Code programmatically (Headless)](https://docs.claude.com/en/docs/claude-code/headless) — 官方 headless 文档，`--output-format`/`stream-json`/`--json-schema`/`--resume`/权限模式的权威说明
2. [Manage sessions](https://docs.claude.com/en/sessions) — `--continue`/`--resume`/`--from-pr` 及 session_id 目录作用域规则
3. [Permission modes](https://docs.claude.com/en/permission-modes) — `default`/`acceptEdits`/`plan`/`bypassPermissions` 四模式语义与 Shift+Tab 切换
4. [Cost tracking with the Claude Agent SDK](https://docs.claude.com/en/agent-sdk/cost-tracking) — `total_cost_usd`、按模型成本拆分、prompt caching 配置
5. [peakflames/claude-print](https://github.com/peakflames/claude-print) — 开源 headless 进度回显封装，本文实时看板设计的现实参照

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