Unified AI Proxy
This article describes setting up a unified AI proxy using LiteLLM router that routes requests to different AI models
1. Introduction#
- Unified AI Proxy acts as a centralized interface between models from OpenAI and Anthropic.
- Uses LiteLLM router
- Provides a single, unified OpenAI-compatible API surface through which you can:
- Automatically route requests to different LLM models
- Use a fallback strategy if the primary model fails
- Manage user permissions using a MySQL database and Redis cache
- Use the same OpenAI clients with other models, such as Claude
2. Architecture#
The proxy works like the OpenAI API (/v1/chat/completions), but:
- Supports provider field to select a model ("claude" or "openai")
- Automatically handles user permissions (user_id from Redis / MariaDB)
- Supports non-streaming responses
- Streaming not tested
3. Main code structure#
- main.py – Main service
# file: main.py
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse, StreamingResponse
from litellm import Router
import os, json, asyncio
import mysql.connector
import redis.asyncio as aioredis
app = FastAPI(title="Unified AI Proxy")
# Loading the model list and fallbacks from ConfigMap (set in Kubernetes with environment variables)
model_list = json.loads(os.getenv("MODELS_JSON", "[]"))
fallbacks = json.loads(os.getenv("FALLBACKS_JSON", "[]"))
# Redis cache (user permissions and caches)
REDIS_HOST = os.getenv("REDIS_HOST", "redis-proxy")
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "")
redis_client = aioredis.Redis(host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD, decode_responses=True)
# LiteLLM Router is responsible for routing and fallbacks for different models
router = Router(
model_list=model_list,
fallbacks=fallbacks,
redis_host=REDIS_HOST,
redis_port=REDIS_PORT,
redis_password=REDIS_PASSWORD,
cache_responses=True,
num_retries=3,
routing_strategy="simple-shuffle"
)
# User management (checking if the user has access)
async def get_allowed_users():
# Tarkistetaan ensin Redisistä
cached = await redis_client.get("allowed_users")
if cached:
return set(json.loads(cached))
# If not cached, fetch from MariaDB and store in Redis cache
try:
connection = mysql.connector.connect(
host=os.getenv("DB_HOST"),
user=os.getenv("DB_USER"),
password=os.getenv("DB_PASSWORD"),
database=os.getenv("DB_NAME")
)
cursor = connection.cursor()
cursor.execute("SELECT user_id FROM allowed_users;")
users = set(row[0] for row in cursor.fetchall())
await redis_client.set("allowed_users", json.dumps(list(users)), ex=60) # TTL 60s
return users
except Exception as e:
print(f"Error connecting to MariaDB: {e}")
return set()
finally:
if 'connection' in locals() and connection.is_connected():
cursor.close()
connection.close()
# OpenAI-compatible main endpoint
@app.post("/v1/chat/completions")
async def chat(request: Request):
"""
A unified endpoint for all ChatCompletion requests.
Uses the provider parameter to select a template.
"""
data = await request.json()
user_id = data.get("user_id")
provider = data.get("provider")
messages = data.get("messages", [])
stream = data.get("stream", False)
model = data.get("model", "openai-gpt-4o-mini")
# Model selection if provider given
if provider == "claude":
model = "claude-opus-4-1-20250805"
elif provider == "openai":
model = "openai-gpt-4o-mini"
# Checking access rights
allowed_users = await get_allowed_users()
if user_id not in allowed_users:
raise HTTPException(status_code=403, detail="Access denied")
# Request processing (stream or non-stream)
try:
if stream:
async def stream_response():
response = await router.acompletion(model=model, messages=messages, stream=True)
async for chunk in response:
yield json.dumps(chunk) + "\n"
await asyncio.sleep(0)
return StreamingResponse(stream_response(), media_type="application/json")
else:
response = await router.acompletion(model=model, messages=messages)
if hasattr(response, "model_dump"):
response = response.model_dump()
return JSONResponse(content=response)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# Health check and metadata
@app.get("/")
def root():
return {
"message": "Unified AI Proxy with Router + Redis",
"models": [m["model_name"] for m in model_list],
"redis": REDIS_HOST,
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", 5000)))
4. Kubernetes deployment guide#
This guide describes how to install the Unified AI Proxy service on a Kubernetes cluster (Rahti OpenShift).
The service consists of four main components:
- Redis Proxy – caching and user list storage
- ConfigMap – template and configuration data
- Proxy Deployment + Service + Route – the actual FastAPI application
- Horizontal Pod Autoscaler (HPA) – automatic scaling based on load
Environment variables summary:
| Variable | Source | Description |
|---|---|---|
REDIS_HOST |
ConfigMap | Redis service name (default: redis-proxy) |
REDIS_PORT |
ConfigMap | Redis port (default: 6379) |
REDIS_PASSWORD |
ConfigMap / Secret | Password if needed |
DB_HOST |
Secret | MariaDB host |
DB_USER |
Secret | MariaDB username |
DB_PASSWORD |
Secret | MariaDB password |
DB_NAME |
Secret | MariaDB database name |
MODELS_JSON |
ConfigMap | List of supported models |
FALLBACKS_JSON |
ConfigMap | Model fallback mapping |
OPENAI_API_KEY |
Secret | API key for OpenAI |
ANTHROPIC_API_KEY |
Secret | API key for Anthropic |
1. Redis Proxy - redis-proxy.yaml#
- LiteLLM Router cache (cache for responses)
- user list (allowed_users) storage with 60 seconds TTL
# file: redis-proxy.yaml
apiVersion: v1
kind: Service
metadata:
name: redis-proxy
spec:
selector:
app: redis-proxy
ports:
- protocol: TCP
port: 6379
targetPort: 6379
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: redis-proxy
spec:
replicas: 1
selector:
matchLabels:
app: redis-proxy
template:
metadata:
labels:
app: redis-proxy
spec:
containers:
- name: redis-proxy
image: redis/redis-stack-server:latest
ports:
- containerPort: 6379
securityContext:
allowPrivilegeEscalation: false
capabilities:
drop: ["ALL"]
seccompProfile:
type: RuntimeDefault
volumeMounts:
- name: redis-data
mountPath: /data
volumes:
- name: redis-data
emptyDir: {}
redis/redis-stack-server:latestcontains Redis + RedisJSON + any additional modulesemptyDir: means that the data is only retained for the lifetime of the pod (enough for caching)- Redis is an internal service (ClusterIP) that is used by Proxy via the environment variable
REDIS_HOST=redis-proxy
2. ConfigMap - proxy-config.yaml#
- Contains all the environment variables that Proxy needs without sensitive information.
# file: proxy-config.yaml
apiVersion: v1
kind: ConfigMap
metadata:
name: proxy-config
data:
REDIS_HOST: "redis-proxy"
REDIS_PORT: "6379"
REDIS_PASSWORD: ""
MODELS_JSON: |
[
{
"model_name": "openai-gpt-4o-mini",
"litellm_params": {"model": "gpt-4o-mini"}
},
{
"model_name": "openai-gpt-4o",
"litellm_params": {"model": "gpt-4o"}
},
{
"model_name": "claude-opus-4",
"litellm_params": {"model": "claude-opus-4-1-20250805"}
},
{
"model_name": "claude-sonnet-3.7",
"litellm_params": {"model": "claude-3-7-sonnet-20250219"}
}
]
FALLBACKS_JSON: |
[
{"claude-opus-4": ["claude-sonnet-3.7"]},
{"openai-gpt-4o-mini": ["openai-gpt-4o"]}
]
MODELS_JSONdefines the models that LiteLLM Router knows and how they are mapped to LLM services.FALLBACKS_JSONtells which model to use as a fallback model if the primary model fails.- The proxy reads these JSONs from environment variables
os.getenv("MODELS_JSON")etc.
3. Proxy Deployment ja Route - proxy-deployment.yaml#
- This is the actual AI Proxy service that provides the /v1/chat/completions interface.
# file: proxy-deployment.yaml
apiVersion: v1
kind: Service
metadata:
name: proxy-service
spec:
selector:
app: proxy
ports:
- protocol: TCP
port: 5000
targetPort: 5000
type: ClusterIP
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: proxy
spec:
replicas: 1
selector:
matchLabels:
app: proxy
template:
metadata:
labels:
app: proxy
spec:
containers:
- name: prestashop
image: <IMAGE_URL>:latest
imagePullPolicy: Always
ports:
- containerPort: 5000
envFrom:
- secretRef:
name: ai-api-keys # Sisältää esim. OpenAI- ja Anthropic API-avaimet
- configMapRef:
name: proxy-config
- secretRef:
name: db-credentials # Sisältää MariaDB yhteystiedot
resources:
requests:
cpu: "100m"
memory: "256Mi"
limits:
cpu: "500m"
memory: "512Mi"
securityContext:
allowPrivilegeEscalation: false
capabilities:
drop: ["ALL"]
seccompProfile:
type: RuntimeDefault
imagePullSecrets:
- name: gitlab-registry-secret
---
apiVersion: route.openshift.io/v1
kind: Route
metadata:
name: proxy
spec:
to:
kind: Service
name: proxy-service
port:
targetPort: 5000
tls:
termination: edge
- Service: creates an internal DNS name proxy-service and port 5000 for the proxy
- Deployment: runs the Docker image ai-api-proxy:latest
- envFrom automatically combines ConfigMap and Secrets
- CPU/memory is limited and protected (allowPrivilegeEscalation: false)
- Route: OpenShift's external HTTPS route (terminates at the TLS edge)
4. Auto Scaling – proxy-hpa.yaml#
- This configures automatic horizontal scaling based on CPU load.
# file: proxy-hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: proxy-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: proxy
minReplicas: 1
maxReplicas: 6
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- If the proxy CPU load exceeds 70% on average, new pods are created (max 6)
- When the load decreases, pods are reduced
- This ensures scalability even when there are many simultaneous LLM calls
5. Secrets (not included in files)#
The proxy uses two Secret objects:
- ai-api-keys
Contains API keys for various model services:
kubectl create secret generic ai-api-keys \
--from-literal=OPENAI_API_KEY='sk-xxxx' \
--from-literal=ANTHROPIC_API_KEY='sk-ant-xxxx'
- db-credentials
Contains MariaDB database connection information:
kubectl create secret generic db-credentials \
--from-literal=DB_HOST=<DB_HOST/SERVER> \
--from-literal=DB_USER=<DB_USERNAME> \
--from-literal=DB_PASSWORD=<DB_PASSWORD> \
--from-literal=DB_NAME=<DATABASE_NAME>
5. Usage examples#
Example 1 — Basic Python request#
- This simple example shows how to make a direct POST request to the Unified AI Proxy in Python using the requests library.
- The proxy works like OpenAI's API, but supports additional fields such as provider (e.g. "claude" or "openai") and user_id,
- which are used for permissions management.
import requests
resp = requests.post("<PROXY_URL>/v1/chat/completions", json={
"user_id": "<ALLOWED-USERNAME>",
"provider": "claude",
"messages": [{"role": "user", "content": "Is this working?"}],
})
print(resp.json())
Example 2 — TypeScript / JavaScript fetch#
- Same idea as the previous one, but in a JavaScript environment.
- This works, for example, in Node.js or a browser when you want to call a proxy directly with the fetch function in an OpenAI-compatible way.
const response = await fetch("<PROXY_URL>/v1/chat/completions", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
user_id: "ae5971",
provider: "claude",
messages: [{ role: "user", content: "Is this working?" }],
}),
});
const data = await response.json();
console.log(data.choices[0].message);
Example 3 — Tool calls with external API (PrestaShop integration)#
- This example shows how to use a proxy with “tool calling” features.
- The proxy forwards calls to a model (e.g. Claude or OpenAI), which can decide to call an external interface – in this case the PrestaShop REST API.
-
This enables intelligent tools that retrieve data from an external service upon request from the model.
3.1 – FastAPI connector (PrestaShop API interface)
# prestashop-connector for a test of an actual tools
from fastapi import FastAPI
import requests
import os
app = FastAPI()
PRESTASHOP_URL = "<PRESTASHOP_URL>/api"
PRESTASHOP_KEY = os.getenv("PS_API_KEY")
# Gets resources from PrestaShop based on OpenAI's query
@app.get("/query/{RESOURCE}")
def get_resources(RESOURCE):
resource_url = f"{PRESTASHOP_URL}/{RESOURCE}?ws_key={PRESTASHOP_KEY}&output_format=JSON&display=full"
resource_r = requests.get(resource_url)
resource_r.raise_for_status()
data = resource_r.json()
return {"results": data}
3.2 – openai_proxy.py (tool call execution via Unified Proxy)
# file: openai_proxy.py
from openai import OpenAI
import requests
import os
import json
client = OpenAI(
api_key="proxy-key",
base_url="<PROXY_URL>"
)
def get_resource(resource):
url = f"http://localhost:8000/query/{resource}"
r = requests.get(url)
r.raise_for_status()
return r.json()
tools = [
{
"type": "function",
"function": {
"name": "get_resource",
"description": "Get information from PrestaShop and from the given resource",
"parameters": {
"type": "object",
"properties": {
"resource": {
"type": "string",
"enum": [
"addresses",
"attachments",
"attachments/file",
"carriers",
"cart_rules",
"carts",
"categories",
"combinations",
"configurations",
"contacts",
"content_management_system",
"countries",
"currencies",
"customer_messages",
"customer_threads",
"customers",
"customizations",
"deliveries",
"employees",
"groups",
"guests",
"image_types",
"images",
"images/general/header",
"images/general/mail",
"images/general/invoice",
"images/general/store_icon",
"images/products",
"images/categories",
"images/manufacturers",
"images/suppliers",
"images/stores",
"images/customizations",
"languages",
"manufacturers",
"messages",
"order_carriers",
"order_details",
"order_histories",
"order_invoices",
"order_payments",
"order_slip",
"order_states",
"orders",
"price_ranges",
"product_customization_fields",
"product_feature_values",
"product_features",
"product_option_values",
"product_options",
"product_suppliers",
"products",
"search",
"shop_groups",
"shop_urls",
"shops",
"specific_price_rules",
"specific_prices",
"states",
"stock_availables",
"stock_movement_reasons",
"stock_movements",
"stocks",
"stores",
"suppliers",
"supply_order_details",
"supply_order_histories",
"supply_order_receipt_histories",
"supply_order_states",
"supply_orders",
"tags",
"tax_rule_groups",
"tax_rules",
"taxes",
"translated_configurations",
"warehouse_product_locations",
"warehouses",
"weight_ranges",
"zones"
],
"description": "PrestaShop resource to be fetched."
}
},
"required": ["resource"]
}
}
}
]
def ask_question(question: str) -> str:
# Makes a query to AI and fetched information from PrestaShop resource if necessary.
messages = [
{"role": "system", "content": "You are an assistant who can search for information in PrestaShop."},
{"role": "user", "content": question}
]
while True:
response = client.chat.completions.create(
model="openai-gpt-4o-mini", # proxy bypasses if provider is given, but required for query
messages=messages,
tools=tools,
tool_choice="auto",
extra_body={
"user_id": "<ALLOWED-USERNAME>",
"provider": "claude" # "openai" or "claude"
}
)
msg = response.choices[0].message
if not msg.tool_calls:
# Return answer as string
return msg.content
# Add message to history and handle tools
messages.append({
"role": "assistant",
"content": msg.content,
"tool_calls": [tc.to_dict() for tc in msg.tool_calls]
})
for call in msg.tool_calls:
if call.function.name == "get_resource":
args = json.loads(call.function.arguments)
resource = args["resource"]
data = get_resource(resource)
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": json.dumps(data)
})
if __name__ == "__main__":
while True:
question = input("Question: ")
answer = ask_question(question)
print("Answer:", answer)
Example 4 — LangChain integration (ChatOpenAI)
- This example uses the LangChain library, which allows for building more complex AI agents and chained functions.
- Since the proxy supports an OpenAI-compatible interface, it can be used directly via the ChatOpenAI class without any additional modifications, even when using Claude
- Note: LangChain also includes the ChatLiteLLM class, but the current proxy still responds in OpenAI format, so ChatOpenAI works directly without any modifications.
from langchain_openai import ChatOpenAI
# Here ChatOpenAI is used.
# LangChain contains also a ChatLiteLLM, but it requires modifications on the proxy side especially when using Claude
# Currently proxy, when using ChatLiteLLM with Claude, returns information in a form that LiteLLM doesn't accept as is
llm = ChatOpenAI(
model="openai-gpt-4o-mini", # proxy bypasses if provider is given, but required for query
api_key="proxy-key",
base_url="<PROXY_URL>/v1",
extra_body={"user_id": "<ALLOWED-USERNAME>", "provider": "claude"},
)
response = llm.invoke("Is this working?")
print("Answer:", response.content)
6. Summary#
Unified AI Proxy enables a single, unified interface for different LLM services.
Its benefits include:
- Compatibility with OpenAI clients and possibility to use claude through that as well
- Permission and cache management (MariaDB + Redis)
- Fallback and routing strategies (LiteLLM Router)
- Scalable and secure Kubernetes implementation
Next steps / further development:
- Add auditing (who asked what)
- Use JWT token instead of
user_id - Add ChatLiteLLM compatibility for LangChain internal support
- Expand the list of models (e.g. Mistral, Gemini)