Build an LLM API Gateway

Build a FastAPI gateway that proxies LLM requests with caching and budget limits.

Introduction

30 Second Summary

Every time your app asks an AI model a question, it costs money. Without a central checkpoint between your code and the model provider, those costs become invisible until the bill arrives.

In this project, you will build a gateway server that intercepts every request to OpenAI. It tracks costs per request, caches duplicate questions to avoid double-billing, and enforces hard budget limits per user.

What You'll Build

You send a chat to your gateway and get back the AI's response alongside an exact dollar-cost breakdown, with duplicate queries returning instantly from cache at zero cost.

By the end of this project, you'll have:

A single proxy endpoint that forwards your chat messages to OpenAI and reports the exact cost in dollars for each request.
A response cache backed by Redis that catches duplicate questions and returns instant answers at zero cost, with proof on your usage dashboard.
A budget enforcement system that rejects requests the moment a user's accumulated spend hits their limit, returning a clear error with the overage details.
Secret Mission: Add sliding-window rate limiting that caps requests per minute per user.

Are there any prerequisites?

You should be comfortable writing Python code. You also need Docker installed on your machine. Step 1 walks you through setting up your OpenAI API key if you don't have one yet.

Before We Start

Before diving into code, let's get clear on what you're building and why it matters. This is the moment you commit to the problem you're solving and the approach you'll take.

Set Up Your Environment

Your gateway needs three pieces of infrastructure before it can handle a single request: a Python environment with the right libraries, a Redis server for caching and cost tracking, and an OpenAI API key to forward requests to.

In this step, you'll set all three up on your local machine. By the end, you'll have a verified foundation ready for the gateway code in Step 2.

In this step, get ready to:

Create a project directory with a Python virtual environment.
Install FastAPI, Redis, and OpenAI dependencies.
Start a Redis Docker container and configure your API key.

Create your project directory and virtual environment

Every Python project needs its own isolated environment so dependencies don't collide with other projects on your machine. You'll create a dedicated folder and a virtual environment inside it.

