FluxMeter

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.4 aligns the first trial, API contract, deployment examples, and capacity claims. Flink projection idempotency stays bounded to its crash-safety horizon rather than 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.4 · Open spec + HTTP SDKs · Synchronous budget admission · Multi-provider

Links: Website · GitHub · PyPI · Docs · API reference · OpenAPI

Who is this for

• 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

Use checks and reservations to reduce overspend risk before requests, then reconcile actual usage through Flink. A check alone creates no hold: concurrent requests, stale fallback data, and estimated streaming tokens mean this is not a zero-overspend guarantee.

Repository layout (OpenCore)

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.

Budget Enforcement (the core feature)

Set a prepaid balance and check admission before a request. Check latency depends on load, Redis, and network conditions; measure it in your deployment:

# 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	Synchronous; deployment-dependent	GET /budget/{id}/check — admission decision only; the caller must honor denial; no funds are held
Reservation	Synchronous Redis operation	POST /budget/{id}/reserve holds an estimated amount before upstream work
Streaming guard	Per chunk	May stop a stream using estimated tokens when provider usage is unavailable
Final settlement	Normally ~10–15s; longer under lag	Flink prices actual reported usage, deducts balance, and reconciles Gateway reservations

The check reads Redis first. On Redis failure it may use a recent in-process cache entry, then applies the configured fail policy when no usable cache remains. A cached decision can be stale, including when the configured policy is closed. Stream guards use character-based estimates if provider token usage is missing; final billed usage and the estimate can differ.

Quick Start

Start the only supported architecture:

git clone https://github.com/10kshuaizhang/fluxmeter.git
cd fluxmeter
make demo          # start stack, wait for /ready, ingest and verify customer usage
make demo-verify   # repeat the usage proof on the running stack; new IDs each run
# Optional deeper control + audit proof (starts the benchmark overlay):
make demo-proof

make demo requires Java 17, Docker Compose, and Python 3.10+. It fails unless the public API shows one unique event, 1,200 input + 400 output tokens, and $0.00042 at the default gpt-4o-mini catalog rates. It polls readiness and projections with a bounded timeout; 202 alone is not settlement. No model-provider key is required. See the complete local trial and troubleshooting.

make 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.tape

See 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

Integration

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=False)
try:
    client.chat.completions.create(model="gpt-4o-mini", messages=[...])
except BudgetExceededError:
    ...  # never hit the provider

The wrapper logs a warning if post-call metering fails; use on_metering_error to route failures to your recovery/alerting path. fail_open controls admission failures, not metering delivery. See the Python SDK.

JavaScript 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.

Query API

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 (no hold; 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

Customer billing queries

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}

Architecture

[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
• Budget check: Redis → recent cache on failure → configured fail policy
• Auditable ClickHouse cold store (ADR-025) on make start-benchmark — not billing truth

Event Schema

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",
  "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.

Durability

Custody, replay and recovery protections have configuration and retention boundaries. They are not an unconditional no-data-loss guarantee:

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

Performance

FluxMeter measures the public HTTP custody boundary separately from the trusted internal Kafka/Flink generator. See docs/load-testing.md.

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.

Integrations

Connect FluxMeter to your billing platform: docs/integrations.md

• Stripe, Lago, OpenMeter, Orb, Metronome, Zuora

SDK publishing: docs/pypi-release.md

Production Deployment

Kubernetes + RocksDB + S3 checkpoints: docs/production-deploy.md

Helm chart: deploy/helm/README.md

There is no validated 100K sustained-capacity or monthly hosting-cost promise. Size from representative end-to-end measurements, retry-identity retention, and current provider pricing.

Makefile

make demo        # Build, start, and verify customer tokens and cost
make demo-verify # Repeat the no-provider-key usage proof
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 + clean

What's next

See 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/client on 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

Requirements

• Docker & Docker Compose
• Java 17 (building the engine)
• Python 3.10+ (local API/tests); container image uses Python 3.11

License

Apache 2.0