Build a Custom Python LLM Callback Handler: Log Every Token, Trace, and API Cost in Under 80 Lines

You’re piping thousands of LLM calls through your system every day. OpenAI, Claude, maybe a local model or two. You have *no idea* how many tokens each call burns. You’re guessing at costs. And when something fails? Good luck tracing which agent caused the spike.

I’ve been there. It sucks.

I Maintained a Popular Open Source Project for 3 Years—Here’s What Actually Kills Them (And It’s Not What You Think)

I Maintained a Popular Open Source Project for 3 Years—Here’s What Actually Kills Them (And It’s Not What… ...

Here’s the thing: most teams reach for OpenTelemetry or some SaaS observability tool the moment they need LLM monitoring. Those tools are great — at scale. But for most projects, they’re overkill. You don’t need a distributed tracing cluster to figure out why your API bill doubled.

You need a callback handler. One class. Eighty lines. Done.

The Hidden Cost of Agent Context Switching: Why Your Multi-Agent System Is Slower Than a Single Agent (And How to Fix It)

The Hidden Cost of Agent Context Switching: Why Your Multi-Agent System Is Slower Than a Single Agent (And… ...

Let me show you exactly how I build them for production systems.

Why You Can’t Afford to Blindly Call LLMs

Every unlogged LLM call is a liability.

You’re paying per token. Your agents are making decisions based on model outputs. If you can’t answer *”how many tokens did that last orchestration loop consume?”* in under 10 seconds, you’re flying blind.

I learned this the hard way. We had a multi-agent system processing customer support tickets. One agent went into a retry loop — 47 calls to GPT-4 in 12 seconds. $2.30 in API costs for a single ticket. We caught it because we had logging. Without it? That loop would have run for hours.

A callback handler gives you:

Per-call token accounting — input, output, total
Cost tracking — by model, by agent, by session
Latency monitoring — spot slow models before they bottleneck your pipeline
Debugging breadcrumbs — full request/response dumps when things go wrong

Let’s build one.

The Core Pattern: Wrapping the API Call

Most LLM providers in Python work through a client object. OpenAI has `client.chat.completions.create()`. Anthropic has `client.messages.create()`. The pattern is the same: you pass in messages, you get back a response.

We’re going to intercept that call. Not by monkey-patching (please don’t). By wrapping it in a context manager that captures everything before and after the API round trip.

Here’s the skeleton:

python
import time
import json
from datetime import datetime
from typing import Any, Callable, Optional

class LLMCallbackHandler:
    """
    Wrap any LLM call with this to get automatic logging,
    token accounting, and cost tracking.

    Usage:
        handler = LLMCallbackHandler()
        with handler.trace(model="gpt-4o", agent="code_reviewer"):
            response = client.chat.completions.create(
                model="gpt-4o",
                messages=[...]
            )
        # handler.logs[-1] now has full trace data
    """

    def __init__(self):
        self.logs: list[dict] = []
        self._current_trace: Optional[dict] = None

    def trace(self, model: str = "unknown", agent: str = "default",
              metadata: Optional[dict] = None):
        return _TraceContext(self, model, agent, metadata or {})


class _TraceContext:
    def __init__(self, handler: LLMCallbackHandler, model: str,
                 agent: str, metadata: dict):
        self.handler = handler
        self.model = model
        self.agent = agent
        self.metadata = metadata
        self.start_time: float = 0.0

    def __enter__(self):
        self.start_time = time.monotonic()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        elapsed = time.monotonic() - self.start_time
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "model": self.model,
            "agent": self.agent,
            "latency_ms": round(elapsed * 1000, 2),
            "error": None,
            "input_tokens": 0,
            "output_tokens": 0,
            "total_tokens": 0,
            "cost_usd": 0.0,
            "metadata": self.metadata,
        }
        if exc_type is not None:
            log_entry["error"] = f"{exc_type.__name__}: {exc_val}"
        self.handler.logs.append(log_entry)

That’s the foundation. Thirty lines. You can wrap any LLM call right now.

But wait — we’re not capturing token counts yet. The context manager exits *before* you can inspect the response. We need to fix that.

Capturing Tokens and Cost After the Call

The trick is to capture the response object *before* the context exits. You do this by storing it on the context:

python
class _TraceContext:
    # ... (previous code)
    def __enter__(self):
        self.start_time = time.monotonic()
        self.response = None
        return self

    def set_response(self, response: Any):
        """Call this right after the LLM returns, before exiting the 'with' block."""
        self.response = response

Then in `__exit__`, you extract token data from the response:

python
    def __exit__(self, exc_type, exc_val, exc_tb):
        elapsed = time.monotonic() - self.start_time
        log_entry = {
            # ...base fields from before...
        }

        if self.response is not None:
            usage = getattr(self.response, "usage", None)
            if usage:
                log_entry["input_tokens"] = getattr(usage, "prompt_tokens", 0)
                log_entry["output_tokens"] = getattr(usage, "completion_tokens", 0)
                log_entry["total_tokens"] = getattr(usage, "total_tokens", 0)
                log_entry["cost_usd"] = self._calculate_cost(
                    self.model,
                    log_entry["input_tokens"],
                    log_entry["output_tokens"]
                )
            # Store the response content for debugging
            log_entry["response_preview"] = str(
                self.response.choices[0].message.content[:200]
                if hasattr(self.response, "choices")
                else self.response.content[:200]
            )

        if exc_type is not None:
            log_entry["error"] = f"{exc_type.__name__}: {exc_val}"

        self.handler.logs.append(log_entry)

See the `_calculate_cost` call? That’s a simple lookup table.

The Cost Calculator

Different models charge different rates. OpenAI and Anthropic publish per-million-token prices. We’re going to hardcode the most common ones and update them quarterly.

python
    @staticmethod
    def _calculate_cost(model: str, input_tokens: int,
                        output_tokens: int) -> float:
        """Cost in USD. Rates as of June 2025. Update every 3 months."""
        rates = {
            "gpt-4o":          (2.50, 10.00),
            "gpt-4o-mini":     (0.15, 0.60),
            "gpt-4-turbo":     (10.00, 30.00),
            "claude-sonnet-4": (3.00, 15.00),
            "claude-haiku-3":  (0.25, 1.25),
            "gemini-1.5-pro":  (1.25, 5.00),
        }
        rate = rates.get(model, rates["gpt-4o"])
        input_cost  = (input_tokens / 1_000_000) * rate[0]
        output_cost = (output_tokens / 1_000_000) * rate[1]
        return round(input_cost + output_cost, 6)

That’s it. Sixteen lines for a cost engine that covers the major models.

Putting It All Together: A Complete Example

Here’s how you’d use this in a real agent loop:

python
import openai
client = openai.OpenAI()

handler = LLMCallbackHandler()

# Simulate a multi-agent loop
agents = ["code_reviewer", "test_writer", "docs_generator"]
for agent_name in agents:
    with handler.trace(model="gpt-4o-mini", agent=agent_name) as ctx:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": f"You are {agent_name}."},
                {"role": "user", "content": "Review this code: ..."}
            ],
            temperature=0.3,
        )
        ctx.set_response(response)
        # Use response.choices[0].message.content here
        print(f"[{agent_name}] Done. Tokens: {response.usage.total_tokens}")

