kostlens

Track every AI call your app makes — tokens, cost, latency, and the tags that matter (feature, end user) — with wrappers designed so they can't break your app.

Part of KostLens, the financial layer for your AI usage.

Install

pip install kostlens

Quickstart (2 lines)

from openai import OpenAI
from kostlens import KostLens, wrap_openai

kl = KostLens(os.environ["KOSTLENS_KEY"])
client = wrap_openai(OpenAI(), kl, tags={"feature": "chat"})

# Use `client` exactly as before. Streaming included.

Also available: wrap_anthropic and wrap_google_genai. Any OpenAI-compatible API (Groq, OpenRouter, DeepSeek, Kimi…) works through wrap_openai — pass provider="deepseek" so pricing uses the honest vendor label.

Async clients (AsyncOpenAI, AsyncAnthropic) are supported too — wrap them the same way and await as usual.

Tag what matters

client = wrap_openai(OpenAI(), kl, tags={
    "feature": "summarize",
    "endUserId": user.id,   # camelCase — reserved key
})

Tag keys travel verbatim: use endUserId, not end_user_id. A snake_case key is stored as a plain custom tag and the Customers page stays empty.

endUserId powers cost-per-customer; feature powers cost-per-feature.

The sacred rule

The SDK is designed so it can never break the host app. track() is non-blocking (the HTTP call happens on a background daemon thread), network failures and KostLens outages degrade to "some events are lost" — never to an exception in your request path. Events are batched, retried with backoff on transient failures, and dropped (with an on_error callback) when permanently rejected.

Options

KostLens(
    "kl_live_…",
    "https://ingest.kostlens.com",  # default; override for self-hosted
    flush_interval=5.0,
    max_batch_size=100,
    max_buffer_size=5000,
    on_error=lambda err: logging.warning("kostlens: %s", err),
)

Call kl.shutdown() before a short-lived process exits to flush what is buffered (it is also registered with atexit).

Docs

Full documentation: kostlens.com/docs

MIT © KostLens