How to Build a Token Metering and Budget Guard for AI APIs (Python Tutorial)

You shipped the demo in a weekend. The chatbot answers questions, the summaries look great, and then the first real month of traffic arrives — and so does an invoice that looks like a phone number. The gap between an AI demo and an AI product that survives contact with real usage is almost always cost engineering, not model quality.

This tutorial builds the exact component that closes that gap: a token metering and budget-guard layer for Python AI apps. It counts input, output, and reasoning tokens per request, attributes them to features and users, and kills runaway requests before they drain your quota. Everything runs against any OpenAI-compatible endpoint, including Qubax AI, where you can test across hundreds of models with one API key.

What We're Building

A production-grade metering layer with four responsibilities:

1. Count — record input/output/reasoning tokens for every call
2. Attribute — tag usage to a feature, user, and model so costs land on the right dashboard
3. Budget — enforce per-request and per-user ceilings, hard-failing before overspend
4. Report — surface a live cost picture good enough to make routing decisions from

Prerequisites

• Python 3.11+
• An OpenAI-compatible API key (get one at qubax.ai)
• pip install openai (v1.x SDK)

The code below is framework-agnostic — drop it into FastAPI, Celery workers, or a plain script.

Step 1: The Token Meter Core

First, the meter itself. It wraps any chat-completion call, extracts usage from the response object, and records it:

import time
import json
import threading
from collections import defaultdict
from dataclasses import dataclass, field
from openai import OpenAI

client = OpenAI(
    api_key="YOUR_QUBAX_KEY",
    base_url="https://api.qubax.ai/v1",  # any OpenAI-compatible endpoint
)

@dataclass
class UsageRecord:
    feature: str
    user: str
    model: str
    input_tokens: int
    output_tokens: int
    reasoning_tokens: int
    latency_ms: float
    ts: float

class TokenMeter:
    def __init__(self):
        self._lock = threading.Lock()
        self.records: list[UsageRecord] = []
        self.by_feature = defaultdict(lambda: [0, 0, 0])  # in, out, reasoning

    def record(self, rec: UsageRecord):
        with self._lock:
            self.records.append(rec)
            agg = self.by_feature[rec.feature]
            agg[0] += rec.input_tokens
            agg[1] += rec.output_tokens
            agg[2] += rec.reasoning_tokens

meter = TokenMeter()

The key detail is reasoning_tokens. Modern "thinking" models bill their internal chain-of-thought as output tokens even though you never see them, and skipping this column is the #1 reason cost dashboards underreport reality. The OpenAI SDK exposes it as response.usage.completion_tokens_details.reasoning_tokens (0 when absent).

Step 2: The Budget Guard

Counting is diagnostics; budgeting is prevention. This guard raises before a call that would blow the limit:

class BudgetExceeded(Exception):
    def __init__(self, user, spent, limit):
        self.user, self.spent, self.limit = user, spent, limit
        super().__init__(f"{user} spent {spent} tokens, limit {limit}")

class BudgetGuard:
    def __init__(self, per_user_daily=2_000_000, per_request_output=4_000):
        self.per_user_daily = per_user_daily
        self.per_request_output = per_request_output
        self._spent = defaultdict(int)
        self._day = time.strftime("%Y-%m-%d")

    def _roll_day(self):
        today = time.strftime("%Y-%m-%d")
        if today != self._day:
            self._day = today
            self._spent.clear()

    def check(self, user: str):
        self._roll_day()
        if self._spent[user] >= self.per_user_daily:
            raise BudgetExceeded(user, self._spent[user], self.per_user_daily)

    def commit(self, user: str, total_tokens: int):
        self._spent[user] += total_tokens

guard = BudgetGuard(per_user_daily=2_000_000, per_request_output=4_000)

Two knobs matter most in practice:

• `per_request_output` — your insurance against a single pathological response (a model that rambles into a loop can easily emit tens of thousands of tokens). Pass it as max_tokens in the API call, not just as a check.
• `per_user_daily` — your insurance against one customer (or one leaked key) consuming your entire margin.

Step 3: The Instrumented Call

Now wire meter and guard around a real completion. This is the function your app actually calls:

MODEL = "deepseek-v4-flash"   # swap freely: "glm-5.2", "claude-sonnet-5", ...

def chat(feature: str, user: str, messages: list, model: str = MODEL, **kwargs):
    guard.check(user)
    t0 = time.time()
    resp = client.chat.completions.create(
        model=model,
        messages=messages,
        max_tokens=guard.per_request_output,   # hard cap at the API level
        **kwargs,
    )
    latency = (time.time() - t0) * 1000

    u = resp.usage
    reasoning = getattr(
        getattr(u, "completion_tokens_details", None),
        "reasoning_tokens", 0,
    ) or 0
    meter.record(UsageRecord(
        feature=feature, user=user, model=model,
        input_tokens=u.prompt_tokens,
        output_tokens=u.completion_tokens - reasoning,  # visible output only
        reasoning_tokens=reasoning,
        latency_ms=latency, ts=time.time(),
    ))
    billed_output = u.completion_tokens  # reasoning IS billed as output
    guard.commit(user, u.prompt_tokens + billed_output)
    return resp.choices[0].message.content

Note the accounting subtlety: we record visible output and reasoning separately for analytics, but both count toward the budget, because that's how the invoice works.

Step 4: Pricing Awareness (Turn Tokens Into Dollars)

