Build a Budget-Capped Streaming LLM Client in Python (Any Model, One API)

Most tutorials show you how to call an LLM API. Almost none show you how to build the part that matters in production: a streaming client that never lets an agent or a runaway retry loop cost you more than you intended. In this guide, we'll build exactly that in Python — a resilient, streaming, budget-capped LLM client that works with any OpenAI-compatible endpoint, including Qubax AI.

By the end you'll have ~150 lines of production-grade code that:

• streams tokens as they arrive (no waiting for full responses),
• tracks exact token usage and cost per request,
• enforces hard per-request and per-day budgets,
• retries transient failures with exponential backoff,
• fails fast when a model starts looping.

Why streaming and budgets belong together

Streaming isn't just about UX. It's your earliest anomaly detector: a healthy completion produces varied tokens at a steady rate, while a degenerate loop repeats itself. By watching the stream, you can cancel a request the moment it goes sideways — before it burns your budget. We've all heard the horror stories: the agent that retried a 200K-token prompt forty times, the summarizer that fell into an infinite "In conclusion, in conclusion" spiral overnight. Every one of those is preventable with the pattern below.

Pair that with a gateway-side spending cap (per-key budgets are available on Qubax) and client-side accounting, and you get defense in depth: the gateway is the hard ceiling, the client is the smart throttle.

The setup

You only need httpx — no heavy SDKs:

pip install httpx
export LLM_API_BASE="https://api.qubax.ai/v1"
export LLM_API_KEY="your-key-here"

import os, time, json, httpx

API_BASE = os.environ.get("LLM_API_BASE", "https://api.qubax.ai/v1")
API_KEY  = os.environ["LLM_API_KEY"]

Because Qubax exposes an OpenAI-compatible /v1/chat/completions endpoint, everything below works unchanged with any model on the marketplace — GPT, Claude, Gemini, DeepSeek, GLM, Kimi, Qwen, Grok — by changing one string. That matters more than it sounds: your budget logic, logging, and retry policy live in one place, and model choice becomes a routing decision rather than an integration project.

The budget tracker

First, a tiny thread-safe daily budget:

import threading
from datetime import date

class Budget:
    def __init__(self, daily_usd: float, per_request_usd: float):
        self.daily = daily_usd
        self.per_request = per_request_usd
        self._lock = threading.Lock()
        self._day = date.today()
        self._spent = 0.0

    def _rollover(self):
        today = date.today()
        if today != self._day:
            self._day, self._spent = today, 0.0

    def can_afford(self, est_usd: float) -> bool:
        with self._lock:
            self._rollover()
            return (self._spent + est_usd <= self.daily
                    and est_usd <= self.per_request)

    def record(self, usd: float):
        with self._lock:
            self._rollover()
            self._spent += usd

The thread-safety matters if your app serves concurrent users — a naive counter shared across FastAPI worker threads will drift and eventually under-count spend.

Cost estimation from model prices

Keep a small price table (per 1M tokens, in/out). Here we use example retail figures — check live prices on Qubax models, where marketplace rates are usually well below retail:

PRICES = {  # USD per 1M tokens: (input, output)
    "gpt-5.6-terra":    (2.50, 15.00),
    "claude-opus-5":    (5.00, 25.00),
    "deepseek-v4-pro":  (1.60, 3.96),
    "glm-5.3":          (1.40, 4.40),
    "kimi-k3":          (3.00, 15.00),
}

def estimate_cost(model: str, prompt_tokens: int, max_tokens: int) -> float:
    pin, pout = PRICES[model]
    return prompt_tokens / 1e6 * pin + max_tokens / 1e6 * pout

The pre-flight estimate uses max_tokens as a worst case for the output. That's deliberately conservative: your budget check should assume the request could get expensive, then reconcile with actuals afterward.

The streaming client

Now the core. We stream with SSE, accumulate tokens, track a rolling window to detect repetition, and abort when the budget or sanity checks trip:

class StreamExceeded(Exception): pass
class BudgetExceeded(Exception): pass

