ai-tokenpulse

# ai-tokenpulse Real-time token economics and context-budget observer for agentic AI systems. Watches every Claude API call your agents make and tells you — live, in your editor — how much you've spent, how full the context window is, and whether the agent is wasting tokens. Exposed over **MCP** so it works natively in Claude Code, Cursor, and any MCP-compatible client. > Most agent failures aren't reasoning failures — they're budget failures dressed up as reasoning failures. Context windows silently saturate; long autonomous loops quietly burn credits; cost overruns surface only at billing time. `ai-tokenpulse` makes the resource layer observable while it's happening. --- ## Install ```bash pip install ai-tokenpulse ``` This gives you three console scripts: `ai-tokenpulsed` (the daemon), `ai-tokenpulse` (CLI), and `ai-tokenpulse-mcp` (the MCP stdio adapter). --- ## How it works ``` Claude Code / Cursor Custom SDK agents │ │ │ stdio│ https│ │ unix socket (MCP)│ (proxy)│ │ (ingest) ▼ ▼ ▼ ┌──────────┐ ┌──────────────────────────────────────┐ │ MCP │ │ ai-tokenpulsed (daemon) │ │ adapter │◄──┤ proxy │ ingest API │ query API │ └──────────┘ │ │ │ │ ▼ │ │ pricing · anomaly · efficiency │ │ ▼ │ │ SQLite (WAL) + .ai-tokenpulse/ JSON exports │ └──────────────────────────────────────┘ ``` A single long-running daemon (`ai-tokenpulsed`) owns all state in a local SQLite database. Three thin adapters connect to it: - **HTTP proxy** — point `ANTHROPIC_BASE_URL` at it; captures every model call from Claude Code or Cursor - **MCP stdio adapter** — spawned by your IDE; lets you ask questions like *"how saturated are we?"* mid-session - **SDK wrapper** (`TrackedClient`) — for custom agents built directly on the Anthropic SDK --- ## Quickstart — Claude Code or Cursor **1. Add the MCP server** to your client config. Claude Code (`.mcp.json` in your project, or `~/.claude/mcp.json`): ```json { "mcpServers": { "ai-tokenpulse": { "command": "ai-tokenpulse-mcp" } } } ``` Cursor (`.cursor/mcp.json` or `~/.cursor/mcp.json`) uses the same shape. **2. Start the daemon** (leave it running): ```bash ai-tokenpulsed --proxy-port 9000 --project my-project ``` **3. Open a session and route your client's traffic through the proxy:** ```bash ai-tokenpulse session start "what-im-working-on" export ANTHROPIC_BASE_URL=http://127.0.0.1:9000 claude # or open Cursor ``` That's it. Every API call now records a step. Inside Claude Code or Cursor, ask questions naturally — *"what's our context saturation right now?"* — and the agent calls `get_context_saturation` over MCP and answers from the live data. When you're done: ```bash ai-tokenpulse session end ``` --- ## Quickstart — Custom Anthropic SDK agents ```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": "..."}], ) # step recorded automatically; response is unmodified ``` `TrackedClient` extracts `usage` from each response and writes a step to the daemon over its Unix socket. The caller gets back the original response object. Pass `fail="closed"` to make calls hard-fail if the daemon is unreachable; the default is `fail="open"` so observability can't break your IDE. --- ## CLI ``` ai-tokenpulse session start <name> open a new session ai-tokenpulse session end close the current session ai-tokenpulse session list recent sessions (active marked) ai-tokenpulse burn [session_id] totals for current (or named) session ai-tokenpulse health ping the daemon # Analytics (mirror the MCP tools, scriptable from a shell) ai-tokenpulse efficiency [session_id] composite efficiency score ai-tokenpulse saturation [session_id] current context window % used ai-tokenpulse models [session_id] per-model cost/token split ai-tokenpulse anomalies [session_id] [--z-threshold N] ai-tokenpulse compaction [session_id] should we compact now? ai-tokenpulse exhaustion [session_id] [--target-cost USD] ai-tokenpulse budget [session_id] --target-cost USD ai-tokenpulse compare <session_a> <session_b> ``` --- ## MCP tools (13) **Session state** - `get_session_burn(session_id?)` — totals: tokens, cost, step count, unpriced count - `get_context_saturation(session_id?)` — % of context window used (latest step) - `get_step_cost(step_id)` — itemized per-step cost - `list_sessions(limit?)` — recent sessions with active/closed flag **Attribution** - `get_tool_costs(session_id?)` — per-tool token + cost breakdown - `get_model_breakdown(session_id?)` — Sonnet vs Opus vs Haiku split **Tagging** - `tag_session(tag, session_id?)` — set or update human-readable tag **Analysis (Phase 4)** - `detect_burn_anomalies(z_threshold=2.0)` — steps with token totals deviating >Nσ from session mean - `suggest_compaction_point()` — fires when utilization > 70% AND projection > 90% in 3 steps - `project_exhaustion(target_cost_usd?)` — `steps_to_context_limit`, `steps_to_cost_limit`, growth rate - `estimate_remaining_budget(target_cost_usd)` — steps left at avg cost - `compare_sessions(session_a, session_b)` — diff with warning when tags differ - `get_efficiency_score()` — composite 0–1 score: burn-rate stability, output yield, decision density `session_id` defaults to `.ai-tokenpulse/current-session` (the session you opened with the CLI), so you rarely have to pass it explicitly. --- ## Data model Each recorded event is a **step** — one model API call: ```python { "schema_version": 2, "session_id": "sess_2026_04_27_abc", "session_tag": "refactor-auth", # optional, for like-for-like compare_sessions "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, priced at 0.1x "tool_use_count": 2, # tool_use blocks emitted in the response "context_window_size": 200000, "context_used": 45230, "cost_usd": 0.0871 # null when model isn't in price table } ``` Persistence is `.ai-tokenpulse/ai-tokenpulse.db` (SQLite, WAL mode) with optional JSON exports per session for human inspection. --- ## Privacy The proxy extracts only `usage`, `model`, `id`, and `tool_use` block counts from responses. **Request bodies, response message content, tool call inputs, and tool call outputs are never logged, persisted, or written to `.ai-tokenpulse`.** They live in proxy memory only for the duration of forwarding. --- ## Configuration | Flag / env var | Default | Purpose | |---|---|---| | `ai-tokenpulsed --proxy-port <N>` | (off) | Run the HTTP proxy on this port | | `ai-tokenpulsed --proxy-host <H>` | `127.0.0.1` | Proxy bind address (loopback only by default; see security note below) | | `ai-tokenpulsed --allow-public-bind` | (off) | Acknowledge non-loopback bind. Refused without this flag | | `ai-tokenpulsed --upstream <URL>` | `https://api.anthropic.com` | Where the proxy forwards | | `ai-tokenpulsed --project <name>` / `AI_TOKENPULSE_PROJECT` | `default` | Project label and **socket name suffix** — see below | | `ai-tokenpulsed --db <path>` | `./.ai-tokenpulse/ai-tokenpulse.db` | SQLite location | | `AI_TOKENPULSE_SESSION` | (unset) | Override active session for the SDK wrapper | | `X-Ai-Tokenpulse-Session` (HTTP header) | (unset) | Per-request session override for the proxy | ### Multiple projects The daemon's Unix socket is named after the project: `~/.ai-tokenpulse/<project>.sock`. To run more than one daemon on the same machine — e.g., one for each open project — give each a distinct `--project` (or shell-level `AI_TOKENPULSE_PROJECT`) and they will coexist. Clients (CLI, MCP adapter, SDK wrapper, proxy) resolve the right socket from the same env var, so a per-project shell environment is enough. ### Security: loopback bind enforced The proxy forwards `Authorization: Bearer …` and `x-api-key` headers transparently to upstream. Binding it to a non-loopback interface would let anyone reachable on that interface issue API calls billed to your account. The daemon refuses non-loopback `--proxy-host` values unless you pass `--allow-public-bind` to acknowledge the risk. Don't pass that flag on shared or untrusted networks. ### What "fail-open" really means The proxy is fail-open with respect to the **daemon**: if a step fails to record (SQLite hiccup, schema error, etc.), the user's API call is unaffected. The proxy is **not** fail-open with respect to **itself** — if `ai-tokenpulsed` crashes or isn't running, `ANTHROPIC_BASE_URL=http://localhost:9000` points at nothing and every IDE call will fail until the daemon is back up. There is no automatic fallback to `api.anthropic.com`. Run the daemon under a supervisor (launchd, systemd, tmux, etc.) if uptime matters. --- ## Limitations - **Linux and macOS only.** The daemon uses Unix domain sockets for the daemon ↔ adapter ↔ SDK-wrapper IPC, which Windows does not support. `pip install ai-tokenpulse` will succeed on Windows (the wheel is platform-agnostic) but the daemon and CLI will fail at runtime. WSL2 works. - **Anthropic only.** Other LLM providers would need separate work. - **Streaming SDK calls aren't yet captured** by `TrackedClient` (the proxy handles streams cleanly; the SDK wrapper warns and falls through if you pass `stream=True`). - **Heuristic efficiency thresholds.** The composite score is calibrated against rough defaults; you'll want to tune them with real session data. Raw component values are always exposed so you can see *why*, not just *what*. - **Local-only store.** No identity, no auth, no team rollups. Per-developer benchmarks across machines require shipping `.ai-tokenpulse/` exports somewhere. - **Pricing data needs manual refresh.** The bundled price table warns when older than 30 days. Unknown models record with `cost_usd: null` rather than a fabricated number. - **Tool attribution is partial.** The proxy counts `tool_use` blocks per turn but doesn't yet attribute cost back to *which* tool. - **Fail-open by default.** If the daemon is down, the proxy keeps your IDE working but silently skips recording. Pass `fail="closed"` to `TrackedClient` for hard failure. --- ## Development ```bash git clone https://github.com/abhinandang90/ai-tokenpulse cd ai-tokenpulse pip install -e . python -m unittest discover -s tests ``` Test layout: - `tests/test_smoke.py` — daemon + SDK wrapper end-to-end - `tests/test_mcp.py` — MCP stdio adapter end-to-end (spawns `ai-tokenpulse-mcp`) - `tests/test_proxy.py` — proxy with stub upstream (buffered + SSE streaming + pass-through) - `tests/test_analytics.py` — pure-stat helpers and Phase 4 derived signals --- ## License MIT — see [LICENSE](LICENSE).