Claude Code动态工作流与自动化编排全实战

# 最基本用法
claude -p "解释当前项目的目录结构"

# 管道输入
cat error.log | claude -p "分析这段错误日志的根本原因"

# 从文件读取Prompt
claude -p "$(cat prompt.txt)"

# text — 纯文本，适合直接阅读
claude -p "生成README" --output-format text

# json — 完整结构化响应，包含cost、duration、session_id等元信息
claude -p "列出所有TODO注释" --output-format json

# stream-json — NDJSON流式输出，每行一个JSON对象，适合实时处理
claude -p "重构src/目录" --output-format stream-json | jq -c 'select(.type == "assistant")'

{"type":"system","subtype":"init","session_id":"abc123","tools":["Read","Write","Bash"]}
{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"正在分析..."}]}}
{"type":"result","subtype":"success","cost_usd":0.023,"duration_ms":4521,"num_turns":3}

# 限制工具集 — 只允许读取，不允许修改
claude -p "审查代码质量" \
  --allowedTools "Read,Glob,Grep" \
  --output-format json

# 限制交互轮数 — 防止无限循环
claude -p "修复lint错误" \
  --max-turns 10 \
  --output-format text

# 延续上次会话 — 实现多轮对话
claude -p --continue "继续上次未完成的重构"

# 指定模型
claude -p "执行复杂架构分析" \
  --model claude-opus-4-6 \
  --output-format json

#!/bin/bash
# 安全的headless调用模板
result=$(claude -p "$1" --output-format json --max-turns 15 2>&1)
exit_code=$?

if [ $exit_code -eq 0 ]; then
  echo "$result" | jq -r '.result'
else
  echo "ERROR (code=$exit_code): $result" >&2
  exit $exit_code
fi

#!/bin/bash
# batch-review.sh — 对所有修改的文件逐一执行AI审查
set -euo pipefail

CHANGED_FILES=$(git diff --name-only main...HEAD)

for file in $CHANGED_FILES; do
  echo "=== 审查: $file ==="

  # 只审查代码文件
  if [[ ! "$file" =~ \.(ts|tsx|py|go|rs|java)$ ]]; then
    echo "跳过非代码文件: $file"
    continue
  fi

  claude -p "审查这个文件的代码质量，关注安全性、性能和可维护性: $(cat "$file")" \
    --allowedTools "Read" \
    --output-format json \
    --max-turns 3 \
    > "reviews/$(basename "$file").json"

  echo "✓ 审查完成: $file"
done

echo "全部审查结果已保存到 reviews/ 目录"

flowchart TD
    A[用户提交Prompt] --> B[UserPromptSubmit Hook]
    B --> C{Hook通过?}
    C -->|exit 0| D[Claude思考并选择工具]
    C -->|exit 2| E[阻断本次操作]
    D --> F[PreToolUse Hook]
    F --> G{Hook通过?}
    G -->|exit 0| H[执行工具]
    G -->|exit 2| I[阻止工具执行]
    H --> J[PostToolUse Hook]
    J --> K[继续Agentic Loop]
    K --> L{任务完成?}
    L -->|否| D
    L -->|是| M[Stop Hook]
    M --> N[Notification Hook]

    style B fill:#4CAF50,color:white
    style F fill:#FF9800,color:white
    style J fill:#2196F3,color:white
    style M fill:#9C27B0,color:white
    style N fill:#F44336,color:white

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [
          {
            "type": "command",
            "command": "python3 .claude/hooks/audit_command.py"
          }
        ]
      }
    ],
    "PostToolUse": [
      {
        "matcher": "Write|Edit",
        "hooks": [
          {
            "type": "command",
            "command": "npx prettier --write $CLAUDE_FILE_PATH 2>/dev/null || true"
          }
        ]
      }
    ],
    "Stop": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "bash .claude/hooks/on_complete.sh"
          }
        ]
      }
    ],
    "Notification": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "curl -X POST $SLACK_WEBHOOK -H 'Content-type: application/json' --data '{\"text\":\"Claude Code任务完成\"}'"
          }
        ]
      }
    ]
  }
}

┌─────────────────┐     JSON on stdin      ┌──────────────────┐
│                  │ ──────────────────────▶ │                   │
│   Claude Code    │                        │   Hook Script     │
│                  │ ◀────────────────────── │                   │
└─────────────────┘     JSON on stdout      └──────────────────┘
                         exit code:
                         0 = 继续
                         2 = 阻断

{
  "session_id": "sess_abc123",
  "tool_name": "Bash",
  "tool_input": {
    "command": "rm -rf /tmp/test",
    "description": "清理临时文件"
  }
}

{
  "decision": "block",
  "reason": "检测到危险的rm -rf命令，已拦截"
}

#!/usr/bin/env python3
# .claude/hooks/pre_tool_use.py — 安全拦截器
import json
import sys

def main():
    data = json.load(sys.stdin)
    tool_name = data.get("tool_name", "")
    tool_input = data.get("tool_input", {})

    # 规则1：拦截危险Bash命令
    if tool_name == "Bash":
        cmd = tool_input.get("command", "")
        dangerous_patterns = [
            "rm -rf /", "DROP TABLE", "format C:",
            "> /dev/sda", "curl.*|.*sh", "chmod 777"
        ]
        import re
        for pattern in dangerous_patterns:
            if re.search(pattern, cmd, re.IGNORECASE):
                print(json.dumps({
                    "decision": "block",
                    "reason": f"拦截危险命令: 匹配模式 '{pattern}'"
                }))
                sys.exit(2)

    # 规则2：保护关键文件不被写入
    if tool_name in ("Write", "Edit"):
        file_path = tool_input.get("file_path", "")
        protected = [".env", "credentials", "secret", "production.yaml"]
        for keyword in protected:
            if keyword in file_path.lower():
                print(json.dumps({
                    "decision": "block",
                    "reason": f"保护敏感文件: {file_path}"
                }))
                sys.exit(2)

    # 通过所有检查
    sys.exit(0)

if __name__ == "__main__":
    main()

{
  "hooks": {
    "PostToolUse": [
      {
        "matcher": "Write|Edit",
        "hooks": [
          {
            "type": "command",
            "command": "bash -c 'FILE=$(cat | jq -r \".tool_input.file_path // .tool_input.filepath // empty\"); if [ -n \"$FILE\" ]; then case \"$FILE\" in *.py) python3 -m black \"$FILE\" 2>/dev/null ;; *.ts|*.tsx) npx prettier --write \"$FILE\" 2>/dev/null ;; *.go) gofmt -w \"$FILE\" 2>/dev/null ;; esac; fi'"
          }
        ]
      }
    ]
  }
}

flowchart LR
    P[用户Prompt] --> O[Orchestrator Agent]
    O -->|生成workflow.js| W[JavaScript运行时]
    W --> S1[SubAgent 1]
    W --> S2[SubAgent 2]
    W --> S3[SubAgent 3]
    W --> S4[SubAgent N...]
    S1 --> R[结果聚合]
    S2 --> R
    S3 --> R
    S4 --> R
    R --> F[最终输出]

    style O fill:#FF6B6B,color:white
    style W fill:#4ECDC4,color:white
    style R fill:#45B7D1,color:white

# 方法1：环境变量
export CLAUDE_CODE_ENABLE_DYNAMIC_WORKFLOWS=1

# 方法2：在.claude/settings.json中配置
# {
#   "experimentalFeatures": {
#     "dynamicWorkflows": true
#   }
# }

# 方法3：headless模式中使用
CLAUDE_CODE_ENABLE_DYNAMIC_WORKFLOWS=1 claude -p \
  "将src/目录下所有组件从Class组件迁移到Hooks" \
  --model claude-opus-4-6

