How to Build Your Own AI API Gateway (Open Source Guide)

Published 2026-07-26 · 18 min read · ★ Linkable Resource

If you're building an application that uses multiple AI models — say GPT-5 for reasoning, Claude for long-context tasks, and DeepSeek for cost-sensitive workloads — you probably have three separate API keys, three SDKs, three billing dashboards, and three places where things can break. An AI API gateway solves this by sitting between your application and the model providers, exposing a single OpenAI-compatible endpoint that transparently routes requests, handles failover, and normalizes responses.

This guide walks through building one from scratch using open-source tools. Whether you want to self-host for full control or use a managed alternative, understanding the architecture will help you make the right decision. We'll cover everything from a 50-line proxy to a production-ready gateway with streaming, caching, and automatic model routing.

Why Build an AI API Gateway?

Before we write any code, let's understand the problems a gateway solves:

1. Vendor Lock-in: If your application is tightly coupled to OpenAI's SDK, switching to Anthropic or DeepSeek means rewriting every API call. A gateway abstracts this away — you change a config value, not application code.

2. Rate Limit Aggregation: Each provider has different rate limits. A gateway can distribute load across providers, effectively multiplying your throughput. If OpenAI returns 429, the gateway can automatically retry with Anthropic.

3. Cost Optimization: Different models have vastly different costs. A simple routing rule like "use DeepSeek for requests under 1,000 tokens, GPT-5 for longer ones" can cut costs by 60-70% without quality loss for many workloads.

4. Unified Billing & Observability: One dashboard for all API usage, with per-request logging, cost tracking, and latency metrics. This is invaluable for teams with multiple developers or projects.

5. Fallback & Reliability: When a provider has an outage (and they do), your application keeps working. The gateway handles the switch transparently.

Architecture Overview

Here's the high-level architecture of an AI API gateway:

┌──────────────────────────────────────────────────┐
│                  Your Application                 │
│         (uses OpenAI SDK with custom base_url)    │
└──────────────────┬───────────────────────────────┘
                   │  POST /v1/chat/completions
                   ▼
┌──────────────────────────────────────────────────┐
│              AI API Gateway (You)                  │
│                                                    │
│  ┌─────────┐  ┌──────────┐  ┌──────────────────┐ │
│  │  Auth &  │→│  Router  │→│  Provider Adapter │ │
│  │  Rate    │  │ (model → │  │  (OpenAI format  │ │
│  │  Limit   │  │ provider)│  │  translation)    │ │
│  └─────────┘  └──────────┘  └────────┬─────────┘ │
│                                       │           │
│  ┌──────────┐  ┌──────────┐          │           │
│  │  Cache   │  │  Logger  │          │           │
│  │  (Redis) │  │  (cost,  │          │           │
│  │          │  │  latency)│          │           │
│  └──────────┘  └──────────┘          │           │
└───────────────────────────────────────┼──────────┘
                                        │
              ┌─────────────────────────┼─────────┐
              ▼                         ▼         ▼
    ┌──────────────┐         ┌──────────────┐  ┌──────────────┐
    │   OpenAI API  │         │ Anthropic API│  │ DeepSeek API │
    │  (GPT-5, etc) │         │ (Claude 4)   │  │  (R1, V3)    │
    └──────────────┘         └──────────────┘  └──────────────┘

The key insight is that your application only ever talks to the gateway using the standard OpenAI API format. The gateway translates requests to each provider's native format, routes them, and normalizes responses back to OpenAI format. Your application code never changes even if you switch all providers.

Prerequisites & Stack Selection

For this guide, we'll use Python + FastAPI. Here's why:

FastAPI is async-first (essential for handling streaming responses), has built-in request validation via Pydantic, auto-generates OpenAPI docs (useful for debugging), and is fast enough for proxy workloads. You could also use Node.js/Express, Go, or Rust — the architecture is the same.

Install the requirements:

# Create a virtual environment
python -m venv gateway-env
source gateway-env/bin/activate  # Windows: gateway-env\Scripts\activate

# Install dependencies
pip install fastapi uvicorn httpx

# You'll also need API keys from each provider:
# OpenAI:     https://platform.openai.com/api-keys
# Anthropic:  https://console.anthropic.com/
# DeepSeek:   https://platform.deepseek.com/

