Skip to content

Instantly share code, notes, and snippets.

@bitsnaps

bitsnaps/openai_proxy.py

Last active August 27, 2026 14:18
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save bitsnaps/59aa161852f387f40a3ab75abce048b8 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save bitsnaps/59aa161852f387f40a3ab75abce048b8 to your computer and use it in GitHub Desktop.
Download ZIP
A simple CLI python3 script to run a local OpenAI-API-compatible proxy server in front of any OpenAI-compatible server with: priority ordering, multi-provider / multi-LLM failover and pluggable custom parameters.
Raw
openai_proxy.py
#!/usr/bin/env python3
"""
openai_proxy.py
================
A modular, extensible, OpenAI-API-compatible local proxy server.
It exposes a local HTTP server that implements the OpenAI API surface
(/v1/chat/completions, /v1/completions, /v1/embeddings, /v1/models, and a
generic passthrough for any other /v1/* route) and forwards every request
to any upstream provider that implements an OpenAI-compatible API
(OpenAI, Azure OpenAI, Groq, Together, Fireworks, OpenRouter, Ollama,
vLLM, LM Studio, text-generation-webui, etc.)
Key design goals
-----------------
1. Spec-correct: request/response bodies are passed through untouched by
default (streaming included), so anything the upstream supports "just
works" through the proxy.
2. Modular / extensible: a small "transform" plugin system lets you
inject, rename, remove or remap request parameters, rewrite response
bodies (including streamed SSE chunks), remap headers, remap model
names, or run arbitrary custom Python logic -- all without touching
this file. You configure this via a JSON config file and/or Python
plugin files.
3. Zero surprises: standard OpenAI error format, standard auth handling
(Bearer tokens), CORS support, health check endpoint.
4. Multi-provider / multi-LLM failover: configure any number of AI
providers (endpoint APIs) and, per provider, a priority-ordered list
of LLMs. If a call to one LLM fails the request is routed to the next
LLM on that provider; if every LLM on a provider fails the request
moves on to the next provider.
5. Virtual "auto" model: any OpenAI-compatible client can select model
"auto" (advertised on GET /v1/models). The proxy then runs the full
provider/LLM priority chain and never forwards the name "auto"
upstream.
Dependencies
------------
pip install fastapi uvicorn httpx
Quick start
-----------
# Proxy to OpenAI itself (just to test)
python3 openai_proxy.py --target-url https://api.openai.com/v1 \\
--target-api-key sk-... --port 8000
# Proxy to Groq, expose a local API key, inject a default temperature,
# and remap "gpt-4o" -> Groq's llama model transparently.
python3 openai_proxy.py \\
--target-url https://api.groq.com/openai/v1 \\
--target-api-key "$GROQ_API_KEY" \\
--proxy-api-key sk-local-anything \\
--model-map '{"gpt-4o": "llama-3.3-70b-versatile"}' \\
--inject-param "chat.completions:top_p=0.9" \\
--port 8000
# Multi-provider failover: try Groq models first, then OpenAI.
# Clients can pick model="auto" to use this chain.
python3 openai_proxy.py \\
--provider name=groq,url=https://api.groq.com/openai/v1,api_key=$GROQ_API_KEY,models=llama-3.3-70b-versatile|llama-3.1-8b-instant \\
--provider name=openai,url=https://api.openai.com/v1,api_key=$OPENAI_API_KEY,models=gpt-4o|gpt-4o-mini \\
--proxy-api-key sk-local-anything \\
--port 8000
Then point any OpenAI SDK / tool at:
base_url = http://localhost:8000/v1
api_key = sk-local-anything (or whatever --proxy-api-key you set)
model = auto (runs the configured failover chain)
Generate a starter config file (declarative transforms) with:
python3 openai_proxy.py --init-config myconfig.json
Write a custom plugin (full programmatic control) with:
python3 openai_proxy.py --init-plugin myplugin.py
Then run with:
python3 openai_proxy.py --target-url ... --config myconfig.json \\
--plugin myplugin.py
"""
from __future__ import annotations
import argparse
import asyncio
import copy
import importlib.util
import json
import logging
import os
import sys
import time
import uuid
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Tuple
try:
from dotenv import load_dotenv
except ImportError: # optional; .env loading is skipped if missing
def load_dotenv(*_args, **_kwargs): # type: ignore[misc]
return False
# --------------------------------------------------------------------------
# Dependency check (kept friendly instead of a raw ImportError traceback)
# --------------------------------------------------------------------------
_MISSING = []
try:
import httpx
except ImportError:
_MISSING.append("httpx")
try:
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
except ImportError:
_MISSING.append("fastapi")
try:
import uvicorn
except ImportError:
_MISSING.append("uvicorn")
if _MISSING:
sys.stderr.write(
"Missing required package(s): {}\n"
"Install them with:\n\n pip install {}\n\n".format(
", ".join(_MISSING), " ".join(_MISSING)
)
)
sys.exit(1)
LOG = logging.getLogger("openai_proxy")
# Unix timestamp used for synthesized model objects (OpenAI requires `created`).
MODEL_CREATED = int(time.time())
DEFAULT_AUTO_MODEL_ID = "auto"
AUTO_MODEL_ALIASES = {
"auto",
"openai-proxy/auto",
"proxy/auto",
"proxy-auto",
}
# ==========================================================================
# Transform registry -- the extensibility backbone of this proxy
# ==========================================================================
RequestTransform = Callable[[Dict[str, Any], Dict[str, Any]], Optional[Dict[str, Any]]]
ResponseTransform = Callable[[Dict[str, Any], Dict[str, Any]], Optional[Dict[str, Any]]]
HeaderTransform = Callable[[Dict[str, str], Dict[str, Any]], Optional[Dict[str, str]]]
class TransformRegistry:
"""
Central place where custom behaviour is registered.
Endpoint keys used throughout this proxy:
"chat.completions", "completions", "embeddings", "models", "*"
"*" transforms run for every endpoint (in addition to the specific one).
"""
def __init__(self) -> None:
self.request_transforms: Dict[str, List[RequestTransform]] = {}
self.response_transforms: Dict[str, List[ResponseTransform]] = {}
self.header_transforms: Dict[str, List[HeaderTransform]] = {}
self.stream_chunk_transforms: Dict[str, List[ResponseTransform]] = {}
# -- registration helpers ---------------------------------------------
def on_request(self, endpoint: str = "*"):
def deco(fn: RequestTransform):
self.request_transforms.setdefault(endpoint, []).append(fn)
return fn
return deco
def on_response(self, endpoint: str = "*"):
def deco(fn: ResponseTransform):
self.response_transforms.setdefault(endpoint, []).append(fn)
return fn
return deco
def on_stream_chunk(self, endpoint: str = "*"):
def deco(fn: ResponseTransform):
self.stream_chunk_transforms.setdefault(endpoint, []).append(fn)
return fn
return deco
def on_headers(self, endpoint: str = "*"):
def deco(fn: HeaderTransform):
self.header_transforms.setdefault(endpoint, []).append(fn)
return fn
return deco
def register_request_transform(self, fn: RequestTransform, endpoint: str = "*") -> None:
self.request_transforms.setdefault(endpoint, []).append(fn)
def register_response_transform(self, fn: ResponseTransform, endpoint: str = "*") -> None:
self.response_transforms.setdefault(endpoint, []).append(fn)
def register_stream_chunk_transform(self, fn: ResponseTransform, endpoint: str = "*") -> None:
self.stream_chunk_transforms.setdefault(endpoint, []).append(fn)
def register_header_transform(self, fn: HeaderTransform, endpoint: str = "*") -> None:
self.header_transforms.setdefault(endpoint, []).append(fn)
# -- application helpers -------------------------------------------
def _run_chain(self, chains: Dict[str, list], endpoint: str, payload, context):
result = payload
for fn in chains.get("*", []):
new = fn(result, context)
if new is not None:
result = new
for fn in chains.get(endpoint, []):
new = fn(result, context)
if new is not None:
result = new
return result
def apply_request(self, endpoint: str, payload: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
return self._run_chain(self.request_transforms, endpoint, payload, context)
def apply_response(self, endpoint: str, payload: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
return self._run_chain(self.response_transforms, endpoint, payload, context)
def apply_stream_chunk(self, endpoint: str, payload: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
return self._run_chain(self.stream_chunk_transforms, endpoint, payload, context)
def apply_headers(self, endpoint: str, headers: Dict[str, str], context: Dict[str, Any]) -> Dict[str, str]:
return self._run_chain(self.header_transforms, endpoint, headers, context)
# ==========================================================================
# Built-in declarative transform factories (used by JSON config)
# ==========================================================================
def make_inject_transform(params: Dict[str, Any], overwrite: bool = False) -> RequestTransform:
def _inject(payload: Dict[str, Any], ctx: Dict[str, Any]) -> Dict[str, Any]:
for k, v in params.items():
if overwrite or k not in payload:
payload[k] = v
return payload
return _inject
def make_rename_transform(mapping: Dict[str, str]) -> RequestTransform:
def _rename(payload: Dict[str, Any], ctx: Dict[str, Any]) -> Dict[str, Any]:
for old, new in mapping.items():
if old in payload:
payload[new] = payload.pop(old)
return payload
return _rename
def make_remove_transform(keys: List[str]) -> RequestTransform:
def _remove(payload: Dict[str, Any], ctx: Dict[str, Any]) -> Dict[str, Any]:
for k in keys:
payload.pop(k, None)
return payload
return _remove
def make_model_map_transform(model_map: Dict[str, str], remember: bool = True) -> RequestTransform:
def _map(payload: Dict[str, Any], ctx: Dict[str, Any]) -> Dict[str, Any]:
# Never remap the virtual auto model -- routing owns that name.
if ctx.get("auto_mode"):
return payload
model = payload.get("model")
if model in model_map:
if remember:
ctx["requested_model"] = model
payload["model"] = model_map[model]
return payload
return _map
def make_restore_model_response_transform() -> ResponseTransform:
def _restore(payload: Dict[str, Any], ctx: Dict[str, Any]) -> Dict[str, Any]:
requested = ctx.get("requested_model")
if requested and isinstance(payload, dict) and "model" in payload:
payload["model"] = requested
return payload
return _restore
def make_header_inject_transform(headers: Dict[str, str]) -> HeaderTransform:
def _inject(hdrs: Dict[str, str], ctx: Dict[str, Any]) -> Dict[str, str]:
hdrs.update(headers)
return hdrs
return _inject
# ==========================================================================
# Settings
# ==========================================================================
@dataclass
class Provider:
"""A single upstream OpenAI-compatible API endpoint and its LLMs."""
name: str
url: str
api_key: Optional[str] = None
models: List[str] = field(default_factory=list) # priority order (first = highest)
extra_headers: Dict[str, str] = field(default_factory=dict)
timeout: Optional[float] = None
priority: int = 0
@dataclass
class Settings:
host: str = "127.0.0.1"
port: int = 8000
target_url: str = ""
target_api_key: Optional[str] = None
proxy_api_key: Optional[str] = None
timeout: float = 600.0
cors_origins: List[str] = field(default_factory=lambda: ["*"])
verbose: bool = False
forward_client_auth: bool = False # if true, pass through incoming Authorization untouched
strip_unknown_params: bool = False
allowed_params: Optional[List[str]] = None
extra_headers: Dict[str, str] = field(default_factory=dict)
restore_model_name: bool = True
providers: List[Provider] = field(default_factory=list)
retry_delay: float = 0.0
auto_model_id: str = DEFAULT_AUTO_MODEL_ID
advertise_auto_model: bool = True
merge_upstream_models: bool = True
def parse_kv_pairs(pairs: List[str]) -> Dict[str, str]:
out = {}
for p in pairs or []:
if ":" not in p and "=" not in p:
raise ValueError(f"Invalid key/value pair: {p!r} (use key=value or key:value)")
sep = "=" if "=" in p else ":"
k, v = p.split(sep, 1)
out[k.strip()] = v.strip()
return out
def parse_endpoint_param(spec: str) -> Tuple[str, str, Any]:
"""
Parses "endpoint:key=value" -> (endpoint, key, value).
If no "endpoint:" prefix is given, endpoint defaults to "*".
"""
endpoint = "*"
rest = spec
if ":" in spec and "=" in spec and spec.index(":") < spec.index("="):
endpoint, rest = spec.split(":", 1)
if "=" not in rest:
raise ValueError(f"Invalid param spec: {spec!r} (expected endpoint:key=value)")
key, value = rest.split("=", 1)
# try to coerce to JSON scalar (numbers/bools/null/objects), fallback to raw string
try:
value = json.loads(value)
except json.JSONDecodeError:
pass
return endpoint.strip(), key.strip(), value
def expand_env_vars(value: Any) -> Any:
"""Expand $VAR / ${VAR} references in strings; leave other types untouched."""
if not isinstance(value, str):
return value
return os.path.expandvars(value)
def parse_provider_spec(spec: str) -> Dict[str, Any]:
"""
Parse a --provider value.
Accepts either a JSON object:
'{"name":"groq","url":"https://...","api_key":"...","models":["a","b"]}'
or a comma-separated key=value list:
name=groq,url=https://...,api_key=...,models=a|b|c,priority=1
"""
raw = spec.strip()
if raw.startswith("{"):
data = json.loads(raw)
if not isinstance(data, dict):
raise ValueError("--provider JSON must be an object")
return data
out: Dict[str, Any] = {}
for part in raw.split(","):
part = part.strip()
if not part:
continue
if "=" not in part:
raise ValueError(
f"Invalid --provider fragment: {part!r} (expected key=value)"
)
key, value = part.split("=", 1)
key, value = key.strip(), value.strip()
if key in ("models", "llms"):
out["models"] = [m.strip() for m in value.split("|") if m.strip()]
elif key == "priority":
out[key] = int(value)
elif key == "timeout":
out[key] = float(value)
else:
out[key] = value
return out
def provider_from_dict(data: Dict[str, Any], default_name: str = "provider") -> Provider:
url = expand_env_vars((data.get("url") or data.get("target_url") or "").strip())
if not url:
raise ValueError(f"Provider {data.get('name', default_name)!r} is missing 'url'")
name = str(data.get("name") or default_name)
api_key = data.get("api_key") or data.get("target_api_key")
if api_key:
api_key = expand_env_vars(str(api_key)) or None
elif data.get("api_key_env"):
api_key = os.environ.get(str(data["api_key_env"]))
models = data.get("models") if data.get("models") is not None else data.get("llms")
if models is None:
models = []
if isinstance(models, str):
splitter = "|" if "|" in models else ","
models = [m.strip() for m in models.split(splitter) if m.strip()]
else:
models = [str(m).strip() for m in models if str(m).strip()]
extra_headers = data.get("extra_headers") or {}
extra_headers = {str(k): expand_env_vars(str(v)) for k, v in extra_headers.items()}
timeout = data.get("timeout")
if timeout is not None:
timeout = float(timeout)
priority = int(data["priority"]) if data.get("priority") is not None else 0
return Provider(
name=name,
url=url,
api_key=api_key,
models=models,
extra_headers=extra_headers,
timeout=timeout,
priority=priority,
)
# ==========================================================================
# Config / plugin loading
# ==========================================================================
SAMPLE_CONFIG = {
"providers": [
{
"name": "groq",
"url": "https://api.groq.com/openai/v1",
"api_key": "${GROQ_API_KEY}",
"models": ["llama-3.3-70b-versatile", "llama-3.1-8b-instant"],
"priority": 1
},
{
"name": "openai",
"url": "https://api.openai.com/v1",
"api_key": "${OPENAI_API_KEY}",
"models": ["gpt-4o", "gpt-4o-mini"],
"priority": 2
}
],
"auto_model_id": "auto",
"advertise_auto_model": True,
"merge_upstream_models": True,
"model_map": {
"gpt-4o": "llama-3.3-70b-versatile",
"gpt-4o-mini": "llama-3.1-8b-instant"
},
"restore_model_name": True,
"inject_params": {
"chat.completions": {"top_p": 0.9},
"*": {}
},
"rename_params": {
"chat.completions": {}
},
"remove_params": {
"chat.completions": []
},
"extra_headers": {
"X-Proxy": "openai-proxy"
},
"allowed_params": None,
"strip_unknown_params": False
}
SAMPLE_PLUGIN = '''"""
Example custom plugin for openai_proxy.py
A plugin module must define a top-level function:
def register(registry, settings, config):
...
`registry` is a TransformRegistry instance -- use it to hook into the
request/response/header pipeline. `settings` is the Settings dataclass.
`config` is the parsed JSON config dict (or {} if none was supplied).
"""
def register(registry, settings, config):
# Example: inject a custom, provider-specific parameter only for
# chat completions requests.
@registry.on_request("chat.completions")
def add_custom_param(payload, ctx):
payload.setdefault("repetition_penalty", 1.1)
return payload
# Example: log every incoming model name.
@registry.on_request("*")
def log_model(payload, ctx):
if isinstance(payload, dict) and "model" in payload:
print(f"[plugin] requested model: {payload['model']}")
return payload
# Example: rewrite non-streamed JSON responses.
@registry.on_response("chat.completions")
def tag_response(payload, ctx):
if isinstance(payload, dict):
payload.setdefault("proxy_tag", "handled-by-openai_proxy")
return payload
# Example: mutate every streamed SSE chunk (already-parsed JSON dict).
@registry.on_stream_chunk("chat.completions")
def tag_stream_chunk(payload, ctx):
return payload
# Example: add/replace outgoing headers sent to the upstream provider.
@registry.on_headers("*")
def add_header(headers, ctx):
headers["X-My-Plugin"] = "1"
return headers
'''
def load_config_file(path: str) -> Dict[str, Any]:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
def load_plugin_file(path: str, registry: TransformRegistry, settings: Settings, config: Dict[str, Any]) -> None:
spec = importlib.util.spec_from_file_location(f"openai_proxy_plugin_{uuid.uuid4().hex}", path)
if spec is None or spec.loader is None:
raise ImportError(f"Could not load plugin from {path}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module) # type: ignore
if not hasattr(module, "register"):
raise AttributeError(f"Plugin {path} must define a register(registry, settings, config) function")
module.register(registry, settings, config)
LOG.info("Loaded plugin: %s", path)
def build_registry_from_config(config: Dict[str, Any], settings: Settings) -> TransformRegistry:
registry = TransformRegistry()
model_map = config.get("model_map") or {}
if model_map:
registry.register_request_transform(make_model_map_transform(model_map), endpoint="*")
if config.get("restore_model_name", settings.restore_model_name):
registry.register_response_transform(make_restore_model_response_transform(), endpoint="*")
registry.register_stream_chunk_transform(make_restore_model_response_transform(), endpoint="*")
elif config.get("restore_model_name", settings.restore_model_name):
# Still restore "auto" (and any other requested_model) even without a map.
registry.register_response_transform(make_restore_model_response_transform(), endpoint="*")
registry.register_stream_chunk_transform(make_restore_model_response_transform(), endpoint="*")
for endpoint, params in (config.get("inject_params") or {}).items():
if params:
registry.register_request_transform(make_inject_transform(params), endpoint=endpoint)
for endpoint, mapping in (config.get("rename_params") or {}).items():
if mapping:
registry.register_request_transform(make_rename_transform(mapping), endpoint=endpoint)
for endpoint, keys in (config.get("remove_params") or {}).items():
if keys:
registry.register_request_transform(make_remove_transform(keys), endpoint=endpoint)
extra_headers = config.get("extra_headers") or {}
if extra_headers:
registry.register_header_transform(make_header_inject_transform(extra_headers), endpoint="*")
return registry
# ==========================================================================
# OpenAI-compatible error helper
# ==========================================================================
def openai_error(message: str, status_code: int = 400, err_type: str = "invalid_request_error",
param: Optional[str] = None, code: Optional[str] = None) -> JSONResponse:
return JSONResponse(
status_code=status_code,
content={
"error": {
"message": message,
"type": err_type,
"param": param,
"code": code,
}
},
)
# ==========================================================================
# Virtual "auto" model + failover helpers
# ==========================================================================
class UpstreamAttemptFailed(Exception):
"""Raised when a single provider+model attempt should be failed over."""
def __init__(
self,
reason: str,
status_code: int = 502,
body: Any = None,
skip_provider: bool = False,
raw_content: Optional[bytes] = None,
media_type: Optional[str] = None,
response_headers: Optional[Dict[str, str]] = None,
) -> None:
super().__init__(reason)
self.reason = reason
self.status_code = status_code
self.body = body
self.skip_provider = skip_provider
self.raw_content = raw_content
self.media_type = media_type
self.response_headers = response_headers or {}
def to_response(self) -> Response:
if isinstance(self.body, dict):
return JSONResponse(
content=self.body,
status_code=self.status_code,
headers=self.response_headers,
)
if self.raw_content is not None:
return Response(
content=self.raw_content,
status_code=self.status_code,
headers=self.response_headers,
media_type=self.media_type or "application/octet-stream",
)
return openai_error(self.reason, status_code=self.status_code, err_type="server_error")
def is_auto_model(name: Optional[str], auto_id: str = DEFAULT_AUTO_MODEL_ID) -> bool:
"""True if the client asked for the virtual router model."""
if not name or not isinstance(name, str):
return False
n = name.strip().lower()
if not n:
return False
if auto_id and n == auto_id.strip().lower():
return True
if n in AUTO_MODEL_ALIASES:
return True
# Accept "something/auto" so UIs that namespace models still work.
if n.endswith("/auto") or n.endswith(":auto"):
return True
return False
def make_model_object(model_id: str, owned_by: str = "openai-proxy",
extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
obj: Dict[str, Any] = {
"id": model_id,
"object": "model",
"created": MODEL_CREATED,
"owned_by": owned_by,
}
if extra:
obj.update(extra)
return obj
def configured_model_ids(providers: List[Provider], auto_id: str) -> List[str]:
"""Unique model ids in display order: auto, then each provider's list."""
seen = set()
out: List[str] = []
if auto_id and auto_id not in seen:
out.append(auto_id)
seen.add(auto_id)
for provider in providers:
for model in provider.models:
if not model or model in seen or is_auto_model(model, auto_id):
continue
out.append(model)
seen.add(model)
return out
def _error_looks_like_auth_failure(body: Any) -> bool:
if not isinstance(body, dict):
return False
err = body.get("error")
if isinstance(err, str):
text = err.lower()
return "api key" in text or "unauthorized" in text or "authentication" in text
if not isinstance(err, dict):
return False
code = str(err.get("code") or "").lower()
err_type = str(err.get("type") or "").lower()
message = str(err.get("message") or "").lower()
if code in {"invalid_api_key", "invalid_api_key_error", "authentication_error"}:
return True
if "auth" in err_type:
return True
if "api key" in message or "unauthorized" in message or "authentication" in message:
return True
return False
def classify_upstream_failure(status_code: int, body: Any) -> str:
"""
Return 'skip_provider' (give up on this provider) or 'failover'
(try the next LLM, then the next provider).
"""
if status_code in (401, 403) or _error_looks_like_auth_failure(body):
return "skip_provider"
return "failover"
def build_attempt_plan(
providers: List[Provider],
payload: Optional[Dict[str, Any]],
auto_mode: bool,
auto_id: str = DEFAULT_AUTO_MODEL_ID,
) -> List[Tuple[Provider, Optional[str]]]:
"""
Build the ordered (provider, model) attempts for one client request.
auto mode -> every provider's configured LLM list, in priority order.
Providers with an empty list are tried once without forcing
a model name (upstream default / passthrough).
specific -> providers that advertise that model, then providers with
no list (passthrough). If nobody advertises it, try the
requested name on every provider.
"""
requested = payload.get("model") if isinstance(payload, dict) else None
if isinstance(requested, str):
requested = requested.strip() or None
plan: List[Tuple[Provider, Optional[str]]] = []
if auto_mode:
for provider in providers:
if provider.models:
for model in provider.models:
if not model or is_auto_model(model, auto_id):
continue
plan.append((provider, model))
else:
# No explicit list: hit the provider and let it pick a default.
# Never send the virtual name "auto" upstream.
plan.append((provider, None))
return plan
matching: List[Tuple[Provider, Optional[str]]] = []
passthrough: List[Tuple[Provider, Optional[str]]] = []
for provider in providers:
if provider.models:
if requested and requested in provider.models:
matching.append((provider, requested))
else:
passthrough.append((provider, requested))
if matching:
return matching + passthrough
# Nobody advertised this id: still try it on every provider so a client
# can call an upstream-only model that was not listed in --provider.
return [(provider, requested) for provider in providers]
def proxy_info_headers(context: Dict[str, Any]) -> Dict[str, str]:
headers: Dict[str, str] = {}
provider = context.get("provider")
model = context.get("upstream_model")
if provider:
headers["x-proxy-provider"] = str(provider)
if model:
headers["x-proxy-model"] = str(model)
if context.get("auto_mode"):
headers["x-proxy-auto"] = "1"
return headers
def parse_models_path(sub_path: str) -> Tuple[str, Optional[str]]:
"""
Return ("list", None) or ("retrieve", model_id) or ("other", None)
for a /v1/{sub_path} or /{sub_path} models route.
"""
path = (sub_path or "").strip("/")
if path.startswith("v1/"):
path = path[3:]
if path == "models":
return "list", None
if path.startswith("models/"):
model_id = path[len("models/"):].strip("/")
if model_id:
return "retrieve", model_id
return "list", None
return "other", None
# ==========================================================================
# httpx timeout helpers
# ==========================================================================
#
# httpx.AsyncClient.send() does NOT accept a `timeout=` keyword. That is
# what produced:
# AsyncClient.send() got an unexpected keyword argument 'timeout'
#
# Timeouts belong on:
# * the AsyncClient constructor (global default)
# * client.request() / get() / post() (per-call, non-streaming)
# * client.build_request(..., timeout=...) (attached to the Request)
# * request.extensions["timeout"] (what send() actually reads)
#
# Never pass timeout= into send().
def make_httpx_timeout(seconds: Optional[float], streaming: bool = False) -> httpx.Timeout:
"""
Build an httpx.Timeout for one upstream attempt.
For streaming SSE (LLM token streams) the *read* deadline is disabled
so a long generation is not killed just because no token arrived
within `seconds`. Connect / write / pool are still bounded.
"""
if seconds is None:
return httpx.Timeout(None)
if streaming:
return httpx.Timeout(seconds, read=None)
return httpx.Timeout(seconds)
def attach_timeout(request: "httpx.Request", timeout: httpx.Timeout) -> "httpx.Request":
"""Write the timeout onto the Request so send() honours it."""
extensions = dict(getattr(request, "extensions", None) or {})
try:
extensions["timeout"] = timeout.as_dict()
except Exception: # noqa: BLE001
extensions["timeout"] = {
"connect": timeout.connect,
"read": timeout.read,
"write": timeout.write,
"pool": timeout.pool,
}
request.extensions = extensions
return request
def build_upstream_request(
client: "httpx.AsyncClient",
method: str,
url: str,
*,
params: Any = None,
headers: Optional[Dict[str, str]] = None,
content: Optional[bytes] = None,
timeout: Any = None,
) -> "httpx.Request":
"""
Build an upstream Request with a per-attempt timeout attached.
Prefer build_request(..., timeout=...) (current httpx). Fall back to
stamping request.extensions["timeout"] on older httpx versions that
reject the timeout kwarg on build_request itself.
"""
timeout_obj = timeout if isinstance(timeout, httpx.Timeout) else make_httpx_timeout(timeout)
try:
request = client.build_request(
method,
url,
params=params,
headers=headers,
content=content,
timeout=timeout_obj,
)
except TypeError:
request = client.build_request(
method,
url,
params=params,
headers=headers,
content=content,
)
attach_timeout(request, timeout_obj)
return request
# ==========================================================================
# Core proxy application
# ==========================================================================
HOP_BY_HOP_HEADERS = {
"connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
"te", "trailers", "transfer-encoding", "upgrade", "content-length",
"content-encoding", "host",
}
ENDPOINT_MAP = {
"/chat/completions": "chat.completions",
"/completions": "completions",
"/embeddings": "embeddings",
"/models": "models",
"/images/generations": "images.generations",
"/images/edits": "images.edits",
"/images/variations": "images.variations",
"/audio/transcriptions": "audio.transcriptions",
"/audio/translations": "audio.translations",
"/audio/speech": "audio.speech",
"/moderations": "moderations",
}
def resolve_endpoint_key(path: str) -> str:
for suffix, key in ENDPOINT_MAP.items():
if path.endswith(suffix):
return key
return "*"
def check_auth(request: Request, settings: Settings) -> Optional[JSONResponse]:
if not settings.proxy_api_key:
return None
auth = request.headers.get("authorization", "")
expected = f"Bearer {settings.proxy_api_key}"
if auth != expected:
return openai_error(
"Incorrect API key provided.",
status_code=401,
err_type="invalid_request_error",
code="invalid_api_key",
)
return None
def build_outgoing_headers(
request: Request,
settings: Settings,
registry: TransformRegistry,
endpoint_key: str,
context: Dict[str, Any],
provider: Optional[Provider] = None,
) -> Dict[str, str]:
headers = {}
for k, v in request.headers.items():
lk = k.lower()
if lk in HOP_BY_HOP_HEADERS or lk == "authorization" or lk == "content-type":
continue
headers[k] = v
headers["content-type"] = "application/json"
if settings.forward_client_auth:
incoming_auth = request.headers.get("authorization")
if incoming_auth:
headers["Authorization"] = incoming_auth
api_key = None
if provider and provider.api_key:
api_key = provider.api_key
elif settings.target_api_key:
api_key = settings.target_api_key
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
for k, v in settings.extra_headers.items():
headers[k] = v
if provider and provider.extra_headers:
headers.update(provider.extra_headers)
headers = registry.apply_headers(endpoint_key, headers, context) or headers
return headers
def create_app(settings: Settings, registry: TransformRegistry) -> FastAPI:
limits = httpx.Limits(max_keepalive_connections=50, max_connections=200)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Create the shared httpx client and store it on app.state so
# route handlers can access it. It is closed on shutdown.
# Client-level timeout is a fallback only; each attempt also
# attaches its own (possibly provider-specific) timeout.
client = httpx.AsyncClient(timeout=make_httpx_timeout(settings.timeout), limits=limits)
app.state.http_client = client
try:
yield
finally:
await client.aclose()
app = FastAPI(title="openai-compatible-proxy", version="1.2.0", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
def local_model_catalog() -> List[Dict[str, Any]]:
"""OpenAI-style model objects the proxy itself knows about."""
items: List[Dict[str, Any]] = []
seen = set()
auto_id = settings.auto_model_id or DEFAULT_AUTO_MODEL_ID
if settings.advertise_auto_model:
items.append(make_model_object(
auto_id,
owned_by="openai-proxy",
extra={"description": "Automatic multi-provider / multi-LLM router"},
))
seen.add(auto_id)
for provider in settings.providers:
for model in provider.models:
if not model or model in seen or is_auto_model(model, auto_id):
continue
items.append(make_model_object(model, owned_by=provider.name))
seen.add(model)
return items
async def fetch_upstream_models(client: "httpx.AsyncClient", provider: Provider) -> List[Dict[str, Any]]:
url = provider.url.rstrip("/") + "/models"
headers: Dict[str, str] = {"accept": "application/json"}
if provider.api_key:
headers["Authorization"] = f"Bearer {provider.api_key}"
elif settings.target_api_key:
headers["Authorization"] = f"Bearer {settings.target_api_key}"
headers.update(provider.extra_headers)
timeout_obj = make_httpx_timeout(min(provider.timeout or settings.timeout, 15.0))
resp = await client.get(url, headers=headers, timeout=timeout_obj)
if not resp.is_success:
return []
data = resp.json()
if isinstance(data, dict) and isinstance(data.get("data"), list):
return [item for item in data["data"] if isinstance(item, dict) and item.get("id")]
if isinstance(data, list):
out = []
for item in data:
if isinstance(item, dict) and item.get("id"):
out.append(item)
elif isinstance(item, str):
out.append(make_model_object(item, owned_by=provider.name))
return out
return []
async def handle_models_list(request: Request) -> JSONResponse:
client = request.app.state.http_client
items = local_model_catalog()
seen = {item["id"] for item in items}
if settings.merge_upstream_models:
for provider in settings.providers:
try:
upstream = await fetch_upstream_models(client, provider)
except Exception as e: # noqa: BLE001
LOG.debug("could not list models from provider=%s: %s", provider.name, e)
continue
for item in upstream:
model_id = str(item.get("id") or "")
if not model_id or model_id in seen or is_auto_model(model_id, settings.auto_model_id):
continue
if "object" not in item:
item["object"] = "model"
if "created" not in item:
item["created"] = MODEL_CREATED
if "owned_by" not in item:
item["owned_by"] = provider.name
items.append(item)
seen.add(model_id)
return JSONResponse({"object": "list", "data": items})
async def handle_models_retrieve(request: Request, model_id: str) -> JSONResponse:
auto_id = settings.auto_model_id or DEFAULT_AUTO_MODEL_ID
if is_auto_model(model_id, auto_id) and settings.advertise_auto_model:
return JSONResponse(make_model_object(
auto_id,
owned_by="openai-proxy",
extra={"description": "Automatic multi-provider / multi-LLM router"},
))
for item in local_model_catalog():
if item["id"] == model_id:
return JSONResponse(item)
if settings.merge_upstream_models:
client = request.app.state.http_client
for provider in settings.providers:
try:
upstream = await fetch_upstream_models(client, provider)
except Exception: # noqa: BLE001
continue
for item in upstream:
if str(item.get("id") or "") == model_id:
item.setdefault("object", "model")
item.setdefault("created", MODEL_CREATED)
item.setdefault("owned_by", provider.name)
return JSONResponse(item)
return openai_error(
f"The model '{model_id}' does not exist",
status_code=404,
err_type="invalid_request_error",
param="model",
code="model_not_found",
)
@app.get("/health")
async def health():
auto_id = settings.auto_model_id or DEFAULT_AUTO_MODEL_ID
return {
"status": "ok",
"target": settings.target_url,
"auto_model": auto_id if settings.advertise_auto_model else None,
"providers": [
{"name": p.name, "url": p.url, "models": p.models, "priority": p.priority}
for p in settings.providers
],
}
async def proxy_request(request: Request, sub_path: str):
auth_err = check_auth(request, settings)
if auth_err:
return auth_err
models_kind, models_id = parse_models_path(sub_path)
if request.method == "GET" and models_kind == "list":
return await handle_models_list(request)
if request.method == "GET" and models_kind == "retrieve" and models_id:
return await handle_models_retrieve(request, models_id)
endpoint_key = resolve_endpoint_key(sub_path)
context: Dict[str, Any] = {
"endpoint": endpoint_key,
"path": sub_path,
"request_id": str(uuid.uuid4()),
"start_time": time.time(),
"auto_mode": False,
}
body_bytes = await request.body()
payload: Optional[Dict[str, Any]] = None
is_json = False
content_type = request.headers.get("content-type", "")
if body_bytes and "application/json" in content_type:
try:
payload = json.loads(body_bytes)
is_json = True
except json.JSONDecodeError:
return openai_error("Request body is not valid JSON.", status_code=400)
if is_json and isinstance(payload, dict):
if "model" in payload and "requested_model" not in context:
context["requested_model"] = payload["model"]
auto_mode = is_auto_model(payload.get("model"), settings.auto_model_id)
context["auto_mode"] = auto_mode
if auto_mode:
# Pin the advertised name so responses restore to "auto",
# then drop it so it is never forwarded upstream.
context["requested_model"] = settings.auto_model_id or DEFAULT_AUTO_MODEL_ID
payload.pop("model", None)
LOG.info(
"auto routing request_id=%s endpoint=%s",
context["request_id"], endpoint_key,
)
try:
payload = registry.apply_request(endpoint_key, payload, context)
except Exception as e: # noqa: BLE001
LOG.exception("request transform failed")
return openai_error(f"Request transform error: {e}", status_code=500, err_type="server_error")
if settings.allowed_params:
allowed = set(settings.allowed_params) | {"model", "messages", "prompt", "input", "stream"}
payload = {k: v for k, v in payload.items() if k in allowed}
elif settings.strip_unknown_params:
LOG.debug("strip_unknown_params set but no allow-list provided; skipping strip")
method = request.method
query = request.url.query
stream_requested = bool(is_json and isinstance(payload, dict) and payload.get("stream"))
auto_mode = bool(context.get("auto_mode"))
providers = settings.providers
if not providers:
return openai_error(
"No upstream providers configured.",
status_code=502,
err_type="server_error",
)
attempt_plan = build_attempt_plan(
providers,
payload if is_json else None,
auto_mode=auto_mode,
auto_id=settings.auto_model_id,
)
if not attempt_plan:
return openai_error(
"No upstream models available for this request.",
status_code=404,
err_type="invalid_request_error",
param="model",
code="model_not_found",
)
last_fail: Optional[UpstreamAttemptFailed] = None
attempt = 0
client = request.app.state.http_client
skip_providers = set()
for provider, model in attempt_plan:
if provider.name in skip_providers:
continue
attempt += 1
context["provider"] = provider.name
context["upstream_model"] = model
context["attempt"] = attempt
if is_json and isinstance(payload, dict):
attempt_payload = copy.deepcopy(payload)
if model:
attempt_payload["model"] = model
else:
attempt_payload.pop("model", None)
body_to_send = json.dumps(attempt_payload).encode("utf-8")
else:
body_to_send = body_bytes
target_headers = build_outgoing_headers(
request, settings, registry, endpoint_key, context, provider
)
target = provider.url.rstrip("/") + "/" + sub_path.lstrip("/")
timeout = provider.timeout if provider.timeout is not None else settings.timeout
if settings.verbose or auto_mode:
LOG.info(
"-> %s %s (provider=%s model=%s endpoint=%s stream=%s auto=%s attempt=%s)",
method, target, provider.name, model, endpoint_key,
stream_requested, auto_mode, attempt,
)
if last_fail is not None and settings.retry_delay > 0:
await asyncio.sleep(settings.retry_delay)
try:
if stream_requested:
return await handle_streaming(
client, method, target, query, target_headers, body_to_send,
endpoint_key, context, registry, timeout,
)
return await handle_non_streaming(
client, method, target, query, target_headers, body_to_send,
endpoint_key, context, registry, timeout,
)
except UpstreamAttemptFailed as e:
LOG.warning(
"upstream attempt failed provider=%s model=%s status=%s skip_provider=%s: %s",
provider.name, model, e.status_code, e.skip_provider, e.reason,
)
last_fail = e
if e.skip_provider:
skip_providers.add(provider.name)
except httpx.ConnectError as e:
LOG.warning(
"could not connect to provider=%s (%s): %s",
provider.name, target, e,
)
last_fail = UpstreamAttemptFailed(
f"Could not connect to upstream target: {e}",
status_code=502,
skip_provider=True,
)
skip_providers.add(provider.name)
except httpx.TimeoutException:
LOG.warning(
"upstream timed out provider=%s model=%s",
provider.name, model,
)
last_fail = UpstreamAttemptFailed(
"Upstream request timed out.",
status_code=504,
)
except Exception as e: # noqa: BLE001
LOG.exception(
"proxy request failed provider=%s model=%s",
provider.name, model,
)
last_fail = UpstreamAttemptFailed(
f"Proxy error: {e}",
status_code=500,
)
if last_fail is not None:
LOG.error(
"all upstream attempts exhausted (%s) last_error=%s",
attempt, last_fail.reason,
)
return last_fail.to_response()
return openai_error(
"All upstream providers failed.",
status_code=502,
err_type="server_error",
)
async def handle_non_streaming(
client, method, target, query, headers, body, endpoint_key, context, registry, timeout,
):
timeout_obj = make_httpx_timeout(timeout, streaming=False)
resp = await client.request(
method, target, params=query or None, headers=headers,
content=body if body else None, timeout=timeout_obj,
)
response_headers = {
k: v for k, v in resp.headers.items()
if k.lower() not in HOP_BY_HOP_HEADERS
}
response_headers.update(proxy_info_headers(context))
media_type = resp.headers.get("content-type", "application/json")
if "application/json" in media_type:
try:
data = resp.json()
except ValueError:
if resp.is_success:
return Response(
content=resp.content, status_code=resp.status_code,
headers=response_headers, media_type=media_type,
)
raise UpstreamAttemptFailed(
reason=f"Upstream returned HTTP {resp.status_code}",
status_code=resp.status_code,
raw_content=resp.content,
media_type=media_type,
response_headers=response_headers,
skip_provider=classify_upstream_failure(resp.status_code, None) == "skip_provider",
)
if resp.is_success:
try:
data = registry.apply_response(endpoint_key, data, context)
except Exception: # noqa: BLE001
LOG.exception("response transform failed")
return JSONResponse(content=data, status_code=resp.status_code, headers=response_headers)
raise UpstreamAttemptFailed(
reason=f"Upstream returned HTTP {resp.status_code}",
status_code=resp.status_code,
body=data,
response_headers=response_headers,
skip_provider=classify_upstream_failure(resp.status_code, data) == "skip_provider",
)
if resp.is_success:
return Response(
content=resp.content, status_code=resp.status_code,
headers=response_headers, media_type=media_type,
)
raise UpstreamAttemptFailed(
reason=f"Upstream returned HTTP {resp.status_code}",
status_code=resp.status_code,
raw_content=resp.content,
media_type=media_type,
response_headers=response_headers,
skip_provider=classify_upstream_failure(resp.status_code, None) == "skip_provider",
)
async def handle_streaming(
client, method, target, query, headers, body, endpoint_key, context, registry, timeout,
):
# IMPORTANT: do not pass timeout= into client.send().
# Attach it to the Request via build_request() / extensions.
timeout_obj = make_httpx_timeout(timeout, streaming=True)
req = build_upstream_request(
client,
method,
target,
params=query or None,
headers=headers,
content=body if body else None,
timeout=timeout_obj,
)
upstream_resp = await client.send(req, stream=True)
if not upstream_resp.is_success:
content = await upstream_resp.aread()
status = upstream_resp.status_code
await upstream_resp.aclose()
try:
data = json.loads(content)
except json.JSONDecodeError:
data = {"error": {"message": content.decode(errors="replace"), "type": "server_error"}}
raise UpstreamAttemptFailed(
reason=f"Upstream returned HTTP {status}",
status_code=status,
body=data,
skip_provider=classify_upstream_failure(status, data) == "skip_provider",
)
response_headers = {
k: v for k, v in upstream_resp.headers.items()
if k.lower() not in HOP_BY_HOP_HEADERS
}
response_headers["cache-control"] = "no-cache"
response_headers["x-accel-buffering"] = "no"
response_headers.update(proxy_info_headers(context))
async def sse_generator():
try:
async for raw_line in upstream_resp.aiter_lines():
if raw_line is None:
continue
if raw_line == "":
yield b"\n"
continue
if raw_line.startswith("data:"):
data_str = raw_line[len("data:"):].strip()
if data_str == "[DONE]":
yield b"data: [DONE]\n\n"
continue
try:
chunk_obj = json.loads(data_str)
except json.JSONDecodeError:
yield (raw_line + "\n").encode("utf-8")
continue
try:
chunk_obj = registry.apply_stream_chunk(endpoint_key, chunk_obj, context)
except Exception: # noqa: BLE001
LOG.exception("stream chunk transform failed")
yield f"data: {json.dumps(chunk_obj)}\n\n".encode("utf-8")
else:
yield (raw_line + "\n").encode("utf-8")
finally:
await upstream_resp.aclose()
return StreamingResponse(
sse_generator(), status_code=upstream_resp.status_code,
headers=response_headers, media_type="text/event-stream",
)
@app.api_route("/v1/{sub_path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])
async def v1_catch_all(sub_path: str, request: Request):
return await proxy_request(request, sub_path)
# Also accept requests without the leading /v1 in case a client is
# configured with a base_url that already omits it.
@app.api_route("/{sub_path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])
async def root_catch_all(sub_path: str, request: Request):
if sub_path in ("health", "favicon.ico"):
return openai_error("Not found", status_code=404, err_type="invalid_request_error")
return await proxy_request(request, sub_path)
return app
# ==========================================================================
# CLI
# ==========================================================================
def build_arg_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="openai_proxy.py",
description="Run a local OpenAI-API-compatible proxy server in front of any "
"OpenAI-compatible AI provider with priority ordering, pluggable custom parameters "
"and multi-provider / multi-LLM failover.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__,
)
p.add_argument("--host", default=os.environ.get("PROXY_HOST", "127.0.0.1"),
help="Host/interface to bind to (default: 127.0.0.1)")
p.add_argument("--port", type=int, default=int(os.environ.get("PROXY_PORT", "8000")),
help="Port to listen on (default: 8000)")
p.add_argument("--target-url", default=os.environ.get("OPENAI_BASE_URL", ""),
help="Base URL of a single upstream OpenAI-compatible API, e.g. "
"https://api.groq.com/openai/v1. Used when no --provider / "
"config providers are given (backward compatible).")
p.add_argument("--target-api-key", default=os.environ.get("OPENAI_API_KEY"),
help="API key to use when calling the --target-url provider")
p.add_argument("--provider", action="append", default=[],
help="Add an upstream provider. Repeatable; order is priority "
"(first = highest) unless 'priority=' is set. Format: "
"name=groq,url=https://...,api_key=...,models=m1|m2 "
"or a JSON object. models/llms are tried in listed order; "
"if all fail the next provider is used.")
p.add_argument("--proxy-api-key", default=os.environ.get("PROXY_API_KEY"),
help="If set, clients must send this key as their Bearer token "
"to use this local proxy")
p.add_argument("--forward-client-auth", action="store_true",
help="Forward the client's own Authorization header upstream "
"instead of / before applying the provider API key")
p.add_argument("--timeout", type=float, default=600.0,
help="Upstream request timeout in seconds (default: 600)")
p.add_argument("--retry-delay", type=float, default=0.0,
help="Seconds to wait between failover attempts (default: 0)")
p.add_argument("--cors-origins", default="*",
help="Comma-separated list of allowed CORS origins (default: *)")
p.add_argument("--config", default=None,
help="Path to a JSON config file with providers and declarative "
"transforms (providers, model_map, inject_params, "
"rename_params, remove_params, extra_headers, "
"allowed_params, strip_unknown_params)")
p.add_argument("--plugin", action="append", default=[],
help="Path to a Python plugin file defining register(registry, "
"settings, config). Can be passed multiple times.")
p.add_argument("--model-map", default=None,
help="Inline JSON object mapping requested model names to "
"upstream model names, e.g. '{\"gpt-4o\":\"llama-3.3-70b\"}'")
p.add_argument("--inject-param", action="append", default=[],
help="Inject a default request parameter. Format: "
"[endpoint:]key=value (value parsed as JSON if possible). "
"Repeatable. endpoint defaults to '*' (all endpoints). "
"Example: --inject-param chat.completions:top_p=0.9")
p.add_argument("--remove-param", action="append", default=[],
help="Remove a request parameter before forwarding. Format: "
"[endpoint:]key. Repeatable.")
p.add_argument("--rename-param", action="append", default=[],
help="Rename a request parameter before forwarding. Format: "
"[endpoint:]old=new. Repeatable.")
p.add_argument("--extra-header", action="append", default=[],
help="Extra header to send upstream. Format: key=value or "
"key:value. Repeatable.")
p.add_argument("--allowed-params", default=None,
help="Comma-separated whitelist of extra params allowed through "
"(model/messages/prompt/input/stream are always allowed). "
"If unset, all params are passed through untouched.")
p.add_argument("--no-restore-model-name", action="store_true",
help="Do not rewrite the response 'model' field back to the "
"client-requested name when --model-map / auto is used")
p.add_argument("--auto-model-id", default=None,
help="Id of the virtual router model advertised on /v1/models "
f"(default: {DEFAULT_AUTO_MODEL_ID})")
p.add_argument("--no-auto-model", action="store_true",
help="Do not advertise the virtual 'auto' model on /v1/models")
p.add_argument("--no-merge-upstream-models", action="store_true",
help="Do not merge GET /v1/models results from upstream providers; "
"only return 'auto' plus models listed in --provider / config")
p.add_argument("--log-level", default="info",
choices=["debug", "info", "warning", "error", "critical"],
help="Logging verbosity (default: info)")
p.add_argument("--verbose", action="store_true",
help="Log every proxied request")
p.add_argument("--reload", action="store_true",
help="Enable uvicorn auto-reload (development only)")
p.add_argument("--init-config", metavar="PATH",
help="Write a sample config JSON file to PATH and exit")
p.add_argument("--init-plugin", metavar="PATH",
help="Write a sample plugin Python file to PATH and exit")
return p
def _collect_providers(args, config: Dict[str, Any]) -> List[Provider]:
"""Build the priority-ordered provider list from config + CLI."""
providers: List[Provider] = []
for i, raw in enumerate(config.get("providers") or []):
if not isinstance(raw, dict):
raise ValueError(f"config providers[{i}] must be an object")
provider = provider_from_dict(raw, default_name=f"provider-{i + 1}")
if raw.get("priority") is None:
provider.priority = i
providers.append(provider)
for i, spec in enumerate(args.provider or []):
raw = parse_provider_spec(spec)
provider = provider_from_dict(raw, default_name=f"cli-provider-{i + 1}")
if raw.get("priority") is None:
provider.priority = 1000 + i
providers.append(provider)
if not providers and args.target_url:
providers.append(
Provider(
name="default",
url=args.target_url,
api_key=args.target_api_key,
models=[],
priority=0,
)
)
elif args.target_url and providers:
LOG.warning(
"--target-url is ignored because one or more providers were "
"configured via --provider / config"
)
providers.sort(key=lambda p: p.priority)
for provider in providers:
if not provider.url:
raise ValueError(f"Provider {provider.name!r} has an empty url")
return providers
def main(argv: Optional[List[str]] = None) -> None:
# Load .env from the current directory, falling back to the user's home
# directory if none is found locally. Existing process env vars take
# precedence over .env values (load_dotenv override=False by default).
cwd_env = os.path.join(os.getcwd(), ".env")
home_env = os.path.join(os.path.expanduser("~"), ".env")
if os.path.isfile(cwd_env):
print("File: .env found in **current** directory.")
load_dotenv(cwd_env)
elif os.path.isfile(home_env):
print("File: .env found in **home** directory.")
load_dotenv(home_env)
parser = build_arg_parser()
args = parser.parse_args(argv)
if args.init_config:
with open(args.init_config, "w", encoding="utf-8") as f:
json.dump(SAMPLE_CONFIG, f, indent=2)
print(f"Sample config written to {args.init_config}")
return
if args.init_plugin:
with open(args.init_plugin, "w", encoding="utf-8") as f:
f.write(SAMPLE_PLUGIN)
print(f"Sample plugin written to {args.init_plugin}")
return
logging.basicConfig(
level=getattr(logging, args.log_level.upper()),
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
config: Dict[str, Any] = {}
if args.config:
config = load_config_file(args.config)
try:
providers = _collect_providers(args, config)
except (ValueError, json.JSONDecodeError) as e:
parser.error(str(e))
if not providers:
parser.error(
"at least one upstream is required: pass --target-url "
"(or set OPENAI_BASE_URL), --provider, or a config file with a "
"providers list. Use --init-config/--init-plugin if you just "
"want sample files."
)
auto_model_id = (
args.auto_model_id
or config.get("auto_model_id")
or DEFAULT_AUTO_MODEL_ID
)
advertise_auto_model = (not args.no_auto_model) and bool(
config.get("advertise_auto_model", True)
)
merge_upstream_models = (not args.no_merge_upstream_models) and bool(
config.get("merge_upstream_models", True)
)
settings = Settings(
host=args.host,
port=args.port,
target_url=providers[0].url,
target_api_key=args.target_api_key or providers[0].api_key,
proxy_api_key=args.proxy_api_key,
timeout=args.timeout,
cors_origins=[o.strip() for o in args.cors_origins.split(",")] if args.cors_origins else ["*"],
verbose=args.verbose,
forward_client_auth=args.forward_client_auth,
allowed_params=[s.strip() for s in args.allowed_params.split(",")] if args.allowed_params else None,
restore_model_name=not args.no_restore_model_name,
providers=providers,
retry_delay=args.retry_delay,
auto_model_id=str(auto_model_id),
advertise_auto_model=advertise_auto_model,
merge_upstream_models=merge_upstream_models,
)
if args.model_map:
config.setdefault("model_map", {}).update(json.loads(args.model_map))
if args.extra_header:
config.setdefault("extra_headers", {}).update(parse_kv_pairs(args.extra_header))
for spec in args.inject_param:
endpoint, key, value = parse_endpoint_param(spec)
config.setdefault("inject_params", {}).setdefault(endpoint, {})[key] = value
for spec in args.remove_param:
if ":" in spec:
endpoint, key = spec.split(":", 1)
else:
endpoint, key = "*", spec
config.setdefault("remove_params", {}).setdefault(endpoint.strip(), []).append(key.strip())
for spec in args.rename_param:
endpoint = "*"
rest = spec
if ":" in spec and "=" in spec and spec.index(":") < spec.index("="):
endpoint, rest = spec.split(":", 1)
if "=" not in rest:
parser.error(f"Invalid --rename-param value: {spec!r} (expected [endpoint:]old=new)")
old, new = rest.split("=", 1)
config.setdefault("rename_params", {}).setdefault(endpoint.strip(), {})[old.strip()] = new.strip()
if "restore_model_name" not in config:
config["restore_model_name"] = settings.restore_model_name
registry = build_registry_from_config(config, settings)
for plugin_path in args.plugin:
load_plugin_file(plugin_path, registry, settings, config)
app = create_app(settings, registry)
chain = " | ".join(
f"{p.name}[{', '.join(p.models) if p.models else '*'}]"
for p in settings.providers
)
LOG.info(
"Starting OpenAI-compatible proxy on http://%s:%s -> %s",
settings.host, settings.port, chain,
)
if settings.advertise_auto_model:
LOG.info(
"Virtual model %r is advertised on GET /v1/models and runs the failover chain",
settings.auto_model_id,
)
if settings.proxy_api_key:
LOG.info("Local proxy requires clients to send Authorization: Bearer <proxy_api_key>")
uvicorn.run(app, host=settings.host, port=settings.port,
log_level=args.log_level, reload=args.reload)
if __name__ == "__main__":
main()
@bitsnaps

bitsnaps commented Aug 16, 2026 •
edited
Loading

Copy link
Copy Markdown
Author

Notes:

  • The API keys will be loaded from var env either from .env in the current directory or from the home user dir.
  • You can use any OpenAI Compatible API (not just Groq as shown in the usage demo).
  • The only required packages are: fastapi uvicorn httpx.
  • Read more by running: python3 openai_proxy.py --help command.
  • Select the virtual "auto" model switch between models automatically by priority order.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment