ai-tokenpulse 0.1.0

Real-time token economics observer for agentic AI systems — exposed over MCP for Claude Code, Cursor, and any MCP-compatible client.

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

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):

{
  "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):

ai-tokenpulsed --proxy-port 9000 --project my-project

3. Open a session and route your client's traffic through the proxy:

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:

ai-tokenpulse session end

---

Quickstart — Custom Anthropic SDK agents

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:

{
  "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

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.