Press the Windows key to open your search bar.
Type PowerShell and press Enter to open it.
Navigate to C:\Projects by running these commands (the first command creates the folder if it doesn't exist):

mkdir "C:\Projects" -Force | Out-Null
cd C:\Projects

Create your project folder, move into it, and set up a virtual environment by running these commands:

mkdir llm-gateway
cd llm-gateway
python -m venv venv

What does this command do?

mkdir llm-gateway creates your project folder in C:\Projects.
cd llm-gateway moves your terminal into that folder.
python -m venv venv creates an isolated Python environment in a subfolder called venv.

Activate the virtual environment by running:

venv\Scripts\activate

Your terminal prompt now shows (venv) at the beginning. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

Make sure you ran the activate command from inside the llm-gateway folder. Run pwd to check your current directory.

If you see an execution policy error, run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned and try activating again.

Help me troubleshoot my virtual environment activation

Install your dependencies

The gateway relies on three libraries: FastAPI for the HTTP server, redis-py for cache and cost storage, and the OpenAI SDK for forwarding requests to the model.

Install all three libraries by running this command:

pip install "fastapi[standard]==0.139.0" "redis==8.0.1" "openai==2.45.0"

What does this command do?

fastapi[standard]==0.139.0 installs FastAPI plus its recommended extras (including uvicorn for serving and pydantic for request validation).
redis==8.0.1 installs the async-capable Redis client for Python.
openai==2.45.0 installs OpenAI's official SDK with async support built in.
Pinning exact versions ensures your gateway behaves identically to this walkthrough.

You'll see pip downloading and installing packages. Once it finishes, confirm everything imported correctly by running this check:

python -c "import fastapi, redis, openai; print('All imports OK')"

You should see All imports OK printed to your terminal.

Seeing an ImportError?

Make sure your virtual environment is active. You should see (venv) in your prompt. If not, run the activate command from the previous substep.

If pip itself failed, check you're connected to the internet and try running the install command again.

Help me fix my import error

Start Redis and configure your API key

Redis is the in-memory data store your gateway uses for caching responses and tracking cost. You'll run it as a Docker container so there's nothing to install on your machine beyond Docker itself.

Start a Redis container by running this command:

docker run -d --name gateway-redis -p 6379:6379 redis:latest

What does this command do?

-d runs the container in the background (detached mode).
--name gateway-redis gives the container a memorable name you'll reference later.
-p 6379:6379 maps the container's Redis port to the same port on your machine so your Python code can connect to localhost:6379.

Docker prints a long container ID string. That confirms the container started.

Verify Redis is responding by running:

docker exec gateway-redis redis-cli ping

You should see PONG in your terminal. This proves Redis is running and reachable.

Seeing "No such container"?

Make sure Docker Desktop is running. If the container failed to start, check if port 6379 is already in use by running docker ps -a to see all containers.

If a container named gateway-redis already exists from a previous attempt, remove it first with docker rm -f gateway-redis and re-run the docker run command.

Help me fix my Redis container issue

Now you need an OpenAI API key so the gateway can forward requests to the model.

Open https://platform.openai.com/api-keys in your browser and sign in to your OpenAI account.
Click Create new secret key.
Copy the key value. You will only see it once.

Why a .env file?

Storing your key in a .env file keeps it out of your source code. The OpenAI SDK reads OPENAI_API_KEY from the environment automatically, so you never hardcode secrets into Python files.

Create the .env file by running this command, replacing your-api-key-here with the key you just copied:

Set-Content .env "OPENAI_API_KEY=your-api-key-here"

Confirm the file was created by running:

Get-Content .env

You should see your OPENAI_API_KEY= line printed with your actual key value.

File is empty or missing?

Make sure you're in the llm-gateway directory. Run pwd to confirm. If the file is empty, re-run the Set-Content command making sure you include the quotes around the value.

Help me fix my .env file

Before you move on, predict: if everything is set up correctly, what should the import check and Redis ping return?

Run the final verification to confirm your full environment:

python -c "import fastapi, redis, openai; print('All imports OK')"; docker exec gateway-redis redis-cli ping

You should see All imports OK followed by PONG on the next line. Both lines confirm your environment is fully ready.

Your foundation is locked in. Next up, you'll write the gateway code that turns this infrastructure into a working LLM proxy with cost tracking from the very first request.

Build the Proxy with Cost Tracking

Your environment is ready. Now it is time to build the core of the gateway: a single endpoint that sits between your application and OpenAI.

Every request passes through this endpoint, and every response comes back with a dollar amount attached. Cost tracking is not a feature you bolt on later. It is baked in from request number one.

In this step, get ready to:

Create a FastAPI app that connects to Redis and OpenAI on startup.
Build a proxy endpoint that forwards chat requests to OpenAI and calculates per-request cost.
Add a usage dashboard endpoint that shows spend by user and feature.

Create the app foundation

The gateway needs two external connections at startup: Redis for storing cost data and the OpenAI client for forwarding requests. FastAPI's lifespan pattern handles both.

Create a file called main.py inside your llm-gateway folder and add the imports, pricing constants, and request models:

import hashlib
import json
import os
import time
from contextlib import asynccontextmanager

import redis.asyncio as redis_async
from fastapi import FastAPI, Header, HTTPException, Request
from openai import AsyncOpenAI
from pydantic import BaseModel

# Per-token pricing for cost calculation
PRICING = {
    "gpt-4o-mini": {"input": 0.15 / 1_000_000, "output": 0.60 / 1_000_000},
}

DEFAULT_MODEL = "gpt-4o-mini"
CACHE_TTL_SECONDS = 300


# Request shape for chat completions
class ChatRequest(BaseModel):
    messages: list[dict]
    model: str = DEFAULT_MODEL


# Request shape for budget management
class BudgetRequest(BaseModel):
    user_id: str
    limit_usd: float

What does this code do?

PRICING maps each model to its per-token cost in dollars. For gpt-4o-mini, input tokens cost $0.15 per million and output tokens cost $0.60 per million.
ChatRequest defines the shape of incoming requests. It expects a list of messages and defaults to gpt-4o-mini as the model.
BudgetRequest defines the shape for setting spending limits. You will use this in a later step.
CACHE_TTL_SECONDS sets how long cached responses live. You will use this when you add caching later.

Below the models in main.py, add the lifespan handler, the app instance, and two helper functions:

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Connect to Redis and initialize OpenAI client on startup
    app.state.redis = redis_async.Redis(
        host="localhost", port=6379, decode_responses=True
    )
    app.state.openai = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
    yield
    # Clean up Redis connection on shutdown
    await app.state.redis.aclose()


app = FastAPI(title="LLM API Gateway", lifespan=lifespan)


def calculate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    # Look up per-token prices and compute total request cost
    prices = PRICING.get(model, PRICING[DEFAULT_MODEL])
    return (prompt_tokens * prices["input"]) + (completion_tokens * prices["output"])


def make_cache_key(messages: list[dict]) -> str:
    # Hash the messages to create a deterministic cache key
    content = json.dumps(messages, sort_keys=True)
    return f"cache:{hashlib.sha256(content.encode()).hexdigest()}"

What does this code do?

The lifespan function runs once at startup (before yield) to connect to Redis and create the OpenAI client. After yield, it runs on shutdown to close connections cleanly.
calculate_cost multiplies token counts by per-token prices. This is how you convert raw usage into a dollar amount.
make_cache_key hashes the messages list into a unique string. You will use this for response caching in a future step.

Save main.py.
Load your API key into the terminal environment and start the server by running these commands:

export $(cat .env | xargs)
fastapi dev main.py

You should see output ending with Application startup complete. and the server listening at http://127.0.0.1:8000. This confirms Redis connected and the OpenAI client initialized.

Why export the .env file?

The code reads your API key with os.environ.get("OPENAI_API_KEY"). The export $(cat .env | xargs) command loads every line from your .env file into your current terminal session as an environment variable.

You only need to run this once per terminal session. The server picks up the key on startup.

🙋‍♀️ Server not starting?

If you see a connection error mentioning Redis, confirm your gateway-redis container is running with docker ps. Start it with docker start gateway-redis if needed.
If you see an import error, make sure your virtual environment is activated (you should see (venv) in your terminal prompt).
Why won't my FastAPI server start?

Add the proxy endpoint with cost tracking

The server is running, but it has no endpoints yet. Now you will add the core proxy route that forwards requests to OpenAI and calculates cost on every single response.

Stop the server with Ctrl+C.
In your editor, add the following endpoint to the bottom of main.py. This is the first half, which defines the route and calls OpenAI:

@app.post("/v1/chat/completions")
async def chat_completions(
    body: ChatRequest,
    x_user_id: str = Header(default="anonymous"),
    x_feature: str = Header(default="default"),
):
    r = app.state.redis

    # Forward to OpenAI
    response = await app.state.openai.chat.completions.create(
        model=body.model,
        messages=body.messages,
    )

    prompt_tokens = response.usage.prompt_tokens
    completion_tokens = response.usage.completion_tokens
    cost = calculate_cost(body.model, prompt_tokens, completion_tokens)

What is happening here?

X-User-Id and X-Feature headers identify who is calling and why, so you can track spend per user and per feature.
The function forwards the messages to OpenAI and waits for the response.
It pulls prompt_tokens and completion_tokens from the response's usage object, then calculates the cost in dollars.

Immediately below, still inside the same function, add the result construction and cost logging:

    result = {
        "id": response.id,
        "model": response.model,
        "content": response.choices[0].message.content,
        "usage": {
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": response.usage.total_tokens,
        },
    }

    # Track cost
    await r.incrbyfloat(f"spent:{x_user_id}", cost)
    await r.incrbyfloat(f"cost:{x_user_id}:{x_feature}", cost)

    return {**result, "cost_usd": round(cost, 8), "cache": "miss"}

What does this section do?

It builds a result dictionary with the model's reply and token usage.
incrbyfloat atomically increments a floating-point value in Redis. This avoids race conditions if multiple requests arrive at once.
The return value includes cost_usd so the caller sees exactly what that request cost in dollars.

Save main.py.
Restart the server by running:

fastapi dev main.py

Before you test, what do you expect to see in the response? Think about what fields the endpoint returns and whether the cost will be a large or tiny number.

Open a second terminal tab (the server occupies your first terminal). Make sure you are in the llm-gateway directory with the venv activated.
Send a test request by running this command:

curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "X-User-Id: alice" \
  -H "X-Feature: summarize" \
  -d '{"messages":[{"role":"user","content":"Say hello"}]}'

You should see a JSON response containing the model's reply in content, a usage object with token counts, and a cost_usd field showing the dollar amount (a tiny fraction of a cent).

Seeing a 500 error or connection refused?

If you see Connection refused, the server is not running. Check your first terminal for errors and restart with fastapi dev main.py.
If you see an internal server error, check the server terminal for traceback details. A missing OPENAI_API_KEY causes an authentication error from the OpenAI SDK. Re-run export $(cat .env | xargs) in the server terminal and restart.
Help me debug my curl request to the gateway.

Add the usage dashboard and budgets endpoint

You can see cost on individual responses, but you also need a way to query total spend across all requests. The usage endpoint aggregates everything Redis has tracked.

Add the following two endpoints to the bottom of main.py:

@app.get("/v1/usage")
async def get_usage():
    r = app.state.redis
    # Scan for all cost tracking keys
    keys = [k async for k in r.scan_iter(match="cost:*")]
    usage = {}
    for key in keys:
        value = await r.get(key)
        parts = key.split(":", 2)
        if len(parts) == 3:
            user_id, feature = parts[1], parts[2]
            if user_id not in usage:
                usage[user_id] = {"total_usd": 0.0, "features": {}}
            amount = float(value)
            usage[user_id]["features"][feature] = round(amount, 8)
            usage[user_id]["total_usd"] = round(
                usage[user_id]["total_usd"] + amount, 8
            )
    return {"usage": usage}


@app.post("/v1/budgets")
async def set_budget(body: BudgetRequest):
    r = app.state.redis
    # Store the budget limit for a user
    await r.set(f"budget:{body.user_id}", str(body.limit_usd))
    return {"user_id": body.user_id, "limit_usd": body.limit_usd, "status": "set"}

What do these endpoints do?

GET /v1/usage scans Redis for all keys matching cost:*. It splits each key into user and feature, then builds a summary showing total spend per user and per feature.
POST /v1/budgets stores a spending limit for a user in Redis. You will wire up the enforcement logic in a later step. For now, it just saves the value.

Save main.py. The server reloads automatically.
Query the usage dashboard by running:

curl http://localhost:8000/v1/usage

You should see a JSON object showing alice's spend under the summarize feature, matching the cost_usd from your earlier request.

Usage showing empty results?

If you see {"usage": {}}, the server may have restarted and lost its Redis keys. Send another request to /v1/chat/completions first, then check usage again. Redis data persists across server restarts but not across Redis container restarts.
Why is my usage endpoint returning empty data?

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and the server is running without errors.

ⓧ I'd like to double check the full code

Your main.py should contain these sections in order. Scroll through to confirm each one is present:

Imports (hashlib, json, os, time, asynccontextmanager, redis.asyncio, FastAPI, Header, HTTPException, Request, AsyncOpenAI, BaseModel).
PRICING dictionary, DEFAULT_MODEL, and CACHE_TTL_SECONDS constants.
ChatRequest and BudgetRequest Pydantic models.
lifespan async context manager connecting to Redis and OpenAI.
app = FastAPI(title="LLM API Gateway", lifespan=lifespan) instance.
calculate_cost and make_cache_key helper functions.
POST /v1/chat/completions endpoint with cost tracking.
GET /v1/usage endpoint.
POST /v1/budgets endpoint.

Your gateway is live. Every request that passes through it gets a dollar amount stamped on the response, and Redis accumulates the running total. Next up, you will discover why this setup still wastes money on duplicate queries.

Experience the Double-Billing Problem

Your gateway is proxying requests and tracking cost. But there is a hidden waste built into the current design.

Right now, every single request hits OpenAI regardless of whether someone asked the same question ten seconds ago. In this step, you will see that problem in action by sending the same request twice and watching your bill grow for no reason.

In this step, get ready to:

Send the same request to your gateway twice and observe both cost money.
Confirm the usage endpoint shows doubled cumulative spend.
Identify the double-billing problem that caching will solve.

Send duplicate requests and observe the cost

You are about to ask your gateway the same question twice. Both requests will travel all the way to OpenAI, and both will cost real money for the exact same answer.

Open a new terminal window (keep your gateway server running in the other one).
Send your first request by running this command:

curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "X-User-Id: bob" \
  -H "X-Feature: faq" \
  -d '{"messages":[{"role":"user","content":"What is the capital of France?"}]}'

You should see a JSON response containing the model's answer, a usage object with token counts, and a cost_usd field showing a small dollar amount (something like 0.0000051).

Now here is the question: what do you think will happen to cost_usd if you send the exact same message again? Will the gateway recognize the duplicate and skip the charge?

Send the exact same request a second time by pressing the up arrow in your terminal and hitting Enter.

The second response also shows a non-zero cost_usd. The gateway forwarded the identical question to OpenAI again and paid for it again.

Confirm the damage by checking your usage dashboard:

curl http://localhost:8000/v1/usage

The total_usd for user bob reflects both requests. You paid twice for the same answer.

Seeing a connection refused error?

Make sure your gateway server is still running in the other terminal window. If it stopped, switch back to that terminal and restart it with fastapi dev main.py.

If you see an OpenAI authentication error, confirm your .env file has a valid key and that the server loaded it on startup.

Help me debug my gateway connection

Name the problem

This is the double-billing problem. Your gateway has zero memory of previous requests.

Why does this matter at scale?

Imagine 100 users asking "What is your refund policy?" in the same hour. Without caching, you pay for 100 separate OpenAI calls that all return the same answer.

At gpt-4o-mini pricing ($0.15 per million input tokens, $0.60 per million output tokens), each individual call is cheap. But repeated identical calls across users and features add up fast with no benefit.

The fix is a response cache that recognizes identical prompts and returns a stored answer without hitting the model. That is exactly what you will build in the next step.

You have seen the double-billing problem firsthand. Every identical request burns money for an answer you already have. Next up, you will add a Redis-backed response cache that eliminates this waste entirely.

Add Response Caching

You just proved that identical requests double your spend. In production, that pattern multiplies across hundreds of users asking the same question.

In this step, you fix the problem by adding a Redis cache in front of OpenAI. Before any request reaches the model, your gateway checks whether an identical request was already answered. If it was, the cached response returns instantly at zero cost.

In this step, get ready to:

Add a cache lookup that intercepts requests before they reach OpenAI.
Store fresh responses in Redis with a 5-minute TTL.
Verify that repeated requests return cached responses at zero cost.

Check the cache before calling OpenAI

The make_cache_key() function already exists in your file. It hashes the request messages into a unique Redis key. Now you need to call it at the top of the endpoint and check whether that key already has a stored response.

In main.py, find the line r = app.state.redis inside the chat_completions function.
Add the following code directly below it:

    # Cache check
    cache_key = make_cache_key(body.messages)
    cached = await r.get(cache_key)
    if cached:
        return {**json.loads(cached), "cost_usd": 0.0, "cache": "hit"}

What does this code do?

make_cache_key(body.messages) serializes the messages list to JSON and hashes it with SHA-256. Identical message lists always produce the same key.
await r.get(cache_key) asks Redis if a response already exists for this exact request.
If a cached response is found, the function returns immediately with cost_usd: 0.0 and cache: "hit". OpenAI is never called.

Save main.py. The dev server auto-reloads.
Confirm the endpoint still works by sending a test request:

curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "X-User-Id: alice" \
  -H "X-Feature: summarize" \
  -d '{"messages":[{"role":"user","content":"What is 2+2?"}]}'

You should see a JSON response with the model's answer, "cache": "miss", and a small cost_usd value. The cache check ran but found nothing stored yet.

Seeing a server error?

Check indentation. The new code must be inside the chat_completions function body, indented with 4 spaces.
Make sure the new block sits between r = app.state.redis and the existing # Forward to OpenAI comment.
help me debug a 500 error in my cache check code

Store fresh responses in the cache

The cache check is in place but nothing populates it yet. You need to store each fresh OpenAI response so that the next identical request finds it.

In main.py, find the # Track cost comment inside chat_completions.
Add the following code directly above it:

    # Cache the response
    await r.setex(cache_key, CACHE_TTL_SECONDS, json.dumps(result))

What does this code do?

r.setex() stores the response JSON in Redis with a key, a time-to-live (TTL), and the value. After 300 seconds (5 minutes), Redis automatically deletes the entry.
The TTL means stale data eventually expires. Fresh answers replace outdated ones without manual cleanup.

Save main.py. The dev server auto-reloads.

Server not reloading?

Confirm you are running the server with fastapi dev main.py. The dev command enables auto-reload on save.
Check the terminal for syntax errors. A missing comma or mismatched parenthesis will prevent reload.
help me fix my FastAPI server reload issue

Test the full caching cycle

Both pieces are in place. The gateway now checks the cache before calling OpenAI and stores fresh responses after receiving them.

Send a fresh request with a new user so you can track the cost cleanly:

curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "X-User-Id: bob" \
  -H "X-Feature: summarize" \
  -d '{"messages":[{"role":"user","content":"Say hello"}]}'

You should see a response with "cache": "miss" and a non-zero cost_usd value. This request went to OpenAI and the response is now stored in Redis.

Before you run the same request again, what do you expect to see in the response this time?

Send the exact same request again:

curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "X-User-Id: bob" \
  -H "X-Feature: summarize" \
  -d '{"messages":[{"role":"user","content":"Say hello"}]}'

This time you should see "cache": "hit" and "cost_usd": 0.0. The model was never called. The response came straight from Redis.

Confirm the cost was only recorded once by checking the usage endpoint:

curl http://localhost:8000/v1/usage

In the response, find bob's entry. The total should reflect only the first request's cost. The cached request added nothing.

Why exact-match caching?

This cache uses an exact SHA-256 hash of the messages. Two requests must be character-for-character identical to share a cache entry. This is simple, deterministic, and has zero false positives.

Semantic caching (returning results for "similar" queries) would require an embedding model and a vector database. That adds cost and complexity without teaching a different gateway concept.

🙋‍♀️ Second request still showing cache miss?

Make sure both curl commands use the exact same messages JSON. Even a trailing space or different quote style changes the hash.
Confirm the r.setex() line is placed before the return statement and after the result dictionary is built.
Check that Redis is still running with docker exec gateway-redis redis-cli ping. You should see PONG.
help me debug why my Redis cache isn't returning hits

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and confirmed the cache hit response.

ⓧ I'd like to double check the full code

Your main.py is longer than 30 lines, so confirm these sections are present in order:

Imports (hashlib, json, os, time, asynccontextmanager, redis.asyncio, FastAPI, AsyncOpenAI, BaseModel).
PRICING dictionary, DEFAULT_MODEL, and CACHE_TTL_SECONDS = 300 constants.
ChatRequest and BudgetRequest Pydantic models.
Lifespan handler connecting Redis and OpenAI.
calculate_cost() and make_cache_key() helper functions.
chat_completions endpoint with: cache check (make_cache_key, r.get, early return on hit), OpenAI call, r.setex to cache, cost tracking, return with cache: "miss".
get_usage endpoint scanning cost:* keys.
set_budget endpoint storing budget limits.

Your gateway now eliminates wasted spend on repeated queries. Next up, you will add budget enforcement so that even unique requests are rejected once a user exceeds their spending limit.

Enforce Budget Limits

Your gateway now tracks every dollar spent and caches identical requests. But tracking cost without enforcing limits is like watching your credit card statement without ever setting a spending cap.

In this step, you add the guardrail that prevents a runaway feature or user from blowing through your spend. The gateway checks each user's accumulated cost against their budget, and rejects the request before it ever reaches OpenAI.

In this step, get ready to:

Add budget enforcement logic that blocks requests once a user's spend exceeds their limit.
Test the full enforcement loop by setting, exceeding, and raising a budget.

Add budget enforcement to the completions endpoint

Your /v1/budgets endpoint already stores each user's budget limit in Redis. Now you need the completions endpoint to actually read that limit and block requests when the user has spent too much.

In main.py, find the chat_completions function. Locate this line:

    r = app.state.redis

    # Cache check

Add the following budget check between r = app.state.redis and # Cache check:

    # Budget check
    budget_raw = await r.get(f"budget:{x_user_id}")
    if budget_raw is not None:
        budget_limit = float(budget_raw)
        spent_raw = await r.get(f"spent:{x_user_id}")
        spent = float(spent_raw) if spent_raw else 0.0
        if spent >= budget_limit:
            raise HTTPException(
                status_code=429,
                detail={
                    "error": "Budget exceeded",
                    "spent": spent,
                    "limit": budget_limit,
                },
            )

What does this code do?

The budget check runs before the cache check and before the OpenAI call. This means even a cache hit won't be returned if the user is over budget.

await r.get(f"budget:{x_user_id}") looks up whether a budget has been set for this user. If no budget exists, the check is skipped entirely.
await r.get(f"spent:{x_user_id}") reads the user's accumulated spend that was tracked by incrbyfloat in previous requests.
If spend is at or over the limit, the gateway raises HTTP 429 (Too Many Requests) with a JSON body showing the current spend and the limit. The request never reaches OpenAI.

💡 Why HTTP 429 instead of 403?

HTTP 429 means "Too Many Requests" and signals that the client should back off. It is semantically correct for rate and quota limits per RFC 6585.

HTTP 403 would mean the user lacks permission entirely. Budget enforcement is a temporary block that lifts when the budget is raised, so 429 is the right signal.

Save main.py.
Switch to the terminal where your FastAPI server is running and restart it by pressing Ctrl+C then running:

fastapi dev main.py

Server won't start?

Check for indentation errors. The budget check block must be indented inside the chat_completions function at the same level as the existing # Cache check comment.
Make sure HTTPException is imported. It should already be in your imports from fastapi.

Help me debug my FastAPI server startup error.

Test the enforcement loop

Now you will set a tiny budget, exhaust it, and watch the gateway reject the next request. This proves the enforcement is working end to end.

Set alice's budget to $0.001 by running this in a separate terminal:

curl -X POST http://localhost:8000/v1/budgets \
  -H "Content-Type: application/json" \
  -d '{"user_id": "alice", "limit_usd": 0.001}'

You should see a JSON response confirming the budget is set.

Now send a chat completion request as alice. Before you run it, predict: will this request succeed or fail?

curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "X-User-Id: alice" \
  -H "X-Feature: summarize" \
  -d '{"messages":[{"role":"user","content":"What is an API gateway?"}]}'

If alice already has accumulated spend from previous steps that exceeds $0.001, the gateway immediately rejects the request with a 429 response. You will see something like:

{"detail":{"error":"Budget exceeded","spent":0.00012,"limit":0.001}}

What if my request went through?

If alice's existing spend from previous steps was below $0.001, the first request might succeed. Send the same request again with a different message to generate a fresh (non-cached) charge. Once alice's cumulative spend crosses the limit, the next request will be rejected.

Now raise alice's budget to $1.00 and confirm requests flow again:

curl -X POST http://localhost:8000/v1/budgets \
  -H "Content-Type: application/json" \
  -d '{"user_id": "alice", "limit_usd": 1.00}'

Send another request as alice:

curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "X-User-Id: alice" \
  -H "X-Feature: summarize" \
  -d '{"messages":[{"role":"user","content":"Explain rate limiting in one sentence"}]}'

This time the request goes through. You see the model's response along with a cost_usd field and "cache": "miss". The budget gate opened because alice's spend is now below the new limit.

Still seeing 429 after raising the budget?

Make sure you restarted the server after saving your code changes. The budget check only runs with the updated code.
Verify the budget was set by checking the response from the POST /v1/budgets call. It should show "limit_usd": 1.0.

Help me debug why my budget enforcement still blocks requests after I raised the limit.

✔️ Awesome, I've got everything!

Great. Make sure you have saved main.py and your server is running.

ⓧ I'd like to double check the full code

Your main.py should have the following sections in order. Scroll up and confirm each is present:

Imports, PRICING constant, DEFAULT_MODEL, CACHE_TTL_SECONDS.
ChatRequest and BudgetRequest Pydantic models.
The lifespan context manager connecting Redis and OpenAI.
calculate_cost and make_cache_key helper functions.
The chat_completions endpoint with the budget check BEFORE the cache check, then the OpenAI call, cache store, and cost tracking.
The get_usage endpoint.
The set_budget endpoint.

Your gateway now has teeth. It tracks cost, caches duplicates, and hard-blocks any user who hits their spend limit. This is the complete cost control layer that production teams rely on to prevent surprise bills.

Secret mission

Add Sliding-Window Rate Limiting

Budget limits protect against total spend, but what if a user blasts 50 requests in one second? Your gateway needs a second layer of protection that caps how many requests a user can send per minute.

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs locally with a Docker container and minimal OpenAI API usage (~$0.01-0.05 during testing), so there are no ongoing costs unless you actively send more requests.

Resources you used:

Redis Docker container named gateway-redis (local, no cost).
Project directory llm-gateway on your Desktop (includes virtual environment, .env file, and main.py).
OpenAI API key (used for test requests during the project).

Keep everything running

No action needed. Choose this if you are still actively building or want to experiment further with the gateway.

Your Redis container continues running in the background. It uses minimal system resources.
Your OpenAI API key remains active. You are only charged when you send requests.
To restart the gateway later, activate your virtual environment and run fastapi dev main.py from inside the llm-gateway folder.

Pause - I'll come back to this later

Shut down running processes to free up memory while keeping your project files intact.

Stop the FastAPI dev server by pressing Ctrl+C in the terminal where it is running.
Stop the Redis container by running:

docker stop gateway-redis

When you are ready to come back, restart Redis and your gateway by running:

docker start gateway-redis
cd ~/Desktop/llm-gateway
source venv/bin/activate
fastapi dev main.py

Your cached responses and cost data persist inside Redis between stops and starts.

Delete - I don't want to use this again

Remove all project resources and start fresh. This is irreversible.

Stop the FastAPI dev server by pressing Ctrl+C in the terminal where it is running.
Remove the Redis Docker container by running:

docker rm -f gateway-redis

Deactivate your virtual environment by running:

deactivate

Delete the project directory. On macOS/Linux, run:

rm -rf ~/Desktop/llm-gateway

On Windows, run:

rmdir /s /q %USERPROFILE%\Desktop\llm-gateway

Optionally, revoke your OpenAI API key. Go to platform.openai.com, find your key in the list, and click the trash icon to delete it.

That's a Wrap!

Nice work. You just built a production-style LLM API gateway that gives you complete visibility and control over every request between your application and OpenAI.

You've learned how to:

Build a FastAPI reverse proxy with per-request cost calculation that logs exactly how much every LLM call costs using OpenAI's token usage fields.
Add Redis-backed response caching that eliminates duplicate spend on identical queries, proving the savings with a cache hit at zero cost.
Implement per-user budget enforcement that rejects requests when a spend threshold is exceeded, with a usage dashboard showing spend by user and feature.
Secret Mission: Added sliding-window rate limiting using Redis sorted sets to cap request frequency per user.

Ready to quiz yourself?