Step 1: The Minimal Proxy (50 Lines)

Let's start with the simplest possible gateway — a reverse proxy that forwards requests to OpenAI and passes responses back. This is your foundation:

# gateway.py — Minimal AI API Proxy
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
import httpx
import os

app = FastAPI(title="AI Gateway")

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENAI_BASE = "https://api.openai.com/v1"

@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def proxy(path: str, request: Request):
    """Forward all requests to OpenAI with auth injection."""
    async with httpx.AsyncClient(timeout=120.0) as client:
        response = await client.request(
            method=request.method,
            url=f"{OPENAI_BASE}/{path}",
            headers={
                "Authorization": f"Bearer {OPENAI_API_KEY}",
                "Content-Type": "application/json",
            },
            content=await request.body(),
        )
        return response.json()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Run it and test with curl:

# Set your API key
export OPENAI_API_KEY="sk-your-key-here"

# Start the gateway
python gateway.py

# Test it — note we're pointing to localhost:8000, not api.openai.com
curl http://localhost:8000/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-5",
    "messages": [{"role": "user", "content": "Hello!"}],
    "max_tokens": 50
  }'

That's it — you have a working proxy. Your application can now use the OpenAI SDK with base_url="http://localhost:8000/v1" and everything just works. But this only talks to OpenAI. Let's make it multi-provider.

Step 2: Adding Multi-Provider Routing

The core of a useful gateway is the routing engine — logic that decides which provider handles each request based on the model name. Here's how to implement it:

# gateway.py — Multi-Provider Routing
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import httpx
import os
import json

app = FastAPI(title="Multi-Provider AI Gateway")

# ─── Provider Configuration ──────────────────────────
PROVIDERS = {
    "openai": {
        "base_url": "https://api.openai.com/v1",
        "api_key": os.getenv("OPENAI_API_KEY"),
        "models": ["gpt-5", "gpt-5-mini", "gpt-4o", "o4-mini"],
        "headers": lambda key: {"Authorization": f"Bearer {key}"},
        "transform_request": lambda body: body,        # No transform needed
        "transform_response": lambda resp: resp,       # Already in OpenAI format
    },
    "anthropic": {
        "base_url": "https://api.anthropic.com/v1",
        "api_key": os.getenv("ANTHROPIC_API_KEY"),
        "models": ["claude-opus-4", "claude-sonnet-4", "claude-3-5-haiku"],
        "headers": lambda key: {
            "x-api-key": key,
            "anthropic-version": "2023-06-01",
        },
        "transform_request": "anthropic_transform_request",
        "transform_response": "anthropic_transform_response",
    },
    "deepseek": {
        "base_url": "https://api.deepseek.com/v1",
        "api_key": os.getenv("DEEPSEEK_API_KEY"),
        "models": ["deepseek-reasoner", "deepseek-chat"],
        "headers": lambda key: {"Authorization": f"Bearer {key}"},
        "transform_request": lambda body: body,
        "transform_response": lambda resp: resp,
    },
}

# Build a reverse lookup: model_name → provider
MODEL_TO_PROVIDER = {}
for provider_name, config in PROVIDERS.items():
    for model in config["models"]:
        MODEL_TO_PROVIDER[model] = provider_name


def route_to_provider(model_name: str) -> str:
    """Determine which provider handles a given model."""
    provider = MODEL_TO_PROVIDER.get(model_name)
    if not provider:
        raise HTTPException(
            status_code=400,
            detail=f"Model '{model_name}' not supported. "
                   f"Available: {list(MODEL_TO_PROVIDER.keys())}"
        )
    return provider


@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    body = await request.json()
    model = body.get("model")
    
    provider_name = route_to_provider(model)
    provider = PROVIDERS[provider_name]
    
    # Transform request if needed (Anthropic has different format)
    if provider["transform_request"] == "anthropic_transform_request":
        body = transform_openai_to_anthropic(body)
    
    async with httpx.AsyncClient(timeout=120.0) as client:
        url = f"{provider['base_url']}/chat/completions"
        
        response = await client.post(
            url,
            json=body,
            headers={
                **provider["headers"](provider["api_key"]),
                "Content-Type": "application/json",
            },
        )
        
        if response.status_code != 200:
            raise HTTPException(
                status_code=response.status_code,
                detail=response.text
            )
        
        result = response.json()
        
        # Transform response back to OpenAI format if needed
        if provider["transform_response"] == "anthropic_transform_response":
            result = transform_anthropic_to_openai(result)
        
        return JSONResponse(content=result)


