#!/usr/bin/env python3 """ Alcozaky AI Proxy — OpenAI-compatible proxy Routes & rotates between multiple free AI providers. Usage: python3 /opt/ai-proxy/proxy.py """ import os, json, time, asyncio, random, hashlib from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse, StreamingResponse from openai import AsyncOpenAI import httpx app = FastAPI(title="Alcozaky AI Proxy") # ============================================================ # PROVIDERS CONFIG # ============================================================ PROVIDERS = [ { "name": "openrouter-1", "client": lambda: AsyncOpenAI( api_key=os.getenv("OPENROUTER_API_KEY_1"), base_url="https://openrouter.ai/api/v1" ), "models": [ "openai/gpt-4o-mini", "google/gemini-2.0-flash-001", "deepseek/deepseek-chat", "anthropic/claude-3-haiku", "meta-llama/llama-3.2-3b-instruct:free", ], "priority": 1, }, { "name": "openrouter-2", "client": lambda: AsyncOpenAI( api_key=os.getenv("OPENROUTER_API_KEY_2"), base_url="https://openrouter.ai/api/v1" ), "models": [ "openai/gpt-4o-mini", "meta-llama/llama-3.2-3b-instruct:free", "google/gemini-2.0-flash-001", ], "priority": 2, }, { "name": "openrouter-3", "client": lambda: AsyncOpenAI( api_key=os.getenv("OPENROUTER_API_KEY_3"), base_url="https://openrouter.ai/api/v1" ), "models": [ "openai/gpt-4o-mini", "deepseek/deepseek-chat", "meta-llama/llama-3.2-3b-instruct:free", ], "priority": 3, }, { "name": "github-1", "client": lambda: AsyncOpenAI( api_key=os.getenv("GITHUB_TOKEN_1"), base_url="https://models.inference.ai.azure.com" ), "models": ["gpt-4o-mini", "gpt-4o"], "priority": 4, }, { "name": "github-2", "client": lambda: AsyncOpenAI( api_key=os.getenv("GITHUB_TOKEN_2"), base_url="https://models.inference.ai.azure.com" ), "models": ["gpt-4o-mini"], "priority": 5, }, { "name": "github-3", "client": lambda: AsyncOpenAI( api_key=os.getenv("GITHUB_TOKEN_3"), base_url="https://models.inference.ai.azure.com" ), "models": ["gpt-4o-mini"], "priority": 6, }, { "name": "openrouter-2", "client": lambda: AsyncOpenAI( api_key=os.getenv("OPENROUTER_API_KEY_2"), base_url="https://openrouter.ai/api/v1" ), "models": [ "openai/gpt-4o-mini", "meta-llama/llama-3.2-3b-instruct:free", "anthropic/claude-3-haiku", ], "priority": 2, }, { "name": "groq-1", "client": lambda: AsyncOpenAI( api_key=os.getenv("GROQ_API_KEY_1"), base_url="https://api.groq.com/openai/v1" ), "models": [ "llama-3.3-70b-versatile", "mixtral-8x7b-32768", "llama-3.1-8b-instant", ], "priority": 10, }, { "name": "groq-2", "client": lambda: AsyncOpenAI( api_key=os.getenv("GROQ_API_KEY_2"), base_url="https://api.groq.com/openai/v1" ), "models": [ "llama-3.1-8b-instant", ], "priority": 11, }, { "name": "groq-3", "client": lambda: AsyncOpenAI( api_key=os.getenv("GROQ_API_KEY_3"), base_url="https://api.groq.com/openai/v1" ), "models": [ "llama-3.1-8b-instant", "llama-3.3-70b-versatile", ], "priority": 12, }, { "name": "deepseek", "client": lambda: AsyncOpenAI( api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com/v1" ), "models": ["deepseek-chat"], "priority": 30, }, { "name": "together", "client": lambda: AsyncOpenAI( api_key=os.getenv("TOGETHER_API_KEY"), base_url="https://api.together.xyz/v1" ), "models": ["meta-llama/Llama-3.2-3B-Instruct-Turbo"], "priority": 40, }, { "name": "huggingface", "client": lambda: AsyncOpenAI( api_key=os.getenv("HUGGINGFACE_API_KEY"), base_url="https://api-inference.huggingface.co/v1" ), "models": ["meta-llama/Llama-3.2-3B-Instruct"], "priority": 50, }, ] # Filter providers with valid API keys active_providers = [] for p in PROVIDERS: try: c = p["client"]() if c.api_key: active_providers.append(p) except Exception: pass provider_failures = {} # name -> timestamp of last failure provider_cooldowns = {} # name -> cooldown until # Model alias mapping MODEL_ALIASES = { "gpt-4o-mini": "openai/gpt-4o-mini", "gpt4": "openai/gpt-4o-mini", "fast": "llama-3.3-70b-versatile", "cheap": "meta-llama/Llama-3.2-3B-Instruct-Turbo", "gemini": "gemini-2.0-flash-exp", "claude": "anthropic/claude-3-haiku", } def get_active_providers(): """Return providers sorted by priority, excluding those in cooldown.""" now = time.time() result = [] for p in sorted(active_providers, key=lambda x: x["priority"]): # Check cooldown if p["name"] in provider_cooldowns: if now < provider_cooldowns[p["name"]]: continue # Still in cooldown else: del provider_cooldowns[p["name"]] result.append(p) return result def resolve_model(model_name): """Resolve model alias to actual model name.""" return MODEL_ALIASES.get(model_name, model_name) @app.get("/health") @app.get("/") async def health(): return {"status": "ok", "providers": len(active_providers)} @app.get("/v1/models") async def list_models(): models = [] for p in get_active_providers(): for m in p["models"]: models.append({ "id": m, "provider": p["name"], "object": "model", "created": int(time.time()), "owned_by": p["name"], }) return {"object": "list", "data": models} @app.post("/v1/chat/completions") async def chat_completions(request: Request): body = await request.json() model = resolve_model(body.get("model", "gpt-4o-mini")) messages = body.get("messages", []) stream = body.get("stream", False) max_tokens = body.get("max_tokens", 4096) temperature = body.get("temperature", 0.7) # Try each provider providers = get_active_providers() first_provider = None last_error = None for provider in providers: if not first_provider: first_provider = provider client = provider["client"]() actual_model = model # Check if this provider has this model # For OpenRouter: model names include prefix (e.g. openai/gpt-4o-mini) # For others: use first available model if model not found if model not in provider["models"]: # Try to find closest match or use first model if model.startswith("openai/") or model.startswith("google/") or model.startswith("anthropic/"): # Only OpenRouter supports these prefixed models if "openrouter" not in provider["name"]: continue else: # Non-prefixed model - check if any provider model contains it matches = [m for m in provider["models"] if model in m] if not matches: continue actual_model = matches[0] try: if stream: return await stream_response(client, actual_model, messages, max_tokens, temperature) else: response = await client.chat.completions.create( model=actual_model, messages=messages, max_tokens=max_tokens, temperature=temperature, ) # Success! Reset failure count if provider["name"] in provider_failures: del provider_failures[provider["name"]] return JSONResponse(content=response.model_dump()) except Exception as e: last_error = str(e) provider_failures[provider["name"]] = time.time() # Cooldown for 30 seconds provider_cooldowns[provider["name"]] = time.time() + 30 continue # All providers failed raise HTTPException( status_code=503, detail={ "error": "All providers failed", "last_error": last_error, "active_providers": len(providers), } ) async def stream_response(client, model, messages, max_tokens, temperature): """Handle streaming responses.""" response = await client.chat.completions.create( model=model, messages=messages, max_tokens=max_tokens, temperature=temperature, stream=True, ) async def generate(): async for chunk in response: yield f"data: {chunk.model_dump_json()}\n\n" yield "data: [DONE]\n\n" return StreamingResponse( generate(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", } ) if __name__ == "__main__": import uvicorn port = int(os.getenv("PORT", "4000")) host = os.getenv("HOST", "0.0.0.0") print(f"🚀 Alcozaky AI Proxy starting on {host}:{port}") print(f"📡 Providers loaded: {len(active_providers)}") for p in active_providers: print(f" ├ {p['name']}: {len(p['models'])} models") print(f"🔀 Rotation: latency-based auto-fallback") uvicorn.run(app, host=host, port=port)