""" title: OpenRouter API author: ross1996 (modded by SolutionsHub.ai) version: 0.4.0 license: MIT """ """Pipeline that lists OpenRouter models and shows FREE models. Non-streaming version. 0.4.0 changes ------------ * Removed streaming functionality to eliminate looping issues * Simplified pipe() method to always return complete responses * Streamlined error handling and debugging * Improved payload preparation logic * Better legacy completion fallback handling """ import json import os import time from typing import Dict, List import requests from pydantic import BaseModel, Field from datetime import datetime DEBUG = True class Pipe: class Valves(BaseModel): OPENROUTER_API_BASE_URL: str = Field( default="https://openrouter.ai/api/v1", description="REST root" ) OPENROUTER_API_KEY: str = Field(default="", description="Bearer key") FREE_ONLY: bool = Field(default=False, description="Show only free models") TIMEOUT_GET: int = Field( default=15, description="Timeout for GET requests in seconds" ) TIMEOUT_POST: int = Field( default=60, description="Timeout for POST requests in seconds" ) MAX_TOKENS: int = Field( default=4000, description="Maximum tokens for responses" ) def __init__(self): self.id = "openrouter" self.name = "openrouter/" self.valves = self.Valves() # --------------------------- helpers ----------------------------------- def _debug(self, msg: str, category: str = "INFO") -> None: if DEBUG: timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] [{category}] {msg}") def _headers(self) -> Dict[str, str]: if not self.valves.OPENROUTER_API_KEY: raise ValueError("OPENROUTER_API_KEY is not set") return { "Authorization": f"Bearer {self.valves.OPENROUTER_API_KEY}", "Content-Type": "application/json", } @staticmethod def _is_free(model: dict) -> bool: pricing = model.get("pricing") or {} if pricing == {}: return True def _vals(d): for v in d.values(): if isinstance(v, dict): yield from _vals(v) else: yield v return all(float(v) == 0.0 for v in _vals(pricing) if v not in (None, "")) @staticmethod def _format_id(mid: str) -> str: """Strip OWUI‑added vendor prefixes so the raw model id is OpenRouter‑ready.""" for prefix in ( "openrouter_api.", "openrouter.", "openroutermodels.", "openrouter/", ): if mid.startswith(prefix): return mid[len(prefix) :] return mid # --------------------- model manifest ---------------------------------- def get_openrouter_models(self) -> List[Dict[str, str]]: url = f"{self.valves.OPENROUTER_API_BASE_URL}/models" try: headers = self._headers() res = requests.get(url, headers=headers, timeout=self.valves.TIMEOUT_GET) res.raise_for_status() data = res.json().get("data", []) if self.valves.FREE_ONLY: data = [m for m in data if self._is_free(m)] return [{"id": m["id"], "name": m.get("name", m["id"])} for m in data] except Exception as exc: self._debug(f"Model fetch failed: {exc}") return [{"id": "openrouter", "name": str(exc)}] # required by Pipelines def pipes(self) -> List[dict]: return self.get_openrouter_models() # --------------------------- http core ---------------------------------- def _post(self, url: str, payload: dict): headers = self._headers() return requests.post( url, json=payload, headers=headers, timeout=self.valves.TIMEOUT_POST, ) # --------------------------- chat interface ----------------------------- def pipe(self, body: dict) -> str: """Main pipeline method - always returns complete response""" model_id = self._format_id(body.get("model", "")) messages = body.get("messages", []) if not model_id: return "Error: Invalid or missing model ID." return self._complete_chat(model_id, messages) def _complete_chat(self, model: str, messages: List[dict]) -> str: """Try chat completions first, fallback to legacy completions""" # Try chat completions endpoint (modern standard) chat_url = f"{self.valves.OPENROUTER_API_BASE_URL}/chat/completions" chat_payload = { "model": model, "messages": messages, "stream": False, "max_tokens": self.valves.MAX_TOKENS, } try: self._debug(f"Attempting chat completions for model: {model}") response = self._post(chat_url, chat_payload) if DEBUG: self._debug(f"Chat completions response status: {response.status_code}") if response.status_code >= 400: try: error_detail = response.text[:500] self._debug(f"Chat completions error detail: {error_detail}") except Exception: pass response.raise_for_status() data = response.json() self._debug(f"Chat completions response structure: {list(data.keys())}") if "choices" in data and data["choices"]: first_choice = data["choices"][0] if "message" in first_choice and "content" in first_choice["message"]: content = first_choice["message"]["content"] self._debug( f"Successfully got chat response: {len(content)} characters" ) return content else: self._debug( f"Unexpected chat completion structure: {first_choice.keys()}" ) return "Error: Unexpected chat completion response structure" else: self._debug(f"No choices in chat completion response: {data}") return "Error: No choices found in chat completion response" except requests.HTTPError as exc: self._debug(f"Chat completions HTTP error: {exc}") if exc.response.status_code == 404: # Model might not support chat completions, try legacy self._debug("Chat completions not supported, trying legacy completion") return self._try_legacy_completion(model, messages) else: return f"Error: {exc.response.status_code} - {exc.response.text[:200]}" except requests.RequestException as exc: self._debug(f"Chat completions request error: {exc}") return self._try_legacy_completion(model, messages) except Exception as exc: self._debug(f"Chat completions unexpected error: {exc}") return self._try_legacy_completion(model, messages) def _try_legacy_completion(self, model: str, messages: List[dict]) -> str: """Fallback to legacy completion endpoint""" completion_url = f"{self.valves.OPENROUTER_API_BASE_URL}/completions" # Convert messages to simple prompt if messages: prompt_parts = [] for msg in messages: role = msg.get("role", "user") content = msg.get("content", "") if content: prompt_parts.append(f"{role}: {content}") prompt = "\n".join(prompt_parts) else: prompt = "Hello" completion_payload = { "model": model, "prompt": prompt, "max_tokens": self.valves.MAX_TOKENS, "stream": False, } try: self._debug(f"Attempting legacy completion for model: {model}") response = self._post(completion_url, completion_payload) if DEBUG: self._debug( f"Legacy completion response status: {response.status_code}" ) if response.status_code >= 400: try: error_detail = response.text[:500] self._debug(f"Legacy completion error detail: {error_detail}") except Exception: pass response.raise_for_status() data = response.json() self._debug(f"Legacy completion response structure: {list(data.keys())}") if "choices" in data and data["choices"]: first_choice = data["choices"][0] if "text" in first_choice: text = first_choice["text"] self._debug( f"Successfully got legacy response: {len(text)} characters" ) return text else: self._debug( f"Unexpected legacy completion structure: {first_choice.keys()}" ) return "Error: Unexpected legacy completion response structure" elif "text" in data: # Sometimes text is at top level text = data["text"] self._debug(f"Got top-level text response: {len(text)} characters") return text else: self._debug(f"No choices or text in legacy response: {data}") return "Error: No text found in legacy completion response" except requests.HTTPError as exc: self._debug(f"Legacy completion HTTP error: {exc}") return f"Error: Legacy completion failed with {exc.response.status_code} - {exc.response.text[:200]}" except requests.RequestException as exc: self._debug(f"Legacy completion request error: {exc}") return f"Error: Legacy completion request failed - {str(exc)[:200]}" except Exception as exc: self._debug(f"Legacy completion unexpected error: {exc}") return f"Error: Both chat and completion endpoints failed. Last error: {str(exc)[:200]}"