不到 500 行 Python把 Claude Code 的请求转发到api.anthropic.com顺手把每个项目消耗了多少 token、按官方价折成多少美元、占了 Max 配额多少份额全部记到 SQLite。适用场景你订了 Claude Max定额套餐不按量计费但你想知道过去一周哪个项目最烧钱、哪个项目最容易让我撞限流。一、要解决什么问题订了 Claude Max 套餐之后会遇到一个尴尬官方账单看不到细分。Max 是包月制账单上只有一个固定数字看不出哪个项目消耗多。配额会被耗尽。Max 不是无限的5 小时/周/月各有上限。撞到限流时你想知道是哪个项目把你拖下水的。多项目并行时责任无法归因。同一台机器上跑 3 个项目的 Claude Code谁的 prompt 最贵全靠拍脑袋。我们想要的东西其实很简单——一只电表不改 Claude Code不要求每个项目都改代码。区分项目维度。把 token 消耗折算成两种可比指标影子美元横向跨项目 ROI 比较、Sonnet 等效 token评估配额份额。出问题能查现场——完整的请求/应答 dump。token-proxy就是为这个目标做的。整体只有一份proxy.py约 470 行 一个requirements.txt3 个依赖FastAPI、httpx、uvicornSQLite 自带零部署成本。二、整体架构插在中间的 FastAPI 反向代理┌──────────────┐ HTTPS ┌───────────────────┐ HTTPS ┌──────────────────┐ │ Claude Code │ ────────► │ token-proxy │ ────────► │ api.anthropic.com │ │ (per repo) │ │ FastAPI httpx │ │ │ └──────────────┘ └─────────┬─────────┘ └──────────────────┘ │ │ │ X-Project-Id: 项目A ├──► usage.db (SQLite按项目聚合) └──► logs/*.json 可选完整请求/响应快照关键设计点完全透明转发客户端把代理当成https://api.anthropic.com用请求体、响应体、状态码、流式分片都原样直传。副作用旁路计量、落盘是在中间偷一份拷贝做的不阻塞流式响应——这是用户体验的关键。项目身份靠一个 HTTP 头X-Project-Id。Claude Code 支持通过ANTHROPIC_CUSTOM_HEADERS注入自定义头我们只用了这一根钩子无需任何 SDK 改动。接入方法在每个项目的.claude/settings.json里加 4 行 env{ env: { ANTHROPIC_BASE_URL: http://127.0.0.1:8787, ANTHROPIC_CUSTOM_HEADERS: X-Project-Id: 项目A } }启动claude时它读 env把请求打到代理上并附带项目标识。代理转发到上游旁路解析usage落库。三、核心实现拆解3.1 单一通配路由吃下所有路径app.api_route(/{path:path}, methods[GET, POST, PUT, DELETE, PATCH, OPTIONS]) async def proxy(path: str, request: Request): body await request.body() project request.headers.get(x-project-id, _default) is_messages ( request.method POST and path.startswith(v1/messages) and not path.endswith(count_tokens) ) stream is_messages and _is_stream_body(body) ...只挂一条路由匹配所有路径所有方法把决策推迟到运行时。这样/v1/messages、/v1/messages/count_tokens、/v1/models、未来还没出来的端点……一并兜住。是否是要计量的请求is_messages和是否流式stream只在POST /v1/messages且非 token 计数时才成立。其他请求走纯透传分支。隐含决策count_tokens是 Claude Code 内部用来估算上下文长度的探测调用不消耗对话 token明确排除。3.2 头部白名单/黑名单避免转发陷阱HTTP 反向代理最容易翻车的地方就是头部。代码里维护了两份小集合_STRIP_REQ_HEADERS { host, content-length, connection, accept-encoding, x-project-id, # 我们消费掉它不让上游看见 } _STRIP_RES_HEADERS { content-length, content-encoding, transfer-encoding, connection, }为什么要剥这些host/content-length转发时由 httpx 重新计算带着旧的会出 400。accept-encoding让 httpx 自己决定要不要 gzip避免双重压缩。transfer-encoding/content-encodingFastAPI/Starlette 在响应阶段会自行处理 chunked 和 gzip原封不动透传会和实际 body 长度对不上。x-project-id这是给我们看的内部头不该污染上游。3.3 流式响应的偷拷贝边转发边解析这是整个项目最值得说的部分。Claude API 的流式响应是 SSE 协议一条 message 由若干data: {...}\n\n事件组成最后一条是[DONE]。 token 用量信息分布在两类事件里message_start携带model、初始input_tokens、cache_creation_input_tokens、cache_read_input_tokens。message_delta携带累计output_tokens以及更新后的缓存统计。朴素思路是把整段 body 都收下来再解析——但那样客户端要等代理收完才能看到第一个字节体验直接退回非流式。所以我们用边走边算async def relay(): usage {input_tokens: 0, output_tokens: 0, cache_creation_input_tokens: 0, cache_read_input_tokens: 0} model_holder [] buf b full bytearray() if LOG_DIR else None try: async for chunk in upstream.aiter_raw(): yield chunk # ① 立刻交还给客户端 buf chunk buf _parse_sse_chunk(buf, usage, model_holder) # ② 偷一份拷贝解析 if full is not None: full.extend(chunk) # ③ 同时收集完整 body 落盘 finally: await upstream.aclose() _log(project, model_holder[0], usage, ...) # ④ 汇总写库 if full is not None: _finish_log(bytes(full), sseTrue)三个动作按 ①②③ 顺序但代价是增加一次bytes拷贝。在 LLM 场景里 token 速率远低于 IOCPU 完全吃得下。_parse_sse_chunk的实现也值得看def _parse_sse_chunk(buf: bytes, usage: dict, model_holder: list[str]) - bytes: while b\n\n in buf: raw, buf buf.split(b\n\n, 1) for line in raw.split(b\n): if not line.startswith(bdata:): continue data line[5:].lstrip() if not data or data b[DONE]: continue try: evt json.loads(data) except Exception: continue ... return buf要点以\n\n分割完整事件剩余不完整的字节留给下一次 chunk 拼接。这是 SSE 解析的标准做法。任何解析失败都吞掉——网络上 SSE 偶尔会有不完整的事件、注释行、心跳行绝不能因为解析失败就让转发也挂掉。这是旁路计量的纪律副作用永远不能影响主路径。message_delta的output_tokens是累计值不是增量。所以代码用usage[output_tokens] u[output_tokens]直接覆盖不是加。被这个细节坑过的人都懂。3.4 非流式的简单路径if not stream: try: data await upstream.aread() finally: await upstream.aclose() try: j json.loads(data) _log(project, j.get(model, ), j.get(usage, {}) or {}, ...) except Exception: pass _finish_log(data, sseFalse) return Response(contentdata, status_codestatus, headersdict(res_headers))非流式响应是一整个 JSON直接json.loads拿usage。两层 try/except 同样是绝不污染主路径——上游真挂了或者格式变了只少一条记账请求该返回还返回。3.5 失败/非计量请求的全透传if not is_messages or status 400: async def passthrough(): collected bytearray() if LOG_DIR else None try: async for chunk in upstream.aiter_raw(): yield chunk if collected is not None: collected.extend(chunk) finally: await upstream.aclose() if collected is not None: _finish_log(bytes(collected), sseFalse) return StreamingResponse(passthrough(), status_codestatus, headersdict(res_headers))两类情况走透传不是 messages 端点比如/v1/models、/v1/messages/count_tokens没必要解析。上游返回 4xx/5xx错误响应里没有 usage但日志要保留方便事后看是 401鉴权挂了还是 429限流了。3.6 单连接复用 lifespanasynccontextmanager async def _lifespan(app: FastAPI): app.state.client httpx.AsyncClient(timeouthttpx.Timeout(None, connect30.0)) try: yield finally: await app.state.client.aclose()整个进程共享一个httpx.AsyncClient。这意味着HTTP/2 连接池常驻不用每次握手 TLS。请求超时用None——LLM 长输出可能跑几分钟固定超时会误杀。但连接超时给了 30 秒避免上游不可达时堆积请求。四、SQLite Schema 与计量CREATE TABLE IF NOT EXISTS usage ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT NOT NULL, project TEXT NOT NULL, model TEXT NOT NULL, input_tokens INTEGER NOT NULL DEFAULT 0, output_tokens INTEGER NOT NULL DEFAULT 0, cache_creation_input_tokens INTEGER NOT NULL DEFAULT 0, cache_read_input_tokens INTEGER NOT NULL DEFAULT 0, stream INTEGER NOT NULL DEFAULT 0, duration_ms INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX idx_usage_project_ts ON usage(project, ts);设计取舍一行 一次 API 调用。聚合在查询时做不在写入时做——保留原始事实未来想加新维度按小时分桶、按 model 切分只需要改 SQL不必迁移历史数据。四种 token 分开存input、output、cache write、cache read。Anthropic 计价不一样、配额份额不一样合并存就丢信息了。stream和duration_ms是免费的现场以后想做流式对延迟的影响分析数据已经在那了。复合索引(project, ts)报表的两种主要过滤维度都覆盖到。写入的零失败容忍def _log(project, model, usage, stream, duration_ms): if not any(usage.get(k) for k in (...)): return # ① usage 全 0 就别记可能是代理路径或者错误响应 with _db() as conn: conn.execute(INSERT ..., (...))with _db() as conn利用 Python 的 sqlite3 上下文异常时自动回滚正常时自动提交。一行解决事务管理。五、两种报表影子美元 vs Sonnet 等效这是 Max 用户特有的设计。Anthropic 的 API 公开价是按量计费的但 Max 是包月——所以美元对你来说不是真账单而是一个虚拟单位方便横向比较。5.1 影子美元_PRICING [ (opus, {in: 15.0, out: 75.0}), (sonnet, {in: 3.0, out: 15.0}), (haiku, {in: 1.0, out: 5.0}), ] ​ def _row_cost(model, in_tok, out_tok, cc, cr): key _model_key(model) if key is None: return 0.0, 0.0 price dict(_PRICING)[key] usd ( in_tok * price[in] out_tok * price[out] cc * price[in] * 1.25 # 缓存写入 输入价 × 1.25 cr * price[in] * 0.1 # 缓存读取 输入价 × 0.1 ) / 1_000_000.0 ...公式直接对应 Anthropic 公开价目。模型识别用 substring 匹配opus in model_id未来出 Opus 5、Sonnet 5 也能直接命中不用改代码。意义当你看到项目 A 这周影子花了 $187项目 B 只花了 $9你能立刻判断 ROI——A 给你创造的价值有没有 20 倍于 B没有的话A 应该优化 prompt 或者改用 Haiku。5.2 Sonnet 等效 token_QUOTA_WEIGHT {opus: 5.0, sonnet: 1.0, haiku: 0.2} ​ sonnet_eq (in_tok out_tok cc cr) * weight这个数字回答另一个问题——哪个项目让我撞限流Max 配额按 model 加权扣减社区经验里 Opus 大概是 Sonnet 的 5x、Haiku 是 0.2x。所以 1M Opus token 占的配额相当于 5M Sonnet token。报表里%quota列展示每个项目占总配额的份额让你一眼看出是谁吃掉了你的窗口。输出长这样project model calls shadow USD sonnet-eq tok %quota ------------------------------------------------------------------- 项目A opus 42 18.4231 3,200,000 62.1% 项目A sonnet 11 0.4520 180,000 3.5% ↳ subtotal 18.8751 3,380,000 项目B sonnet 80 3.1200 1,500,000 29.1% ↳ subtotal 3.1200 1,500,000 ------------------------------------------------------------------- TOTAL 22.0000 5,150,000一目了然项目 A 用 Opus 跑 42 次就吃了 62% 的配额但绝对调用量比项目 B 少一半。如果 A 的产出不足以匹配这个消耗就该考虑降级一些不需要 Opus 推理深度的子任务。六、可选的请求/响应日志调试现场设置PROXY_LOG_DIR./logs后每次请求生成一个 JSON 文件logs/20260427T091203Z-a3f8b1c4-项目A.json文件名格式 UTC 时间戳 8 位随机 ID 项目名路径不安全字符替换为_。这样天然按时间排序ls就是时间线。随机 ID 防同一秒并发碰撞。项目名留在文件名里肉眼一眼看出是哪个项目的请求。文件内容是结构化 JSON请求和响应都记下来{ id: a3f8b1c4, project: 项目A, request: { ts: 2026-04-27T09:12:03Z, method: POST, url: https://api.anthropic.com/v1/messages, headers: { authorization: ***, x-api-key: ***, ... }, body: { model: claude-opus-4-7, messages: [...], stream: true } }, response: { status: 200, stream: true, duration_ms: 8421, truncated: false, body: [ { event: message_start, data: { ... } }, { event: content_block_delta, data: { ... } }, { event: message_delta, data: { usage: { output_tokens: 512 } } } ] } }几个细节敏感头自动打码。authorization、x-api-key、anthropic-api-key、proxy-authorization都被替换成***但保留键名以便确认头确实存在。流式响应被解析成事件数组。比原始data: ...\n\n文本可读得多——你可以直接看到 token 是怎么逐步到达的、哪一步耗时最久。PROXY_LOG_MAX_BYTES控制单条响应最大字节数。长上下文 流式回复可能产生几 MB 的 SSE 文本开了限额就只截响应正文不影响请求和元信息。写日志失败只打印 stderr不抛。同样是副作用纪律。def _write_log(rid, project, req, res): ... try: path.write_text(json.dumps(payload, ensure_asciiFalse, indent2), encodingutf-8) except Exception as e: print(f[proxy] log write failed: {e}, filesys.stderr)七、设计哲学旁路、留痕、零侵入把整个项目浓缩成三条原则1. 副作用绝不阻塞主路径每一处try/except都是这条原则的实例解析 SSE 失败 → 吞掉转发继续。写库失败 → 抛出后被外层捕获转发继续。写日志失败 → stderr 一行转发继续。上游 4xx/5xx → 透传给客户端让它自己处理。代理坏了用户立刻就知道但计量坏了用户毫无感觉——所以计量要假定自己永远可能坏永远不能拖累代理本职。2. 留原始事实不留聚合结果数据库里每一行都是一次原始 API 调用从未做过预聚合。报表 SQL 是SUM ... GROUP BY ...现场算的。这个选择换来的是加新维度不用迁移历史数据。单价或者权重表错了改了就重跑原始 token 数永远不变。想做按小时画消耗曲线SQL 一句话。3. 零侵入零 SDK 依赖整个方案只依赖 Claude Code 已经支持的两个环境变量ANTHROPIC_BASE_URL、ANTHROPIC_CUSTOM_HEADERS。这意味着任何按 Anthropic SDK 标准实现的客户端都能接入。Claude Code 升级不会影响代理。Anthropic 加了新 API 端点也能直接转发不需要更新代理。八、能扩展到哪里代码故意保持小而清晰留了几个扩展点扩展方向改动量团队级聚合把_db()换成 PostgreSQL/ClickHouseschema 不变SQL 兼容实时仪表盘在 FastAPI 里加/dashboard路由读 SQLite前端用任何框架配额预警在_log()里加阈值判断超限调 webhook 推送到 Slack/邮件模型切换策略在请求转发前根据 body 决定改model字段把不需要 Opus 的请求降级多上游路由按X-Project-Id路由到不同的PROXY_UPSTREAM比如开发用 self-host生产用官方缓存命中分析cache_read_input_tokens / (input cache_*)算每个项目的 prompt 缓存命中率每一项都不需要伤筋动骨——这是一份小文件 原始事实存储带来的红利。九、价值小结如果只让说一句话它把我用了多少 Claude这件事从感觉变成了数据。更具体地ROI 量化。项目 A 这个月让我虚拟花掉 $200但它没产出对应价值——这种判断以前是直觉现在有数。配额归因。撞限流时不再骂运气能直接定位到罪魁祸首项目针对性优化。prompt 优化反馈环。改了 system prompt 之后第二天对比cache_read比例就知道缓存有没有失效。故障现场永远在。PROXY_LOG_DIR开着事后任何可疑响应都能翻出原始 SSE 流复盘。零迁移成本。Claude Code 不感知它的存在关掉代理回归官方零摩擦。不到 500 行 Python做这些事够用了。附录完整文件清单token-proxy/ ├── proxy.py 主程序FastAPI 服务 report/cost 子命令约 470 行 ├── requirements.txt fastapi / httpx / uvicorn[standard] ├── usage.db SQLite 计费库首次启动自动创建 └── logs/ 请求日志设了 PROXY_LOG_DIR 才有启动一条命令、配置每个项目两行 env、查报表两个子命令。Done。项目过于简单不再上传代码了。