内容
# ai-tokenpulse
实时代币经济和上下文预算观察器，用于代理AI系统。监视您的代理发出的每个Claude API调用，并实时在您的编辑器中告诉您 - 您已经花费了多少，上下文窗口的利用率是多少，以及代理是否浪费代币。
通过 **MCP** 公开，因此它在Claude Code、Cursor和任何MCP兼容的客户端中原生工作。
> 大多数代理失败不是推理失败，而是预算失败伪装成推理失败。上下文窗口悄悄饱和；长时间自主循环悄悄燃烧积分；成本超支只在计费时才显现。`ai-tokenpulse` 在发生时使资源层可观察。
---
## 安装
```bash
pip install ai-tokenpulse
```
这为您提供了三个控制台脚本：`ai-tokenpulsed`（守护进程）、`ai-tokenpulse`（CLI）和 `ai-tokenpulse-mcp`（MCP stdio适配器）。
---
## 工作原理
```
Claude Code / Cursor 自定义SDK代理
│ │ │
stdio│ https│ │ unix socket
(MCP)│ (代理)│ │ (ingest)
▼ ▼ ▼
┌──────────┐ ┌──────────────────────────────────────┐
│ MCP │ │ ai-tokenpulsed (守护进程) │
│ 适配器 │◄──┤ 代理 │ ingest API │ query API │
└──────────┘ │ │ │
│ ▼ │
│ 定价 · 异常 · 效率 │
│ ▼ │
│ SQLite (WAL) + .ai-tokenpulse/ JSON导出 │
└──────────────────────────────────────┘
```
一个单独的长期运行的守护进程（`ai-tokenpulsed`）拥有本地SQLite数据库中的所有状态。三个薄适配器连接到它：
- **HTTP代理** - 将 `ANTHROPIC_BASE_URL` 指向它；捕获Claude Code或Cursor的每个模型调用
- **MCP stdio适配器** - 由您的IDE启动；允许您在会话中询问诸如 *"我们当前的上下文饱和度是多少?"* 之类的问题
- **SDK包装器** (`TrackedClient`) - 用于直接在Anthropic SDK上构建的自定义代理
---
## 快速入门 — Claude Code 或 Cursor
**1. 将MCP服务器** 添加到您的客户端配置中。
Claude Code (`.mcp.json` 在您的项目中，或 `~/.claude/mcp.json`）：
```json
{
"mcpServers": {
"ai-tokenpulse": {
"command": "ai-tokenpulse-mcp"
}
}
}
```
Cursor (`.cursor/mcp.json` 或 `~/.cursor/mcp.json`) 使用相同的格式。
**2. 启动守护进程** （保持运行）：
```bash
ai-tokenpulsed --proxy-port 9000 --project my-project
```
**3. 打开会话并将客户端流量路由到代理：**
```bash
ai-tokenpulse session start "what-im-working-on"
export ANTHROPIC_BASE_URL=http://127.0.0.1:9000
claude # 或打开Cursor
```
就是这样。每个API调用现在都记录了一个步骤。在Claude Code或Cursor中，自然地提问 - *"我们当前的上下文饱和度是多少?"* - 代理调用 `get_context_saturation` 通过MCP并从实时数据中回答。
当您完成时：
```bash
ai-tokenpulse session end
```
---
## 快速入门 — 自定义Anthropic SDK代理
```python
import anthropic
from ai_tokenpulse import TrackedClient
raw = anthropic.Anthropic()
client = TrackedClient(raw, session_id="sess_abc", project="my-project")
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "..."}],
)
# 步骤自动记录；响应未修改
```
`TrackedClient` 从每个响应中提取 `usage` 并通过Unix套接字将步骤写入守护进程。调用者返回原始响应对象。
传递 `fail="closed"` 以使调用在守护进程无法访问时硬失败；默认值为 `fail="open"`，因此可观察性不会破坏您的IDE。
---
## CLI
```
ai-tokenpulse session start <name> 打开新会话
ai-tokenpulse session end 关闭当前会话
ai-tokenpulse session list 最近的会话（活动标记）
ai-tokenpulse burn [session_id] 当前（或命名）会话的总计
ai-tokenpulse health ping守护进程
# 分析（镜像MCP工具，可从shell脚本）
ai-tokenpulse efficiency [session_id] 综合效率得分
ai-tokenpulse saturation [session_id] 当前上下文窗口使用百分比
ai-tokenpulse models [session_id] 每模型成本/代币分割
ai-tokenpulse anomalies [session_id] [--z-threshold N]
ai-tokenpulse compaction [session_id] 我们应该现在压缩吗？
ai-tokenpulse exhaustion [session_id] [--target-cost USD]
ai-tokenpulse budget [session_id] --target-cost USD
ai-tokenpulse compare <session_a> <session_b>
```
---
## MCP工具（13）
**会话状态**
- `get_session_burn(session_id?)` — 总计：代币，成本，步骤数，未定价计数
- `get_context_saturation(session_id?)` — 上下文窗口使用百分比（最新步骤）
- `get_step_cost(step_id)` — 逐步骤成本明细
- `list_sessions(limit?)` — 最近的会话，带有活动/关闭标志
**归属**
- `get_tool_costs(session_id?)` — 每工具代币 + 成本明细
- `get_model_breakdown(session_id?)` — Sonnet vs Opus vs Haiku 分割
**标签**
- `tag_session(tag, session_id?)` — 设置或更新人类可读标签
**分析（第4阶段）**
- `detect_burn_anomalies(z_threshold=2.0)` — 步骤的代币总数超过会话均值 >Nσ
- `suggest_compaction_point()` — 当利用率 > 70% 且投影 > 90% 在3步内时触发
- `project_exhaustion(target_cost_usd?)` — `steps_to_context_limit`, `steps_to_cost_limit`, 增长率
- `estimate_remaining_budget(target_cost_usd)` — 在平均成本下剩余步骤
- `compare_sessions(session_a, session_b)` — 当标签不同的diff警告
- `get_efficiency_score()` — 综合0–1得分：烧伤率稳定性，输出产量，决策密度
`session_id` 默认值为 `.ai-tokenpulse/current-session` （您使用CLI打开的会话），因此您很少需要显式传递它。
---
## 数据模型
每个记录的事件都是一个 **步骤** - 一个模型API调用：
```python
{
"schema_version": 2,
"session_id": "sess_2026_04_27_abc",
"session_tag": "refactor-auth", # 可选，用于类似比较会话
"step_id": "step_042",
"timestamp": "2026-04-27T14:32:11Z",
"model": "claude-opus-4-7",
"input_tokens": 12450,
"output_tokens": 380,
"cached_tokens": 8200, # cache_read_input_tokens，价格为0.1x
"tool_use_count": 2, # 工具使用块在响应中发出
"context_window_size": 200000,
"context_used": 45230,
"cost_usd": 0.0871 # null 当模型不在价格表中
}
```
持久性存储在 `.ai-tokenpulse/ai-tokenpulse.db` （SQLite，WAL模式）中，可选地为每个会话导出JSON以供人工检查。
---
## 隐私
代理仅从响应中提取 `usage`、`model`、`id` 和 `tool_use` 块计数。**请求体、响应消息内容、工具调用输入和工具调用输出永远不会被记录、持久化或写入 `.ai-tokenpulse`.** 它们仅在转发期间在代理内存中存在。
---
## 配置
| 标志 / 环境变量 | 默认值 | 目的 |
|---|---|---|
| `ai-tokenpulsed --proxy-port <N>` | （关闭） | 在此端口上运行HTTP代理 |
| `ai-tokenpulsed --proxy-host <H>` | `127.0.0.1` | 代理绑定地址（默认仅限环回；参见下面的安全说明） |
| `ai-tokenpulsed --allow-public-bind` | （关闭） | 确认非环回绑定。未提供此标志则拒绝 |
| `ai-tokenpulsed --upstream <URL>` | `https://api.anthropic.com` | 代理转发到的位置 |
| `ai-tokenpulsed --project <name>` / `AI_TOKENPULSE_PROJECT` | `default` | 项目标签和 **套接字名称后缀** - 参见下文 |
| `ai-tokenpulsed --db <path>` | `./.ai-tokenpulse/ai-tokenpulse.db` | SQLite位置 |
| `AI_TOKENPULSE_SESSION` | （未设置） | 为SDK包装器重写活动会话 |
| `X-Ai-Tokenpulse-Session` （HTTP头） | （未设置） | 每个请求的会话重写代理 |
### 多个项目
守护进程的Unix套接字以项目命名：`~/.ai-tokenpulse/<project>.sock`。要在同一台机器上运行多个守护进程 - 例如，每个打开的项目一个 - 给每个守护进程一个不同的 `--project` （或shell级别的 `AI_TOKENPULSE_PROJECT`），它们将共存。客户端（CLI，MCP适配器，SDK包装器，代理）从同一个环境变量解析正确的套接字，因此一个按项目的shell环境就足够了。
### 安全：强制环回绑定
代理将 `Authorization: Bearer …` 和 `x-api-key` 头透明地转发到上游。将其绑定到非环回接口将允许任何可访问该接口的人发出API调用并计费到您的帐户。守护进程拒绝非环回 `--proxy-host` 值，除非您传递 `--allow-public-bind` 以确认风险。请勿在共享或不受信任的网络上传递该标志。
### “失败开放”的真正含义
代理相对于 **守护进程** 是失败开放的：如果步骤记录失败（SQLite hiccup，模式错误等），用户的API调用不受影响。代理相对于 **自身** 不是失败开放的 - 如果 `ai-tokenpulsed` 崩溃或未运行，`ANTHROPIC_BASE_URL=http://localhost:9000` 指向空值，每个IDE调用都将失败，直到守护进程恢复运行。没有自动回退到 `api.anthropic.com`。如果需要，请在supervisor（launchd，systemd，tmux等）下运行守护进程。
---
## 限制
- **仅限Linux和macOS.** 守护进程使用Unix域套接字进行守护进程 ↔ 适配器 ↔ SDK包装器IPC，而Windows不支持。 `pip install ai-tokenpulse` 将在Windows上成功（轮子是平台无关的），但守护进程和CLI将在运行时失败。WSL2有效。
- **仅限Anthropic.** 其他LLM提供商需要单独的工作。
- **流式SDK调用尚未被 `TrackedClient` 捕获** （代理干净地处理流；SDK包装器警告并在您传递 `stream=True`时回退）。
- **启发式效率阈值.** 综合得分针对粗略默认值进行校准；您需要使用实际会话数据调整它们。原始组件值始终公开，因此您可以查看 *原因* 而不仅仅是 *结果*。
- **本地存储.** 没有身份验证，没有身份验证，没有团队汇总。跨机器的开发人员基准需要将 `.ai-tokenpulse/` 导出到某个地方。
- **定价数据需要手动刷新.** 捆绑的价格表在30天后过期。未知模型记录为 `cost_usd: null` 而不是虚构的数字。
- **工具归因是部分的.** 代理按回合计算 `tool_use` 块，但尚未将成本归因于 *哪个* 工具。
- **默认失败开放.** 如果守护进程关闭，代理保持您的IDE工作，但静默跳过记录。传递 `fail="closed"` 到 `TrackedClient` 进行硬失败。
---
## 开发
```bash
git clone https://github.com/abhinandang90/ai-tokenpulse
cd ai-tokenpulse
pip install -e .
python -m unittest discover -s tests
```
测试布局：
- `tests/test_smoke.py` — 守护进程 + SDK包装器端到端
- `tests/test_mcp.py` — MCP stdio适配器端到端（启动 `ai-tokenpulse-mcp`）
- `tests/test_proxy.py` — 代理与存根上游（缓冲 + SSE流 + 直通）
- `tests/test_analytics.py` — 纯统计助手和第4阶段派生信号
---
## License
MIT — 参见 [LICENSE](LICENSE)。
