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 时开发者面对的“限额”并不只有一种。常见的情况可以分成三类限制类型常见表现对开发的影响费用额度账户余额、月度预算、单次项目预算超额后继续调用会产生额外费用或直接无法调用速率限制每分钟、每小时请求数限制短时间内频繁调用会收到限流提示任务被中断会话预算单次任务希望控制的 token 或成本上限长任务可能越跑越远消耗超预期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。事件名称触发时机AgentObs 的用途PreToolUse工具执行之前检查用量决定是否 blockPostToolUse工具执行之后记录本次工具调用累计消耗UserPromptSubmit用户提交提示词时可选记录会话开始时间NotificationClaude Code 发送通知时可选触发告警Stop一次完整生成结束时可选做会话级汇总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 脚本不依赖外部服务。使用它之前需要先准备好以下环境依赖项说明Python 3.8AgentObs 使用标准库实现不需要额外安装第三方包Claude Code CLI需要在系统中能够正常启动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 }各字段含义如下字段含义max_requests_per_hour每小时允许的 hook 计数请求数超过阈值则拦截max_tokens_per_session单次会话估算 token 上限max_cost_usd_per_month月度估算成本上限单位美元warning_threshold预警系数0.9 表示用量到 90% 就触发failure_policy脚本异常时默认动作allow 表示放行block 表示阻断state_path / log_path状态文件和日志文件路径支持使用 ~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. 核心代码读写状态、判断阈值、返回 block4.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, encodingutf-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, encodingutf-8) as f: json.dump(state, f, ensure_asciiFalse, indent2) os.replace(tmp_path, path)注意json.dump里的ensure_asciiFalse这是为了让包含中文的tool_input在日志和状态文件中可读。Windows 环境下还要确保打开文件时指定encodingutf-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, encodingutf-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_asciiFalse))在 decide 模式中脚本只做判断不修改状态。这样可以避免一个问题如果工具并没有执行却把状态计数累加了会导致后续误判越来越准最终把所有工具都拦住。4.5 record 模式记录请求与估算消耗record 模式由 PostToolUse 事件触发。只有工具真正执行完才把这次请求计入状态文件。def estimate_tokens(tool_name, tool_input): text json.dumps(tool_input, ensure_asciiFalse) 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_asciiFalse)) except Exception as e: with open(config[log_path], a, encodingutf-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_asciiFalse)) if __name__ __main__: main()异常兜底里failure_policy为allow时即使 AgentObs 自身出现问题Claude Code 仍能继续工作代价是成本控制失效failure_policy为block时问题可能导致所有匹配工具都被拦截。这个取舍应该由团队根据自己的成本敏感度来决定。5. 挂载到 Claude Codesettings.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 已经加载配置好之后不要急着写复杂逻辑。先做一次最小验证在 Claude Code 中发送一个会让模型调用 Bash 的提示词例如“列出当前目录文件”。查看~/.agentobs/agent_obs.log是否有新日志。查看~/.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。这时工具调用会被 blockClaude Code 会展示拦截原因模型会收到一个无法执行工具的反馈。它可能会重新选择其他工具也可能提示用户需要手动处理。这就是 AgentObs 的核心价值把“突然断线”变成“有提示地停下”。开发者可以在状态文件里看到原因在日志里看到是哪一步触发了拦截从而判断是调整预算还是降低工具调用频率。7. 常见问题排查hook 不生效、