Claude Code 第六课：给无人值守的 Agent 装门禁和仪表盘

生命不息，折腾不止。Agent 敢让它碰服务器，就得先给它装上刹车和计价器。

第五课我们把 Claude Code 包成了 FastAPI 服务，谁都能发个请求让它干活。上线第二天我就睡不着了：这玩意儿手里握着服务器权限、能跑任意 shell、能改任意文件，而我连它昨晚跑了多少条命令、烧了多少钱都不知道。

这一课不写新功能，专门补两个「生产环境才想起来、出事就来不及」的东西：门禁和仪表盘。

一、先把风险摆上桌

一个无人值守的 Agent，出事就三种：

1. 手滑型：rm -rf 路径拼错、git push --force 把同事的提交冲掉；
2. 越界型：改 .env、改 ~/.ssh/、往 /etc/ 里写东西；
3. 烧钱型：一个死循环把对话刷到 200 轮，账单比工资高。

这三种，靠「提示词里求它别这样」是没用的。大模型是概率机器，你说一百遍「小心点」，它总有一次会忘。能兜住概率的，只有确定性代码——这就是 Claude Code 的 Hooks 存在的意义：在工具真正执行之前，跑一段你写的脚本，由脚本说了算。

官方文档对它的定位很直白：Hooks 提供的是「确定性控制」，让某些操作总是发生，而不是指望模型自己想起来。这句我认同，把它当成 Agent 的中间件防火墙来用就对了。

二、门禁第一层：PreToolUse 拦危险命令

2.1 事件挂在哪儿

PreToolUse 在工具调用执行前触发，是唯一能在动手前掐断的位置。配置写在设置文件里，三个位置对应三种生效范围：

文件	范围
~/.claude/settings.json	当前用户所有项目（我们服务用这个）
.claude/settings.json	当前项目，会进 git，团队共用
.claude/settings.local.json	当前项目，只给自己

注册长这样，matcher 是工具名，多个用 | 隔开：

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash|Write|Edit|MultiEdit",
        "hooks": [
          {
            "type": "command",
            "command": "python3 /opt/agent/hooks/guard.py",
            "timeout": 10
          }
        ]
      }
    ]
  }
}

两个细节：matcher 区分大小写，写 bash 是不生效的；timeout 单位是秒，门禁脚本必须够快，卡十秒以上会明显拖慢每一轮对话。

2.2 写一个 guard.py

Hook 脚本的输入是 stdin 上的一段 JSON，里面有 session_id、tool_name、tool_input。拿 Bash 举例，tool_input.command 就是它准备执行的那条命令。

下面这版只用 Python 标准库，不依赖 jq，可直接扔到服务器上跑：

#!/usr/bin/env python3
"""guard.py —— 危险命令拦截 + 敏感文件保护 + 全量审计"""
import json, os, re, sys, time, pathlib

HOME = os.path.expanduser("~")
AUDIT = pathlib.Path(HOME) / "claude-audit.jsonl"

PROTECTED_DIRS = ["/etc/", "/boot/", "/dev/", "/proc/",
                  f"{HOME}/.ssh/", f"{HOME}/.claude/"]
PROTECTED_NAMES = {".env", ".env.local", ".cf-env",
                   "id_rsa", "id_ed25519", "authorized_keys"}


def check_bash(cmd: str):
    compact = re.sub(r"\s+", " ", cmd or "").strip()

    # rm 递归删除：单独取目标判断，比一条万能正则靠谱得多
    if re.search(r"\brm\b[^|;&]*\s-[a-zA-Z]*[rf]", compact):
        target = compact.split()[-1].strip("'\"")
        if target in ("/", "/*", "~", "~/", "$HOME", "*", "./*"):
            return f"拒绝执行：rm 递归删除 {target}"

    rules = [
        (r"\bgit\s+push\b[^|;&]*(--force\b|\s-f\b)", "拒绝执行：强制推送，远端历史会被覆盖"),
        (r"\b(mkfs(\.\w+)?|fdisk|parted)\b", "拒绝执行：磁盘分区/格式化"),
        (r"\bdd\b[^|;&]*of=/dev/", "拒绝执行：dd 直接写块设备"),
        (r"\b(curl|wget)\b[^|;&]*\|\s*(sudo\s+)?(ba|z)?sh\b", "拒绝执行：网络内容直接管道给 shell，先下载再人工审"),
        (r"\b(DROP|TRUNCATE)\s+(DATABASE|TABLE)\b", "拒绝执行：删库/清表语句"),
        (r"\b(shutdown|reboot|halt|poweroff)\b", "拒绝执行：关机或重启"),
        (r"\bchmod\s+-R\s+0?777\s+/", "拒绝执行：对根目录递归 777"),
    ]
    for pattern, reason in rules:
        if re.search(pattern, compact, re.IGNORECASE):
            return reason
    return None


