Claude Code hooks实战：AgentObs实现用量预警与自动拦截

Claude Code 这类 AI 编程工具在提升日常开发效率的同时，也让不少开发者开始关心另一个问题：使用配额或预算额度会在什么时候耗尽。AgentObs 正是一个针对这个场景设计的 hook 程序，它通过 Claude Code 的 hooks 机制，在工具真正执行之前检查当前用量，一旦达到预警阈值就返回 block 决策，从而在额度边界前主动停下来，而不是等 API 报错后被迫中断。

这篇内容会围绕 AgentObs 从零讲清楚：Claude Code hooks 到底是什么，AgentObs 是如何实现“先判断、后执行、再记账”的，以及怎样把它配置到 Claude Code、验证拦截效果、排查常见的 hook 失效问题。文中给出的代码和配置都以最小可运行的方式组织，适合直接复制到本机做实验，再根据实际项目调整阈值、事件范围和状态文件路径。

1. 先理解 AgentObs 要解决的限额问题

1.1 用量限制通常来自三个层面

使用 Claude Code 时，开发者面对的“限额”并不只有一种。常见的情况可以分成三类：

AgentObs 的设计目标不是替代 Anthropic 官方后台，而是在 Claude Code 这一层提供一道“提前刹车”的守护逻辑。它不关心你用的是按量付费还是订阅包，只关心状态文件里记录的累计用量是否已经进入危险区间。

1.2 等 API 报错再处理，问题往往已经发生

很多开发者第一次接触超限，是在 Claude Code 执行到一半时看到错误提示。这个时候整个会话可能已经被中断，代码可能只写到一半，文件状态也处于中间态。更关键的是，一次失败的调用往往已经产生了费用，而不是零成本失败。

AgentObs 的核心思路是把判断前移。不是在 API 返回错误后再处理，而是在 Claude Code 准备执行 Bash、Write、Edit 这类高成本工具之前，先读取本地用量状态文件，发现接近阈值就直接返回 block，让模型换一条低消耗路径，或者提示开发者手动确认。

1.3 AgentObs 在整个调用链中的位置

Claude Code 的执行流程可以简化为：用户输入提示词，模型决定调用某个工具，工具执行，模型拿到结果继续推理。AgentObs 挂在“模型决定调用工具”和“工具真正执行”之间的 PreToolUse 事件上，同时用 PostToolUse 事件记录本次消耗。

这里需要特别说明：AgentObs 并不能阻止模型 API 请求本身发生。模型已经在生成回答时消耗了 token，这一点无法由 hook 完全拦截。AgentObs 能阻止的是工具侧继续产生新的高成本动作，例如继续执行 bash 命令、继续大段写入文件、继续发起网络请求。这种“高成本动作被阻断”的价值在于，避免模型在一个失控循环里把预算迅速耗尽。

2. Claude Code hooks 是 AgentObs 的运行基础

2.1 hook 本质上是事件回调

Claude Code 提供了一套 hook 机制，允许开发者把自定义命令挂到特定生命周期事件上。你不需要修改 Claude Code 的源码，只需要在配置文件里声明“当某个事件发生时，帮我执行某个命令”。这个命令可以由 Python、Node、Shell 等任意可执行程序实现。

AgentObs 选择用 Python 实现，主要是因为它对 JSON 处理、文件并发和跨平台路径处理都比较直接。如果你更习惯 Node 或 Go，也可以按同样的协议改造，关键不取决于语言，而取决于 hook 的输入输出规则。

2.2 常用 hook 事件和 AgentObs 的关心点

Claude Code 的 hook 事件并不是只有一个。不同事件承担不同职责，AgentObs 至少会用到其中的 PreToolUse 和 PostToolUse。

PreToolUse 是关键节点。因为它发生在工具执行之前，返回 block 可以阻止这次工具调用。PostToolUse 则适合做状态累计，因为只有工具真正执行了，才应该计入请求次数和估算消耗。

2.3 hook 脚本的输入输出协议

Claude Code 执行 hook 时，会把事件数据以 JSON 形式写到命令的标准输入。事件数据通常包含 session_id、hook_event_name、tool_name、tool_input 等字段。具体字段名会随版本变化，所以 AgentObs 的代码不会假设所有字段都存在，而是用字典的 get 方法做兼容处理。

hook 脚本通过标准输出返回决策。如果要放行，可以输出：

{"decision": "allow"}

如果要阻断，输出：

