You can stand up a fastapi proxy openai-compatible api in an afternoon to centralize API keys, inject logging, and route requests across providers. This guide walks through a production-shaped implementation using FastAPI and httpx, covering streaming, header forwarding, and a verification path you can run locally.
Step 1: Scaffold the app and dependencies
Install the minimal stack: FastAPI, Uvicorn, httpx, and python-dotenv.
pip install fastapi uvicorn httpx python-dotenvCreate a .env file with your upstream base URL and key. We target any OpenAI-compatible server, so the request and response shapes are fixed by the OpenAI spec.
UPSTREAM_BASE_URL=https://api.openai.com/v1
UPSTREAM_API_KEY=sk-your-key
PROXY_PORT=8000Now build the skeleton. Use a single httpx.AsyncClient for connection pooling across requests. Do not create a client per request; that leaks sockets.
import os
from dotenv import load_dotenv
from fastapi import FastAPI, Request, Response
import httpx
load_dotenv()
app = FastAPI()
UPSTREAM_BASE_URL = os.getenv("UPSTREAM_BASE_URL")
UPSTREAM_API_KEY = os.getenv("UPSTREAM_API_KEY")
client = httpx.AsyncClient(
base_url=UPSTREAM_BASE_URL,
timeout=httpx.Timeout(60.0, connect=10.0),
)Step 2: Proxy the chat completions route
The core of a fastapi proxy openai-compatible api is a pass-through for POST /v1/chat/completions. Every OpenAI-compatible backend exposes that path, so we forward the raw body and return the upstream response unchanged. This keeps the JSON contract intact for any SDK.
@app.post("/v1/chat/completions")
async def proxy_chat_completions(request: Request):
body = await request.body()
headers = {
"Authorization": f"Bearer {UPSTREAM_API_KEY}",
"Content-Type": "application/json",
}
upstream = await client.post(
"/chat/completions",
content=body,
headers=headers,
)
return Response(
content=upstream.content,
status_code=upstream.status_code,
headers=dict(upstream.headers),
)For non-streaming calls this is enough. Note we overwrite Authorization server-side; the client never sends a key to the upstream. Strip any incoming Authorization header from the client to avoid confusion.
Step 3: Support streaming responses
OpenAI uses Server-Sent Events (SSE) when stream: true. You must stream the upstream bytes back to the client without buffering. FastAPI’s StreamingResponse handles this, but you need to avoid letting httpx decode the SSE frames.
from fastapi.responses import StreamingResponse
@app.post("/v1/chat/completions")
async def proxy_chat_completions(request: Request):
body = await request.body()
try:
payload = await request.json()
except Exception:
payload = {}
stream = payload.get("stream", False)
headers = {
"Authorization": f"Bearer {UPSTREAM_API_KEY}",
"Content-Type": "application/json",
}
if not stream:
upstream = await client.post("/chat/completions", content=body, headers=headers)
return Response(
content=upstream.content,
status_code=upstream.status_code,
headers=dict(upstream.headers),
)
async def event_generator():
async with client.stream(
"POST", "/chat/completions", content=body, headers=headers
) as upstream:
async for chunk in upstream.aiter_raw():
yield chunk
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
)aiter_raw() yields exactly what the upstream sent, preserving data: prefixes and the [DONE] terminator. Setting media_type="text/event-stream" stops FastAPI from applying JSON framing.
Step 4: Forward client headers and add observability
You often need to pass trace IDs or accept upstream cache hints. Forward a safe subset of incoming headers and stamp your own. Never forward the client’s Authorization—the proxy owns upstream credentials.
def build_upstream_headers(request: Request) -> dict:
headers = {
"Authorization": f"Bearer {UPSTREAM_API_KEY}",
"Content-Type": "application/json",
}
if "x-trace-id" in request.headers:
headers["X-Trace-Id"] = request.headers["x-trace-id"]
return headersAdd a middleware to log latency and status. This is lightweight and sufficient for local debugging; replace with structured logging in production.
@app.middleware("http")
async def log_requests(request: Request, call_next):
import time
start = time.perf_counter()
response = await call_next(request)
duration = time.perf_counter() - start
print(f"{request.method} {request.url.path} -> {response.status_code} ({duration:.2f}s)")
return responseStep 5: Model routing and fallback
A fastapi proxy openai-compatible api becomes useful when you route by model name. Suppose you have a secondary upstream for open-source models hosted elsewhere.
ROUTES = {
"gpt-": os.getenv("UPSTREAM_BASE_URL"),
"mistral": "https://api.mistral.ai/v1",
}
def resolve_base_url(model: str) -> str:
for prefix, url in ROUTES.items():
if model.startswith(prefix):
return url
return os.getenv("UPSTREAM_BASE_URL")In the endpoint, pick the base URL per request. For production-grade fallback, wrap the upstream call and retry on 429 or 503:
async def forward_with_fallback(payload: dict, headers: dict):
primary = resolve_base_url(payload.get("model", ""))
try:
async with httpx.AsyncClient(base_url=primary, timeout=60.0) as c:
r = await c.post("/chat/completions", json=payload, headers=headers)
if r.status_code in (429, 503):
raise httpx.HTTPStatusError("degraded", request=r.request, response=r)
return r
except (httpx.HTTPStatusError, httpx.RequestError):
secondary = os.getenv("FALLBACK_BASE_URL")
async with httpx.AsyncClient(base_url=secondary, timeout=60.0) as c:
return await c.post("/chat/completions", json=payload, headers=headers)If you’d rather not operate that logic, an OpenAI-compatible endpoint like n4n.ai addresses 240+ models with automatic fallback when a provider is rate-limited or degraded, and honors client routing directives while forwarding provider cache-control hints.
Step 6: Run and verify end-to-end
Start the server:
uvicorn main:app --port ${PROXY_PORT:-8000}Point the OpenAI Python SDK at your proxy. The SDK requires an api_key, but since the proxy injects the real one, any dummy string works.
import openai
client = openai.OpenAI(
base_url="http://localhost:8000/v1",
api_key="dummy",
)
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Say hi."}],
)
print(resp.choices[0].message.content)For streaming, add stream=True and iterate. Verify at the protocol level with curl:
curl -N http://localhost:8000/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}],"stream":true}'You should see SSE chunks beginning with data: and a final data: [DONE].
Verification checklist
- Non-streaming request returns valid JSON with
choicesandusagefields. - Streaming request emits SSE frames and terminates with
[DONE]. - Upstream API key never reaches the client; only the proxy holds it.
- Logs show latency and status for each call.
- Routing sends
mistral-prefixed models to the alternate base URL (confirm via server logs or mock).
Step 7: Hardening notes
Close the httpx client on shutdown to free sockets:
@app.on_event("shutdown")
async def shutdown():
await client.aclose()If you deploy behind Nginx or a load balancer, set proxy_headers=True in uvicorn and trust forwarded headers deliberately. For per-token metering, parse the usage object from the upstream response or rely on a gateway that already provides per-token usage metering. Add CORS middleware only if browser clients will call the proxy directly, and restrict allowed origins.
The pattern above is a complete fastapi proxy openai-compatible api: it terminates client connections, injects server-side secrets, streams SSE correctly, and routes by model. Extend it with rate limiting (e.g., slowapi) or request validation (pydantic models) as your backend matures.