def check_file(path: str):
    if not path:
        return None
    p = os.path.abspath(os.path.expanduser(path))
    for d in PROTECTED_DIRS:
        if p.startswith(os.path.abspath(d)):
            return f"拒绝修改：{p} 在受保护目录内"
    if os.path.basename(p) in PROTECTED_NAMES:
        return f"拒绝修改：{p} 是敏感文件（凭据/密钥）"
    return None


def main():
    try:
        payload = json.load(sys.stdin)
    except Exception:
        return  # 输入都读不懂：退出 0 不带 JSON，不做决策，走正常权限流程

    tool = payload.get("tool_name", "")
    ti = payload.get("tool_input") or {}
    reason = None

    if tool == "Bash":
        reason = check_bash(ti.get("command", ""))
    elif tool in ("Write", "Edit", "MultiEdit", "NotebookEdit"):
        reason = check_file(ti.get("file_path") or ti.get("notebook_path"))

    record = {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "session": payload.get("session_id", "")[:8],
        "tool": tool,
        "decision": "deny" if reason else "allow",
        "reason": reason,
        "input": ti,
    }
    try:
        with AUDIT.open("a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False)[:4000] + "\n")
    except Exception:
        pass  # 审计写不进去，不该拖垮业务

    if reason:
        print(json.dumps({"hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "deny",
            "permissionDecisionReason": reason,
        }}, ensure_ascii=False))


if __name__ == "__main__":
    main()

然后 chmod +x /opt/agent/hooks/guard.py。文件其实用 python3 调用，不加执行位也能跑，但加上省事——别的脚本引用时不用再想路径。

2.3 为什么用 JSON deny，而不是 exit 2

官方给的两条路：脚本 exit 2 把 stderr 当理由反馈给模型；或者 exit 0 往 stdout 打一个 JSON 决策。我选后者，原因有三：

• 能区分三档：permissionDecision 可以是 deny（直接拒）、ask（弹给人确认）、allow（跳过交互提示）。exit 2 只有「拒」一种语气。
• 理由能说清楚：permissionDecisionReason 会原样喂回给 Claude，它看得懂就不会一根筋重试，会换个思路。
• 最硬的一点：文档写明 deny 在所有权限模式下都生效，连 bypassPermissions 和 --dangerously-skip-permissions 都绕不过去。

最后这条是整套方案的命门。我们的服务为了让它少卡在权限提示上，本来就用宽权限在跑；要是门禁能被宽权限模式绕过，那就是个摆设。

2.4 立刻验一下

不用开服务，直接拿 JSON 喂脚本：

echo '{"tool_name":"Bash","tool_input":{"command":"rm -rf /"}}' | python3 /opt/agent/hooks/guard.py

应该看到 "permissionDecision": "deny"。再试 ls -la，什么都不输出——不输出就是放行，这条规矩要记住，后面排查全靠它。

接着在 Claude Code 里输入 /hooks，能看到刚注册的 hook 挂在 PreToolUse 下面，说明配置被读进去了。改完设置文件一般自动热加载，不生效就重开会话。

三、门禁第二层：审计日志和配置变更

拦截只能挡住「你想到的坏事」，剩下靠记录。

~/claude-audit.jsonl 现在是每行一条 JSON，随手用 jq 就能查：

# 今天被拦了几次、都拦了什么
jq -r 'select(.decision=="deny") | "\(.ts) \(.tool) \(.reason)"' ~/claude-audit.jsonl | tail -20