{"decision": "block", "reason": "monthly cost threshold reached"}

这里有一个非常重要的工程习惯：hook 脚本不要往 stdout 打印任何日志，否则 Claude Code 在解析 JSON 时会被额外内容干扰。所有调试日志都应该写到 stderr，或者像 AgentObs 一样写到独立的日志文件。

3. AgentObs 的最小实现：环境、目录和状态设计

3.1 环境要求

AgentObs 是一个本地运行的 hook 脚本，不依赖外部服务。使用它之前，需要先准备好以下环境：

如果只是在学习环境测试，不要求服务器或云数据库。AgentObs 的所有状态都保存在本机 JSON 文件中。生产环境如果担心多机或多用户协作，可以改造状态存储为 SQLite 或 Redis，但核心判断逻辑不变。

3.2 目录结构

推荐把 AgentObs 独立安装在用户目录下，避免和项目代码混在一起：

~/.agentobs/ ├── bin/ │ └── agent_obs.py ├── config.json ├── state.json ├── agent_obs.log └── README.md

其中config.json是限额配置，state.json是运行状态，agent_obs.log是调试日志。目录名称和路径不是强制要求，但保持独立目录会让后续升级和维护更清晰。

3.3 配置文件设计

config.json用来声明“哪些指标达到多少算危险”。下面是一个最小配置示例：

{ "limits": { "max_requests_per_hour": 200, "max_tokens_per_session": 100000, "max_cost_usd_per_month": 20.0 }, "warning_threshold": 0.9, "failure_policy": "allow", "state_path": "~/.agentobs/state.json", "log_path": "~/.agentobs/agent_obs.log" }

各字段含义如下：

warning_threshold的作用非常关键。直接把限额设成 100% 并不安全，因为工具调用一旦发起，后续还可能继续消耗。预留 10% 到 20% 的缓冲空间，可以让 Claude Code 在真正没有额度之前先停下来。

对应的state.json初始状态可以写成：

{ "minute_window": {"count": 0, "start": "2026-01-01T00:00:00"}, "hour_window": {"count": 0, "start": "2026-01-01T00:00:00"}, "session_tokens": 0, "month_cost_usd": 0, "total_requests": 0 }

start字段用于时间窗口重置。AgentObs 会先判断当前时间与窗口起始时间是否超过窗口长度，如果超过就把count重置为 0。

4. 核心代码：读写状态、判断阈值、返回 block

4.1 从 stdin 读取 hook 事件

AgentObs 每次启动都由 Claude Code 拉起，并通过 stdin 接收事件 JSON。先实现一个最基础的读取函数：

import json import sys def read_event(): raw = sys.stdin.read() if not raw.strip(): return None try: return json.loads(raw) except json.JSONDecodeError: return None

这里有两个关键点。第一，stdin.read()会等待所有输入结束，适合 hook 这种一次性命令场景。第二，读取失败时返回None，由上层决定是放行还是阻断，不能让脚本直接崩溃。

4.2 状态读写与原子写入

状态文件会被多个 hook 进程并发访问，因此不能直接覆盖写入。先用临时文件写入，再用os.replace原子替换，可以避免 Claude Code 同时触发多个事件时读到半截文件。

import os def load_state(path): try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except FileNotFoundError: return {} except json.JSONDecodeError: return {} def save_state(path, state): tmp_path = path + ".tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, indent=2) os.replace(tmp_path, path)

注意json.dump里的ensure_ascii=False，这是为了让包含中文的tool_input在日志和状态文件中可读。Windows 环境下，还要确保打开文件时指定encoding="utf-8"，否则系统默认编码可能引发 Unicode 错误。

4.3 配置读取与失败策略

配置文件可能不存在，也可能被写坏。AgentObs 采用“出错时读默认值”的策略，并用一个failure_policy字段控制异常兜底：

DEFAULT_CONFIG = { "limits": {}, "warning_threshold": 1.0, "failure_policy": "allow", "state_path": "~/.agentobs/state.json", "log_path": "~/.agentobs/agent_obs.log" } def load_config(path): config = dict(DEFAULT_CONFIG) try: with open(path, "r", encoding="utf-8") as f: loaded = json.load(f) if isinstance(loaded, dict): config.update(loaded) except (FileNotFoundError, json.JSONDecodeError): pass return config

