Website: fluxmeter.dev · Docs: fluxmeter.dev/docs · Blog: Agent cost control
Open-source, self-hostable real-time AI token metering and budget enforcement. Call GET /budget/{id}/check before every LLM request. All public usage events enter through HTTP; 202 means Kafka acknowledged the event and FluxMeter finalized its tenant-scoped retry identity before Flink performs billing and aggregation. v4.8.3 keeps Flink projection idempotency bounded to its crash-safety horizon instead of retaining a second 30-day event registry.
When to use FluxMeter: prepaid token wallets, agent loop cost control, self-hosted LLM metering, export to Stripe/Lago/Orb/Metronome.
fluxmeter.dev — overview, quick start, architecture · v4.8.3 · Open spec + HTTP SDKs · <10ms budget check · Multi-provider
Links: Website · GitHub · PyPI · Docs · API reference · OpenAPI
- AI app builders shipping LLM wrappers, agent platforms, or code assistants that bill per token
- Platform teams that need real-time cost visibility across OpenAI, Anthropic, and Google models
- Anyone who's been burned by a runaway agent loop spending $500 in 30 seconds before the billing system noticed
If your customers prepay for tokens and you need to cut them off the instant they run out — not 30 seconds later — FluxMeter does that.
| Layer | Path | Purpose |
|---|---|---|
| Spec | spec/ |
Event schema, OpenAPI, semantic conventions |
| SDKs | sdk/python/, sdk/js/ |
Python + JS clients |
| Community | contrib/ |
Provider mappings, pricing, connectors |
| Engine | src/ |
Flink reference implementation (aggregation, budget enforcement) |
| Demo | make demo |
HTTP→Kafka→Flink→Redis |
The runtime is organized around four deep modules: Custody accepts retry-safe events, Pricing validates and quotes token categories, Reservation owns temporary holds, and Budget authorizes spend. HTTP routes, the Gateway, Redis, and Kafka implementations adapt to those interfaces rather than duplicating their state machines. See ADR-026.
Set a prepaid balance. FluxMeter enforces it in <10ms per request:
# Set $50 budget, alert at $5 remaining, max 100 requests/minute
curl -X POST localhost:8000/budget/cust_123 \
-H 'Content-Type: application/json' \
-d '{"balance_usd": 50.0, "alert_threshold_usd": 5.0, "max_rpm": 100}'
# Pre-request check — call this BEFORE every LLM request
curl "localhost:8000/budget/cust_123/check?estimated_cost_usd=0.05"
# → {"allowed": true, "balance_usd": 47.23, "held_usd": 0.0, "effective_balance_usd": 47.23, ...}
# → {"allowed": false, "reason": "budget_exhausted", "source": "redis"}
# → {"allowed": false, "reason": "rate_limited", "max_rpm": 100}Two-layer enforcement:
| Layer | Latency | What it does |
|---|---|---|
| Pre-request check | <10ms | GET /budget/{id}/check — blocks request before tokens are burned |
| Post-window deduction | 10-15s | Flink aggregates → atomic Lua deduction → Kafka kill signal |
The pre-request check uses a three-layer resilience stack (in-process cache → Redis → configurable fail policy) so it never blocks your agent's hot path, even during Redis outages.
Start the only supported architecture:
git clone https://github.com/10kshuaizhang/fluxmeter.git
cd fluxmeter
make demo # API + Gateway + Kafka + Flink + Redis + Grafana
make demo-proof # deterministic reserve → meter → kill → audit; no provider keymake demo-proof starts the benchmark audit overlay, serves a local OpenAI-compatible stream, and fails unless it observes the temporary hold, metered token/cost receipt, mid-stream termination, Flink settlement, and matching ClickHouse raw event.
Record the terminal GIF (optional, requires vhs):
make demo && make demo-record # writes demo.gif from demo.tapeSee also: docs/intelligence-api.md · docs/gateway.md
Starts Kafka, Flink, Redis, and the API. Open:
- API docs: http://localhost:8000/docs
- Flink UI: http://localhost:8081
- Grafana: http://localhost:3000
Python SDK (HTTP, retry-safe event IDs):
from fluxmeter import FluxMeter
meter = FluxMeter(api_url="http://localhost:8000")
meter.track_openai("cust_123", openai_response, latency_ms=1200)Wrap (path activation — check before every call):
from openai import OpenAI
from fluxmeter import FluxMeter, wrap, BudgetExceededError
meter = FluxMeter(api_url="http://localhost:8000")
client = wrap(OpenAI(), meter, customer_id="cust_123", fail_open=True)
try:
client.chat.completions.create(model="gpt-4o-mini", messages=[...])
except BudgetExceededError:
... # never hit the providerJavaScript SDK (HTTP):
import { FluxMeter } from "@fluxmeter/client";
const meter = new FluxMeter({ apiUrl: "http://localhost:8000" });
await meter.trackOpenAI("cust_123", openaiResponse);HTTP API (zero dependencies — any language, curl, serverless):
curl -X POST localhost:8000/ingest \
-H 'Content-Type: application/json' \
-d '{"eventId":"completion-01J...","customerId":"cust_123","modelId":"gpt-4o","inputTokens":500,"outputTokens":150}'Supply a stable eventId for retry-safe delivery. If it is omitted, the API creates an ID for that request, but a later retry cannot be recognized as the same event. The SDKs and Gateway always supply stable IDs.
Kafka is an internal transport. Customer SDKs do not accept broker configuration, and the base deployment does not expose a broker port. The benchmark overlay retains a trusted operator producer for load and recovery tooling. On Kafka outage: HTTP /ingest returns retryable 503; the Gateway may buffer via Redis outbox until Kafka recovers.
| Endpoint | Description |
|---|---|
GET /usage/global |
Total events, tokens, cost |
GET /usage/customer/{id} |
Per-customer breakdown |
GET /usage/customer/{id}/period/{YYYY-MM} |
Monthly usage (UTC calendar) |
GET /usage/customer/{id}/day/{YYYY-MM-DD} |
Daily usage |
GET /usage/session/{id} |
Session/project aggregated cost |
GET /usage/customer/{id}/model/{model} |
Per-model detail |
GET /usage/span/{id} |
Agent span cost (total cost of an agent run) |
GET /budget/{id}/check |
Pre-request allow/deny (<10ms, uses balance - held) |
POST /budget/{id} |
Set balance + threshold + rate limit |
POST /budget/{id}/topup |
Add credits |
POST /budget/{id}/reserve |
Hold estimate for streaming (does not deduct balance) |
POST /budget/{id}/reconcile |
Release hold after stream ends |
POST /budget/{id}/webhook |
Configure HTTPS alerts (EXHAUSTED / LOW) |
GET /pricing |
Current pricing catalog |
POST /admin/customers/{id}/api-keys |
Create customer-scoped API key |
GET /admin/reconciliation |
Balance drift snapshot |
POST /ingest |
HTTP event ingest |
POST /ingest/batch |
Batch ingest (up to 1000) |
POST /rerate/preview |
Preview price change impact |
POST /rerate/apply |
Apply retroactive re-rating |
Full reference: docs/api-reference.md
Expose usage to end users without a separate warehouse:
# Token reseller: monthly / daily spend
curl localhost:8000/usage/customer/cust_123/period/2026-07
curl localhost:8000/usage/customer/cust_123/day/2026-07-05
# Agent platform: cost of one run (set parentSpanId on every child LLM call)
curl localhost:8000/usage/span/span_agent_42
# Multi-turn project (set sessionId on each event)
curl localhost:8000/usage/session/sess_456| Use case | Field on ingest | Query |
|---|---|---|
| Per-model lifetime | — | GET /usage/customer/{id}/model/{model} |
| Monthly invoice | — | GET /usage/customer/{id}/period/{YYYY-MM} |
| Today's spend | — | GET /usage/customer/{id}/day/{YYYY-MM-DD} |
| One agent task | parentSpanId |
GET /usage/span/{id} |
| Conversation / project | sessionId |
GET /usage/session/{id} |
[Your App / SDK / Gateway] → [HTTP API] → [Kafka] → [Flink] → [Redis]
│ │
budget alerts Query API
│
ClickHouse cold store (benchmark overlay)
Key design choices:
- Incremental aggregation — O(keys) memory, not O(events)
- Atomic budget deduction via Redis Lua script
- Microdollar precision (long) — no float accumulation errors
- Sink idempotency (SHA-256 + SET NX) — no double-billing on replay
- Three-layer budget check (cache → Redis → fail policy) — never blocks
- Auditable ClickHouse cold store (ADR-025) on
make start-benchmark— not billing truth
Each event = one LLM API call:
{
"customerId": "cust_123",
"modelId": "gpt-4o",
"provider": "openai",
"inputTokens": 1250,
"outputTokens": 847,
"cacheReadTokens": 200,
"reasoningTokens": 0,
"parentSpanId": "span_agent_42",
"sessionId": "sess_123",
"timestamp": 1718534400000,
"latencyMs": 1340
}Providers: OpenAI (gpt-4o, gpt-4o-mini, o1, o3-mini), Anthropic (claude-opus-4, claude-sonnet-4, claude-haiku-4), Google (gemini-1.5-pro, gemini-1.5-flash)
Token categories: input, output, cached, reasoning, embedding — each priced independently.
No single-component failure loses billing data:
| Failure | Protection |
|---|---|
| Kafka down | HTTP returns retryable 503; Gateway retains a Redis outbox record |
| Broker crash / ACK timeout | HTTP returns 503 custody_uncertain; a late delivery callback finalizes or releases the identity |
| Flink restart | Checkpoints restore state + offsets exactly |
| Flink replay | Sink idempotency (SET NX) prevents double-counting |
| Redis restart | AOF persistence + named volume |
| Duplicate events | Compact tenant-sharded HTTP identity registry (30-day retry window) plus idempotent sinks; Flink keeps only a 10-minute crash-window safety dedup |
| Late events | Routed to DLQ topic, not silently dropped |
FluxMeter measures the public HTTP custody boundary separately from the trusted internal Kafka/Flink generator. See docs/load-testing.md.
| Environment | 10K eps | 50K eps | 500K+ target |
|---|---|---|---|
| Local docker-compose (1 TM, 4 slots) | ~9K avg / ~18K peak | ~49K avg / ~92K peak | ~40–45K avg (Redis/Flink bound) |
| Historical internal engine runs (not HTTP ingress) | Stable | Stable | 500K indefinite; 1M bursts |
The public gates are 10K single-event eps (p50 ≤50ms, p99 ≤200ms; 25/100ms stretch) and 100K batch-event eps (1,000 items, p99 ≤500ms), each after 5 minutes warmup for a 30-minute measurement. They are not yet passed as full-pipeline sustained claims. A v4.8.2 35-minute Custody run accepted 18,063,140 events at 10,034.90 eps with p50 36ms / p99 173ms and no rejection or transport error, but the 2,000-slot generator dropped 0.149% of offers and the co-located Flink/Redis pipeline accumulated lag. Correcting benchmark Flink parallelism from 2 to 12 raised the full-stack 60-second sample to 7,772 eps with near-zero end lag; a split-Redis A/B peaked at 9,197 eps but missed latency. The 100K batch stage reached 32,567 eps. Historical 1M figures are internal burst benchmarks, not public HTTP or a current sustained claim.
Connect FluxMeter to your billing platform: docs/integrations.md
- Stripe, Lago, OpenMeter, Orb, Metronome, Zuora
SDK publishing: docs/pypi-release.md
Kubernetes + RocksDB + S3 checkpoints: docs/production-deploy.md
Helm chart: deploy/helm/README.md
Estimated cost: ~$1,550/month for 100K events/sec on AWS.
make demo # Build and start the only architecture
make demo-record # Re-record demo.gif (requires vhs)
make demo-gateway # Gateway mock self-check only
make start # Start API, Gateway, Kafka, Flink, Redis, workers, Grafana
make start-benchmark # Scale overlay and expose Kafka for trusted operator tools
make generate # Trusted internal load generator (benchmark overlay)
make load-test # Staged load test 10K→1M
make load-test-quick # Staged 10K→500K
make test-e2e # Integration + v2 E2E tests
make test-unit # Python + Java unit tests (no Docker)
make test-java # Java unit tests only
make benchmark # Streaming vs batch comparison
make apply-cold-store-init # ClickHouse cold-store DDL (benchmark overlay)
make test-cold-store # ADR-025 acceptance A1–A7 (kafka + clickhouse)
make validate-spec # Validate schema + OpenAPI artifacts
make stop # Stop containers
make clean # Stop + remove volumes + cleanSee ROADMAP.md for the full plan. Highlights:
- Tiered pricing (flat / volume / graduated) in Flink — see
contrib/pricing/tiered-example.json - Full multi-tenant RBAC / org model
- Wrap SDK + mid-stream kill (
wrap(OpenAI()),StreamKilledError) + Gateway proxy (/v1/chat/completions) -
@fluxmeter/clienton npm - Webhook delivery for budget alerts
- Customer-scoped API keys
- Single public HTTP entrance backed by Kafka/Flink (v4.0+)
- Auditable ClickHouse cold store (v4.4.0)
- Python and JavaScript HTTP SDKs
- Docker & Docker Compose
- Java 17 (building the engine)
- Python 3.9+ (SDK and API)
Apache 2.0