Claude Code Routines定时编排：三级调度全实战

graph TB
    subgraph "三级调度体系"
        L["/loop<br/>本地循环<br/>Session内"]
        D["Desktop Scheduled<br/>桌面定时<br/>Cron表达式"]
        C["Cloud Routines<br/>云端编排<br/>GitHub/API/定时"]
    end

    L -->|适合| L1["实时监控<br/>开发辅助<br/>短周期任务"]
    D -->|适合| D1["本地定时<br/>日常维护<br/>个人工作流"]
    C -->|适合| C1["团队协作<br/>事件驱动<br/>7×24运行"]

    style L fill:#4CAF50,color:#fff
    style D fill:#FF9800,color:#fff
    style C fill:#2196F3,color:#fff

# 基本语法
/loop <间隔> <提示词>

# 每60秒检查一次CI状态
/loop 60s 检查最近的GitHub Actions运行状态，如果有失败告诉我

# 每5分钟检查PR审查状态
/loop 5m 查看当前分支的PR状态，列出新的review comments

# 每30秒监控日志异常
/loop 30s 检查 logs/app.log 最后50行，如果有ERROR级别日志就报告

# 步骤1：启动Claude Code
claude

# 步骤2：在会话中设置loop
/loop 45s 运行 gh run list --limit 3 --json status,conclusion,name,headBranch 
并告诉我最新的workflow run状态。如果全部通过或全部完成，停止loop并总结结果。

