🛡️
● API Online
LiteLLM Gateway API
Intelligent LLM routing with DeBERTa PII shielding.
Connect your Streamlit dashboard to start routing queries.
Open Dashboard →
API Docs
""", status_code=200)
# -----------------------------------------------------------
# INFRASTRUCTURE
# -----------------------------------------------------------
@self.app.get("/health")
async def health():
"""Health check for Railway / Kubernetes / Fly.io."""
return {
"status": "healthy",
"timestamp": time.time(),
"routing_strategy": self.config.routing_strategy,
"endpoints_loaded": len(self.config.endpoints)
}
@self.app.get("/metrics")
async def metrics():
"""Real-time load-balancing and token metrics."""
return {
"timestamp": time.time(),
"metrics": self.router.get_metrics(),
"active_routing_rules": {
"strategy": self.config.routing_strategy,
"retries": self.config.num_retries,
"timeout": self.config.timeout
},
"registered_virtual_models": list(set(e.model_name for e in self.config.endpoints)),
"registered_physical_providers": list(set(e.model.split("/")[0] for e in self.config.endpoints))
}
# -----------------------------------------------------------
# LLM API ENDPOINTS
# -----------------------------------------------------------
@self.app.get("/v1/models")
async def list_models():
"""Lists virtual and physical backend models."""
data = []
for vm in set(e.model_name for e in self.config.endpoints):
data.append({
"id": vm, "object": "model", "created": 1686935002,
"owned_by": "proxy-system", "type": "virtual"
})
for ep in self.config.endpoints:
data.append({
"id": ep.model, "object": "model", "created": 1686935002,
"owned_by": ep.model.split("/")[0], "type": "physical",
"tpm_limit": ep.tpm, "rpm_limit": ep.rpm, "tpr_limit": ep.tpr
})
return {"object": "list", "data": data}
@self.app.post("/v1/chat/completions")
async def chat_completions(request: ChatCompletionRequest):
"""OpenAI-compatible chat completion with load balancing and PII shielding."""
messages_dict = [{"role": m.role, "content": m.content} for m in request.messages]
virtual_models = set(e.model_name for e in self.config.endpoints)
if request.model not in virtual_models:
raise HTTPException(
status_code=404,
detail=f"Model '{request.model}' not found. Available: {list(virtual_models)}"
)
try:
return await self.router.execute_chat_completion(
model=request.model,
messages=messages_dict,
temperature=request.temperature,
max_tokens=request.max_tokens,
mock_sandbox=request.mock_sandbox
)
except ValueError as ve:
raise HTTPException(status_code=400, detail=str(ve))
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# -----------------------------------------------------------
# PREFERENCE / CREDIT ROUTING ENDPOINTS
# -----------------------------------------------------------
@self.app.get("/preference-config")
async def get_preference_config():
return {
"preference_enabled": self.router.preference_enabled,
"preference_list": self.router.preference_list,
"credit_limits": self.router.credit_limits,
"accumulated_spend": self.router.accumulated_spend,
"available_physical_models": list(set(e.model for e in self.config.endpoints))
}
@self.app.post("/preference-config")
async def update_preference_config(config: dict = Body(...)):
self.router.preference_enabled = config.get("preference_enabled", self.router.preference_enabled)
self.router.preference_list = config.get("preference_list", self.router.preference_list)
for m, limit in config.get("credit_limits", {}).items():
self.router.credit_limits[m] = float(limit)
return {"status": "success", "message": "Preference routing configuration synchronized."}
@self.app.post("/preference-config/reset")
async def reset_preference_spend():
for m in list(self.router.accumulated_spend.keys()):
self.router.accumulated_spend[m] = 0.0
return {"status": "success", "message": "Spend counters reset to zero."}
# -----------------------------------------------------------
# PII GUARDRAIL CONFIGURATION ENDPOINTS
# -----------------------------------------------------------
@self.app.get("/ui/pii-config")
async def get_pii_config():
return {
"pii_enabled": getattr(self.router, "pii_enabled", False),
"pii_action": getattr(self.router, "pii_action", "MASK"),
"pii_policy": getattr(self.router, "pii_policy", None)
}
@self.app.post("/ui/pii-config")
async def update_pii_config(config: UIPiiConfig):
self.router.pii_enabled = config.pii_enabled
self.router.pii_action = config.pii_action
self.router.pii_policy = config.pii_policy
return {
"status": "success",
"message": f"PII Guardrail updated: enabled={config.pii_enabled}, action={config.pii_action}."
}
def get_app(self) -> FastAPI:
"""Returns the FastAPI instance for use with ASGI servers."""
return self.app