企业官网建设流程全解析

不到 500 行 Python，把 Claude Code 的请求转发到api.anthropic.com，顺手把每个项目消耗了多少 token、按官方价折成多少美元、占了 Max 配额多少份额，全部记到 SQLite。

适用场景：你订了 Claude Max，定额套餐不按量计费，但你想知道"过去一周哪个项目最烧钱、哪个项目最容易让我撞限流"。

一、要解决什么问题

订了 Claude Max 套餐之后会遇到一个尴尬：

• 官方账单看不到细分。Max 是包月制，账单上只有一个固定数字，看不出哪个项目消耗多。

• 配额会被耗尽。Max 不是无限的，5 小时/周/月各有上限。撞到限流时，你想知道是哪个项目把你拖下水的。

• 多项目并行时责任无法归因。同一台机器上跑 3 个项目的 Claude Code，谁的 prompt 最贵，全靠拍脑袋。

我们想要的东西其实很简单——一只电表：

1. 不改 Claude Code，不要求每个项目都改代码。

2. 区分项目维度。

3. 把 token 消耗折算成两种可比指标：影子美元（横向跨项目 ROI 比较）、Sonnet 等效 token（评估配额份额）。

4. 出问题能查现场——完整的请求/应答 dump。

token-proxy就是为这个目标做的。整体只有一份proxy.py（约 470 行）+ 一个requirements.txt（3 个依赖：FastAPI、httpx、uvicorn），SQLite 自带，零部署成本。

二、整体架构：插在中间的 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-encoding：FastAPI/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), sse=True)

三个动作按 ①②③ 顺序，但代价是增加一次bytes拷贝。在 LLM 场景里 token 速率远低于 IO，CPU 完全吃得下。

_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(b"data:"): 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, sse=False) return Response(content=data, status_code=status, headers=dict(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), sse=False) return StreamingResponse(passthrough(), status_code=status, headers=dict(res_headers))

两类情况走透传：

• 不是 messages 端点：比如/v1/models、/v1/messages/count_tokens，没必要解析。

• 上游返回 4xx/5xx：错误响应里没有 usage，但日志要保留，方便事后看是 401（鉴权挂了）还是 429（限流了）。

3.6 单连接复用 + lifespan

@asynccontextmanager async def _lifespan(app: FastAPI): app.state.client = httpx.AsyncClient(timeout=httpx.Timeout(None, connect=30.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_ascii=False, indent=2), encoding="utf-8") except Exception as e: print(f"[proxy] log write failed: {e}", file=sys.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 端点也能直接转发，不需要更新代理。

八、能扩展到哪里

代码故意保持小而清晰，留了几个扩展点：

每一项都不需要伤筋动骨——这是"一份小文件 + 原始事实存储"带来的红利。

九、价值小结

如果只让说一句话：它把"我用了多少 Claude"这件事从感觉变成了数据。

更具体地：

1. ROI 量化。"项目 A 这个月让我虚拟花掉 $200，但它没产出对应价值"——这种判断以前是直觉，现在有数。

2. 配额归因。撞限流时不再骂运气，能直接定位到罪魁祸首项目，针对性优化。

3. prompt 优化反馈环。改了 system prompt 之后，第二天对比cache_read比例就知道缓存有没有失效。

4. 故障现场永远在。PROXY_LOG_DIR开着，事后任何可疑响应都能翻出原始 SSE 流复盘。

5. 零迁移成本。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。

项目过于简单，不再上传代码了。