Einen OpenAI-API-Proxy mit Python bauen: erst selbst schreiben, dann alle Modelle anbinden

TL;DR

• Ein OpenAI-API-Proxy ist ein kleiner Server zwischen deiner Anwendung und der Modell-API. Er verwahrt den echten Schlüssel und ergänzt Fähigkeiten, die die API selbst nicht bietet: kundenspezifische Schlüssel, Rate Limiting, Logging und Schlüsselrotation. Unten steht eine funktionsfähige FastAPI-Version mit etwa 50 Zeilen.
• Die Text-API von HiAPI ist OpenAI-kompatibel: POST /v1/chat/completions und POST /v1/responses liegen unter https://api.hiapi.ai/v1. Wenn du das offizielle openai-SDK oder deinen Proxy auf diese Base-URL richtest, kannst du mit einem einzigen Schlüssel die Textmodelle von DeepSeek, Claude, GPT, Kimi, GLM und Grok aufrufen.
• Bilder und Videos sind anders: Sie brauchen Sekunden bis Minuten und laufen daher über das asynchrone POST /v1/tasks von HiAPI. Solche Anfragen gehören nicht in einen synchronen Proxy.

Schritt 1: Ein minimaler Durchleitungs-Proxy

Der Proxy prüft den internen Schlüssel, den der Client sendet, ersetzt ihn durch den echten Upstream-Schlüssel und streamt die Antwort zurück.

pip install fastapi uvicorn httpx

# proxy.py
import os
import httpx
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
from starlette.background import BackgroundTask

UPSTREAM = os.environ.get("UPSTREAM_BASE", "https://api.openai.com/v1")
UPSTREAM_KEY = os.environ["UPSTREAM_KEY"]
INTERNAL_KEYS = set(os.environ["INTERNAL_KEYS"].split(","))  # comma-separated

app = FastAPI()
client = httpx.AsyncClient(timeout=120.0)

def auth(request: Request) -> str:
    header = request.headers.get("authorization", "")
    key = header.removeprefix("Bearer ").strip()
    if key not in INTERNAL_KEYS:
        raise HTTPException(401, "invalid internal key")
    return key

@app.post("/v1/{path:path}")
async def forward(path: str, request: Request):
    tenant = auth(request)
    body = await request.body()
    upstream_req = client.build_request(
        "POST", f"{UPSTREAM}/{path}", content=body,
        headers={"Authorization": f"Bearer {UPSTREAM_KEY}",
                 "Content-Type": request.headers.get("content-type", "application/json")},
    )
    upstream = await client.send(upstream_req, stream=True)
    # Stream back so long answers and SSE don't time out; close the upstream stream when done
    return StreamingResponse(upstream.aiter_raw(), status_code=upstream.status_code,
                             media_type=upstream.headers.get("content-type"),
                             background=BackgroundTask(upstream.aclose))

So startest du ihn:

UPSTREAM_KEY=your-upstream-key INTERNAL_KEYS=tenant-a,tenant-b \
uvicorn proxy:app --host 0.0.0.0 --port 8000

Der Client verwendet wie gewohnt das OpenAI-SDK und setzt die Proxy-Adresse als Base-URL:

from openai import OpenAI

client = OpenAI(api_key="tenant-a", base_url="http://localhost:8000/v1")
resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}],
)
print(resp.choices[0].message.content)

Chats, Streaming und Embeddings laufen alle über denselben Handler, weil der Proxy niemals den Inhalt des Request-Bodys parst.

Schritt 2: Fähigkeiten ergänzen, die die Upstream-API nicht bietet

Rate Limiting pro Tenant. Ohne dieses kann ein einzelner ausgelasteter Kunde dein gesamtes Kontingent aufbrauchen.

import time
from collections import defaultdict

_window: dict[str, list[float]] = defaultdict(list)
LIMIT = 60  # requests per 60 s