# 场景：将100+文件从JavaScript迁移到TypeScript
CLAUDE_CODE_ENABLE_DYNAMIC_WORKFLOWS=1 claude -p \
  "将src/目录下所有.js文件迁移到TypeScript：
   1. 添加类型注解
   2. 创建对应的.ts文件
   3. 更新所有import路径
   4. 确保每个迁移后的文件通过tsc --noEmit检查
   每个文件分配给一个独立的子Agent并行处理" \
  --model claude-opus-4-6 \
  --output-format stream-json

// workflow.js (由Claude自动生成)
const files = await glob("src/**/*.js");

const tasks = files.map(file => ({
  agent: {
    prompt: `将 ${file} 迁移为TypeScript。添加完整的类型注解，确保通过 tsc --noEmit。`,
    tools: ["Read", "Write", "Edit", "Bash"],
    model: "claude-sonnet-4-6"
  }
}));

const results = await Promise.all(tasks.map(t => runAgent(t)));
await aggregate(results);

# 启用ultracode
CLAUDE_CODE_ENABLE_DYNAMIC_WORKFLOWS=1 claude -p \
  "重构整个项目的错误处理模式" \
  --model claude-opus-4-6

# ultracode会在settings中配置
# "ultracode": "xhigh"  — 最高并行度，适合大批量任务
# "ultracode": "high"   — 高并行度
# "ultracode": "medium" — 中等并行度（默认）

# Python
pip install claude-agent-sdk

# TypeScript
npm install @anthropic-ai/claude-agent-sdk

import asyncio
from claude_code_sdk import query, ClaudeCodeOptions

async def main():
    # 基础调用
    result = await query(
        prompt="分析当前项目的架构，找出可以优化的瓶颈",
        options=ClaudeCodeOptions(
            allowed_tools=["Read", "Glob", "Grep", "Bash"],
            max_turns=10,
            output_format="json"
        )
    )

    # 处理流式响应
    async for event in query_stream(
        prompt="重构auth模块",
        options=ClaudeCodeOptions(
            allowed_tools=["Read", "Write", "Edit", "Bash"],
            max_turns=20
        )
    ):
        if event.type == "assistant":
            print(f"助手: {event.message.content}")
        elif event.type == "result":
            print(f"完成! 耗时: {event.duration_ms}ms, 费用: ${event.cost_usd}")

asyncio.run(main())

from claude_code_sdk import query, ClaudeCodeOptions, Session

async def code_review_pipeline():
    # 创建持久会话
    session = Session()

    # 第一轮：全局分析
    r1 = await session.query("分析这个项目的测试覆盖率")

    # 第二轮：基于分析结果深入
    r2 = await session.query("为覆盖率最低的三个模块编写测试")

    # 第三轮：验证
    r3 = await session.query("运行所有测试，确保新测试通过")

    return r3

asyncio.run(code_review_pipeline())

import asyncio
from claude_code_sdk import query, ClaudeCodeOptions

async def parallel_review():
    """并行审查多个文件"""
    files = ["src/auth.py", "src/api.py", "src/db.py", "src/utils.py"]

    options = ClaudeCodeOptions(
        allowed_tools=["Read", "Grep"],
        max_turns=5,
        output_format="json"
    )

    # 并发启动4个审查任务
    tasks = [
        query(
            prompt=f"审查文件 {f} 的安全漏洞、性能问题和代码风格。输出JSON格式。",
            options=options
        )
        for f in files
    ]

    results = await asyncio.gather(*tasks)

    # 聚合结果
    for f, r in zip(files, results):
        print(f"\n{'='*50}")
        print(f"文件: {f}")
        print(r)

asyncio.run(parallel_review())

# .github/workflows/claude-review.yml
name: Claude Code Review

on:
  pull_request:
    types: [opened, synchronize]