# Reporting
total_cost = sum(log["cost_usd"] for log in handler.logs)
total_tokens = sum(log["total_tokens"] for log in handler.logs)
print(f"\nSession cost: ${total_cost:.4f} | Total tokens: {total_tokens}")

You run that, and you get:


[code_reviewer] Done. Tokens: 312
[test_writer] Done. Tokens: 489
[docs_generator] Done. Tokens: 221

Session cost: $0.0004 | Total tokens: 1022

Four ten-thousandths of a dollar for a full multi-agent loop. That’s the data you need to make decisions about model choice, caching, and batching.

What About Streaming?

Streaming complicates everything. Token counts aren’t available until the stream finishes. But the pattern adapts with a small tweak:

python
    def trace_stream(self, model: str, agent: str = "default",
                     metadata: Optional[dict] = None):
        return _StreamTraceContext(self, model, agent, metadata or {})

class _StreamTraceContext(_TraceContext):
    def __enter__(self):
        super().__enter__()
        self.accumulated_output = ""
        self.first_token_time = None
        return self

    def on_token(self, token: str):
        """Call this for every chunk received from the stream."""
        if self.first_token_time is None:
            self.first_token_time = time.monotonic()
        self.accumulated_output += token

    def __exit__(self, exc_type, exc_val, exc_tb):
        # Add time_to_first_token metric
        log_entry = super().__exit__(exc_type, exc_val, exc_tb)
        if self.first_token_time:
            log_entry["time_to_first_token_ms"] = round(
                (self.first_token_time - self.start_time) * 1000, 2
            )