# 某个会话到底跑了哪些命令
jq -r 'select(.session=="a1b2c3d4" and .tool=="Bash") | .input.command' ~/claude-audit.jsonl

还有一个容易漏的口子：模型没动你的文件，但它改了配置。Claude Code 有个 ConfigChange 事件，外部进程或编辑器改设置文件、skills 文件时触发，一行就能记下来：

{
  "hooks": {
    "ConfigChange": [
      {
        "matcher": "",
        "hooks": [
          {
            "type": "command",
            "command": "jq -c '{ts: now | todate, source: .source, file: .file_path}' >> ~/claude-config-audit.log"
          }
        ]
      }
    ]
  }
}

matcher 留空表示全记，也可以只盯 user_settings、project_settings、skills 这几类。有人偷偷把门禁那行删了，这里会留痕——审计日志的第一价值不是抓模型，是抓人。

四、仪表盘：把每次调用记成账

门禁解决「别乱来」，仪表盘解决「花了多少」。

4.1 成本数据藏在返回里

第五课我们用 claude -p 跑无头模式，那时只取了 result 文本。其实加 --output-format json 后，返回的是一整个结果对象，成本就在里面：

{
  "type": "result",
  "subtype": "success",
  "result": "……最终回答……",
  "session_id": "abc-123-def",
  "num_turns": 4,
  "duration_ms": 2847,
  "total_cost_usd": 0.0034,
  "is_error": false
}

四个字段直接够做报表：total_cost_usd 花了多少、num_turns 转了几轮、duration_ms 花了多久、session_id 串起整条链路。

一句提醒：这个金额是客户端按公开价目估算的，跟你实际账单可能有出入。当趋势看、当限额依据够用，别拿去跟财务对账。

4.2 记账 + 日限额熔断

在第五课那个 FastAPI 里插两段就成。先是记账：

import json, sqlite3, subprocess, time, datetime

DB = "/opt/agent/usage.db"

def init_db():
    with sqlite3.connect(DB) as c:
        c.execute("""CREATE TABLE IF NOT EXISTS usage(
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id TEXT, day TEXT, session_id TEXT,
            cost_usd REAL, turns INTEGER, duration_ms INTEGER, ts INTEGER)""")

def record_usage(user_id: str, data: dict):
    day = datetime.date.today().isoformat()
    with sqlite3.connect(DB) as c:
        c.execute("INSERT INTO usage(user_id,day,session_id,cost_usd,turns,duration_ms,ts)"
                  " VALUES(?,?,?,?,?,?,?)",
                  (user_id, day, data.get("session_id"),
                   float(data.get("total_cost_usd") or 0),
                   int(data.get("num_turns") or 0),
                   int(data.get("duration_ms") or 0), int(time.time())))

再是调用 + 熔断，注意 --max-budget-usd 这个参数，它是单次调用的硬上限，模型烧到这个数会被强制停：

DAILY_QUOTA = 2.0   # 每人每天 2 美元

def run_claude(user_id: str, prompt: str) -> dict:
    day = datetime.date.today().isoformat()
    with sqlite3.connect(DB) as c:
        spent = c.execute("SELECT COALESCE(SUM(cost_usd),0) FROM usage"
                          " WHERE user_id=? AND day=?", (user_id, day)).fetchone()[0]
    if spent >= DAILY_QUOTA:
        raise RuntimeError(f"今日额度已用完：${spent:.2f}/${DAILY_QUOTA}")

    proc = subprocess.run(
        ["claude", "-p", prompt, "--output-format", "json",
         "--max-turns", "12", "--max-budget-usd", "0.50"],
        capture_output=True, text=True, timeout=900,
    )
    line = proc.stdout.strip().splitlines()[-1]
    data = json.loads(line)
    record_usage(user_id, data)
    return {"text": data.get("result", ""), "cost_usd": data.get("total_cost_usd")}

三层防线叠起来：日限额挡「今天别再用」、--max-budget-usd 挡「单次别失控」、--max-turns 挡「别无限转圈」。少一层都会在某个深夜被教育。

4.3 加一个 /usage 接口