jobs:
  review:
    runs-on: ubuntu-latest
    permissions:
      contents: read
      pull-requests: write

    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - name: Claude Code Review
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          # 获取变更文件列表
          CHANGED=$(git diff --name-only origin/main...HEAD)

          # 逐一审查
          for file in $CHANGED; do
            echo "审查: $file"
            claude -p \
              "作为代码审查专家，审查文件 $file 的变更。
               关注：安全性、性能、可维护性、测试覆盖。
               用中文输出审查意见，格式为GitHub评论。" \
              --allowedTools "Read,Grep,Glob" \
              --output-format text \
              --max-turns 5 \
              >> review_comments.md

            echo -e "\n---\n" >> review_comments.md
          done

      - name: Post Review Comment
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const review = fs.readFileSync('review_comments.md', 'utf8');
            github.rest.issues.createComment({
              owner: context.repo.owner,
              repo: context.repo.repo,
              issue_number: context.issue.number,
              body: review
            });

# .github/workflows/auto-fix.yml
name: Auto-fix CI Failures

on:
  workflow_run:
    workflows: ["Test"]
    types: [completed]
    conclusion: failure

jobs:
  auto-fix:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Get Failure Logs
        run: |
          gh run view ${{ github.event.workflow_run.id }} --log > failure.log
        env:
          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}

      - name: Claude Auto-fix
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          claude -p \
            "CI测试失败了，错误日志如下：$(cat failure.log)
             请分析失败原因，修复代码，确保测试通过。
             只修改必要的代码，不要过度重构。" \
            --allowedTools "Read,Write,Edit,Bash" \
            --max-turns 15 \
            --output-format json \
            > fix_result.json

          # 检查修复是否成功
          if [ "$(jq -r '.exit_code // 0' fix_result.json)" = "0" ]; then
            echo "fix_applied=true" >> $GITHUB_OUTPUT
          fi

      - name: Commit and Push Fix
        if: steps.auto-fix.outputs.fix_applied == 'true'
        run: |
          git config user.name "Claude Code Bot"
          git config user.email "[email protected]"
          git add -A
          git commit -m "fix: 自动修复CI测试失败

          Co-Authored-By: Claude <[email protected]>"
          git push

# .github/workflows/quality-scan.yml
name: Weekly Quality Scan

on:
  schedule:
    - cron: '0 2 * * 1'  # 每周一凌晨2点

jobs:
  scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Full Codebase Analysis
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          claude -p \
            "对整个代码库进行全面质量审计：
             1. 安全漏洞扫描（OWASP Top 10）
             2. 性能瓶颈识别
             3. 死代码检测
             4. 依赖安全审计
             5. 代码复杂度分析
             输出结构化JSON报告。" \
            --allowedTools "Read,Glob,Grep,Bash" \
            --max-turns 30 \
            --output-format json \
            > quality_report.json

      - name: Create Issue
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const report = JSON.parse(fs.readFileSync('quality_report.json', 'utf8'));
            github.rest.issues.create({
              owner: context.repo.owner,
              repo: context.repo.repo,
              title: `Weekly Quality Report - ${new Date().toISOString().split('T')[0]}`,
              body: report.result || 'See attached report',
              labels: ['quality', 'automated']
            });

flowchart TB
    subgraph Phase1["阶段1: 分析"]
        A[扫描项目结构] --> B[识别Class组件]
        B --> C[生成迁移清单]
    end

    subgraph Phase2["阶段2: 迁移"]
        C --> D1[Agent 1: 组件A]
        C --> D2[Agent 2: 组件B]
        C --> D3[Agent 3: 组件C]
        C --> D4[Agent N: ...]
    end

    subgraph Phase3["阶段3: 验证"]
        D1 --> E[TypeScript编译检查]
        D2 --> E
        D3 --> E
        D4 --> E
        E --> F[运行测试套件]
        F --> G{通过?}
        G -->|是| H[生成PR]
        G -->|否| I[自动修复]
        I --> F
    end

    subgraph Phase4["阶段4: 报告"]
        H --> J[迁移统计]
        J --> K[质量报告]
    end

    style A fill:#E8F5E9
    style D1 fill:#FFF3E0
    style D2 fill:#FFF3E0
    style D3 fill:#FFF3E0
    style D4 fill:#FFF3E0
    style E fill:#E3F2FD
    style H fill:#F3E5F5

#!/bin/bash
# migrate-react.sh — 全自动React迁移流水线
set -euo pipefail

