cd ~/repos/my-service
claude -p "用三句话总结这个仓库的架构，并指出最可能的技术债。"

claude -p "..." > review.txt

claude -p \
  --output-format json \          # 结构化输出，便于 jq 解析
  --verbose \                     # json 模式下附带每一步的会话信息
  --max-turns 15 \                # 限制自主轮数，防止失控
  --allowedTools "Read,Grep,Glob,Bash"\  # 收紧权限白名单
  --model claude-sonnet-4-6 \     # 指定模型
  "为 src/ 下所有导出的函数补上 JSDoc 注释，只改注释不改逻辑"

claude -p --output-format stream-json --verbose "列出所有 TODO 注释的位置" | \
  jq -r 'select(.type=="tool_use" and .name=="Grep") | .input.pattern'

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

# Python
pip install claude-agent-sdk

import { query } from "@anthropic-ai/claude-agent-sdk";

async function main() {
  const messages = [{ role: "user", content: "列出 src/ 下所有 TypeScript 文件并统计每个文件的行数" }];

  // query() 是一个异步生成器，逐个 yield 事件
  for await (const message of query({
    prompt: "分析本仓库的测试覆盖率现状，给出提升到 80% 的优先动作清单",
    options: {
      cwd: "/home/me/repos/my-service",  // 工作目录，决定 Claude 能操作哪个仓库
      allowedTools: ["Read", "Grep", "Glob", "Bash"],  // 工具白名单
      maxTurns: 15,
      model: "claude-sonnet-4-6",
      permissionMode: "acceptEdits",
    },
  })) {
    // 每条 message 是一个流事件：assistant 文本、tool_use、tool_result、result...
    if (message.type === "assistant") {
      for (const block of message.message.content) {
        if (block.type === "text") process.stdout.write(block.text);
      }
    }
  }
}

main();

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        cwd="/home/me/repos/my-service",
        allowed_tools=["Read", "Grep", "Glob", "Bash"],
        max_turns=15,
        model="claude-sonnet-4-6",
        permission_mode="acceptEdits",
    )

    async for message in query(
        prompt="找出所有未处理的 Promise rejection 和 bare except，列出文件:行号",
        options=options,
    ):
        if getattr(message, "type", None) == "assistant":
            for block in message.message.content:
                if block.type == "text":
                    print(block.text, end="")

anyio.run(main)

# 第一轮：让它调研
result = await query(prompt="调研迁移到 ESM 的可行性", options=opts)
session_id = result....  # 从结果事件中取出

# （人工审核中间结论后）

# 第二轮：基于同一会话继续
await query(prompt="好，现在按你建议的方案动手改 package.json 和 tsconfig", 
            options={**opts, "resume": session_id})

// 注册一个自定义工具：查内部 SLO 面板
const options = {
  mcpServers: {
    "slo-panel": {
      type: "stdio",
      command: "node",
      args: ["./slo-mcp-server.js"],  // 你自己实现的 MCP server
    },
  },
  // Claude 会像调用 Read/Bash 一样调用 slo-panel 暴露的工具
};

# 概念性结构（实际以本机版本配置格式为准）
name: weekly-security-scan
prompt: |
  扫描本仓库的依赖清单与代码，找出已知 CVE、硬编码密钥、
  不安全的反序列化点。输出一份 Markdown 报告，按严重程度排序。
repositories:
  - /home/me/repos/billing-service
  - /home/me/repos/auth-gateway
  - /home/me/repos/notification-worker
options:
  allowedTools: [Read, Grep, Glob, Bash]
  maxTurns: 20
  model: claude-sonnet-4-6
  permissionMode: acceptEdits
  outputFormat: json

#!/usr/bin/env bash
set -euo pipefail

REPO_LIST="repos.txt"        # 每行一个仓库绝对路径
MAX_PARALLEL=4
OUT_DIR="reports/$(date +%Y%m%d)"
mkdir -p "$OUT_DIR"

# 关键：-P 控制并发数；每个仓库一个独立 Claude 进程
cat "$REPO_LIST" | xargs -I {} -P "$MAX_PARALLEL" bash -c '
  repo="$1"
  name=$(basename "$repo")
  out="'"$OUT_DIR"'/${name}.json"
  cd "$repo"
  claude -p \
    --output-format json \
    --max-turns 15 \
    --allowedTools "Read,Grep,Glob,Bash" \
    --permission-mode acceptEdits \
    "审查本仓库的安全性，列出高/中/低危问题各不超过 5 条，给出文件:行号" \
    > "$out" 2> "${out%.json}.err"
  echo "✓ $name (exit=$?)"
