首页 › tutorials › 流式JSON：打造命令行编排看板

概述

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

前置条件

# 环境自检一行流
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。

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。

常见问题：

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

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

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

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 至少包含这几类：

常见问题：

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

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

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

# (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}。

常见问题：

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

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

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

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 等元数据一起返回。

常见问题：

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

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

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

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 应立刻熔断报警。

常见问题：

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

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

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

# 第一步：开 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 工作流，而每一步都能独立观测、独立计费。

常见问题：

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

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

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

# 只读巡视（最安全）：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 事件照常带成本。

常见问题：

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

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

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

#!/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 留存完整事件流供事后审计。

常见问题：

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

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

#!/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 自动停。

工作流程图

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

延伸阅读

训练营适用性评估

维度 评估
适合的训练营 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) — 官方 headless 文档，--output-format/stream-json/--json-schema/--resume/权限模式的权威说明
  2. Manage sessions — --continue/--resume/--from-pr 及 session_id 目录作用域规则
  3. Permission modes — default/acceptEdits/plan/bypassPermissions 四模式语义与 Shift+Tab 切换
  4. Cost tracking with the Claude Agent SDK — total_cost_usd、按模型成本拆分、prompt caching 配置
  5. 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学通信知识星球海报-扫码加入

📎 本分类最新文章