PROJECT_DIR="$1"
LOG_DIR="migration_logs_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$LOG_DIR"

echo "========================================="
echo " React → TypeScript 迁移流水线"
echo " 项目: $PROJECT_DIR"
echo "========================================="

# ---- 阶段1: 分析 ----
echo "[Phase 1] 分析项目结构..."

INVENTORY=$(claude -p \
  "分析 $PROJECT_DIR 项目的React组件结构。
   列出所有Class组件，按复杂度排序。
   输出JSON格式: [{file, name, complexity, dependencies}]" \
  --allowedTools "Read,Glob,Grep" \
  --max-turns 10 \
  --output-format json)

echo "$INVENTORY" | jq '.' > "$LOG_DIR/inventory.json"
COMPONENT_COUNT=$(echo "$INVENTORY" | jq '.result | length')
echo "发现 $COMPONENT_COUNT 个Class组件"

# ---- 阶段2: 并行迁移 ----
echo "[Phase 2] 启动并行迁移..."

# 使用Dynamic Workflows或手动并行
claude -p \
  "将以下Class组件迁移为TypeScript函数组件+Hooks:
   $(cat "$LOG_DIR/inventory.json")

   迁移规则：
   - this.state → useState/useReducer
   - componentDidMount → useEffect
   - componentDidUpdate → useEffect  
   - componentWillUnmount → useEffect cleanup
   - this.props → 函数参数+TypeScript接口
   - PropTypes → TypeScript类型
   - 每个文件一个独立的子Agent处理
   迁移后确保TypeScript编译通过" \
  --allowedTools "Read,Write,Edit,Bash,Glob,Grep" \
  --max-turns 50 \
  --output-format stream-json \
  > "$LOG_DIR/migration_stream.jsonl" || true

# ---- 阶段3: 验证 ----
echo "[Phase 3] 运行验证..."

# TypeScript编译检查
cd "$PROJECT_DIR"
if npx tsc --noEmit 2>&1 | tee "$LOG_DIR/tsc_errors.txt"; then
  echo "✓ TypeScript编译通过"
else
  echo "✗ TypeScript编译错误，尝试自动修复..."
  claude -p \
    "修复以下TypeScript编译错误：$(cat "$LOG_DIR/tsc_errors.txt")" \
    --allowedTools "Read,Write,Edit" \
    --max-turns 15 \
    --output-format text
fi

# 运行测试
if npm test 2>&1 | tee "$LOG_DIR/test_results.txt"; then
  echo "✓ 所有测试通过"
else
  echo "✗ 测试失败，尝试自动修复..."
  claude -p \
    "修复以下测试失败：$(cat "$LOG_DIR/test_results.txt")" \
    --allowedTools "Read,Write,Edit,Bash" \
    --max-turns 20 \
    --output-format text
fi

cd -

# ---- 阶段4: 生成报告 ----
echo "[Phase 4] 生成迁移报告..."

claude -p \
  "基于以下日志生成迁移报告:
   - 清单: $(cat "$LOG_DIR/inventory.json")
   - 迁移流: $(head -100 "$LOG_DIR/migration_stream.jsonl")
   - TSC结果: $(cat "$LOG_DIR/tsc_errors.txt")
   - 测试结果: $(cat "$LOG_DIR/test_results.txt")

   报告包含：迁移统计、成功/失败列表、遗留问题、建议后续步骤" \
  --allowedTools "Read" \
  --max-turns 5 \
  --output-format text \
  > "$LOG_DIR/migration_report.md"

echo "========================================="
echo " 迁移完成! 报告: $LOG_DIR/migration_report.md"
echo "========================================="

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [{
          "type": "command",
          "command": "python3 .claude/hooks/safety_check.py"
        }]
      }
    ],
    "PostToolUse": [
      {
        "matcher": "Write|Edit",
        "hooks": [
          {
            "type": "command",
            "command": "npx prettier --write \"$CLAUDE_FILE_PATH\" 2>/dev/null || true"
          },
          {
            "type": "command",
            "command": "npx eslint --fix \"$CLAUDE_FILE_PATH\" 2>/dev/null || true"
          }
        ]
      }
    ],
    "Stop": [
      {
        "hooks": [{
          "type": "command",
          "command": "curl -s -X POST \"$SLACK_WEBHOOK\" -H 'Content-Type: application/json' -d '{\"text\": \"✅ 迁移任务完成! 查看报告: '$LOG_DIR/migration_report.md\"}' || true"
        }]
      }
    ]
  }
}

