LLM Gateway in Python: Calling a Backend Endpoint

What a thin gateway in Python does

A useful llm gateway python setup is smaller than people expect. You do not need a full platform. A thin gateway is a small web service that sits between your apps and a backend endpoint and adds four things: one shared auth token for your own clients, a timeout and retry policy, a cap on concurrent upstream calls, and clean streaming passthrough. Your apps talk to it as if it were OpenAI, and it forwards to the real backend.

This article builds that with FastAPI and httpx, using https://api.openllmgateway.com/v1 as the backend and the model ID uncensored. If you are still deciding whether you need a gateway at all, read what an LLM gateway is first. The point here is the code.

Setup and the non-streaming route

Install the dependencies and set two environment variables: the backend key, and a comma-separated list of tokens you hand out to your own apps. Your apps never see the backend key.

Copypip install fastapi uvicorn httpx openai
export BACKEND_KEY="your-backend-key"
export GATEWAY_TOKENS="app-one-token,app-two-token"

The first half of the service covers configuration, auth and the plain JSON route with retries:

Copyimport asyncio
import os

import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse, Response, StreamingResponse

BACKEND = "https://api.openllmgateway.com/v1"
BACKEND_KEY = os.environ["BACKEND_KEY"]
TOKENS = set(os.environ["GATEWAY_TOKENS"].split(","))

app = FastAPI()
http = httpx.AsyncClient(
    timeout=httpx.Timeout(connect=5, read=120, write=30, pool=10)
)
slots = asyncio.Semaphore(8)  # matches the full-key concurrency limit
UP = {"Authorization": "Bearer " + BACKEND_KEY, "Content-Type": "application/json"}


def check_auth(req: Request):
    token = req.headers.get("authorization", "").removeprefix("Bearer ").strip()
    if token not in TOKENS:
        raise HTTPException(401, "invalid gateway token")


@app.post("/v1/chat/completions")
async def chat(req: Request):
    check_auth(req)
    body = await req.json()
    body["model"] = "uncensored"
    if body.get("stream"):
        return await relay_stream(body)
    async with slots:
        for attempt in range(4):
            try:
                r = await http.post(BACKEND + "/chat/completions", json=body, headers=UP)
            except (httpx.ConnectError, httpx.ReadTimeout):
                await asyncio.sleep(2 ** attempt)
                continue
            if r.status_code == 429:
                await asyncio.sleep(float(r.headers.get("Retry-After", "2")))
                continue
            if r.status_code >= 500:
                await asyncio.sleep(2 ** attempt)
                continue
            return JSONResponse(r.json(), status_code=r.status_code)
    raise HTTPException(502, "backend unavailable after retries")

Notice that the gateway forces model to uncensored. Clients cannot request anything else, which is a simple way to keep one policy in one place.

Streaming passthrough that does not leak slots

Streaming is where thin gateways go wrong. If you read the whole upstream body and then send it, the user sees nothing until the end. If you forget to release the concurrency slot when a client disconnects, you eventually block yourself. The API treats a streamed request as occupying its concurrency slot until the stream ends, so your local semaphore should do the same.

Copyasync def relay_stream(body):
    await slots.acquire()
    try:
        req = http.build_request("POST", BACKEND + "/chat/completions", json=body, headers=UP)
        upstream = await http.send(req, stream=True)
    except httpx.HTTPError:
        slots.release()
        raise HTTPException(502, "backend unreachable")

    if upstream.status_code >= 300:
        data = await upstream.aread()
        await upstream.aclose()
        slots.release()
        return Response(data, status_code=upstream.status_code, media_type="application/json")

    async def pump():
        try:
            async for chunk in upstream.aiter_raw():
                yield chunk
        finally:
            await upstream.aclose()
            slots.release()

    return StreamingResponse(pump(), media_type="text/event-stream")

The finally block runs when the stream completes, when the client drops, or when an error occurs, so the slot always comes back. Retries are deliberately absent here. Once bytes have gone to the client you cannot restart cleanly, so retry only before the first chunk, and in this version the status check above is the only gate.

Calling your gateway from a client

Run it with uvicorn gateway:app --port 8000. Your apps now use the standard OpenAI client with the gateway as the base URL and one of your own tokens as the key:

Copyfrom openai import OpenAI

# Talks to the local gateway, which forwards to https://api.openllmgateway.com/v1
client = OpenAI(base_url="http://localhost:8000/v1", api_key="app-one-token")

for chunk in client.chat.completions.create(
    model="uncensored",
    messages=[{"role": "user", "content": "Give me three names for a hiking app."}],
    stream=True,
):
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

You can add the other routes the same way. GET /v1/models and POST /v1/responses are available on the backend, and tool calling and JSON mode need no special handling because the gateway forwards the body untouched. Max request body is 8 MB, so set a matching limit in front of the gateway if it faces the internet.

Handling 429 correctly

The backend answers 429 in three situations, and your gateway should tell them apart. A rate_limited error code means you exceeded requests per minute: 300 on a full key, 60 on a trial key. A concurrency code means too many requests at once: 8 on a full key, 2 on a trial key. Temporary overload also returns 429 with Retry-After: 5.

The semaphore above prevents the concurrency case in the first place, as long as the gateway is the only client on that key. For requests per minute, add a simple token bucket or a counter per minute. The code above honours Retry-After on every 429, which covers overload. If you see concurrency errors anyway, another process is using the same key, or an old stream is still open.

Log the X-Request-Id header for each upstream call so you can trace a bad response, and log X-Balance-USD to watch your credit. A gateway is a good place to alert when the balance drops below a threshold. The AI gateway production checklist covers the rest: secrets, timeouts, health checks and monitoring.

Cost tracking inside the gateway

Because every response carries token usage, the gateway can record spend per client token. Price is $0.15 per 1M input tokens and $0.50 per 1M output tokens. For a call that used 1,200 prompt tokens and 350 completion tokens, the cost is 1,200 x $0.15 / 1M plus 350 x $0.50 / 1M, which is $0.00018 plus $0.000175, or $0.000355.

Copydef cost_usd(usage):
    return usage["prompt_tokens"] * 0.15 / 1e6 + usage["completion_tokens"] * 0.50 / 1e6

Add it after the JSON response is parsed and store it with the client token. For streams, ask the client to include usage in the last chunk with stream_options. Failed or refused requests are not charged, so only count successful responses. Current rates and top-up bonuses are on the pricing page.

Next steps

You now have a gateway with auth, retries, a concurrency cap and streaming passthrough in about 80 lines. Natural extensions are per-token budgets, request logging to your own storage, and a second backend for failover. Keep the code small; once you need many of those at once, a dedicated gateway project may fit better.

To run it, you need a backend key. Sign up and claim the $0.50 trial credit with one click on the API key page, then check parameters and error formats in the API docs. Remember that a trial key allows only 2 concurrent requests, so set the semaphore to 2 while testing.

More guides

• LLM Gateway vs OpenRouter: Which Do You Need?
• AI Gateway Checklist: Uncensored LLM API