' _ {}

#!/usr/bin/env bash
set -euo pipefail

OUT_DIR="reports/$(date +%Y%m%d)"
mkdir -p "$OUT_DIR"
export OUT_DIR

# parallel 支持 --jobs、--load（按系统负载限流）、--retries（失败重试）
parallel --jobs 4 --retries 1 --joblog "$OUT_DIR/joblog.txt" '
  repo={}
  name=$(basename "$repo")
  cd "$repo"
  claude -p --output-format json --max-turns 15 \
    --allowedTools "Read,Grep,Glob" \
    "生成本仓库的架构概述（500 字内）" \
    > "'"$OUT_DIR"'/${name}.json"
' :::: repos.txt

# joblog 记录每个任务的退出码、耗时，便于事后审计

import asyncio, json, os
from pathlib import Path
from claude_agent_sdk import query, ClaudeAgentOptions

REPOS = Path("repos.txt").read_text().splitlines()
MAX_CONCURRENCY = 4
PROMPT = "审查安全性，按严重程度列出问题，输出 JSON：{issues:[{severity,file,line,desc}]}"

async def scan_one(repo: str, sem: asyncio.Semaphore):
    async with sem:
        opts = ClaudeAgentOptions(
            cwd=repo,
            allowed_tools=["Read", "Grep", "Glob", "Bash"],
            max_turns=15,
            model="claude-sonnet-4-6",
            permission_mode="acceptEdits",
        )
        final_text = ""
        try:
            async for msg in query(prompt=PROMPT, options=opts):
                if getattr(msg, "type", None) == "result":
                    final_text = msg.result
            return {"repo": Path(repo).name, "ok": True, "raw": final_text}
        except Exception as e:
            return {"repo": Path(repo).name, "ok": False, "error": str(e)}

async def main():
    sem = asyncio.Semaphore(MAX_CONCURRENCY)  # 信号量控并发
    results = await asyncio.gather(*[scan_one(r, sem) for r in REPOS])

    # 按严重程度排序输出总报告
    sev_order = {"high": 0, "medium": 1, "low": 2}
    all_issues.sort(key=lambda x: sev_order.get(x.get("severity", "low"), 3))
    Path("reports/summary.json").write_text(json.dumps(all_issues, indent=2, ensure_ascii=False))
    print(f"完成：{len(REPOS)} 个仓库，{len(all_issues)} 条问题")

asyncio.run(main())

