首页 › tutorials › Claude Code 无头-p批量任务实战

概述

本教程聚焦 Claude Code 的无头模式（headless -p）——把交互式终端变成可被脚本、CI 流水线、批处理程序调用的命令行工具。学完你将掌握：用 -p 跑非交互任务、用 --bare 保证 CI 可复现、用 --output-format json/--json-schema 拿到结构化结果、用 --allowedTools/--permission-mode 圈定权限、用 --continue/--resume 续跑会话，以及用 Python 编排并发批量任务。适合已写过一两个 Skill 或用过交互模式的开发者，想把 Claude Code 嵌入自动化链路。

前置条件

正文

第 1 步：认识无头模式 -p 与 --bare

交互模式下，Claude Code 会渲染一个 TUI（终端界面），等待你逐条输入。无头模式用 -p（或 --print）跳过 TUI，把提示词作为参数传入，结果直接打到 stdout：

claude -p "What does the auth module do?"

预期结果：终端直接打印一段纯文本说明，无界面、无颜色码污染，调用结束即退出。

CI 场景下，强烈建议加上 --bare。它的作用是跳过自动发现 hooks、skills、plugins、MCP 服务器、auto memory 和 CLAUDE.md，只认你显式传入的参数。这意味着同事 ~/.claude 里的某个 hook、或项目 .mcp.json 里的某个 MCP 服务器，在 bare 模式下都不会被加载——从而保证「同一台机器/同一个 commit，每次结果一致」。

claude --bare -p "Summarize this file" --allowedTools "Read"

bare 模式下 Claude 仍可使用 Bash、文件读、文件写三类工具；需要其它上下文时用 flag 注入：

想加载的内容 对应 flag
追加系统提示 --append-system-prompt / --append-system-prompt-file
设置文件 --settings <path>
MCP 服务器 --mcp-config <path>
自定义子代理 --agents <path>
插件 --plugin-dir <path> / --plugin-url <url>

注意：bare 模式会跳过 OAuth 与 keychain 读取，鉴权必须来自 ANTHROPIC_API_KEY 或 --settings 里的 apiKeyHelper。官方已表态 --bare 未来会成为 -p 的默认行为，现在就养成加 --bare 的习惯最省心。

常见问题：脚本里不写 --bare，结果本地跑得好好的，到了 CI 就行为飘忽——多半是 CI 机器上存在某个本地 hook 或 .mcp.json 被加载了。统一加 --bare 即可消除。

第 2 步：管道与重定向——把 Claude 当 Unix 工具

-p 模式会读取 stdin，因此你可以像对待 grep、awk 一样把数据管道进来、把结果重定向出去：

cat build-error.txt | claude -p 'concisely explain the root cause of this build error' > output.txt

把 diff 喂给 Claude 当「拼写检查器」，写进 package.json 的脚本里：

{
  "scripts": {
    "lint:claude": "git diff main | claude -p \"you are a typo linter. for each typo in this diff, report filename:line on one line and the issue on the next. return nothing else.\""
  }
}

把 diff 管道进来有个好处：Claude 不需要 Bash 权限去读它，省去一次权限交互。

容量红线：从 v2.1.128 起，管道传入的 stdin 上限是 10MB。超过会以非零退出码报错。遇到超大输入（如巨型日志），别管道，改写成文件后在 prompt 里引用路径。

# 错误：超大日志直接管道会触发 10MB 上限
# cat huge.log | claude -p "analyze"   # ❌ 退出码非 0

# 正确：写文件，在 prompt 里点名
# claude -p "Analyze the log at ./huge.log" --allowedTools "Read"  # ✅

常见问题：脚本跑完发现 stdout 里混入了一行提示性文本。检查是否误用了交互式 claude（没带 -p）——那会渲染 TUI。-p 输出永远是干净的。

第 3 步：结构化输出 --output-format 与 --json-schema

文本输出对人友好，对程序不友好。--output-format 给你三种模式：

claude -p "Summarize this project" --output-format json

用 jq 抽字段：

# 取文本结果
claude -p "Summarize this project" --output-format json | jq -r '.result'

# 取成本
claude -p "Summarize this project" --output-format json | jq '.total_cost_usd'

要拿到符合固定 schema 的结构化数据，组合 --output-format json + --json-schema，结果会出现在 structured_output 字段里：