# ─── Anthropic Format Transformers ───────────────────
def transform_openai_to_anthropic(body: dict) -> dict:
    """Convert OpenAI chat format to Anthropic Messages API format."""
    messages = body.get("messages", [])
    system_msg = None
    chat_msgs = []
    
    for msg in messages:
        if msg["role"] == "system":
            system_msg = msg["content"]
        else:
            chat_msgs.append({
                "role": msg["role"],
                "content": msg["content"],
            })
    
    anthropic_body = {
        "model": body["model"],
        "messages": chat_msgs,
        "max_tokens": body.get("max_tokens", 4096),
    }
    if system_msg:
        anthropic_body["system"] = system_msg
    if "temperature" in body:
        anthropic_body["temperature"] = body["temperature"]
    
    return anthropic_body


def transform_anthropic_to_openai(resp: dict) -> dict:
    """Convert Anthropic response to OpenAI format."""
    return {
        "id": resp.get("id", ""),
        "object": "chat.completion",
        "model": resp.get("model", ""),
        "choices": [{
            "index": 0,
            "message": {
                "role": "assistant",
                "content": resp["content"][0]["text"]
                           if resp.get("content") else "",
            },
            "finish_reason": resp.get("stop_reason", "stop"),
        }],
        "usage": resp.get("usage", {}),
    }


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Now your application can call any model through a single endpoint:

# GPT-5 — routed to OpenAI
curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model": "gpt-5", "messages": [{"role":"user","content":"Hi"}]}'

# Claude Opus 4 — routed to Anthropic (format auto-translated!)
curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model": "claude-opus-4", "messages": [{"role":"user","content":"Hi"}]}'

# DeepSeek R1 — routed to DeepSeek
curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model": "deepseek-reasoner", "messages": [{"role":"user","content":"Hi"}]}'

Step 3: Streaming Support (SSE)

Streaming responses (Server-Sent Events) are essential for good UX in chat applications. The OpenAI format uses stream: true and returns chunks. Here's how to proxy streaming responses while maintaining format compatibility:

# gateway.py — Add streaming endpoint
from fastapi.responses import StreamingResponse
import asyncio

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    body = await request.json()
    model = body.get("model")
    stream = body.get("stream", False)
    
    provider_name = route_to_provider(model)
    provider = PROVIDERS[provider_name]
    
    if provider["transform_request"] == "anthropic_transform_request":
        body = transform_openai_to_anthropic(body)
    
    if stream:
        return StreamingResponse(
            stream_response(provider, body),
            media_type="text/event-stream",
        )
    
    # ... non-streaming code from Step 2 ...


async def stream_response(provider: dict, body: dict):
    """Proxy streaming responses, converting to OpenAI SSE format."""
    async with httpx.AsyncClient(timeout=120.0) as client:
        async with client.stream(
            "POST",
            f"{provider['base_url']}/chat/completions",
            json=body,
            headers={
                **provider["headers"](provider["api_key"]),
                "Content-Type": "application/json",
            },
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    chunk = line[6:]  # Strip "data: " prefix
                    if chunk.strip() == "[DONE]":
                        yield f"data: [DONE]\n\n"
                        return
                    
                    # For Anthropic, transform each chunk to OpenAI format
                    if "anthropic" in provider["base_url"]:
                        try:
                            data = json.loads(chunk)
                            openai_chunk = {
                                "id": data.get("id", ""),
                                "object": "chat.completion.chunk",
                                "model": body.get("model", ""),
                                "choices": [{
                                    "index": 0,
                                    "delta": {"content": 
                                        data.get("delta", {}).get("text", "")},
                                    "finish_reason": None,
                                }],
                            }
                            yield f"data: {json.dumps(openai_chunk)}\n\n"
                        except json.JSONDecodeError:
                            pass
                    else:
                        # OpenAI/DeepSeek already in correct format
                        yield f"data: {chunk}\n\n"

Test streaming:

curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -N \
  -d '{
    "model": "claude-sonnet-4",
    "messages": [{"role": "user", "content": "Write a haiku about APIs"}],
    "stream": true,
    "max_tokens": 100
  }'

