""" title: OpenRouter API author: ross1996 (modded by SolutionsHub.ai) version: 0.3.1 license: MIT """ """Pipeline that lists OpenRouter models and shows FREE models. Tested: 04-30-25 0.3.1 fixes ------------ * `_format_id` now lives at module level (was nested/indented incorrectly). * Removed stray `return mid.strip()[len(prefix):]` which broke logic. * Flake8‑clean import order. """ import json import os import time from typing import Dict, Generator, List, Union 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") def __init__(self): # Removed self.type = "manifold" self.id = "openrouter" self.name = "openrouter/" self.valves = self.Valves() # Removed environment variable fetching for OPENROUTER_API_KEY and FREE_ONLY # Removed error message for missing OPENROUTER_API_KEY, as it is now handled in the UI # --------------------------- helpers ----------------------------------- # Enhanced debug messages with timestamps and categories 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: # Use the headers helper 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, stream: bool = False): # Use the headers helper headers = self._headers() return requests.post( url, json=payload, headers=headers, stream=stream, timeout=self.valves.TIMEOUT_POST, ) # Refactored redundant code in _stream_chat and _complete_chat # Moved shared logic to a helper function def _prepare_payload(self, model: str, messages: List[dict], stream: bool) -> List[tuple]: payload_chat = {"model": model, "messages": messages, "stream": stream} payload_completion = { "model": model, "prompt": "\n".join(m.get("content", "") for m in messages) or "Hello", "max_tokens": 512, "stream": stream, } return [ (f"{self.valves.OPENROUTER_API_BASE_URL}/chat/completions", payload_chat), (f"{self.valves.OPENROUTER_API_BASE_URL}/completions", payload_completion), ] # --------------------------- chat interface ----------------------------- def pipe(self, body: dict) -> Union[str, Generator[str, None, None]]: model_id = self._format_id(body.get("model", "")) messages = body.get("messages", []) stream = body.get("stream", False) # Handle the case where the model_id might still be empty or invalid after formatting if not model_id: return "Error: Invalid or missing model ID." if stream: return self._stream_chat(model_id, messages) return self._complete_chat(model_id, messages) def _endpoint_try_order(self): base = self.valves.OPENROUTER_API_BASE_URL # Prioritize chat completions as it's the standard return [f"{base}/chat/completions", f"{base}/completions"] # --------------------- streaming chat ----------------------------------- def _stream_chat( self, model: str, messages: List[dict] ) -> Generator[str, None, None]: # Prepare payloads for both possible endpoints endpoints_with_payloads = self._prepare_payload(model, messages, stream=True) # Loop through endpoints and try posting for endpoint, payload in endpoints_with_payloads: try: self._debug( f"Attempting streaming post to: {endpoint} with payload: {payload}" ) # Call _post with stream=True res = self._post(endpoint, payload, stream=True) # Check status code for debugging *after* the call if DEBUG: self._debug(f"{endpoint} response status: {res.status_code}") if res.status_code >= 400: # Try reading a bit of the response text for error details try: error_detail = res.text[:500] # Read first 500 chars self._debug(f"Error detail from {endpoint}: {error_detail}") except Exception as read_exc: self._debug(f"Failed to read error detail: {read_exc}") res.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) # Process streaming response for line in res.iter_lines(): if not line: continue ln = line.decode().lstrip("data: ") if ln == "[DONE]": break try: delta = json.loads(ln) if "choices" in delta: # chat format # Check if delta['choices'] exists and is not empty if delta["choices"] and "delta" in delta["choices"][0]: out = delta["choices"][0]["delta"].get("content") else: # Handle cases where choices/delta might be structured differently or empty self._debug( f"Unexpected delta structure (chat): {delta}" ) continue # Skip this delta if structure is unexpected else: # legacy completion stream sends raw strings or text object out = ( delta.get("text") or delta ) # Handles text key or raw string if out: yield out except json.JSONDecodeError: # Handle lines that are not valid JSON, might be error messages or other data self._debug( f"JSONDecodeError on line: {ln}. Yielding raw line." ) yield ln except Exception as parse_exc: self._debug( f"Error processing stream chunk: {parse_exc} on line: {ln}" ) # Optionally yield an error indicator or stop # yield f"Stream Parse Error: {parse_exc}" # break # uncomment to stop streaming on first parse error # If we successfully processed a response, return return except requests.exceptions.RequestException as exc: # Catch specific requests exceptions like HTTPError, ConnectionError, Timeout self._debug(f"Request failed for {endpoint}: {exc}") if isinstance(exc, requests.HTTPError): # If it's an HTTP error and specifically 400 or 404, try the next endpoint if exc.response.status_code in (400, 404): self._debug( f"Received {exc.response.status_code}, trying next endpoint." ) continue # Try the next endpoint in the loop else: # For other HTTP errors, yield the error and stop yield f"Error: {exc.response.status_code} - {exc.response.text}" return else: # For other request exceptions (connection, timeout etc.), yield the error and stop yield f"Error: Request failed - {exc}" return except Exception as exc: # Catch any other unexpected errors during the loop/processing self._debug( f"Unexpected error in stream processing for {endpoint}: {exc}" ) yield f"Error: Unexpected stream error - {exc}" return # If the loop finishes without returning, both endpoints failed yield "Error: Failed to get a successful response from both chat and completion endpoints." # --------------------- non‑stream chat ----------------------------------- def _complete_chat(self, model: str, messages: List[dict]) -> str: # Prepare payloads for both possible endpoints (stream=False here) endpoints_with_payloads = self._prepare_payload(model, messages, stream=False) for endpoint, payload in endpoints_with_payloads: try: self._debug( f"Attempting non-streaming post to: {endpoint} with payload: {payload}" ) # Call _post with stream=False res = self._post(endpoint, payload, stream=False) # Check status code for debugging *after* the call if DEBUG: self._debug(f"{endpoint} response status: {res.status_code}") if res.status_code >= 400: try: error_detail = res.text[:500] self._debug(f"Error detail from {endpoint}: {error_detail}") except Exception as read_exc: self._debug(f"Failed to read error detail: {read_exc}") res.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) data = res.json() self._debug(f"Received data (non-stream): {data}") # Parse based on endpoint/response structure if endpoint.endswith("/chat/completions"): # Chat completion format if "choices" in data and data["choices"]: first_choice = data["choices"][0] if ( "message" in first_choice and "content" in first_choice["message"] ): return first_choice["message"]["content"] else: self._debug(f"Unexpected chat completion structure: {data}") return f"Error: Unexpected chat completion structure received from {endpoint}" else: self._debug( f"No choices found in chat completion response from {endpoint}: {data}" ) return f"Error: No choices found in chat completion response from {endpoint}" elif endpoint.endswith("/completions"): # Legacy completion format if "choices" in data and data["choices"]: first_choice = data["choices"][0] if "text" in first_choice: return first_choice["text"] else: self._debug(f"Unexpected completion structure: {data}") return f"Error: Unexpected completion structure received from {endpoint}" elif ( "text" in data ): # Sometimes the raw text might be at the top level? (less common) return data["text"] else: self._debug( f"No choices or text found in completion response from {endpoint}: {data}" ) return f"Error: No choices or text found in completion response from {endpoint}" else: # Should not happen with the current endpoint list self._debug(f"Unknown endpoint processed: {endpoint}") return f"Error: Processed unknown endpoint {endpoint}" except requests.exceptions.RequestException as exc: self._debug(f"Request failed for {endpoint}: {exc}") if isinstance(exc, requests.HTTPError): if exc.response.status_code in (400, 404): self._debug( f"Received {exc.response.status_code}, trying next endpoint." ) continue # Try the next endpoint else: # For other HTTP errors, return the error and stop return ( f"Error: {exc.response.status_code} - {exc.response.text}" ) else: # For other request exceptions, return the error and stop return f"Error: Request failed - {exc}" except Exception as exc: # Catch any other unexpected errors self._debug( f"Unexpected error in non-stream processing for {endpoint}: {exc}" ) return f"Error: Unexpected non-stream error - {exc}" # If the loop finishes without returning, both endpoints failed return "Error: Failed to get a successful response from both chat and completion endpoints."