Building an OpenAI-Compatible Gateway for Local LLMs

Modern AI development is powered by a standard dialect: the OpenAI Chat Completions API. From IDE extensions like Cursor and VS Code Copilot to agentic frameworks like LangChain, AutoGen, and CrewAI, almost every tool is hardcoded to expect the /v1/chat/completions endpoint. When you want to run open-source models locally—whether it is DeepSeek-R1, Llama 3, or Mistral—you run into a fragmentation problem. Every local runner (Ollama, Llama.cpp, vLLM, Hugging Face TGI) has its own slightly different API quirk, port, or payload format.

This is where a local LLM proxy comes in. By placing a lightweight, compatible gateway between your local consumers and your model runners, you achieve a "one port, many consumers" architecture. While platforms like n1n.ai provide unified access to global cloud models, a local proxy ensures your offline workflow remains seamless and cost-effective.

In this guide, we will design and implement a production-ready, OpenAI-compatible local LLM gateway. We will cover the compatibility contract, Server-Sent Events (SSE) streaming, prompt-hash caching, and multi-provider failover routing.

The Compatibility Contract

To trick client SDKs (like the official openai Python or Node.js packages) into communicating with a local model, your gateway must mimic the OpenAI API specification exactly. This contract involves three main pillars:

1. The Endpoint Structure: You must expose /v1/chat/completions for chat and /v1/models for model discovery.
2. The Request Payload: Your gateway must parse fields like messages, model, temperature, max_tokens, and stream without throwing serialization errors.
3. The Response Payload: The JSON response must match the nested structure containing choices, message, role, content, and usage statistics.

Handling Strict Client Validation

Many developer tools perform a handshake by calling /v1/models first. If your proxy returns an empty list or does not include the exact model name the tool expects, the tool will crash. To bypass this, your proxy must dynamically intercept model requests. It should either report a spoofed list of standard models (e.g., gpt-4o, gpt-3.5-turbo) and map them internally to your local models, or accept any arbitrary model string passed by the client.

Here is a basic FastAPI implementation of the compatibility skeleton:

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import time

app = FastAPI()

# Map incoming requested models to local targets
MODEL_MAP = {
    "gpt-4o": "deepseek-r1:8b",
    "gpt-3.5-turbo": "llama3:8b",
    "default": "llama3:8b"
}

@app.get("/v1/models")
async def list_models():
    return {
        "object": "list",
        "data": [
            {"id": "gpt-4o", "object": "model", "created": int(time.time()), "owned_by": "system"},
            {"id": "gpt-3.5-turbo", "object": "model", "created": int(time.time()), "owned_by": "system"}
        ]
    }

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    body = await request.json()
    requested_model = body.get("model", "default")
    local_model = MODEL_MAP.get(requested_model, MODEL_MAP["default"])
    
    # Extract other OpenAI parameters
    messages = body.get("messages", [])
    temperature = body.get("temperature", 0.7)
    stream = body.get("stream", False)
    
    # Forwarding logic goes here...
    return JSONResponse(content={"status": "ready_to_forward", "target": local_model})

Streaming Without Breaking Clients

Streaming is the difference between an application feeling instantaneous and feeling broken. The OpenAI API uses Server-Sent Events (SSE) to stream tokens as they are generated. If your proxy buffers the upstream response and returns it all at once, the client interface will freeze, and timeouts may trigger.

Implementing Proper SSE Streaming

To implement streaming correctly, your proxy must read the upstream chunk-by-chunk, parse the incoming bytes, format them into the standard data: {...} payload structure, and flush the buffer immediately.

An SSE stream must:

• Set the HTTP header Content-Type: text/event-stream.
• Set Cache-Control: no-cache and Connection: keep-alive.
• Format every message with a data: prefix, followed by the JSON payload, followed by two newlines (\n\n).
• Terminate the stream with data: [DONE] to signal the client parser to close the connection.

Here is how to handle SSE streaming using httpx in Python:

import httpx
from fastapi.responses import StreamingResponse
import json

async def forward_stream_to_client(upstream_url: str, payload: dict):
    headers = {"Content-Type": "application/json"}
    
    async with httpx.AsyncClient() as client:
        async with client.stream("POST", upstream_url, json=payload, headers=headers, timeout=60.0) as response:
            if response.status_code != 200:
                yield f"data: {json.dumps({'error': 'Upstream error'})}\n\n"
                yield "data: [DONE]\n\n"
                return
            
            async for line in response.aiter_lines():
                # Clean and parse upstream line
                if not line.strip():
                    continue
                if line.startswith("data: "):
                    data_content = line[6:].strip()
                    if data_content == "[DONE]":
                        yield "data: [DONE]\n\n"
                        break
                    
                    # Yield formatted token stream
                    yield f"data: {data_content}\n\n"

The Silent Killer: Timeouts and Keep-Alives

Local inference models, especially large ones run on consumer GPUs, can experience long Time-To-First-Token (TTFT) periods due to context pre-filling. If the upstream runner takes longer than 15-30 seconds to evaluate a long system prompt, intermediate proxies, load balancers, or client libraries will drop the connection.

To prevent this, your proxy should run a background keep-alive task. While waiting for the first token, it should periodically send SSE comments (lines starting with a colon :), which client parsers safely ignore but keep the TCP connection alive:

# SSE Keep-alive comment syntax
yield ": keep-alive\n\n"

Caching as the Cost and Latency Killer

In local and hybrid setups, identical prompts are sent repeatedly. Developers constantly save files in IDEs, triggering fresh context evaluations. Agent loops often repeat system prompts and scratchpads.

Implementing a caching layer at the proxy level dramatically reduces latency and saves API costs when falling back to paid providers.

The Design Decision: Exact Match vs. Semantic Caching

For developer tools and local agents, semantic caching is where correctness dies. A slightly modified variable name in a code file should not trigger a cached response from a completely different function. Therefore, your local gateway should use exact prompt-hash caching with a strict Time-to-Live (TTL).

import hashlib
import json

def generate_request_hash(messages: list, temperature: float, model: str) -> str:
    # Normalize payload keys to ensure consistent hashes
    normalized_payload = {
        "messages": messages,
        "temperature": temperature,
        "model": model
    }
    payload_bytes = json.dumps(normalized_payload, sort_keys=True).encode('utf-8')
    return hashlib.sha256(payload_bytes).hexdigest()

Store these hashes in a local in-memory store (like a Python dictionary or Redis) with a TTL of 10-30 minutes. If the exact same prompt is sent within that window, return the cached JSON response instantly.

Intelligent Routing and the Fallback Ladder

Running models locally is great until your VRAM runs out, a model crashes, or you need capabilities that local models cannot provide (e.g., complex reasoning from OpenAI o3 or massive context windows from Claude 3.5 Sonnet).

An intelligent gateway should support a fallback ladder. If a local model fails or returns a 5xx/429 error, the proxy should automatically route the request to a cloud provider. Integrating an aggregator like n1n.ai as the ultimate tier in your fallback ladder ensures you have a highly available, multi-model cloud fallback ready at all times.

Incoming Request 
     │
     ▼
┌────────────────────────┐
│   Local LLM Runner     │ ──(Fails / Timeout)──┐
│  (Ollama / Llama.cpp)  │                     │
└────────────────────────┘                     ▼
                                   ┌────────────────────────┐
                                   │   Cloud Aggregator     │
                                   │      (n1n.ai)          │
                                   └────────────────────────┘

Implementing a Fallback Router with Exponential Backoff

Here is the logic structure for an automatic fallback mechanism:

import httpx
import asyncio

LOCAL_ENDPOINT = "http://localhost:11434/v1/chat/completions"
CLOUD_ENDPOINT = "https://api.n1n.ai/v1/chat/completions"
API_KEY = "YOUR_N1N_API_KEY"

async def route_request_with_fallback(payload: dict):
    async with httpx.AsyncClient() as client:
        # Step 1: Attempt local execution
        try:
            response = await client.post(LOCAL_ENDPOINT, json=payload, timeout=10.0)
            if response.status_code == 200:
                return response.json()
        except (httpx.RequestError, httpx.TimeoutException):
            # Log failure, prepare fallback
            print("Local runner failed. Routing to cloud fallback via n1n.ai...")
        
        # Step 2: Fallback to cloud aggregator (n1n.ai)
        headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
        # Adjust model name for cloud fallback
        payload["model"] = "deepseek-chat" # Example cloud target
        
        for attempt in range(3):
            try:
                response = await client.post(CLOUD_ENDPOINT, json=payload, headers=headers, timeout=30.0)
                if response.status_code == 200:
                    return response.json()
                elif response.status_code == 429:
                    # Exponential backoff on rate limits
                    await asyncio.sleep(2 ** attempt)
            except Exception as e:
                print(f"Cloud fallback failed: {e}")
                
        raise HTTPException(status_code=502, detail="All upstream providers exhausted")

Operational Details: Production-Grade Setup

To make your proxy a permanent fixture of your development environment, you must handle system-level integration, structured logging, and health checks.

1. System Supervision

You do not want to run this script manually in a terminal window. Use a process manager like systemd (on Linux) or launchd (on macOS) to ensure the proxy runs in the background and restarts on boot.

Example systemd unit file (/etc/systemd/system/llm-proxy.service):

[Unit]
Description=Local LLM Proxy Gateway
After=network.target

[Service]
Type=simple
User=developer
WorkingDirectory=/opt/llm-proxy
ExecStart=/opt/llm-proxy/venv/bin/uvicorn main:app --host 127.0.0.1 --port 8000
Restart=always
RestartSec=3

[Install]
WantedBy=multi-user.target

2. Structured Logging

Your gateway is the single source of truth for your machine's AI usage. Implement JSON logging to capture key metrics for every request:

• Requested Model vs. Routed Model
• Request Latency (Time to First Token and total duration)
• Token consumption (Prompt tokens, completion tokens)
• Cache hits vs. Cache misses

3. Health Checks

Expose a /health endpoint that checks the socket status of local runners. If the local runner is down, notify your monitoring script or automatically redirect all traffic to n1n.ai until the local GPU resources are freed up.

Conclusion

Building a local LLM proxy bridges the gap between modern AI tools and local execution. It gives you absolute control over prompts, caching, and model configurations while guaranteeing zero code changes on your clients. By combining local gateways with robust aggregators like n1n.ai for hybrid failover, you build a resilient, high-speed AI engineering environment that works anywhere, anytime.

Get a free API key at n1n.ai