# Output (OpenAI-format SSE chunks, even though Claude is generating):
# data: {"choices":[{"delta":{"content":"Silent"},"index":0}]}
# data: {"choices":[{"delta":{"content":" wires"},"index":0}]}
# data: {"choices":[{"delta":{"content":" hum"},"index":0}]}
# data: [DONE]

Step 4: Automatic Failover & Retry

This is where the gateway earns its keep. When a provider fails — whether it's a 429 rate limit, a 500 server error, or a network timeout — the gateway should automatically retry or fail over to a backup provider:

# gateway.py — Failover Engine
import asyncio
import random
from datetime import datetime

# Define fallback chains: if primary fails, try these in order
FALLBACK_CHAINS = {
    "gpt-5": ["gpt-5", "claude-sonnet-4", "deepseek-chat"],
    "claude-opus-4": ["claude-opus-4", "gpt-5", "deepseek-reasoner"],
    "deepseek-reasoner": ["deepseek-reasoner", "gpt-5", "claude-sonnet-4"],
}

# Track provider health
provider_health = {name: {"failures": 0, "last_failure": None} 
                   for name in PROVIDERS}
CIRCUIT_BREAKER_THRESHOLD = 5
CIRCUIT_BREAKER_RESET_SECONDS = 60


def is_provider_healthy(provider_name: str) -> bool:
    """Check if a provider's circuit breaker is tripped."""
    health = provider_health[provider_name]
    if health["failures"] >= CIRCUIT_BREAKER_THRESHOLD:
        if health["last_failure"]:
            elapsed = (datetime.now() - health["last_failure"]).total_seconds()
            if elapsed < CIRCUIT_RESET_SECONDS:
                return False
            # Reset circuit breaker
            health["failures"] = 0
    return True


def record_failure(provider_name: str):
    """Record a provider failure for circuit breaker logic."""
    provider_health[provider_name]["failures"] += 1
    provider_health[provider_name]["last_failure"] = datetime.now()


def record_success(provider_name: str):
    """Reset failure count on success."""
    provider_health[provider_name]["failures"] = 0


async def call_with_failover(model: str, body: dict) -> dict:
    """Try the primary model, fall back to alternatives on failure."""
    chain = FALLBACK_CHAINS.get(model, [model])
    
    for i, fallback_model in enumerate(chain):
        provider_name = MODEL_TO_PROVIDER.get(fallback_model)
        if not provider_name or not is_provider_healthy(provider_name):
            continue
        
        provider = PROVIDERS[provider_name]
        request_body = dict(body)
        request_body["model"] = fallback_model
        
        try:
            # Exponential backoff retry within same provider
            result = await retry_with_backoff(
                provider, request_body, max_retries=3
            )
            record_success(provider_name)
            
            if i > 0:
                print(f"⚠️ Failover: {model} → {fallback_model}")
            
            return result
            
        except Exception as e:
            record_failure(provider_name)
            print(f"Provider {provider_name} failed: {e}")
            continue
    
    raise HTTPException(
        status_code=503,
        detail=f"All providers in failover chain failed for {model}"
    )


async def retry_with_backoff(provider: dict, body: dict, 
                              max_retries: int = 3):
    """Retry with exponential backoff + jitter."""
    for attempt in range(max_retries):
        try:
            async with httpx.AsyncClient(timeout=120.0) as client:
                if body.get("transform") == "anthropic":
                    body = transform_openai_to_anthropic(body)
                
                response = await client.post(
                    f"{provider['base_url']}/chat/completions",
                    json=body,
                    headers={
                        **provider["headers"](provider["api_key"]),
                        "Content-Type": "application/json",
                    },
                )
                
                if response.status_code == 429:
                    # Rate limited — exponential backoff
                    wait = (2 ** attempt) + random.uniform(0, 1)
                    await asyncio.sleep(wait)
                    continue
                
                response.raise_for_status()
                
                result = response.json()
                if "anthropic" in provider["base_url"]:
                    result = transform_anthropic_to_openai(result)
                
                return result
                
        except httpx.TimeoutException:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)
    
    raise Exception(f"Max retries ({max_retries}) exceeded")

