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

# inside your request handler, where you know the customer
client = wrap_openai(OpenAI(), kl, tags={
    "feature": "summarize",
    "endUserId": user.id,   # camelCase, a 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.

Tags belong to the client you wrap. feature rarely changes, so wrapping once per module is fine. endUserId changes per request, so that wrap goes in the handler.

Give each set of tags its own client. Wrapping replaces client.chat.completions.create in place and marks the client, so handing the same client over again returns it untouched and every call keeps the tags it started with. The mark is what stops a second wrap from layering over the first and reporting every call once per layer. It also means one shared module-level client cannot carry a different endUserId per request: construct the client inside the handler.

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