class StreamingLLM:
    def __init__(self, budget: Budget):
        self.budget = budget
        self.client = httpx.Client(
            base_url=API_BASE,
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=httpx.Timeout(120.0, connect=10.0),
        )

    def chat(self, model: str, messages: list, max_tokens: int = 1024):
        prompt_tokens = sum(len(m["content"].split()) * 4 // 3
                            for m in messages)
        est = estimate_cost(model, prompt_tokens, max_tokens)
        if not self.budget.can_afford(est):
            raise BudgetExceeded(f"estimate ${est:.4f} exceeds budget")

        payload = {"model": model, "messages": messages,
                   "max_tokens": max_tokens, "stream": True}
        out_text, usage = [], None
        window = []  # rolling text for loop detection

        with self.client.stream("POST", "/chat/completions",
                                json=payload) as r:
            r.raise_for_status()
            for line in r.iter_lines():
                if not line.startswith("data: "):
                    continue
                data = line[6:]
                if data.strip() == "[DONE]":
                    break
                chunk = json.loads(data)
                delta = chunk["choices"][0].get("delta", {})
                piece = delta.get("content") or ""
                if piece:
                    out_text.append(piece)
                    window.append(piece)
                    if self._looks_looped(window):
                        raise StreamExceeded(
                            "repetition detected; aborting stream")
                if chunk.get("usage"):
                    usage = chunk["usage"]  # final chunk, if enabled

        actual = estimate_cost(
            model,
            usage["prompt_tokens"] if usage else prompt_tokens,
            usage["completion_tokens"] if usage else
            len("".join(out_text)) * 4 // 3)
        self.budget.record(actual)
        return "".join(out_text), actual

    def _looks_looped(self, window, n=12):
        tail = "".join(window[-120:])
        if len(tail) < 80:
            return False
        frag = tail[-24:]
        return tail.count(frag) >= n

A few design notes:

• `_looks_looped` is intentionally naive. It catches the common degenerate case (a short fragment repeating many times). If you need something smarter, score the tail with a compression ratio — degenerate text compresses absurdly well.
• Aborting the stream mid-flight saves money because most providers bill only the tokens generated up to cancellation.
• Prefer actual usage over estimates when the API returns it. Estimate-based accounting is the fallback.

Retries with backoff — but budget-aware

Retries are where costs silently multiply. Retry only transient failures, and re-check the budget every attempt:

def chat_with_retry(llm: StreamingLLM, model: str, messages: list,
                    attempts: int = 3):
    delay = 1.0
    for i in range(attempts):
        try:
            return llm.chat(model, messages)
        except (httpx.TransportError, httpx.HTTPStatusError) as e:
            status = getattr(getattr(e, "response", None),
                             "status_code", None)
            if status and status < 500 and status != 429:
                raise  # client error: retrying won't help
            if i == attempts - 1:
                raise
            time.sleep(delay)
            delay *= 2

The single most important line is the early raise: retrying a 400 Bad Request is pure waste — the request will fail identically every time, and each attempt still costs prompt-processing tokens on some providers.

Putting it together

if __name__ == "__main__":
    budget = Budget(daily_usd=2.00, per_request_usd=0.10)
    llm = StreamingLLM(budget)

    answer, cost = chat_with_retry(llm, "deepseek-v4-pro", [
        {"role": "system", "content": "You are a concise assistant."},
        {"role": "user", "content": "Explain MoE routing in 3 sentences."},
    ])
    print(f"[${cost:.5f}] {answer}")
    print(f"spent today: ${budget._spent:.4f} / ${budget.daily}")

That's it: streaming output, exact cost accounting, daily and per-request caps, loop detection, and budget-aware retries — in about 150 lines you fully understand because you built it.

Hardening checklist for production

• Store the price table server-side and refresh it — marketplace prices move weekly. Qubax exposes live prices per model on its models page.
• Log every request with model, token counts, and cost to a time-series store; alert on spend-rate anomalies, not just totals. A budget that's exceeded at 3am should page someone by 3:05.
• Set gateway-side caps too. Client budgets can be bypassed by buggy code; a per-key hard limit on your API gateway cannot.
• Prefer cheaper models with fallback. Route drafts to a cheap MoE model (DeepSeek V4 Pro, GLM 5.3) and escalate to flagships only when quality checks fail. On an open marketplace this "cascade" pattern typically cuts cost 60–90% for mixed workloads.
• Cache aggressively. Even a naive semantic cache on repeated questions eliminates a large share of spend in chat products.
• Test failure paths. Unit-test BudgetExceeded and StreamExceeded like any other exception — they're your financial safety net.

FAQ

Does streaming cost more than non-streaming?

No — you pay for tokens, not delivery mechanics. Streaming just lets you see (and abort) problems sooner.

How accurate is my client-side cost estimate?

Within a few percent if you use actual usage fields from the response; estimate-based accounting is conservative, which is what you want for budgets.

Can I use this with OpenAI directly?

Yes — the code targets the OpenAI-compatible chat completions format, so swapping API_BASE and the key is all it takes. The advantage of a marketplace like Qubax is that the same client reaches every major model, often at below-retail rates.

What's the fastest way to cut my LLM bill?

Route to frontier MoE models for most work and reserve flagship dense models for the hardest 10% of requests — then enforce it with the budget class above.