Building a Flask API proxy for OpenAI-compatible models

A flask api proxy openai-compatible endpoint lets you sit between your apps and any LLM provider that speaks the OpenAI protocol, enforcing auth and logging without rewriting client code. This tutorial builds a working proxy in Flask that forwards chat completions, streams tokens, and passes through model lists. You will be able to point an existing OpenAI SDK at your own URL and have it work unchanged.

Prerequisites

• Python 3.10 or newer
• pip install flask requests
• An upstream OpenAI-compatible base URL and API key. For example, https://api.openai.com/v1 with an OpenAI key, or a gateway endpoint.
• curl and the OpenAI Python client for local tests.

Create a virtual environment and install dependencies:

python -m venv .venv
source .venv/bin/activate
pip install flask requests

Export your upstream credentials:

export UPSTREAM_BASE="https://api.openai.com/v1"
export UPSTREAM_KEY="sk-..."
export PROXY_KEY="secret-proxy"

Step 1: A minimal non-streaming proxy

Start with the smallest useful surface: forward POST /v1/chat/completions to the upstream. The flask api proxy openai-compatible pattern here is just a thin HTTP relay. We read the JSON body, attach the upstream bearer token, and return the raw response.

import os
import requests
from flask import Flask, request, Response

app = Flask(__name__)
UPSTREAM_BASE = os.environ["UPSTREAM_BASE"]
UPSTREAM_KEY = os.environ["UPSTREAM_KEY"]

@app.route("/v1/chat/completions", methods=["POST"])
def chat_completions():
    payload = request.get_json(silent=True) or {}
    headers = {
        "Authorization": f"Bearer {UPSTREAM_KEY}",
        "Content-Type": "application/json",
    }
    resp = requests.post(
        f"{UPSTREAM_BASE}/chat/completions",
        headers=headers,
        json=payload,
        timeout=60,
    )
    return Response(
        resp.content,
        status=resp.status_code,
        content_type=resp.headers.get("Content-Type", "application/json"),
    )

if __name__ == "__main__":
    app.run(port=8000)

Run it:

python app.py

In another shell, send a request using the standard OpenAI curl style:

curl -s http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model":"gpt-3.5-turbo","messages":[{"role":"user","content":"Say hi"}]}'

Expected output (truncated):

{
  "id": "chatcmpl-...",
  "object": "chat.completion",
  "choices": [
    {"message": {"role": "assistant", "content": "Hi there!"}}
  ]
}

Step 2: Streaming support

Most chat UIs need token streaming. The OpenAI streaming format is text/event-stream with data: {json}\n\n lines and a final data: [DONE]. We must not buffer the whole response. Use requests with stream=True and yield chunks straight to the Flask response generator.

@app.route("/v1/chat/completions", methods=["POST"])
def chat_completions():
    payload = request.get_json(silent=True) or {}
    headers = {
        "Authorization": f"Bearer {UPSTREAM_KEY}",
        "Content-Type": "application/json",
    }
    if payload.get("stream"):
        upstream = requests.post(
            f"{UPSTREAM_BASE}/chat/completions",
            headers=headers,
            json=payload,
            stream=True,
            timeout=60,
        )
        def generate():
            for line in upstream.iter_lines():
                if line:
                    yield line + b"\n"
        return Response(
            generate(),
            status=upstream.status_code,
            content_type="text/event-stream",
        )
    resp = requests.post(
        f"{UPSTREAM_BASE}/chat/completions",
        headers=headers,
        json=payload,
        timeout=60,
    )
    return Response(
        resp.content,
        status=resp.status_code,
        content_type=resp.headers.get("Content-Type", "application/json"),
    )

Test streaming:

curl -s http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model":"gpt-3.5-turbo","stream":true,"messages":[{"role":"user","content":"Count to 3"}]}'

You should see SSE frames:

data: {"id":"chatcmpl-...","choices":[{"delta":{"content":"1"}}]}
data: {"id":"chatcmpl-...","choices":[{"delta":{"content":" 2"}}]}
data: [DONE]

The iter_lines call already splits on newline; we re-append \n because Flask expects bytes. This preserves the exact upstream framing, which the OpenAI SDK expects.

Step 3: Pass through models and routing headers