def check_rate_limit(tenant: str):
    now = time.time()
    _window[tenant] = [t for t in _window[tenant] if now - t < 60]
    if len(_window[tenant]) >= LIMIT:
        raise HTTPException(429, "rate limit exceeded")
    _window[tenant].append(now)

Rufe diese Funktion direkt nach auth() auf. Wenn mehrere Serverprozesse laufen, verschiebe die Zähler nach Redis (INCR plus EXPIRE).

Logging. Protokolliere Tenant, Pfad, Statuscode, Latenz und Token-Anzahl. Protokolliere niemals den Authorization-Header.

Schlüsselrotation. Lade UPSTREAM_KEY aus einem Secret-Management-Dienst und aktualisiere ihn regelmäßig. Wenn der Upstream-Schlüssel im Hintergrund gewechselt wird, verwenden die Clients weiterhin ihre jeweiligen Tenant-Schlüssel.

Kostenregeln. Parse den Request-Body und lehne Anfragen ab, die deine Richtlinien verletzen, zum Beispiel teure Modelle im Free-Tarif oder ein max_tokens über deinem Limit.

Schritt 3: Mehrere Modellanbieter hinter einem Proxy bündeln

Ein Proxy für nur einen Anbieter ist einfach. Anthropic, DeepSeek oder Moonshot hinzuzufügen bedeutet unterschiedliche Auth-Header, Request-Strukturen und Fehlerformate – ein schlanker Proxy wird so schnell zur Übersetzungsschicht.

Die Text-API von HiAPI verwendet für alle ihre Textmodelle die Request-Struktur von OpenAI, sodass der obige Proxy ohne Änderungen funktioniert. Setze zwei Umgebungsvariablen:

UPSTREAM_BASE=https://api.hiapi.ai/v1 UPSTREAM_KEY=your-hiapi-key INTERNAL_KEYS=tenant-a,tenant-b \
uvicorn proxy:app --host 0.0.0.0 --port 8000

Oder überspringe den Proxy und verwende das SDK direkt:

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["HIAPI_KEY"], base_url="https://api.hiapi.ai/v1")
resp = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[{"role": "user", "content": "Explain sparse attention in two sentences."}],
    max_tokens=400,
)
print(resp.choices[0].message.content)

Wir haben diesen Code mit dem offiziellen openai-Python-Paket (3.35.1) ausgeführt, einmal als regulären Aufruf und einmal mit stream=True; beide Aufrufe liefen fehlerfrei durch, und das SDK hat die Antworten wie gewohnt geparst. Eine Stolperfalle beim Test: Bei max_tokens=20 war die Antwort leer, weil deepseek-v4-flash alle 20 Token für die Inferenz aufgebraucht hat (usage.completion_tokens_details.reasoning_tokens war 22). Lass Reasoning-Modellen ein paar Hundert Token Spielraum.

Einige Textmodelle auf HiAPI:

Modell-ID	Route	Eingabe / Ausgabe, USD pro 1 Mio. Token
deepseek-v4-flash	Chat Completions	$0.14 / $0.28
deepseek-v4.1-flash	Chat Completions und Responses	$0.44 / $1.32
gpt-5.6-luna	Nur Responses	$0.05 / $0.30
grok-4.6	Chat Completions	$0.84 / $2.52
deepseek-v4-pro	Chat Completions und Responses	$1.50 / $4.50
claude-opus-5-5	Chat Completions	$2.02 / $10.08
kimi-k3	Chat Completions	$2.32 / $11.60
claude-sonnet-4-6	Chat Completions	$3.00 / $15.00

Live-Preise von HiAPI, abgefragt am 20. Oktober 2026. GET https://api.hiapi.ai/v1/models listet alle Modelle auf, die mit deinem Schlüssel verfügbar sind.

GPT-Modelle wie gpt-5.6-luna werden nur über die Responses API angeboten (client.responses.create), daher deckt ein Proxy, der /v1/{path} weiterleitet, beide ab.

