import os import secrets import httpx from fastapi import FastAPI, Request, HTTPException, Depends from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, HTMLResponse app = FastAPI() app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) # ── Auth ────────────────────────────────────────────────────────────────────── MOHAN_API_KEYS_RAW = os.environ.get("MOHAN_API_KEY", "") VALID_KEYS = set(k.strip() for k in MOHAN_API_KEYS_RAW.split(",") if k.strip()) def verify_key(key): return any(secrets.compare_digest(key.strip(), v) for v in VALID_KEYS) # ── Provider Keys (A = primary, B = backup) ─────────────────────────────────── GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "") GEMINI_API_KEY_B = os.environ.get("GEMINI_API_KEY_B", "") GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "") GROQ_API_KEY_B = os.environ.get("GROQ_API_KEY_B", "") CEREBRAS_API_KEY = os.environ.get("CEREBRAS_API_KEY", "") CEREBRAS_API_KEY_B = os.environ.get("CEREBRAS_API_KEY_B", "") MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY", "") MISTRAL_API_KEY_B = os.environ.get("MISTRAL_API_KEY_B", "") NVIDIA_API_KEY = os.environ.get("NVIDIA_API_KEY", "") NVIDIA_API_KEY_B = os.environ.get("NVIDIA_API_KEY_B", "") GEMINI_BASE = "https://generativelanguage.googleapis.com/v1beta/models" GROQ_BASE = "https://api.groq.com/openai/v1" CEREBRAS_BASE = "https://api.cerebras.ai/v1" MISTRAL_BASE = "https://api.mistral.ai/v1" NVIDIA_BASE = "https://integrate.api.nvidia.com/v1" # ── Models List (A = primary, B = backup with _b prefix) ───────────────────── ALL_MODELS = [ # ── Gemini (A) ── {"id": "gemini/gemini-2.5-flash", "name": "Gemini 2.5 Flash", "provider": "Gemini"}, {"id": "gemini/gemini-2.5-flash-lite", "name": "Gemini 2.5 Flash Lite", "provider": "Gemini"}, {"id": "gemini/gemini-3-flash-preview", "name": "Gemini 3 Flash Preview", "provider": "Gemini"}, # ── Gemini (B) ── {"id": "gemini_b/gemini-2.5-flash", "name": "Gemini 2.5 Flash [B]", "provider": "Gemini-B"}, {"id": "gemini_b/gemini-2.5-flash-lite", "name": "Gemini 2.5 Flash Lite [B]", "provider": "Gemini-B"}, {"id": "gemini_b/gemini-3-flash-preview", "name": "Gemini 3 Flash Preview [B]", "provider": "Gemini-B"}, # ── Groq (A) ── {"id": "groq/llama-3.1-8b-instant", "name": "LLaMA 3.1 8B Instant", "provider": "Groq"}, {"id": "groq/meta-llama/llama-4-scout-17b-16e-instruct", "name": "LLaMA 4 Scout 17B", "provider": "Groq"}, # ── Groq (B) ── {"id": "groq_b/llama-3.1-8b-instant", "name": "LLaMA 3.1 8B Instant [B]","provider": "Groq-B"}, {"id": "groq_b/meta-llama/llama-4-scout-17b-16e-instruct", "name": "LLaMA 4 Scout 17B [B]", "provider": "Groq-B"}, # ── Cerebras (A) ── {"id": "cerebras/gemma-4-31b", "name": "Gemma 4 31B (Cerebras ⚡)", "provider": "Cerebras"}, # ── Cerebras (B) ── {"id": "cerebras_b/gemma-4-31b", "name": "Gemma 4 31B (Cerebras ⚡) [B]", "provider": "Cerebras-B"}, # ── Mistral (A) ── {"id": "mistral/mistral-small-latest", "name": "Mistral Small 4", "provider": "Mistral"}, {"id": "mistral/open-mistral-nemo", "name": "Mistral Nemo", "provider": "Mistral"}, {"id": "mistral/ministral-8b-latest", "name": "Ministral 8B", "provider": "Mistral"}, {"id": "mistral/codestral-latest", "name": "Codestral (Code)", "provider": "Mistral"}, # ── Mistral (B) ── {"id": "mistral_b/mistral-small-latest", "name": "Mistral Small 4 [B]", "provider": "Mistral-B"}, {"id": "mistral_b/open-mistral-nemo", "name": "Mistral Nemo [B]", "provider": "Mistral-B"}, {"id": "mistral_b/ministral-8b-latest", "name": "Ministral 8B [B]", "provider": "Mistral-B"}, {"id": "mistral_b/codestral-latest", "name": "Codestral (Code) [B]", "provider": "Mistral-B"}, # ── NVIDIA (A) ── {"id": "nvidia/meta/llama-3.1-8b-instruct", "name": "LLaMA 3.1 8B (NIM)", "provider": "NVIDIA"}, {"id": "nvidia/meta/llama-3.3-70b-instruct", "name": "LLaMA 3.3 70B (NIM)", "provider": "NVIDIA"}, {"id": "nvidia/mistralai/mistral-small-4-119b-2603", "name": "Mistral Small 4 119B (NIM)", "provider": "NVIDIA"}, {"id": "nvidia/qwen/qwen2.5-coder-32b-instruct", "name": "Qwen2.5 Coder 32B (NIM)", "provider": "NVIDIA"}, {"id": "nvidia/deepseek-ai/deepseek-r1-distill-llama-8b", "name": "DeepSeek R1 8B (NIM)", "provider": "NVIDIA"}, # ── NVIDIA (B) ── {"id": "nvidia_b/meta/llama-3.1-8b-instruct", "name": "LLaMA 3.1 8B (NIM) [B]", "provider": "NVIDIA-B"}, {"id": "nvidia_b/meta/llama-3.3-70b-instruct", "name": "LLaMA 3.3 70B (NIM) [B]", "provider": "NVIDIA-B"}, {"id": "nvidia_b/mistralai/mistral-small-4-119b-2603", "name": "Mistral Small 4 119B (NIM) [B]","provider": "NVIDIA-B"}, {"id": "nvidia_b/qwen/qwen2.5-coder-32b-instruct", "name": "Qwen2.5 Coder 32B (NIM) [B]","provider": "NVIDIA-B"}, {"id": "nvidia_b/deepseek-ai/deepseek-r1-distill-llama-8b", "name": "DeepSeek R1 8B (NIM) [B]", "provider": "NVIDIA-B"}, ] # ── Provider detection helpers ──────────────────────────────────────────────── def is_groq(m): return m.startswith("groq/") def is_groq_b(m): return m.startswith("groq_b/") def is_gemini(m): return m.startswith("gemini/") def is_gemini_b(m): return m.startswith("gemini_b/") def is_cerebras(m): return m.startswith("cerebras/") def is_cerebras_b(m): return m.startswith("cerebras_b/") def is_mistral(m): return m.startswith("mistral/") def is_mistral_b(m): return m.startswith("mistral_b/") def is_nvidia(m): return m.startswith("nvidia/") def is_nvidia_b(m): return m.startswith("nvidia_b/") def strip_prefix(m): return m.split("/", 1)[1] def is_quota_error(status_code: int, body: dict) -> bool: """Detect quota/rate-limit errors from any provider.""" if status_code == 429: return True # Gemini quota error if status_code == 200: return False error_msg = str(body.get("error", "")).lower() if any(k in error_msg for k in ["quota", "rate limit", "rate_limit", "exceeded", "too many"]): return True return False # ── Gemini format helpers ───────────────────────────────────────────────────── def openai_messages_to_gemini(messages): contents = [] for m in messages: role = "user" if m["role"] == "user" else "model" content = m["content"] if isinstance(content, str): parts = [{"text": content}] elif isinstance(content, list): parts = [] for item in content: if item.get("type") == "text": parts.append({"text": item["text"]}) elif item.get("type") == "image_url": url = item["image_url"]["url"] if url.startswith("data:"): header, b64data = url.split(",", 1) mime = header.split(":")[1].split(";")[0] parts.append({"inline_data": {"mime_type": mime, "data": b64data}}) else: parts = [{"text": str(content)}] contents.append({"role": role, "parts": parts}) return contents def gemini_response_to_openai(r, model_id): try: text = r["candidates"][0]["content"]["parts"][0]["text"] except: text = r.get("error", {}).get("message", "No response") return { "id": "gemini-chat", "object": "chat.completion", "model": model_id, "choices": [{"index": 0, "message": {"role": "assistant", "content": text}, "finish_reason": "stop"}], "usage": { "prompt_tokens": r.get("usageMetadata", {}).get("promptTokenCount", 0), "completion_tokens": r.get("usageMetadata", {}).get("candidatesTokenCount", 0), "total_tokens": r.get("usageMetadata", {}).get("totalTokenCount", 0), } } # ── Generic OpenAI-compatible caller with A→B fallback ─────────────────────── async def call_openai_compat(base_url: str, key_a: str, key_b: str, model_name: str, body: dict): """Try key_a first; if quota error, automatically fallback to key_b.""" payload = {**body, "model": model_name} async with httpx.AsyncClient(timeout=60) as client: # ── Try A ── if key_a: resp = await client.post( f"{base_url}/chat/completions", headers={"Authorization": f"Bearer {key_a}", "Content-Type": "application/json"}, json=payload ) data = resp.json() if not is_quota_error(resp.status_code, data): return JSONResponse(content=data, status_code=resp.status_code) # ── Fallback to B ── if key_b: resp = await client.post( f"{base_url}/chat/completions", headers={"Authorization": f"Bearer {key_b}", "Content-Type": "application/json"}, json=payload ) return JSONResponse(content=resp.json(), status_code=resp.status_code) return JSONResponse({"error": "Both A and B keys failed or not configured."}, status_code=429) # ── Gemini caller with A→B fallback ────────────────────────────────────────── async def call_gemini(model_name: str, key_a: str, key_b: str, body: dict, model_id: str): payload = { "contents": openai_messages_to_gemini(body.get("messages", [])), "generationConfig": { "maxOutputTokens": body.get("max_tokens", 1024), "temperature": body.get("temperature", 1.0) } } async with httpx.AsyncClient(timeout=60) as client: # ── Try A ── if key_a: url = f"{GEMINI_BASE}/{model_name}:generateContent?key={key_a}" resp = await client.post(url, headers={"Content-Type": "application/json"}, json=payload) data = resp.json() if not is_quota_error(resp.status_code, data): return JSONResponse(content=gemini_response_to_openai(data, model_id), status_code=200) # ── Fallback to B ── if key_b: url = f"{GEMINI_BASE}/{model_name}:generateContent?key={key_b}" resp = await client.post(url, headers={"Content-Type": "application/json"}, json=payload) return JSONResponse(content=gemini_response_to_openai(resp.json(), model_id), status_code=200) return JSONResponse({"error": "Both Gemini A and B keys failed or not configured."}, status_code=429) # ── HTML UI ─────────────────────────────────────────────────────────────────── HTML_UI = """ Mohan AI
Live
Mohan AI
Powered by 20+ frontier models
across Gemini, Groq, Mistral, NVIDIA & more
✨ Write a poem
🧠 Explain quantum computing
💻 Debug my code
🌍 Translate to Hindi
0 tokens
Choose Model
""" DASHBOARD_HTML = """ Mohan AI | Dashboard
Admin Dashboard
Enter your API key to access
""" # ── Auth Dependency ─────────────────────────────────────────────────────────── async def require_auth(request: Request): key = request.headers.get("X-API-Key", "").strip() if not VALID_KEYS: raise HTTPException(status_code=500, detail="MOHAN_API_KEY not configured on server.") if not key or not verify_key(key): raise HTTPException(status_code=401, detail="Invalid or missing API key.") return key # ── Routes ──────────────────────────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) def home(): return HTML_UI @app.get("/dashboard", response_class=HTMLResponse) def dashboard(): return DASHBOARD_HTML @app.post("/auth/verify") async def auth_verify(request: Request): key = request.headers.get("X-API-Key", "").strip() if not VALID_KEYS: raise HTTPException(status_code=500, detail="MOHAN_API_KEY not configured on server.") if key and verify_key(key): return {"ok": True} raise HTTPException(status_code=401, detail="Invalid key. Access denied.") @app.get("/models") def get_models(key: str = Depends(require_auth)): return {"models": ALL_MODELS, "total": len(ALL_MODELS)} @app.get("/debug") def debug(key: str = Depends(require_auth)): return { "openrouter": {"key_loaded": bool(OPENROUTER_API_KEY)}, "openrouter_b": {"key_loaded": bool(OPENROUTER_API_KEY_B)}, "gemini": {"key_loaded": bool(GEMINI_API_KEY)}, "gemini_b": {"key_loaded": bool(GEMINI_API_KEY_B)}, "groq": {"key_loaded": bool(GROQ_API_KEY)}, "groq_b": {"key_loaded": bool(GROQ_API_KEY_B)}, "cerebras": {"key_loaded": bool(CEREBRAS_API_KEY)}, "cerebras_b": {"key_loaded": bool(CEREBRAS_API_KEY_B)}, "mistral": {"key_loaded": bool(MISTRAL_API_KEY)}, "mistral_b": {"key_loaded": bool(MISTRAL_API_KEY_B)}, "nvidia": {"key_loaded": bool(NVIDIA_API_KEY)}, "nvidia_b": {"key_loaded": bool(NVIDIA_API_KEY_B)}, } @app.post("/chat/completions") async def chat(request: Request, key: str = Depends(require_auth)): body = await request.json() model_id: str = body.get("model", "") try: # ── Gemini A (with auto B fallback) ── if is_gemini(model_id): if not GEMINI_API_KEY and not GEMINI_API_KEY_B: return JSONResponse({"error": "GEMINI_API_KEY not configured."}, status_code=500) return await call_gemini(strip_prefix(model_id), GEMINI_API_KEY, GEMINI_API_KEY_B, body, model_id) # ── Gemini B (direct) ── if is_gemini_b(model_id): if not GEMINI_API_KEY_B: return JSONResponse({"error": "GEMINI_API_KEY_B not configured."}, status_code=500) return await call_gemini(strip_prefix(model_id), GEMINI_API_KEY_B, "", body, model_id) # ── Groq A (with auto B fallback) ── if is_groq(model_id): if not GROQ_API_KEY and not GROQ_API_KEY_B: return JSONResponse({"error": "GROQ_API_KEY not configured."}, status_code=500) return await call_openai_compat(GROQ_BASE, GROQ_API_KEY, GROQ_API_KEY_B, strip_prefix(model_id), body) # ── Groq B (direct) ── if is_groq_b(model_id): if not GROQ_API_KEY_B: return JSONResponse({"error": "GROQ_API_KEY_B not configured."}, status_code=500) return await call_openai_compat(GROQ_BASE, GROQ_API_KEY_B, "", strip_prefix(model_id), body) # ── Cerebras A (with auto B fallback) ── if is_cerebras(model_id): if not CEREBRAS_API_KEY and not CEREBRAS_API_KEY_B: return JSONResponse({"error": "CEREBRAS_API_KEY not configured."}, status_code=500) return await call_openai_compat(CEREBRAS_BASE, CEREBRAS_API_KEY, CEREBRAS_API_KEY_B, strip_prefix(model_id), body) # ── Cerebras B (direct) ── if is_cerebras_b(model_id): if not CEREBRAS_API_KEY_B: return JSONResponse({"error": "CEREBRAS_API_KEY_B not configured."}, status_code=500) return await call_openai_compat(CEREBRAS_BASE, CEREBRAS_API_KEY_B, "", strip_prefix(model_id), body) # ── Mistral A (with auto B fallback) ── if is_mistral(model_id): if not MISTRAL_API_KEY and not MISTRAL_API_KEY_B: return JSONResponse({"error": "MISTRAL_API_KEY not configured."}, status_code=500) return await call_openai_compat(MISTRAL_BASE, MISTRAL_API_KEY, MISTRAL_API_KEY_B, strip_prefix(model_id), body) # ── Mistral B (direct) ── if is_mistral_b(model_id): if not MISTRAL_API_KEY_B: return JSONResponse({"error": "MISTRAL_API_KEY_B not configured."}, status_code=500) return await call_openai_compat(MISTRAL_BASE, MISTRAL_API_KEY_B, "", strip_prefix(model_id), body) # ── NVIDIA A (with auto B fallback) ── if is_nvidia(model_id): if not NVIDIA_API_KEY and not NVIDIA_API_KEY_B: return JSONResponse({"error": "NVIDIA_API_KEY not configured."}, status_code=500) return await call_openai_compat(NVIDIA_BASE, NVIDIA_API_KEY, NVIDIA_API_KEY_B, strip_prefix(model_id), body) # ── NVIDIA B (direct) ── if is_nvidia_b(model_id): if not NVIDIA_API_KEY_B: return JSONResponse({"error": "NVIDIA_API_KEY_B not configured."}, status_code=500) return await call_openai_compat(NVIDIA_BASE, NVIDIA_API_KEY_B, "", strip_prefix(model_id), body) # ── OpenRouter B (direct) ── if is_openrouter_b(model_id): if not OPENROUTER_API_KEY_B: return JSONResponse({"error": "OPENROUTER_API_KEY_B not configured."}, status_code=500) async with httpx.AsyncClient(timeout=60) as client: real_model = strip_prefix(model_id) resp = await client.post( f"{OPENROUTER_BASE}/chat/completions", headers={ "Authorization": f"Bearer {OPENROUTER_API_KEY_B}", "Content-Type": "application/json", "HTTP-Referer": "https://senpai315-mohan-ai.hf.space", "X-Title": "Mohan-AI Playground", }, json={**body, "model": real_model} ) return JSONResponse(content=resp.json(), status_code=resp.status_code) # ── OpenRouter A (with auto B fallback) — default fallback ── if not OPENROUTER_API_KEY and not OPENROUTER_API_KEY_B: return JSONResponse({"error": "OPENROUTER_API_KEY not configured."}, status_code=500) async with httpx.AsyncClient(timeout=60) as client: headers_a = { "Authorization": f"Bearer {OPENROUTER_API_KEY}", "Content-Type": "application/json", "HTTP-Referer": "https://senpai315-mohan-ai.hf.space", "X-Title": "Mohan-AI Playground", } if OPENROUTER_API_KEY: resp = await client.post(f"{OPENROUTER_BASE}/chat/completions", headers=headers_a, json=body) data = resp.json() if not is_quota_error(resp.status_code, data): return JSONResponse(content=data, status_code=resp.status_code) # Fallback to B if OPENROUTER_API_KEY_B: headers_b = {**headers_a, "Authorization": f"Bearer {OPENROUTER_API_KEY_B}"} resp = await client.post(f"{OPENROUTER_BASE}/chat/completions", headers=headers_b, json=body) return JSONResponse(content=resp.json(), status_code=resp.status_code) return JSONResponse({"error": "All OpenRouter keys exhausted."}, status_code=429) except Exception as e: return JSONResponse({"error": str(e)}, status_code=500)