#!/bin/bash
# resume-pipeline.sh — 支持中断恢复的流水线

SESSION_FILE=".claude_session_id"

if [ -f "$SESSION_FILE" ]; then
  SESSION_ID=$(cat "$SESSION_FILE")
  echo "恢复会话: $SESSION_ID"
  claude -p --resume "$SESSION_ID" \
    "继续之前的任务" \
    --output-format json
else
  echo "启动新会话..."
  result=$(claude -p \
    "开始重构任务..." \
    --output-format json)

  # 保存session_id
  echo "$result" | jq -r '.session_id' > "$SESSION_FILE"
fi

#!/bin/bash
# retry-wrapper.sh — 带重试的headless调用

MAX_RETRIES=3
RETRY_DELAY=5

run_claude_with_retry() {
  local prompt="$1"
  local attempt=0

  while [ $attempt -lt $MAX_RETRIES ]; do
    attempt=$((attempt + 1))
    echo "尝试 $attempt/$MAX_RETRIES..."

    if result=$(claude -p "$prompt" --output-format json --max-turns 20 2>&1); then
      echo "$result"
      return 0
    fi

    if [ $attempt -lt $MAX_RETRIES ]; then
      delay=$((RETRY_DELAY * 2 ** (attempt - 1)))
      echo "失败，${delay}秒后重试..."
      sleep $delay
    fi
  done

  echo "所有重试均失败" >&2
  return 1
}

# 使用Sonnet处理简单任务（降低成本）
claude -p "格式化这个文件" \
  --model claude-sonnet-4-6 \
  --max-turns 3

# 使用Opus处理复杂架构决策
claude -p "重新设计认证系统的架构" \
  --model claude-opus-4-6 \
  --max-turns 30

# 限制工具减少API调用
claude -p "审查代码" \
  --allowedTools "Read,Grep" \
  --max-turns 5

# agent-team.sh — 启动多个Claude Code实例协作
# 需要 Claude Code ≥ 2.1.150

# 在独立worktree中启动多个Agent
claude -p \
  "作为前端Agent，负责React组件开发" \
  --output-format stream-json &
PID_FRONTEND=$!

claude -p \
  "作为后端Agent，负责API开发" \
  --output-format stream-json &
PID_BACKEND=$!

claude -p \
  "作为测试Agent，等待前两个完成后编写测试" \
  --output-format stream-json &
PID_TEST=$!

# 等待所有Agent完成
wait $PID_FRONTEND $PID_BACKEND $PID_TEST
echo "所有Agent已完成"

# 使用plan权限模式跳过确认
claude -p "任务" --permission-mode plan

# 或者使用bypassPermissions（仅CI环境）
claude -p "任务" --dangerously-skip-permissions

# 降低并行度
# 在CLAUDE.md中添加：
# dynamicWorkflows:
#   maxConcurrency: 5

# 或使用环境变量
export CLAUDE_MAX_CONCURRENCY=5

# 检查配置文件位置
ls -la .claude/settings.json
cat .claude/settings.json | jq '.hooks'

# 确保Hook脚本有执行权限
chmod +x .claude/hooks/*.py .claude/hooks/*.sh

# 测试Hook脚本
echo '{"tool_name":"Bash","tool_input":{"command":"ls"}}' | \
  python3 .claude/hooks/pre_tool_use.py
echo "Exit code: $?"

# 使用jq的-R选项逐行处理
claude -p "任务" --output-format stream-json | \
  while IFS= read -r line; do
    echo "$line" | jq -c 'select(.type == "result") // empty' 2>/dev/null
  done