这里的容错思路是：AgentObs 的配置坏了，不应该让 Claude Code 挂掉。如果failure_policy是allow，脚本异常时输出放行决策；如果配置要求严格，则可以输出 block。生产环境应该显式设置这个字段，避免团队成员各自安装后行为不一致。

4.4 decide 模式：判断是否拦截

decide 模式由 PreToolUse 事件触发。脚本读取状态后，依次检查小时请求数、会话 token 估算值、月度成本估算值。

def check_limits(config, state, now): reasons = [] limits = config.get("limits", {}) threshold = config.get("warning_threshold", 1.0) hour_window = state.get("hour_window", {"count": 0}) hour_limit = limits.get("max_requests_per_hour") if hour_limit and hour_window.get("count", 0) >= hour_limit * threshold: reasons.append("hourly request threshold reached") session_tokens = state.get("session_tokens", 0) token_limit = limits.get("max_tokens_per_session") if token_limit and session_tokens >= token_limit * threshold: reasons.append("session token threshold reached") month_cost = state.get("month_cost_usd", 0) cost_limit = limits.get("max_cost_usd_per_month") if cost_limit and month_cost >= cost_limit * threshold: reasons.append("monthly cost threshold reached") return reasons

这里有个细节：threshold默认值是 1.0，也就是没有配置预警系数时，达到 100% 才拦截。如果配置了 0.9，那么 90% 就会触发。之所以单独作为一个字段而不是写死在代码里，是为了让不同项目可以采用不同的风险偏好。

decide 的主流程如下：

def cmd_decide(config, event): state = load_state(config["state_path"]) now = time.time() reasons = check_limits(config, state, now) if reasons: decision = { "decision": "block", "reason": "AgentObs: " + "; ".join(reasons) } else: decision = {"decision": "allow"} print(json.dumps(decision, ensure_ascii=False))

在 decide 模式中，脚本只做判断，不修改状态。这样可以避免一个问题：如果工具并没有执行，却把状态计数累加了，会导致后续误判越来越准，最终把所有工具都拦住。

4.5 record 模式：记录请求与估算消耗

record 模式由 PostToolUse 事件触发。只有工具真正执行完，才把这次请求计入状态文件。

def estimate_tokens(tool_name, tool_input): text = json.dumps(tool_input, ensure_ascii=False) char_count = max(1, len(text)) base_tokens = char_count // 4 if tool_name == "Bash": return base_tokens + 200 if tool_name in ("Write", "Edit"): return base_tokens + 100 return base_tokens

这段代码的意图很明确：Bash 执行风险高，估算权重更高；Write 和 Edit 会改动文件，也可能触发后续检查；Read 只读，权重低。需要注意的是，这只是一个本地估算，不能当作 Anthropic 官方账单数据。如果希望精确计算，可以把账单系统导出的 token 数据定时写入状态文件。

record 主流程如下：

def cmd_record(config, event): state = load_state(config["state_path"]) now = time.time() tool_name = event.get("tool_name", "") tool_input = event.get("tool_input", {}) hour_window = state.get("hour_window", {"count": 0, "start": now}) if now - hour_window.get("start", now) > 3600: hour_window = {"count": 0, "start": now} hour_window["count"] = hour_window.get("count", 0) + 1 state["hour_window"] = hour_window tokens = estimate_tokens(tool_name, tool_input) state["session_tokens"] = state.get("session_tokens", 0) + tokens state["total_requests"] = state.get("total_requests", 0) + 1 state["last_event_at"] = time.strftime("%Y-%m-%d %H:%M:%S") save_state(config["state_path"], state)

时间窗口重叠的问题是常见的坑。如果窗口跨天或跨小时，状态文件里的start必须随着重置而更新，不能只重置count。否则窗口过期后，count 会被清零，但start还是旧时间，下一个事件又触发清零，计数永远起不来。

4.6 主入口与异常兜底

主入口把 decide 和 record 两个模式接到命令行参数上，并统一处理异常：

import time def main(): config = load_config("~/.agentobs/config.json") config["state_path"] = os.path.expanduser(config["state_path"]) config["log_path"] = os.path.expanduser(config["log_path"]) try: mode = sys.argv[1] if len(sys.argv) > 1 else "decide" event = read_event() if event is None: return if mode == "decide": cmd_decide(config, event) elif mode == "record": cmd_record(config, event) else: decision = {"decision": "block", "reason": "unknown mode: " + mode} print(json.dumps(decision, ensure_ascii=False)) except Exception as e: with open(config["log_path"], "a", encoding="utf-8") as f: f.write(time.strftime("%Y-%m-%d %H:%M:%S ") + "unhandled error: " + str(e) + "\n") decision = {"decision": config.get("failure_policy", "allow")} print(json.dumps(decision, ensure_ascii=False)) if __name__ == "__main__": main()