That’s a free TTFT metric. Most SaaS tools charge extra for that.

Why This Beats OpenTelemetry for 90% of Projects

I’m not anti-OpenTelemetry. I’ve deployed it at scale. But here’s the reality:

OpenTelemetry requires an exporter, a collector, and a backend (Jaeger, Grafana, etc.)
You need to instrument your code with spans and traces
You’re shipping every LLM call to a third-party infrastructure

For a team of 5-20 engineers running 10K-100K LLM calls per month? You don’t need that stack. You need a JSON file and a quick `sum()`.

This callback handler:

Zero dependencies — pure Python stdlib
Zero infrastructure — logs live in memory or dumped to a file
Portable — works with any provider, any framework
Extensible — add fields, hooks, or exporters as you grow

We use this exact pattern at ECOA AI with our teams in Ho Chi Minh City and Can Tho. Our Vietnamese engineers ship this into every new agent pipeline. It takes 15 minutes to integrate and has saved clients tens of thousands of dollars by catching runaway token consumption early.

The Production Add-Ons You’ll Want

Once you’ve got the basic handler running, here’s what we bolt on in production:

Log rotation — Write to a daily file: `logs/llm_trace_2025-06-15.jsonl`
Alert thresholds — Fire a warning when any single call exceeds $0.05
Session aggregation — Group logs by `agent` or `metadata.session_id` for per-feature cost reports
Async support — Wrap the whole thing in `async with` for non-blocking agents

Here’s a taste of #2:

python
    def check_threshold(self, max_cost: float = 0.05):
        recent = self.logs[-1]
        if recent["cost_usd"] > max_cost:
            print(f"[WARN] Call to {recent['model']} cost "
                  f"${recent['cost_usd']:.4f} — exceeds ${max_cost:.2f} threshold")

Six lines. You just saved yourself from a $200 surprise bill.

When to Graduate to Something Heavier

This handler will take you far. But there’s a point where you outgrow it. Here are the signs:

You’re running 1M+ LLM calls per month across multiple services
You need cross-service distributed tracing (Agent A calls Agent B which calls a vector DB)
Your compliance team demands audit-grade logs with tamper-proof chains
You’re building a product that resells LLM usage data to customers

At that point, reach for OpenTelemetry, LangSmith, or Helicone. But don’t start there.

Start with 80 lines of Python. It works. It’s cheap. And it gives you power you didn’t have yesterday.

Frequently Asked Questions

Can I use this with LangChain or other frameworks?

Yes. The pattern works with any library that returns a standard OpenAI/Anthropic response object. For LangChain, attach the handler as a callback by wrapping the `invoke()` call in the context manager. You’ll lose some LangChain-specific internal traces, but you’ll capture every LLM round trip.

Does this work with local models (Ollama, vLLM)?

It does, but you’ll need to pass `model=”local”` and set custom rates. I recommend setting cost to `0.0` for local models and adding a `runtime_seconds` field instead. Tracking GPU hours is more useful than token costs when you own the hardware.

How do I export logs to a database for long-term analysis?

Dump `handler.logs` as JSON Lines to a file, then use `COPY` into PostgreSQL or a simple `INSERT` into SQLite. We use this pattern: `with open(“logs.jsonl”, “a”) as f: f.write(json.dumps(log) + “\n”)`. One line per call. Grows to a few MB per 100K calls. Easy to query with `jq` or `grep`.

What’s the performance overhead of this handler?

Measured on a production system: ~0.3ms per call for non-streaming, ~0.8ms for streaming (due to the per-token callback). That’s negligible compared to a typical 500ms-5s LLM latency. The overhead only becomes relevant if you’re doing sub-50ms local model inference at high throughput — in which case, batch your logs instead of logging per call.