total=$(cat reports/*.json | jq -s '[.[].cost_usd // 0] | add')
echo "本次总花费: $total USD"

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=2, min=4, max=60),
       retry=lambda e: "429" in str(e) or "529" in str(e))
async def scan_one_with_retry(repo, sem):
    ...

flowchart TB
    subgraph 输入
        R[repos.txt<br/>仓库清单]
        P[Prompt / Routine<br/>统一指令]
        C[配置: maxTurns,<br/>allowedTools, model]
    end

    subgraph 编排层
        SEM{并发控制器<br/>Semaphore -P 4}
        DISP[任务分发器<br/>xargs / parallel / SDK]
    end

    subgraph 执行层 每仓库一个独立进程
        W1[Claude 进程 1<br/>repo-A 独立会话]
        W2[Claude 进程 2<br/>repo-B 独立会话]
        W3[Claude 进程 3<br/>repo-C 独立会话]
        WN[Claude 进程 N<br/>repo-N 独立会话]
    end

    subgraph 结果层
        J1[repo-A.json<br/>含 cost_usd]
        J2[repo-B.json]
        J3[repo-C.json]
        AGG[汇总器<br/>jq / Python]
        RPT[总报告<br/>summary.json]
    end

    R --> DISP
    P --> DISP
    C --> DISP
    DISP --> SEM
    SEM --> W1
    SEM --> W2
    SEM --> W3
    SEM --> WN
    W1 --> J1
    W2 --> J2
    W3 --> J3
    WN --> AGG
    J1 --> AGG
    J2 --> AGG
    J3 --> AGG
    AGG --> RPT

    style SEM fill:#ffe4b5
    style AGG fill:#e0ffe0
    style RPT fill:#e0ffe0

vuln-patrol/
├── repos.txt                  # 待巡检仓库清单
├── routine.yaml               # 巡检指令与配置（Routine 思路）
├── run.sh                     # xargs 并行入口（方案 A）
├── orchestrator.py            # SDK 编排入口（方案 C，含重试与汇总）
└── reports/
    └── 20260615/
        ├── billing-service.json
        ├── auth-gateway.json
        └── summary.md

/home/me/repos/billing-service
/home/me/repos/auth-gateway
/home/me/repos/notification-worker
/home/me/repos/inventory-api
/home/me/repos/search-service

你是依赖安全审计员。请完成以下步骤：
1. 读取本仓库的依赖清单文件（package.json / requirements.txt / go.mod / pom.xml 等，按实际存在为准）
2. 用 Grep 搜索代码中是否使用了已知有漏洞的模式（eval、pickle.loads、不安全的反序列化、硬编码密钥）
3. 对每个发现，给出：severity(high/medium/low)、file、line、description、suggested_fix
4. 严格输出 JSON：{"repo":"<仓库名>","issues":[...],"scanned_files":N}
不要输出 JSON 以外的内容。

import asyncio, json, sys
from pathlib import Path
from datetime import date
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from claude_agent_sdk import query, ClaudeAgentOptions

REPOS = [l.strip() for l in Path("repos.txt").read_text().splitlines() if l.strip()]
PROMPT = Path("audit_prompt.txt").read_text()
MAX_CONCURRENCY = 4
BUDGET_USD = 5.0  # 整批预算上限

class RateLimited(Exception): pass

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=2, min=4, max=60),
    retry=retry_if_exception_type(RateLimited),
)
async def audit_one(repo: str, sem: asyncio.Semaphore):
    async with sem:
        opts = ClaudeAgentOptions(
            cwd=repo,
            allowed_tools=["Read", "Grep", "Glob", "Bash"],
            max_turns=20,
            model="claude-sonnet-4-6",
            permission_mode="acceptEdits",
        )
        result_text, cost = "", 0.0
        async for msg in query(prompt=PROMPT, options=opts):
            t = getattr(msg, "type", None)
            if t == "result":
                result_text = getattr(msg, "result", "")
                cost = getattr(msg, "cost_usd", 0.0) or 0.0
            if t == "system" and "429" in str(getattr(msg, "subtype", "")):
                raise RateLimited(repo)
        return {"repo": Path(repo).name, "cost": cost, "raw": result_text}

async def main():
    sem = asyncio.Semaphore(MAX_CONCURRENCY)
    spent = 0.0
    all_issues = []

    tasks = [audit_one(r, sem) for r in REPOS]
    for coro in asyncio.as_completed(tasks):  # 先完成先处理，便于提前熔断
        res = await coro
        spent += res["cost"]
        if spent > BUDGET_USD:
            print(f"⚠️ 预算熔断：已花费 ${spent:.2f} > ${BUDGET_USD}", file=sys.stderr)
            # 取消尚未完成的任务
            for t in tasks:
                t.cancel()
            break
        try:
            data = json.loads(res["raw"])
            for iss in data.get("issues", []):
                iss["repo"] = res["repo"]
                all_issues.append(iss)
        except json.JSONDecodeError:
            print(f"⚠️ {res['repo']} 输出非 JSON，跳过", file=sys.stderr)

    # 生成 Markdown 总报告
    sev = {"high": [], "medium": [], "low": []}
    for i in all_issues:
        sev.setdefault(i.get("severity", "low"), []).append(i)

    md = [f"# 依赖漏洞巡检报告 {date.today()}\n",
          f"扫描仓库 {len(REPOS)} 个，发现问题 {len(all_issues)} 个，花费 ${spent:.2f}\n"]
    for level in ["high", "medium", "low"]:
        md.append(f"\n## {level.upper()} ({len(sev[level])})\n")
        for i in sev[level]:
            md.append(f"- **{i['repo']}** `{i['file']}:{i.get('line','?')}` — {i['description']}")

    out = Path(f"reports/{date.today()}/summary.md")
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text("\n".join(md))
    print(f"报告已生成：{out}")

asyncio.run(main())

# 每周一 06:17 跑依赖巡检
17 6 * * 1 cd /home/me/vuln-patrol && /usr/bin/python3 orchestrator.py >> logs/cron.log 2>&1