tokenops

Attribute LLM spend to your customers with one line.

TokenOps records what each LLM call cost and which of your customers it was for, so you can see cost-vs-revenue per customer. This is the Python client.

Install

pip install lovie-tokenops

Requires Python 3.9+. No runtime dependencies (uses the standard library).

Quick start — auto-instrument Anthropic

import os
from tokenops import TokenOps, wrap_anthropic
from anthropic import Anthropic

tokenops = TokenOps(
    company_id="<your-company-uuid>",
    secret_key=os.environ["TOKENOPS_SECRET_KEY"],  # sk_live_...
    ingest_url="https://api.lovie.co",
)

# One line. Every non-streaming call is now attributed.
anthropic = wrap_anthropic(Anthropic(), tokenops)

anthropic.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "hi"}],
    customer_id="acme-inc",  # <- attributes the spend; stripped before the real call
)

wrap_anthropic tracks on a background worker: it never blocks or fails your LLM call. Pass on_error to observe background tracking failures. Both sync (Anthropic) and async (AsyncAnthropic) messages.create calls are instrumented; streaming calls (stream=True) and messages.stream() are passed through untracked. Wrapping the same client twice is a no-op — it never double-tracks.

Quick start — auto-instrument OpenAI

import os
from tokenops import TokenOps, wrap_openai
from openai import OpenAI

tokenops = TokenOps(
    company_id="<your-company-uuid>",
    secret_key=os.environ["TOKENOPS_SECRET_KEY"],  # sk_live_...
    ingest_url="https://api.lovie.co",
)

# One line. Every non-streaming call is now attributed.
openai = wrap_openai(OpenAI(), tokenops)

openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "hi"}],
    customer_id="acme-inc",  # <- attributes the spend; stripped before the real call
)

wrap_openai instruments chat.completions.create and, when present, responses.create (a no-op on older SDKs without it). It tracks on a background worker: it never blocks or fails your LLM call. Pass on_error to observe background tracking failures. Both sync (OpenAI) and async (AsyncOpenAI) calls are instrumented; streaming calls (stream=True) are passed through untracked. Wrapping the same client twice is a no-op — it never double-tracks.

Flushing before exit

Tracking runs on a background worker, so a short-lived script or serverless invocation can exit before in-flight events are sent. Events are drained automatically on normal interpreter exit, but for serverless or before a hard exit, flush explicitly:

tokenops.flush()  # block until queued events are sent
tokenops.close()  # flush, then stop the background worker

Track events directly

from tokenops import TokenOpsEvent

tokenops.track(TokenOpsEvent(
    vendor="openai",
    model="gpt-4o",
    input_tokens=1200,
    output_tokens=350,
    customer_id="acme-inc",
    provider_observation_id="resp_123",  # idempotent de-dup on re-send
))

tokenops.track_batch([event1, event2])  # batches of >500 are split automatically

provider_observation_id is optional: set it to the provider's response id to get idempotent de-duplication on re-send. When omitted it is auto-generated so the event is always accepted.

Failed sends retry with exponential backoff on network errors, 429, and 5xx; a 4xx (e.g. a bad key) raises TokenOpsError immediately.

License

MIT