@app.get("/usage/{user_id}")
def usage(user_id: str):
    day = datetime.date.today().isoformat()
    with sqlite3.connect(DB) as c:
        rows = c.execute("SELECT day, SUM(cost_usd), COUNT(*), SUM(turns)"
                         " FROM usage WHERE user_id=? GROUP BY day"
                         " ORDER BY day DESC LIMIT 7", (user_id,)).fetchall()
    return [{"day": d, "cost_usd": round(c or 0, 4), "calls": n, "turns": t}
            for d, c, n, t in rows]

前端随便画个柱状图，「这个月谁在烧钱」一目了然。这套东西不高级，但真出问题时能救命：上周我这边一次异常重试，就是靠 calls 突然翻十倍发现的。

五、进阶：接 OpenTelemetry 做团队级看板

如果 Agent 不止一个服务、还想接 Grafana，Claude Code 原生支持 OTel 导出，不用自己埋点。环境变量四行搞定：

export CLAUDE_CODE_ENABLE_TELEMETRY=1
export OTEL_METRICS_EXPORTER=otlp
export OTEL_EXPORTER_OTLP_PROTOCOL=grpc
export OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:4317

导出的指标里，跟我关心的直接相关这几个：

指标	含义
claude_code.cost.usage	会话成本，单位美元
claude_code.token.usage	token 消耗，按类型分
claude_code.session.count	会话数
claude_code.active_time.total	真实活跃时长，不含闲置

想按人/按团队拆分，加 OTEL_RESOURCE_ATTRIBUTES：

export OTEL_RESOURCE_ATTRIBUTES="user.id=wufeng,team.id=ops,cost_center=main"

这些键会挂到每一条指标上，Grafana 里直接 group by 就行。

三个坑先说清楚，省得你踩：

1. 写在仓库里的 .claude/settings.json 里的 OTEL 变量会被忽略。这是故意的，防止一个 clone 下来的仓库偷偷把你的遥测导到别处。要写在 ~/.claude/settings.json 的 env 块或托管设置里。
2. OTEL_* 不会传给子进程，包括 Bash 工具和 hooks。你在 hook 里调自己的埋点，得单独设一遍。
3. 验证是否通了，看后端有没有 claude_code.session.count；本地也可以先用 export OTEL_METRICS_EXPORTER=console 打到终端上看看。

顺便提一句，交互模式下 /usage 命令能直接看当场会话的 token 用量，临时对一下账很方便，但它只是当场的，做不了历史统计——所以我们才自己记 SQLite。

六、验收清单

上线前照着走一遍，别只看「配置没报错」：

1. rm -rf / 被拒，理由出现在对话里 → 门禁生效；
2. 让它改 .env，被拒 → 文件保护生效；
3. ls -la 正常执行 → 没误伤，这条最容易翻车；
4. ~/claude-audit.jsonl 里有对应记录，decision 字段对得上；
5. 跑一次正常任务，/usage/{user_id} 里出现一条 cost_usd > 0 的记录 → 记账生效；
6. 手动把某用户当天额度改小，请求返回额度不足 → 熔断生效。

还有两个容易踩的坑，都是排查了一下午换来的：

• hook 打印的 JSON 前面不能有别的东西。有些机器的 ~/.bashrc 里有一句无脑 echo "Shell ready"，hook 在非交互 shell 里跑也会加载 profile，结果 stdout 变成「Shell ready」+ JSON，Claude Code 一看不是 { 开头就当纯文本扔了，而且退出码是 0，连报错都没有。修法是把那句 echo 包进 if [[ $- == *i* ]]; then ... fi，只在交互式 shell 里打印。
• permissionDecision 必须在 hookSpecificOutput 里面，放顶层照样是合法 JSON，Claude Code 静默忽略。这类「没报错但没生效」的问题，用 claude --debug-file /tmp/cc.log 起一次，日志里搜 unrecognized keys 立刻现形。

最后说句实在的：这套门禁不防「蓄意攻击」，防的是「模型在你不看着的时候手滑」。真要跑不受信任的代码，还得上容器和沙箱，Hooks 只是第一道闸。

生命不息，折腾不止。门禁和仪表盘齐了，下一步该让这一个服务学会「分工」——下一课用 subagent 加 Skills，把一个大任务拆给几个专职小助手，顺带把单次成本再压一半。Claude Code 第七课：让 Agent 自己带团队。