A real flask api proxy openai-compatible deployment should also expose /v1/models so SDKs can auto-discover. Additionally, some gateways accept routing directives via request headers. If you forward client-supplied headers like X-Route-To or Cache-Control, the upstream can honor them.

@app.route("/v1/models", methods=["GET"])
def list_models():
    headers = {"Authorization": f"Bearer {UPSTREAM_KEY}"}
    resp = requests.get(f"{UPSTREAM_BASE}/models", headers=headers, timeout=30)
    return Response(resp.content, status=resp.status_code,
                    content_type=resp.headers.get("Content-Type", "application/json"))

@app.route("/v1/chat/completions", methods=["POST"])
def chat_completions():
    payload = request.get_json(silent=True) or {}
    headers = {
        "Authorization": f"Bearer {UPSTREAM_KEY}",
        "Content-Type": "application/json",
    }
    for h in ("X-Route-To", "Cache-Control"):
        if h in request.headers:
            headers[h] = request.headers[h]
    # streaming / non-streaming logic as above

Model name remapping

You may want to expose internal aliases. Intercept the model field before forwarding:

MODEL_MAP = {"my-gpt": "gpt-3.5-turbo"}
payload["model"] = MODEL_MAP.get(payload.get("model"), payload.get("model"))

If your upstream is a gateway such as n4n.ai, it already provides automatic fallback when a provider is rate-limited or degraded and per-token usage metering; forwarding those headers is enough, no custom retry loop needed in the proxy.

Step 4: Centralize client auth

You do not want to hand out the upstream key to every service. Put a separate proxy key in the Authorization header from clients, and swap it for the real one server-side.

PROXY_KEY = os.environ.get("PROXY_KEY", "secret-proxy")

@app.before_request
def check_auth():
    if request.path.startswith("/v1/"):
        client_key = request.headers.get("Authorization", "")
        if client_key != f"Bearer {PROXY_KEY}":
            return Response("Unauthorized", status=401)

Now clients use Bearer secret-proxy; the proxy injects UPSTREAM_KEY. This keeps the flask api proxy openai-compatible surface clean and lets you rotate upstream credentials without client changes.

Step 5: Error handling and timeouts

Upstream failures should propagate with minimal distortion. Wrap requests in try/except and return a JSON error in OpenAI format.

from flask import jsonify

@app.errorhandler(Exception)
def handle_err(e):
    return jsonify({"error": {"message": str(e), "type": "proxy_error"}}), 502

Also set explicit timeouts (already done) and consider Response(..., direct_passthrough=True) for large streams to avoid memory bloat. Never let a hung upstream tie up a worker indefinitely.

Step 6: Run under a production server

Flask’s dev server is single-threaded by default. Use gunicorn:

pip install gunicorn
gunicorn -w 4 -b 0.0.0.0:8000 app:app

Test concurrency with the OpenAI Python client pointed at your proxy:

import openai
client = openai.OpenAI(base_url="http://localhost:8000/v1", api_key="secret-proxy")
resp = client.chat.completions.create(model="gpt-3.5-turbo", messages=[{"role":"user","content":"Hi"}])
print(resp.choices[0].message.content)

This confirms the flask api proxy openai-compatible contract holds with official SDKs. The client thinks it is talking to OpenAI; it is talking to your proxy.

Step 7: Observability and logging

Add minimal request logging to capture latency and status:

import time
from flask import g

@app.before_request
def start_timer():
    g.start = time.time()

@app.after_request
def log_request(response):
    dt = time.time() - g.start
    app.logger.info(f"{request.method} {request.path} -> {response.status_code} {dt:.2f}s")
    return response

For token metering, parse the usage object from non-streaming responses or the final SSE chunk, and emit to your metrics pipeline.

Hardening checklist

• Add CORS if browser clients call the proxy directly (flask-cors).
• Log request IDs and token usage from upstream responses for observability.
• Strip sensitive headers from client input except whitelisted routing ones.
• Use a reverse proxy (nginx) for TLS termination.
• Rate limit per API key using flask-limiter if exposed externally.

The code above is roughly 100 lines and covers the core contract. Extend it with response caching or request queuing as your traffic demands. A flask api proxy openai-compatible layer is the right seam for cross-cutting LLM infrastructure concerns.