异常兜底里，failure_policy为allow时，即使 AgentObs 自身出现问题，Claude Code 仍能继续工作，代价是成本控制失效；failure_policy为block时，问题可能导致所有匹配工具都被拦截。这个取舍应该由团队根据自己的成本敏感度来决定。

5. 挂载到 Claude Code：settings.json 配置

5.1 用户级配置和项目级配置

Claude Code 支持两类 hook 配置。用户级配置放在~/.claude/settings.json，对所有项目生效；项目级配置放在项目根目录下的.claude/settings.json，随项目一起提交版本库，适合团队统一约束。

AgentObs 的推荐用法是：本地测试时放在用户级配置，方便随时开关；团队推广时放到项目级配置，并配合 README 说明限额规则。

5.2 完整配置示例

下面是一个挂载示例，PreToolUse 和 PostToolUse 都覆盖 Bash、Write、Edit 三个高风险工具：

{ "hooks": { "PreToolUse": [ { "matcher": "Bash|Write|Edit", "hooks": [ { "type": "command", "command": "python3 /home/user/.agentobs/bin/agent_obs.py decide" } ] } ], "PostToolUse": [ { "matcher": "Bash|Write|Edit", "hooks": [ { "type": "command", "command": "python3 /home/user/.agentobs/bin/agent_obs.py record" } ] } ] } }

这里的matcher是一个正则表达式字符串，用来匹配工具名。Bash|Write|Edit表示 Bash、Write、Edit 这三个工具都会触发同一个 hook。command是实际执行的命令，建议使用绝对路径，避免 Claude Code 启动时的 PATH 环境变量差异导致找不到 python3 或脚本。

5.3 如何确认 hook 已经加载

配置好之后，不要急着写复杂逻辑。先做一次最小验证：

1. 在 Claude Code 中发送一个会让模型调用 Bash 的提示词，例如“列出当前目录文件”。
2. 查看~/.agentobs/agent_obs.log是否有新日志。
3. 查看~/.agentobs/state.json中的total_requests是否增加。

如果这三个检查点都没有变化，说明 hook 配置没有被加载，或者 matcher 没有匹配到工具名。常见原因包括配置路径写错、JSON 语法错误、Claude Code 进程没有重启。

如果只想测试脚本本身，也可以不启动 Claude Code，直接手动喂入事件 JSON。

6. 运行验证：从命令行模拟到真实交互

6.1 手动测试 decide 模式

在不启动 Claude Code 的情况下，可以这样验证 AgentObs 是否正常工作：

echo '{"hook_event_name":"PreToolUse","tool_name":"Bash","tool_input":{"command":"ls"}}' | python3 ~/.agentobs/bin/agent_obs.py decide

如果还没有超限，预期输出是：

{"decision": "allow"}

如果状态文件中的用量已经超过阈值，预期输出是：

{"decision": "block", "reason": "AgentObs: monthly cost threshold reached"}

这条命令的意义在于：把 Claude Code 的 hook 协议单独拉出来测试，不需要一次次启动 Claude Code 等待模型响应，调试速度更快。

6.2 手动模拟超限场景

要验证 block 分支，可以手动把state.json中的month_cost_usd改成一个超过限制的值，再运行上面的 decide 命令。例如：

{ "limits": { "max_cost_usd_per_month": 20.0 } }

然后在state.json中设置：

{ "month_cost_usd": 21.0 }

再运行 decide 命令，就能看到 block 输出。验证完后要记得把状态文件恢复，否则 AgentObs 会一直拦截。

6.3 在 Claude Code 中观察真实效果

把状态文件改成超限状态后，回到 Claude Code 里让模型继续执行 Bash 或 Write。这时工具调用会被 block，Claude Code 会展示拦截原因，模型会收到一个无法执行工具的反馈。它可能会重新选择其他工具，也可能提示用户需要手动处理。

这就是 AgentObs 的核心价值：把“突然断线”变成“有提示地停下”。开发者可以在状态文件里看到原因，在日志里看到是哪一步触发了拦截，从而判断是调整预算，还是降低工具调用频率。

7. 常见问题排查：hook 不生效、