claude -p "Extract the main function names from auth.py" \
  --output-format json \
  --json-schema '{"type":"object","properties":{"functions":{"type":"array","items":{"type":"string"}}},"required":["functions"]}' \
  | jq '.structured_output'

流式模式适合长输出/需要进度反馈的场景，配合 --verbose 和 --include-partial-messages 可逐 token 拿到增量：

claude -p "Explain recursion" \
  --output-format stream-json --verbose --include-partial-messages

只挑文本增量打印：

claude -p "Write a poem" \
  --output-format stream-json --verbose --include-partial-messages \
  | jq -rj 'select(.type == "stream_event" and .event.delta.type? == "text_delta") | .event.delta.text'

流里还有两类系统事件值得在脚本里处理： - system/init：会话第一条事件，上报 model、tools、MCP servers、已加载 plugins。可在脚本里据此校验插件是否加载成功（CI 里失败即阻断）。 - system/api_retry：可重试错误发生时上报 attempt、max_retries、retry_delay_ms、error_status、error 类别（rate_limit/overloaded/billing_error 等）。可用于自定义退避或展示重试进度。

常见问题：解析 result 时发现换行变成字面 \n。这是 JSON 正确转义，用 jq -r（raw）输出即可还原多行。

第 4 步：权限围栏——--allowedTools 与 --permission-mode

无头模式无人值守，权限是第一道安全闸。两条路径：

(1) --allowedTools 白名单，使用权限规则语法：

claude -p "Run the test suite and fix any failures" \
  --allowedTools "Bash,Read,Edit"

细到具体命令前缀（注意 * 前的空格）：

claude -p "Look at my staged changes and create an appropriate commit" \
  --allowedTools "Bash(git diff *),Bash(git log *),Bash(git status *),Bash(git commit *)"

空格很关键：Bash(git diff *) 只匹配 git diff 开头的命令；写成 Bash(git diff*) 会连 git diff-index 也放行。

(2) --permission-mode 设基线： - acceptEdits：允许写文件，并自动放行 mkdir/touch/mv/cp 等常见文件系统命令；其它 shell 命令与网络请求仍需 --allowedTools 或 permissions.allow 规则，否则一尝试就中止。 - dontAsk：除 permissions.allow 规则和只读命令集外一律拒绝，适合锁定型 CI。

claude -p "Apply the lint fixes" --permission-mode acceptEdits

强烈不建议在 CI 里直接用 --dangerously-skip-permissions 或 bypassPermissions 裸奔——它等于让 Claude 无限制执行任意命令。仅在你完全隔离的沙箱、且明确知道后果时使用。生产链路应优先 acceptEdits + 精确白名单。

常见问题：脚本跑到一半报「permission needed」就中止了。说明触发了未放行的工具。检查 --allowedTools 是否覆盖了 Claude 实际用到的命令（例如它会 git add，但白名单里只有 git commit *）。无头模式下不会弹窗等你看，所以白名单要一次配全。

第 5 步：会话连续性 --continue 与 --resume

每次 -p 默认是独立会话。要分多步推进同一任务，用续跑：

# 首轮
claude -p "Review this codebase for performance issues"
# 续最近一次会话
claude -p "Now focus on the database queries" --continue
claude -p "Generate a summary of all issues found" --continue

--continue 续的是当前目录下最近的会话。若同时跑多个会话，要捕获 session_id 精确续跑：

session_id=$(claude -p "Start a review" --output-format json | jq -r '.session_id')
claude -p "Continue that review" --resume "$session_id"

作用域：--resume 的会话查找限定在当前项目目录及其 git worktrees。两条命令要在同一目录下运行。

常见问题：脚本里 --resume 报找不到会话。多半是工作目录变了（比如在临时 clone 目录里拿到的 session_id，又在另一个目录去 resume）。确保续跑命令与首轮同目录。

第 6 步：批量任务编排——Python SIGINT 循环与并发

当任务量超过单次上下文承载（经验上单会话处理约 20–25 个同类项目后开始「变糊」），就要分批：每批跑一会、优雅停掉、记录进度、再开下一批。

串行分批（SIGINT 优雅停）：

# run_claude_in_batches.py
import subprocess, time, signal