[loop #3] 检查CI状态...
  ✅ test-suite (main) — completed
  🔄 lint-check (main) — in_progress  
  ⏳ build (main) — queued

  等待中，45秒后再次检查...

[loop #7] 所有CI运行已完成：
  ✅ test-suite — passed (3m 22s)
  ✅ lint-check — passed (1m 05s)  
  ✅ build — passed (4m 18s)

  Loop已自动停止。总等待时间：约5分钟。

# 基本语法
/schedule <cron表达式> <提示词>

# 每天早上9点生成代码报告
/schedule "0 9 * * 1-5" 分析昨天所有的git commits，生成一份简要的日报

# 每小时检查依赖安全
/schedule "0 * * * *" 运行 npm audit，如果有critical级别漏洞就报告详情

# 每周一凌晨2点清理分支
/schedule "0 2 * * 1" 列出所有本地分支，标记超过30天未活动的分支建议删除

┌─── 分钟 (0-59)
│ ┌─── 小时 (0-23)
│ │ ┌─── 日 (1-31)
│ │ │ ┌─── 月 (1-12)
│ │ │ │ ┌─── 星期 (0-6, 0=周日)
│ │ │ │ │
* * * * *

# 配置定时任务
/schedule "0 9 * * 1-5" 
执行以下检查并生成报告：
1. 运行 git log --since="yesterday" --oneline 统计昨日提交数
2. 运行 npm outdated 检查过期依赖数量
3. 运行 npm audit --production 检查生产依赖安全
4. 检查 test-coverage/coverage-summary.json 的覆盖率
5. 用中文汇总为一份简洁的健康报告

📊 项目健康日报 — 2026-05-30

📝 提交活动：昨天 12 次提交（5位贡献者）
📦 依赖状态：3个过期包（无critical），0个安全漏洞
🧪 测试覆盖率：87.3%（+0.5% vs 上周）
⚠️ 关注事项：[email protected] 建议升级到 4.17.21

总体评级：🟢 良好

# 列出所有已配置的定时任务
/schedule list

# 删除特定任务
/schedule remove <task-id>

# 暂停/恢复任务
/schedule pause <task-id>
/schedule resume <task-id>

# 方法1：使用 claude routine create 命令
claude routine create \
  --name "pr-auto-review" \
  --repo "owner/repo" \
  --trigger "github:pull_request:opened" \
  --prompt "审查这个PR的代码变更，重点关注：1. 安全漏洞 2. 性能问题 3. 代码风格。用中文给出审查意见。"

# 方法2：通过配置文件创建
claude routine create --config routine-config.json

{
  "name": "daily-health-check",
  "description": "每日项目健康检查",
  "repo": "myorg/myproject",
  "connector": {
    "type": "github",
    "repo": "myorg/myproject"
  },
  "trigger": {
    "type": "schedule",
    "cron": "0 8 * * 1-5"
  },
  "prompt": "执行以下项目健康检查：\n1. 检查最近24小时的CI/CD运行结果\n2. 分析开着的Issue数量和优先级分布\n3. 检查依赖更新建议（npm outdated）\n4. 汇总为中文报告并创建一个Issue",
  "model": "claude-sonnet-4-20250514",
  "maxTurns": 30,
  "allowedTools": ["Bash", "Read", "Glob", "Grep", "WebSearch"]
}

sequenceDiagram
    participant Dev as 开发者
    participant GH as GitHub
    participant AH as Anthropic Cloud
    participant CC as Claude Code Routine
    participant Repo as 代码仓库

    Dev->>GH: 创建Pull Request
    GH->>AH: Webhook: pull_request.opened
    AH->>CC: 启动Routine
    CC->>Repo: 读取PR diff
    CC->>CC: 分析代码变更
    CC->>GH: 发布Review评论
    GH->>Dev: 收到自动审查反馈

    Note over AH,CC: 全程云端运行<br/>无需本地机器

{
  "name": "smart-pr-reviewer",
  "repo": "myorg/backend",
  "trigger": {
    "type": "github",
    "events": ["pull_request:opened", "pull_request:synchronize"],
    "filter": {
      "base_branch": ["main", "develop"],
      "files_changed_min": 1
    }
  },
  "prompt": "你是代码审查专家。请分析这个PR：\n\n1. **安全性**：检查是否有SQL注入、XSS、硬编码密钥等安全风险\n2. **性能**：识别N+1查询、大循环内的内存分配等问题\n3. **架构**：评估变更是否符合现有设计模式\n4. **测试**：检查是否有足够的测试覆盖\n\n用中文输出审查报告，对每个发现标注严重级别（🔴 严重 / 🟡 警告 / 🔵 建议）。\n最后给出整体评价：APPROVE / REQUEST_CHANGES / COMMENT",
  "allowedTools": ["Read", "Bash", "Grep", "Glob"],
  "maxTurns": 50
}

# 创建API触发Routine
claude routine create \
  --name "incident-responder" \
  --trigger "api" \
  --prompt "根据传入的事件数据分析根因，搜索相关日志和代码，生成故障报告"

import requests
import json

def trigger_incident_analysis(incident_data):
    """触发Claude Code Routine进行故障分析"""
    response = requests.post(
        "https://api.anthropic.com/v1/routines/{routine_id}/trigger",
        headers={
            "Authorization": f"Bearer {os.environ['ANTHROPIC_API_KEY']}",
            "Content-Type": "application/json"
        },
        json={
            "input": {
                "incident_id": incident_data["id"],
                "error_message": incident_data["error"],
                "service": incident_data["service"],
                "timestamp": incident_data["timestamp"]
            }
        }
    )
    return response.json()["execution_id"]

# 监控系统调用
incident = {
    "id": "INC-2026-0530-001",
    "error": "Database connection pool exhausted",
    "service": "payment-api",
    "timestamp": "2026-05-30T08:15:30Z"
}
exec_id = trigger_incident_analysis(incident)
print(f"分析任务已触发: {exec_id}")

# 列出所有Routines
claude routine list

# 查看特定Routine详情
claude routine get <routine-id>

# 查看执行历史
claude routine history <routine-id> --limit 10

# 查看特定执行的详细输出
claude routine logs <execution-id>

# 手动触发一次执行
claude routine trigger <routine-id>

# 更新Routine配置
claude routine update <routine-id> --config updated-config.json

# 删除Routine
claude routine delete <routine-id>

flowchart LR
    subgraph "输入"
        P["提示词"]
        S["stdin"]
    end

    subgraph "Claude Code -p"
        CC["claude -p --output-format stream-json"]
    end

    subgraph "输出事件流"
        E1["system event"]
        E2["assistant event"]
        E3["tool_use event"]
        E4["tool_result event"]
        E5["result event"]
    end

    P --> CC
    S --> CC
    CC --> E1
    CC --> E2
    CC --> E3
    CC --> E4
    CC --> E5

# 输出为stream-json格式
claude -p --output-format stream-json "分析当前项目的目录结构"

# 配合jq解析特定事件类型
claude -p --output-format stream-json "列出所有TODO注释" | \
  jq -c 'select(.type == "result") | .result'

# 同时输出到文件和终端
claude -p --output-format stream-json "生成API文档" | \
  tee output.jsonl | \
  jq -c 'select(.type == "assistant") | .message.content[0].text'

#!/bin/bash
# batch-review.sh — 批量PR审查脚本

REPO="myorg/backend"
OPEN_PRS=$(gh pr list --repo "$REPO" --state open --json number --jq '.[].number')

for pr_number in $OPEN_PRS; do
    echo "=== 审查 PR #$pr_number ==="

    # 获取PR diff
    DIFF=$(gh pr diff "$pr_number" --repo "$REPO")

    # 使用Claude Code审查，输出JSON
    RESULT=$(echo "$DIFF" | claude -p \
        --output-format json \
        --max-turns 5 \
        "审查这个PR的代码变更。输出JSON格式：{\"approved\": true/false, \"issues\": [...], \"summary\": \"...\"}" 2>/dev/null)

    # 解析结果
    APPROVED=$(echo "$RESULT" | jq -r '.approved // false')
    SUMMARY=$(echo "$RESULT" | jq -r '.summary // "无摘要"')

    echo "  结果: $([ "$APPROVED" = "true" ] && echo '✅ 通过' || echo '❌ 需修改')"
    echo "  摘要: $SUMMARY"

    # 如果不通过，自动添加评论
    if [ "$APPROVED" != "true" ]; then
        ISSUES=$(echo "$RESULT" | jq -r '.issues[] | "- " + .description')
        gh pr comment "$pr_number" --repo "$REPO" \
            --body "🤖 **自动审查反馈**\n\n$SUMMARY\n\n**问题列表：**\n$ISSUES"
    fi

    echo "---"
done

echo "批量审查完成。共审查 $(echo "$OPEN_PRS" | wc -l) 个PR。"

#!/usr/bin/env python3
"""Claude Code stream-json 实时处理器"""

import subprocess
import json
import sys

def run_claude_stream(prompt: str, cwd: str = None):
    """运行Claude Code并实时处理stream-json事件"""
    cmd = [
        "claude", "-p",
        "--output-format", "stream-json",
        "--max-turns", "10",
        prompt
    ]

    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        cwd=cwd,
        text=True
    )

    full_response = []
    tools_used = []
    total_cost = 0

    for line in proc.stdout:
        line = line.strip()
        if not line:
            continue

        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            continue

        event_type = event.get("type", "")

        if event_type == "assistant":
            # 提取文本内容
            content = event.get("message", {}).get("content", [])
            for block in content:
                if block.get("type") == "text":
                    text = block["text"]
                    full_response.append(text)
                    print(f"[文本] {text[:100]}...")

        elif event_type == "tool_use":
            tool_name = event.get("tool", {}).get("name", "unknown")
            tools_used.append(tool_name)
            print(f"[工具] 调用: {tool_name}")

        elif event_type == "tool_result":
            is_error = event.get("is_error", False)
            status = "❌ 失败" if is_error else "✅ 成功"
            print(f"[结果] {status}")

        elif event_type == "result":
            total_cost = event.get("cost_usd", 0)
            duration = event.get("duration_ms", 0)
            print(f"\n[完成] 耗时: {duration/1000:.1f}s | 费用: ${total_cost:.4f}")
            print(f"[工具] 共使用 {len(tools_used)} 次工具调用")

    proc.wait()
    return {
        "response": "\n".join(full_response),
        "tools_used": tools_used,
        "cost_usd": total_cost,
        "exit_code": proc.returncode
    }


# 使用示例
if __name__ == "__main__":
    result = run_claude_stream(
        "分析 src/ 目录下的所有TypeScript文件，找出潜在的性能问题",
        cwd="/home/user/myproject"
    )
    print(f"\n最终响应:\n{result['response'][:500]}")

graph TB
    subgraph "第一级：/loop 开发中"
        L1["监控CI构建<br/>间隔: 30s"]
        L2["监控热重载<br/>间隔: 10s"]
    end

    subgraph "第二级：Desktop定时"
        D1["早9点: 代码健康报告"]
        D2["午12点: 依赖安全检查"]
        D3["晚6点: 今日工作总结"]
    end

    subgraph "第三级：Cloud Routines"
        C1["PR审查Bot<br/>触发: PR opened"]
        C2["Issue分类Bot<br/>触发: Issue opened"]
        C3["周报生成<br/>触发: 每周五17:00"]
    end

    L1 -->|开发时使用| DEV["开发者终端"]
    D1 -->|日常执行| LAPTOP["本地电脑"]
    D2 --> LAPTOP
    D3 --> LAPTOP
    C1 -->|7×24运行| CLOUD["Anthropic Cloud"]
    C2 --> CLOUD
    C3 --> CLOUD

# 开发功能时，实时监控CI
/loop 30s 
运行以下命令检查CI状态：
1. gh run list --limit 1 --json status,conclusion,name
2. 如果status不是in_progress，报告最终结果
3. 如果有失败，分析失败日志 gh run view --log-failed
如果所有run都completed，停止loop

# 每日健康报告
/schedule "0 9 * * 1-5"
执行项目健康检查：
1. git log --since="1 day ago" --oneline | wc -l 获取提交数
2. npm outdated 2>/dev/null | tail -n +2 | wc -l 获取过期包数
3. npm audit --production 2>/dev/null | grep -c "critical" 获取严重漏洞数
4. find src -name "*.test.*" | wc -l 获取测试文件数
5. 汇总为中文健康报告

# 每周五总结
/schedule "0 17 * * 5"
生成本周工作总结：
1. git log --since="7 days ago" --format="%s" | sort | uniq -c | sort -rn
2. 按功能模块分类提交
3. 列出本周关闭的Issue和合并的PR
4. 生成中文周报

{
  "name": "pr-reviewer",
  "repo": "myorg/project",
  "trigger": {
    "type": "github",
    "events": ["pull_request:opened", "pull_request:synchronize"],
    "filter": {
      "base_branch": ["main"]
    }
  },
  "prompt": "你是一位资深代码审查员。请审查这个PR并：\n1. 识别安全风险（注入、XSS、密钥泄露）\n2. 检查错误处理完整性\n3. 评估代码可读性和维护性\n4. 验证是否有相应的测试\n\n输出格式：\n- 对每个发现标注严重级别\n- 给出具体的改进建议和代码示例\n- 最终给出APPROVE或REQUEST_CHANGES",
  "allowedTools": ["Read", "Bash", "Grep", "Glob"],
  "maxTurns": 40
}

{
  "name": "weekly-engineering-report",
  "repo": "myorg/project",
  "trigger": {
    "type": "schedule",
    "cron": "0 17 * * 5"
  },
  "prompt": "生成本周工程报告：\n1. 使用 gh pr list --state merged --search 'merged:>=$(date -d \"7 days ago\" +%Y-%m-%d)' 获取已合并PR\n2. 使用 gh issue list --state closed 获取关闭的Issue\n3. 分析代码变更统计（新增/删除/修改行数）\n4. 识别本周活跃贡献者\n5. 生成结构化中文周报并创建为GitHub Issue",
  "allowedTools": ["Bash", "Read", "Grep", "WebSearch"],
  "maxTurns": 30
}

# 部署PR审查Routine
claude routine create --config routine-pr-reviewer.json
# ✅ Routine created: pr-reviewer (routine_abc123)

# 部署周报Routine
claude routine create --config routine-weekly-report.json
# ✅ Routine created: weekly-engineering-report (routine_def456)

# 验证部署
claude routine list
# NAME                    TRIGGER              STATUS    LAST RUN
# pr-reviewer             github:pr_opened     active    —
# weekly-engineering-report  schedule (Fri 17:00)  active    —

# 查看执行历史
claude routine history routine_abc123 --limit 5

# 输出示例：
# EXECUTION_ID          STATUS    TRIGGER          STARTED           DURATION
# exec_001              ✅ done   PR #142 opened   2026-05-30 10:15  2m 32s
# exec_002              ✅ done   PR #143 opened   2026-05-30 11:42  1m 58s
# exec_003              ❌ error  PR #144 opened   2026-05-30 14:05  0m 12s

# 查看失败执行的日志
claude routine logs exec_003
# Error: max turns exceeded. The routine used 50 turns without completing.
# Suggestion: Increase maxTurns or simplify the prompt.

flowchart LR
    subgraph "实时层：Hooks"
        H1["PreToolUse<br/>拦截危险操作"]
        H2["PostToolUse<br/>自动格式化"]
        H3["PreCommit<br/>提交前检查"]
    end

    subgraph "调度层：Routines"
        R1["定时健康检查"]
        R2["PR自动审查"]
        R3["周报生成"]
    end

    H1 -->|保护| CC["Claude Code<br/>执行环境"]
    H2 --> CC
    H3 --> CC
    R1 -->|触发| CC
    R2 -->|触发| CC

    style H1 fill:#e74c3c,color:#fff
    style H2 fill:#f39c12,color:#fff
    style H3 fill:#27ae60,color:#fff

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "command": "if echo '$TOOL_INPUT' | grep -qE 'rm -rf|drop table|DELETE FROM'; then echo 'BLOCKED: 危险命令被拦截'; exit 1; fi"
      }
    ],
    "PostToolUse": [
      {
        "matcher": "Edit|Write",
        "command": "npx prettier --write '$FILE_PATH' 2>/dev/null || true"
      }
    ]
  }
}

# 1. 使用 --max-turns 限制工具调用轮次
claude -p --max-turns 5 "快速检查代码风格"

# 2. 使用 --allowedTools 限制可用工具
claude -p --allowedTools "Read,Grep,Glob" "只做代码分析，不修改文件"

# 3. 选择合适的模型
claude -p --model haiku "简单的代码格式检查"  # 便宜快速
claude -p --model sonnet "复杂的架构分析"       # 平衡性价比

# 4. Routine中设置cost上限
# 在routine配置中添加：
{
  "maxCostUsd": 0.50,
  "maxTurns": 15
}

# 查看Routine执行成本
claude routine history <routine-id> --format json | \
  jq -r '.[] | "\(.started_at) | $\(.cost_usd) | \(.duration_ms/1000)s"'