When you expose a generative endpoint, fastapi rate limiting slowapi llm traffic is non-negotiable: a single client can burn your provider quota in seconds. This guide walks through standing up a FastAPI service that proxies to an LLM provider, then layering request and token budgets with slowapi so your backend survives a noisy neighbor.
Step 1: Scaffold the project and install dependencies
Create a clean working directory and a virtual environment. We’ll keep the app in a single main.py for clarity, but the same patterns apply to a packaged service.
mkdir fastapi-llm-rl && cd fastapi-llm-rl
python -m venv .venv && source .venv/bin/activate
pip install fastapi uvicorn slowapi httpx redis tiktokenhttpx handles async forwarding to the upstream model endpoint. redis backs a token bucket for cost accounting. tiktoken gives a rough token count without calling the provider. If you deploy on Kubernetes, bake these into your image; don’t pip install at runtime.
Step 2: Initialize FastAPI with the slowapi Limiter
slowapi is a thin wrapper around the limits library. It integrates with FastAPI by storing a Limiter on app.state and registering an exception handler for RateLimitExceeded. The key function decides what constitutes a client—IP address is the default, but API keys are better for authenticated LLM APIs.
from fastapi import FastAPI, Request, HTTPException
from slowapi import Limiter
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address
from slowapi import _rate_limit_exceeded_handler
app = FastAPI(title="llm-proxy")
limiter = Limiter(key_func=get_remote_address, default_limits=["200/day"])
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)The default_limits array applies to every route unless explicitly overridden or exempted. We set a generous daily cap so internal health checks don’t get blocked. Note that slowapi’s in-memory storage is per-process; with multiple uvicorn workers you must use a shared store (covered in Step 7).
Step 3: Build a minimal LLM proxy route
We do not want to reimplement the OpenAI SDK; we just forward the JSON body and return the upstream response. For simplicity this example uses non-streaming calls. Streaming adds backpressure complexity but the rate-limit logic is identical.
import httpx
from pydantic import BaseModel
UPSTREAM = "https://api.openai.com/v1/chat/completions"
API_KEY = "sk-your-key" # load from env in real life
class ChatReq(BaseModel):
model: str
messages: list[dict]
max_tokens: int = 512
@app.post("/v1/chat")
async def chat(req: ChatReq, request: Request):
headers = {"Authorization": f"Bearer {API_KEY}"}
async with httpx.AsyncClient(timeout=30) as client:
r = await client.post(UPSTREAM, json=req.model_dump(), headers=headers)
return r.json()This works locally, but nothing stops a client from looping this endpoint. That’s where fastapi rate limiting slowapi llm controls come in.
Step 4: Apply request-rate limits with slowapi
Decorate the route with @limiter.limit. The decorator must be outermost (above @app.post) and the function must accept request: Request as a parameter—slowapi pulls the key from it.
@app.post("/v1/chat")
@limiter.limit("10/minute")
async def chat(req: ChatReq, request: Request):
...The limit string follows the limits grammar: "10/minute", "100/hour", "5/second". You can stack multiple limits by passing a list. For authenticated traffic, key by header instead of IP:
def get_api_key(request: Request):
return request.headers.get("x-api-key", get_remote_address(request))
limiter = Limiter(key_func=get_api_key)Now a leaked key is throttled independently of the originating IP, which matters when many users sit behind one NAT.
Step 5: Enforce token budgets, not just request counts
Ten requests per minute is meaningless if each sends a 100k-token document. LLM cost is per token, so we add a Redis-backed token bucket that deducts an estimate of prompt plus completion tokens per call. Use tiktoken for a prompt estimate; trust max_tokens for the completion side.
import redis
import tiktoken
r = redis.Redis(host="localhost", port=6379, db=0)
enc = tiktoken.get_encoding("cl100k_base")
def token_cost(req: ChatReq) -> int:
prompt = " ".join(m["content"] for m in req.messages)
return len(enc.encode(prompt)) + req.max_tokens
async def check_token_budget(key: str, cost: int):
bucket = f"tok:{key}"
limit = 100_000 # per hour
used = r.incrby(bucket, cost)
if used == cost:
r.expire(bucket, 3600)
if used > limit:
r.decrby(bucket, cost)
raise HTTPException(429, "token budget exceeded")Call it inside the route before proxying:
@app.post("/v1/chat")
@limiter.limit("10/minute")
async def chat(req: ChatReq, request: Request):
key = get_api_key(request)
await check_token_budget(key, token_cost(req))
...This gives dual-axis protection: velocity and volume. For strict atomicity under concurrency, replace the incrby/refund with a Lua script that checks and increments in one round trip.
Step 6: Verify the limits with curl and tests
Run the service:
uvicorn main:app --port 8000 --reloadSend 12 rapid requests. The first 10 return 200; the rest return 429 with slowapi’s default body.
for i in $(seq 1 12); do
curl -s -o /dev/null -w "%{http_code}\n" -X POST localhost:8000/v1/chat \
-H "content-type: application/json" \
-d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}]}'
doneCheck the token bucket by temporarily lowering the hourly limit to 50 and sending one request with max_tokens: 100. You should see your own 429.
Add a pytest smoke test to lock the behavior:
def test_rate_limit(client):
payload = {"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hi"}]}
for _ in range(10):
assert client.post("/v1/chat", json=payload).status_code == 200
assert client.post("/v1/chat", json=payload).status_code == 429slowapi also emits X-RateLimit-Remaining headers; assert on those in integration tests to catch silent limit misconfigurations.
Step 7: Production hardening
If you run behind a load balancer, FastAPI sees the LB IP unless you trust X-Forwarded-For. Configure uvicorn with --forwarded-allow-ips or use a middleware that sets request.client.host from the header. Otherwise all clients collapse into one limiter key and you throttle your whole user base simultaneously.
For multi-worker or multi-host deployments, swap slowapi’s memory storage for Redis:
from slowapi.storage import RedisStorage
limiter = Limiter(key_func=get_api_key, storage=RedisStorage(r))If you front an OpenAI-compatible gateway such as n4n.ai, it already provides automatic fallback when a provider is degraded and per-token metering, but you still need edge rate limits to reject abusive clients before they hit your proxy. Your fastapi rate limiting slowapi llm layer stays the first line of defense.
Exempt health checks and metrics endpoints with @limiter.exempt so orchestration probes don’t consume quota. Log rejected requests at warning level and export a counter to Prometheus. Never return your exact limit values in the error body; attackers tune bursts to just under the threshold.
Gotchas we hit in production
- The
@limiter.limitdecorator must wrap the FastAPI route, not a dependency or a nested function. WithAPIRouter, decorate the method exactly as you would onapp. - Forgetting
request: Requestin the signature raises aTypeErrorat call time, not import time. Write a unit test that imports the route to catch it early. - tiktoken encoding costs a couple of milliseconds. Instantiate the encoder once at module load; per-request instantiation will add measurable p99 latency.
- Redis
incrbyplusdecrbyis not perfectly race-free. Under heavy contention two requests can both pass before either refunds. A Lua script eliminates the gap. - Default limits apply globally. If you add a
/adminroute, explicitly set a tighter limit or exempt it; don’t assume defaults are safe.
You now have a FastAPI LLM proxy with dual-axis throttling, verified end to end with curl and pytest. Ship it behind your gateway, watch the 429s, and stop worrying about one client blowing your provider bill.