def one_go(repo, wait_minutes=5):
    cmd = [
        "claude", "--bare", "-p",
        f"Review repo {repo} tracked in ./progress.json; process the next 5 items, update progress, then stop.",
        "--permission-mode", "acceptEdits",
        "--allowedTools", "Read,Edit,Bash",
        "--output-format", "json",
    ]
    proc = subprocess.Popen(cmd, cwd=".", text=True)
    print(f"[{repo}] started PID={proc.pid}, wait {wait_minutes}m")
    time.sleep(wait_minutes * 60)
    proc.send_signal(signal.SIGINT)           # 优雅停，让 Claude 落盘进度
    try:
        proc.wait(timeout=5)
    except subprocess.TimeoutExpired:
        proc.kill()                            # 5 秒不退则强杀
    return proc.returncode

if __name__ == "__main__":
    for repo in open("repos.txt"):
        one_go(repo.strip())

要点：让 Claude 把进度写进文件（如 progress.json），下一轮 prompt 里告诉它「接着进度继续」，从而实现跨批次的状态延续。

并发批量（asyncio + 信号量）：互相独立的任务可并发跑，省墙钟时间。用 --output-format json 收集每个调用的成本并汇总：

# batch_concurrent.py
import asyncio, json, pathlib

async def review_one(repo: str, sem: asyncio.Semaphore) -> dict:
    async with sem:
        proc = await asyncio.create_subprocess_exec(
            "claude", "--bare", "-p",
            f"Review {repo} for security issues; output a short JSON of findings.",
            "--allowedTools", "Read,Bash",
            "--output-format", "json",
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        )
        out, _ = await proc.communicate()
        try:
            data = json.loads(out)
            return {"repo": repo, "cost": data.get("total_cost_usd", 0),
                    "result": data.get("result"), "is_error": data.get("is_error")}
        except json.JSONDecodeError:
            return {"repo": repo, "cost": 0, "result": out.decode(errors="replace"),
                    "is_error": True}

async def main():
    repos = [l.strip() for l in open("repos.txt") if l.strip()]
    sem = asyncio.Semaphore(4)                          # 同时最多 4 个
    results = await asyncio.gather(*(review_one(r, sem) for r in repos))
    pathlib.Path("reports.json").write_text(json.dumps(results, ensure_ascii=False, indent=2))
    total = sum(r["cost"] for r in results)
    print(f"Done. {len(results)} repos, total cost ≈ ${total:.4f}")

asyncio.run(main())

并发上限别拉满。Claude Code 调用共享你的订阅/API 配额，过高的并发会触发 rate_limit（第 3 步提到的 api_retry 事件即对应此类）。建议从 2–4 起步观察。

关于官方 Agent SDK：若你要的不是「调几次 CLI」而是「构建长生命周期 agent 系统」（需要 hooks、子代理、MCP 全家桶、结构化消息对象），直接用官方 Claude Agent SDK（Python/TypeScript），其核心是异步生成器 query()：

# pip install claude-agent-sdk
from claude_agent_sdk import query

async for message in query(prompt="Find and fix the bug in auth.py",
                           options={"allowed_tools": ["Read", "Edit", "Bash"]}):
    print(message)

SDK 底层仍包装 CLI 二进制，但提供类型化消息、hooks、会话 resume/fork 等。轻量脚本用裸 -p 即可（零依赖、~50 行），生产级 agent 用 SDK。

常见问题：批处理中途某条卡死。无头模式下，Claude 启动的后台 Bash 任务（如 dev server、watch build）会在 Claude 返回最终结果、stdin 关闭后约 5 秒被回收；子代理/工作流不受 5 秒限制（其结果属于最终输出），但默认最多等 10 分钟（CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS 可调，设 0 为无限）。脚本里务必给每个调用设 timeout，避免一个卡死的子进程拖垮整条流水线。

第 7 步：无头 + Hooks 协作（CI 安全护栏）

Hooks 在交互模式与无头模式下都会触发（除非 --bare）。在自动化链路里，最有价值的两枚钩子是：

在 .claude/settings.json 里挂一个 PreToolUse 守卫示例（拒绝危险 Bash）：

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [{
          "type": "command",
          "command": "node ~/.claude/guard-block-dangerous.js"
        }]
      }
    ]
  }
}

注意：无头批处理若加了 --bare，这些项目级/用户级 hooks 会被跳过。要么不用 --bare（接受本地配置介入），要么把护栏逻辑移进 --allowedTools 白名单与 prompt 约束。CI 通常选 --bare + 严格白名单的组合，可复现性优先。