With this in place, a call to gpt-5 that hits a rate limit will automatically fail over to claude-sonnet-4, then to deepseek-chat — all transparently to your application.

Step 5: Response Caching

Many AI workloads are repetitive — same prompt, same response. Caching can save 30-50% on API costs for certain applications (classifiers, content moderation, FAQ bots). Here's a simple Redis-backed cache:

# gateway.py — Semantic Cache Layer
import hashlib
import redis.asyncio as redis
import os

redis_client = redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379"))
CACHE_TTL = 3600  # 1 hour


def get_cache_key(body: dict) -> str:
    """Generate deterministic cache key from request body."""
    # Sort keys for deterministic hashing
    cacheable = {
        "model": body["model"],
        "messages": body["messages"],
        "temperature": body.get("temperature", 1.0),
        "max_tokens": body.get("max_tokens", 4096),
    }
    raw = json.dumps(cacheable, sort_keys=True)
    return f"ai:cache:{hashlib.sha256(raw.encode()).hexdigest()}"


@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    body = await request.json()
    
    # Skip cache for streaming or high-temperature requests
    if not body.get("stream", False) and body.get("temperature", 1.0) <= 0.3:
        cache_key = get_cache_key(body)
        cached = await redis_client.get(cache_key)
        if cached:
            print("✅ Cache hit!")
            return JSONResponse(content=json.loads(cached))
    
    # ... call provider ...
    result = await call_with_failover(body["model"], body)
    
    # Cache deterministic responses (temperature ≤ 0.3)
    if body.get("temperature", 1.0) <= 0.3:
        await redis_client.setex(
            cache_key, CACHE_TTL, json.dumps(result)
        )
    
    return JSONResponse(content=result)

Step 6: Docker Deployment

Containerize everything for production deployment:

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY gateway.py .
EXPOSE 8000

CMD ["uvicorn", "gateway:app", "--host", "0.0.0.0", "--port", "8000", \
     "--workers", "4"]

# requirements.txt
fastapi==0.115.0
uvicorn==0.30.0
httpx==0.27.0
redis==5.0.0

# docker-compose.yml
version: "3.9"

services:
  gateway:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - DEEPSEEK_API_KEY=${DEEPSEEK_API_KEY}
      - REDIS_URL=redis://cache:6379
    depends_on:
      - cache
    restart: unless-stopped

  cache:
    image: redis:7-alpine
    volumes:
      - redis-data:/data
    restart: unless-stopped

volumes:
  redis-data:

# .env file (create this, don't commit it!)
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
DEEPSEEK_API_KEY=sk-...

# Deploy!
docker-compose up -d

# Test production deployment
curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model":"gpt-5","messages":[{"role":"user","content":"Hello"}]}'

Existing Open-Source Projects

If you'd rather use an existing project than build from scratch (entirely reasonable), here are the best open-source AI API gateways available in 2026:

Each of these handles the format translation, routing, and failover we built above — plus additional features like usage analytics, team management, and API key generation. The trade-off is configuration complexity and the need to self-host.

Self-Host vs Managed: Decision Matrix

You've now seen what goes into building an AI API gateway. The question is: should you self-host, or use a managed service? Here's an honest comparison:

Choose self-hosting if: You need full data sovereignty, have specific compliance requirements, want zero markup on per-token pricing, or enjoy infrastructure work. For teams with DevOps capacity, this is a legitimate choice.

Choose managed if: You want to focus on your application, not infrastructure. You need instant access to new models without managing separate provider relationships. You want unified billing and a single dashboard. Get started with DrAI free →

Conclusion

Building an AI API gateway is a weekend project that pays dividends for as long as you use AI in production. The architecture — proxy layer, routing engine, format adapters, failover, cache — is conceptually simple even if production-grade implementations have thousands of edge cases.

Whether you build your own, deploy an open-source project like LiteLLM, or use DrAI as a managed gateway, the important thing is to stop coupling your application directly to individual AI providers. The abstraction layer is the single most impactful architectural decision for any AI-powered application in 2026.

If you found this guide useful, consider trying DrAI — we handle all of this for you, plus give you instant access to GPT-5, Claude 4, DeepSeek R1, and 15+ other models through a single API key. No credit card needed to start.