Tokens are an abstraction; money is the metric. Keep a small price table (per million tokens) and join it to your usage records. Live prices for popular models are always available at qubax.ai/models — here's a realistic snapshot:

# USD per 1M tokens {model: (input, output)}
PRICES = {
    "deepseek-v4-flash": (0.019, 0.038),
    "glm-5.2":           (0.0075, 0.0473),
    "claude-sonnet-5":   (0.675, 3.375),
    "claude-opus-5":     (1.875, 9.375),
}

def cost_of(rec: UsageRecord) -> float:
    pin, pout = PRICES.get(rec.model, (0, 0))
    return (rec.input_tokens * pin + (rec.output_tokens + rec.reasoning_tokens) * pout) / 1e6

def spend_report():
    rows = []
    for feat, (tin, tout, treas) in meter.by_feature.items():
        rows.append((feat, tin, tout, treas))
    return rows

Run it after a day of traffic and you'll typically find that one feature dominates output volume — a chatty summarizer, an agent loop that re-sends its whole transcript every iteration, or a thinking model whose reasoning tokens dwarf its answers. Those are your optimization targets, in priority order.

Step 5: Alerting — Know Before the Invoice Does

A meter nobody reads is a log file. Add a threshold check after every commit:

ALERT_THRESHOLD = 0.80  # of daily budget

def maybe_alert(user: str):
    if guard._spent[user] > guard.per_user_daily * ALERT_THRESHOLD:
        print(f"[ALERT] {user} at {guard._spent[user]/guard.per_user_daily:.0%} of daily budget")
        # in production: send to Slack/PagerDuty, not stdout

The 80% threshold catches both organic growth (time to upsell or tune routing) and abuse (time to rate-limit) while there's still runway to act.

Step 6: From Metering to Routing

Once you can see per-feature costs, the highest-leverage move is routing: cheap models for easy work, frontier models only when needed. A minimal difficulty router:

EASY = {"classify", "extract", "format", "summarize-short"}

def route(feature: str) -> str:
    if feature in EASY:
        return "glm-5.2"            # about $0.0075 in / $0.0473 out per 1M tokens
    return "claude-sonnet-5"        # frontier quality for hard tasks

Teams routinely cut total spend 60–80% with routing alone, before touching prompt engineering — because most production traffic is classification, extraction, and formatting that frontier models are massively overqualified for.

Production Notes

• Persist records — the in-memory lists are for the demo; ship UsageRecord rows to your DB or warehouse. The schema above maps 1:1 to a table.
• Meter streaming too — with stream=True, usage arrives in the final chunk when you pass stream_options={"include_usage": True}; sum it the same way.
• Watch latency with cost — a model can be cheap per token but slow enough to hurt UX; record both (we do, in UsageRecord.latency_ms).
• Re-check prices periodically — model pricing moves fast; the Qubax models page always has current numbers.

The Whole Picture

Metering turns AI costs from a monthly surprise into an engineering variable you control: per-feature attribution shows where money goes, budget guards stop runaway spend, and routing decides which model deserves each task. Together they're the difference between a demo and a product with a sane gross margin.

Grab an API key, drop the TokenMeter and BudgetGuard classes into your project, and instrument one endpoint today — the first cost report is usually enlightening. Full API reference and quickstart guides live in the Qubax docs.

FAQ

What is token metering?

Token metering is the practice of recording the input, output, and reasoning tokens for every AI API call, attributed to the feature and user that caused it. It's the foundation of AI cost engineering — you can't optimize, budget, or allocate spend you don't measure.

How do I count reasoning tokens?

In the OpenAI SDK (v1.x), reasoning tokens appear at response.usage.completion_tokens_details.reasoning_tokens. They're billed as output tokens, so add them to your output column for costing even though the model's visible answer excludes them.

How do I set an AI budget per user?

Keep a per-user counter keyed by day, check it before each call, and reject (or downgrade the model) when the ceiling is hit. Also pass max_tokens on every request so a single runaway response can't skip your limit — API-level caps are the only hard guarantee.

How much does it cost to run an AI feature?

It depends on your token mix: cost = input times input price, plus output times output price (per million tokens), plus reasoning tokens on the output side. A summarizer doing 500 requests/day at roughly 400 input and 120 output tokens costs pennies per month on budget models like GLM 5.2 or DeepSeek V4 Flash, and tens of dollars on frontier models — which is why routing matters.

Which model should I use for cheap high-volume tasks?

For classification, extraction, formatting, and short summaries, budget models like GLM 5.2 (about $0.0075/M input and $0.0473/M output) and DeepSeek V4 Flash are usually indistinguishable from frontier models in quality. Compare live pricing across the catalog at qubax.ai/models.

Does metering work with streaming responses?

Yes. Request usage in the final stream chunk with stream_options={"include_usage": True}, then record it exactly as in the non-streaming path. Token counts arrive once, after the stream closes.

Can I use this with providers other than Qubax?

Yes — the code targets the OpenAI-compatible chat-completions interface, which Qubax, OpenAI, and most aggregators implement. Point base_url at your provider and keep the price table synchronized with that provider's rates.

Where do I get an API key to try this?

Create an account at qubax.ai, generate a key, and replace YOUR_QUBAX_KEY in the setup snippet. The same key works across every model in the catalog, so you can A/B cost and quality without juggling provider accounts.