（hooks 的完整配置语法不是本篇重点，前序教程「双钩护栏」「全事件钩子编排」已深入展开，此处只聚焦其与无头模式的协作边界。）

第 8 步：接入 CI/CD（GitHub Actions / GitLab）

无头模式天然适配 CI。三条铁律：加 --bare（可复现）、严格白名单（安全）、解析 JSON 拿成本与错误（可观测）。

GitHub Actions 片段——对 PR diff 做安全审查，失败即阻断：

name: Claude Security Review
on: { pull_request: { types: [opened, synchronize] } }
jobs:
  review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with: { fetch-depth: 0 }
      - run: gh pr diff "$PR" > pr.diff
        env: { GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}, PR: ${{ github.event.pull_request.number }} }
      - name: Review
        env: { ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }} }
        run: |
          set -euo pipefail
          cat pr.diff | claude --bare -p \
            --append-system-prompt "You are a security engineer. Review for vulnerabilities." \
            --output-format json > review.json
          jq -e '.is_error == false' review.json
          jq -r '.result' review.json

GitLab CI 的 script: 段同理，把 ANTHROPIC_API_KEY 放进 CI/CD 变量即可。官方也提供 GitHub Actions 与 GitLab CI/CD 的 Agent SDK 集成文档。

常见问题：CI 里 claude 提示未登录。bare 模式不读 OAuth/keychain，必须在 CI 注入 ANTHROPIC_API_KEY（或用 --settings 传 apiKeyHelper）。

实战案例：多仓库批量安全审查流水线

目标：给定一份 repos.txt，逐个 clone、审查、产出结构化报告，最后汇总成本与高危项。可串行也可并发，带超时与重试。

目录结构：

batch-review/
├── repos.txt              # 每行一个 git 仓库地址
├── run_review.sh          # 单仓库审查（核心调用）
├── batch.py               # 并发编排 + 汇总
└── reports/               # 每仓库一份 .json

run_review.sh——单个仓库的原子审查单元，输入仓库 URL，输出 reports/<name>.json：

#!/usr/bin/env bash
set -euo pipefail
URL="$1"
NAME="$(basename "$URL" .git)"
WORK="/tmp/rev-$NAME"
rm -rf "$WORK"; git clone --depth 1 "$URL" "$WORK" >/dev/null 2>&1

SCHEMA='{"type":"object","properties":{
  "summary":{"type":"string"},
  "critical":{"type":"array","items":{"type":"string"}},
  "recommendations":{"type":"array","items":{"type":"string"}}
},"required":["summary","critical","recommendations"]}'

claude --bare -p \
  "Audit the codebase in the current directory for security vulnerabilities. \
   Focus on injection, auth flaws, secret leakage. Be concise." \
  --add-dir "$WORK" \
  --allowedTools "Read,Bash(git log *),Bash(grep *)" \
  --output-format json \
  --json-schema "$SCHEMA" \
  > "reports/$NAME.raw.json"

# 组装最终报告：附成本与会话
jq --arg repo "$NAME" '{
  repo: $repo,
  structured: .structured_output,
  cost_usd: .total_cost_usd,
  session_id: .session_id,
  is_error: .is_error
}' "reports/$NAME.raw.json" > "reports/$NAME.json"
rm -f "reports/$NAME.raw.json"
echo "✔ $NAME  cost=\$(jq -r '.cost_usd' reports/$NAME.json)"

batch.py——并发驱动 + 汇总（带每任务 8 分钟超时）：

#!/usr/bin/env python3
import asyncio, json, pathlib, sys

REPORTS = pathlib.Path("reports"); REPORTS.mkdir(exist_ok=True)

async def run_one(url: str, sem: asyncio.Semaphore):
    name = url.rstrip("/").split("/")[-1].replace(".git", "")
    async with sem:
        try:
            proc = await asyncio.create_subprocess_exec(
                "./run_review.sh", url,
                stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
            )
            await asyncio.wait_for(proc.communicate(), timeout=8*60)
            return name, proc.returncode
        except asyncio.TimeoutError:
            proc.kill()
            return name, -1   # 超时