Unbekannte Modell-IDs werden im Fehlerstil von OpenAI als HTTP 400 zurückgegeben, zusammen mit einer request_id für die Kontaktaufnahme mit dem Support:

{"error": {"code": "model_not_found", "type": "hiapi_error",
           "message": "The requested model does not exist or is not available. Check the exact model ID with GET /v1/models and try again. ...",
           "request_id": "..."}}

Schritt 4: Bilder und Videos gehören nicht in einen synchronen Proxy

Ein synchroner Proxy hält die Verbindung offen, bis die Antwort zurückkommt. Bei Text ist das kein Problem. Bilder brauchen Dutzende Sekunden, Videos mehrere Minuten – länger, als die meisten Load Balancer und Browser warten möchten.

HiAPI betreibt Bild- und Videomodelle über asynchrone Endpunkte: Aufgabe einreichen, Task-ID erhalten, dann pollen oder Callback empfangen.

curl -s -X POST https://api.hiapi.ai/v1/tasks \
  -H "Authorization: Bearer $HIAPI_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "z-image",
    "input": {"prompt": "a glass data center at golden hour, isometric", "aspect_ratio": "16:9"},
    "callback": {"url": "https://your-app.example.com/hiapi/callback", "when": "final"}
  }'
# => {"code":200,"message":"success","data":{"taskId":"tk-hiapi-..."}}

Dein Proxy kann den Einreichungsaufruf weiterhin durchleiten; er kehrt sofort zurück. Das Ergebnis kommt anschließend über deine Callback-URL oder GET /v1/tasks/{taskId}. Ausgabelinks verfallen nach etwa 7 Tagen, also speichere die Dateien.

Checkliste für die Produktion

• Schlüssel aus Logs entfernen. Protokolliere Authorization niemals, weder eingehend noch ausgehend.
• Callbacks für Medienaufgaben verwenden. Das Pollen von Tausenden von Aufgaben verschwendet Anfragen. Schreibe den Callback-Handler idempotent und verwende die taskId als Schlüssel, da Zustellungen wiederholt werden können.
• Task-ID vor dem Pollen speichern. Wenn der Prozess abstürzt, setze das Pollen fort, statt die Aufgabe erneut einzureichen und doppelt zu bezahlen.
• Platz für Reasoning-Token einplanen. Setze max_tokens deutlich höher als die erwartete Antwortlänge.
• Prüfen, was der Schlüssel kann. GET /v1/models gibt die zu diesem Schlüssel passende Liste zurück.

Häufig gestellte Fragen

Kann ich das OpenAI-Python-SDK mit HiAPI verwenden?

Ja, für Textmodelle. Setze base_url="https://api.hiapi.ai/v1" und verwende deinen HiAPI-Schlüssel. Bilder und Videos laufen über /v1/tasks, diesen Endpunkt deckt das OpenAI-SDK nicht ab; rufe ihn mit requests oder httpx auf.

Brauche ich noch meinen eigenen Proxy?

Nur aus denselben Gründen wie bei jedem Anbieter: mandantenspezifische Schlüssel, Rate Limiting und Audit-Logs. Innerhalb des Proxys ist keine anbieterspezifische Übersetzung mehr nötig.

Funktioniert Streaming?

Ja. stream=True gibt Server-Sent Events zurück, die mit data: [DONE] enden; der obige Proxy leitet sie ungepuffert weiter.

Wie rotiere ich einen HiAPI-Schlüssel ohne Unterbrechung?

Erstelle in der Konsole einen neuen Schlüssel, deploye ihn in deinen Proxy oder Schlüsselspeicher und widerrufe dann den alten Schlüssel. Während der Umstellung sind beide Schlüssel gültig.

Weitere Details: Chat-Completions-Dokumentation, Responses-Dokumentation sowie HiAPI als Alternative zu einem OpenAI-API-Proxy.