async def main(concurrency: int):
    sem = asyncio.Semaphore(concurrency)
    urls = [l.strip() for l in open("repos.txt") if l.strip()]
    done = await asyncio.gather(*(run_one(u, sem) for u in urls))

    rows, total, critical = [], 0.0, 0
    for name, rc in done:
        f = REPORTS / f"{name}.json"
        if f.exists():
            d = json.loads(f.read_text())
            total += float(d.get("cost_usd") or 0)
            n = len((d.get("structured") or {}).get("critical") or [])
            critical += n
            rows.append({"repo": name, "rc": rc, "critical": n,
                         "cost": d.get("cost_usd"), "error": d.get("is_error")})
        else:
            rows.append({"repo": name, "rc": rc, "error": True})

    summary = {"total_cost_usd": round(total, 4), "critical_total": critical, "rows": rows}
    (REPORTS / "_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2))
    print(json.dumps(summary, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    asyncio.run(main(int(sys.argv[1]) if len(sys.argv) > 1 else 3))

运行：

chmod +x run_review.sh
python3 batch.py 3        # 并发 3

预期产出：reports/<repo>.json 每仓库一份（含 summary/critical/recommendations 结构化字段 + 成本 + session_id），外加 reports/_summary.json 汇总总成本与高危项总数。超时的仓库标记 rc: -1 不会让整条流水线崩。

可扩展点：把 --resume "$session_id" 接进 run_review.sh 可对同一个仓库做「初审 → 深挖」的多轮续跑；把 _summary.json 的 critical_total 作为 CI 的失败阈值（如 >0 即阻断合并）。

工作流程图

下图展示一条无头批量流水线的完整时序：编排器并发拉起多个 claude --bare -p 子进程，每个进程独立完成「鉴权 → 工具调用（受白名单/permission-mode 约束）→ 产出 JSON」，再由编排器汇总。

flowchart LR
    A[repos.txt 任务清单] --> B[Python 编排器 batch.py]
    B -->|信号量限并发| S1[claude --bare -p 子进程 1]
    B --> S2[claude --bare -p 子进程 2]
    B --> S3[claude --bare -p 子进程 N]

    S1 --> P1{权限围栏}
    S2 --> P2{权限围栏}
    S3 --> P3{权限围栏}

    subgraph 围栏["--allowedTools / --permission-mode"]
      P1
      P2
      P3
    end

    P1 --> R1[Read/Bash 受限执行]
    P2 --> R2[Read/Bash 受限执行]
    P3 --> R3[Read/Bash 受限执行]

    R1 --> J1[JSON: result + structured_output + cost + session_id]
    R2 --> J2[JSON 结果]
    R3 --> J3[JSON 结果]

    J1 --> SUM[reports/*.json + _summary.json 汇总成本/高危项]
    J2 --> SUM
    J3 --> SUM
    SUM --> CI{CI 阈值判断: critical_total>0 ?}
    CI -->|是| FAIL[阻断合并]
    CI -->|否| PASS[放行]

会话续跑时，从首条响应里取出 session_id，在后续 --resume 调用中复用，保持上下文连贯：

sequenceDiagram
    participant Script as 编排脚本
    participant CC as claude -p
    participant FS as 本地文件

    Script->>CC: -p "首轮审查" --output-format json
    CC-->>Script: {result, session_id, total_cost_usd}
    Note over Script: 解析出 session_id
    Script->>CC: -p "深挖数据库查询" --resume "$session_id"
    CC->>FS: 读项目文件(受白名单约束)
    FS-->>CC: 代码内容
    CC-->>Script: 续跑结果(沿用首轮上下文)
    Script->>FS: 写 reports/<repo>.json

延伸阅读

训练营适用性评估

参考来源

  1. Run Claude Code programmatically（官方无头模式文档） — -p/--bare/--output-format/--allowedTools/--continue/--resume 等核心特性的权威说明
  2. Automate work with routines（官方文档，research preview） — 云端托管的定时/API/GitHub 事件触发自动化
  3. Using Claude Code Programmatically: A Complete Guide（社区 Gist） — --print --output-format=json 程序化封装、Node 包装类、与官方 SDK 的取舍对比
  4. Running Claude Code in Batches for Automated Workflows（DevGuide） — Python SIGINT/kill 分批循环模式与适用场景
  5. Claude Agent SDK Overview（官方） — 生产级 query() 异步生成器、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学通信知识星球海报-扫码加入

📎 本分类最新文章