""" title: OpenRouter Integration for OpenWebUI version: 0.5.0 description: Integration with OpenRouter for OpenWebUI with Free Model Filtering and optional field improvements original author: kevarch original author_url: https://github.com/kevarch original contributor: Eloi Marques da Silva (https://github.com/eloimarquessilva) credits: rburmorrison (https://github.com/rburmorrison), Google Gemini Pro 2.5 license: MIT Changelog: - Version 0.5.0: * Deprecated "including_reasoning" * Replaced with "reasoing" in the payload * Added constant status updates to keep the connection awake - Version 0.4.1: * Contribution by Eloi Marques da Silva * Added FREE_ONLY parameter to optionally filter and display only free models in OpenWebUI. * Changed MODEL_PREFIX and MODEL_PROVIDERS from required (str) to optional (Optional[str]), allowing null values. """ import re import requests import json import traceback # Import traceback for detailed error logging from typing import ( Optional, List, Union, Generator, Iterator, Callable, Awaitable, AsyncGenerator, ) from pydantic import BaseModel, Field import time import httpx import asyncio # --- Helper function for citation text insertion --- def _insert_citations(text: str, citations: list[str]) -> str: """ Replace citation markers [n] in text with markdown links to the corresponding citation URLs. Args: text: The text containing citation markers like [1], [2], etc. citations: A list of citation URLs, where index 0 corresponds to [1] in the text Returns: Text with citation markers replaced with markdown links """ if not citations or not text: return text pattern = r"\[(\d+)\]" def replace_citation(match_obj): try: num = int(match_obj.group(1)) if 1 <= num <= len(citations): url = citations[num - 1] return f"[[{num}]]({url})" else: return match_obj.group(0) except (ValueError, IndexError): return match_obj.group(0) try: return re.sub(pattern, replace_citation, text) except Exception as e: print(f"Error during citation insertion: {e}") return text # --- Helper function for formatting the final citation list --- def _format_citation_list(citations: list[str]) -> str: """ Formats a list of citation URLs into a markdown string. Args: citations: A list of citation URLs. Returns: A formatted markdown string (e.g., "\n\n---\nCitations:\n1. url1\n2. url2") or an empty string if no citations are provided. """ if not citations: return "" try: citation_list = [f"{i+1}. {url}" for i, url in enumerate(citations)] return "\n\n---\nCitations:\n" + "\n".join(citation_list) except Exception as e: print(f"Error formatting citation list: {e}") return "" # --- Main Pipe class --- class Pipe: class Valves(BaseModel): # User-configurable settings OPENROUTER_API_KEY: str = Field( default="", description="Your OpenRouter API key (required)." ) # DEPRECATED: include_reasoning is no longer supported by the API # INCLUDE_REASONING: bool = Field( # default=True, # description="(DEPRECATED) Request reasoning tokens from models that support it.", # ) REASONING_EFFORT: str = Field( default="medium", description="Reasoning effort level: 'high', 'medium', or 'low'. Set to empty string to disable reasoning.", ) MODEL_PREFIX: Optional[str] = Field( default=None, description="Optional prefix for model names in Open WebUI (e.g., 'OR: ').", ) # NEW: Configurable request timeout REQUEST_TIMEOUT: int = Field( default=90, description="Timeout for API requests in seconds.", gt=0, ) MODEL_PROVIDERS: Optional[str] = Field( default=None, description="Comma-separated list of model providers to include or exclude. Leave empty to include all providers.", ) INVERT_PROVIDER_LIST: bool = Field( default=False, description="If true, the above 'Model Providers' list becomes an *exclude* list instead of an *include* list.", ) ENABLE_CACHE_CONTROL: bool = Field( default=False, description="Enable OpenRouter prompt caching by adding 'cache_control' to potentially large message parts. May reduce costs for supported models (e.g., Anthropic, Gemini) on subsequent calls with the same cached prefix. See OpenRouter docs for details.", ) FREE_ONLY: bool = Field( default=False, description="If true, only free models will be available.", ) def __init__(self): self.type = "manifold" # Specifies this pipe provides multiple models self.valves = self.Valves() if not self.valves.OPENROUTER_API_KEY: print("Warning: OPENROUTER_API_KEY is not set in Valves.") def pipes(self) -> List[dict]: """ Fetches available models from the OpenRouter API. This method is called by OpenWebUI to discover the models this pipe provides. """ if not self.valves.OPENROUTER_API_KEY: return [ {"id": "error", "name": "Pipe Error: OpenRouter API Key not provided"} ] try: headers = {"Authorization": f"Bearer {self.valves.OPENROUTER_API_KEY}"} response = requests.get( "https://openrouter.ai/api/v1/models", headers=headers, timeout=self.valves.REQUEST_TIMEOUT, ) response.raise_for_status() models_data = response.json() raw_models_data = models_data.get("data", []) models: List[dict] = [] # --- Provider Filtering Logic --- provider_list_str = (self.valves.MODEL_PROVIDERS or "").lower() invert_list = self.valves.INVERT_PROVIDER_LIST target_providers = { p.strip() for p in provider_list_str.split(",") if p.strip() } # --- End Filtering Logic --- for model in raw_models_data: model_id = model.get("id") if not model_id: continue # Apply Provider Filtering if target_providers: provider = ( model_id.split("/", 1)[0].lower() if "/" in model_id else model_id.lower() ) provider_in_list = provider in target_providers keep = (provider_in_list and not invert_list) or ( not provider_in_list and invert_list ) if not keep: continue # Apply Free Only Filtering if self.valves.FREE_ONLY and "free" not in model_id.lower(): continue model_name = model.get("name", model_id) prefix = self.valves.MODEL_PREFIX or "" models.append({"id": model_id, "name": f"{prefix}{model_name}"}) if not models: if self.valves.FREE_ONLY: return [{"id": "error", "name": "Pipe Error: No free models found"}] elif target_providers: return [ { "id": "error", "name": "Pipe Error: No models found matching the provider filter", } ] else: return [ { "id": "error", "name": "Pipe Error: No models found on OpenRouter", } ] return models except requests.exceptions.Timeout: print("Error fetching models: Request timed out.") return [{"id": "error", "name": "Pipe Error: Timeout fetching models"}] except requests.exceptions.HTTPError as e: error_msg = f"Pipe Error: HTTP {e.response.status_code} fetching models" try: error_detail = e.response.json().get("error", {}).get("message", "") if error_detail: error_msg += f": {error_detail}" except json.JSONDecodeError: pass print(f"Error fetching models: {error_msg} (URL: {e.request.url})") return [{"id": "error", "name": error_msg}] except requests.exceptions.RequestException as e: print(f"Error fetching models: Request failed: {e}") return [ { "id": "error", "name": f"Pipe Error: Network error fetching models: {e}", } ] except Exception as e: print(f"Unexpected error fetching models: {e}") traceback.print_exc() return [{"id": "error", "name": f"Pipe Error: Unexpected error: {e}"}] async def pipe( self, body: dict, __event_emitter__=None ) -> Union[str, Generator, Iterator]: """ Processes incoming chat requests. This is the main function called by OpenWebUI when a user interacts with a model provided by this pipe. """ if not self.valves.OPENROUTER_API_KEY: error_chunk = self._format_sse_chunk( "Pipe Error: OpenRouter API Key is not configured." ) yield error_chunk yield "data: [DONE]\n\n" return try: payload = body.copy() if "model" in payload and payload["model"] and "." in payload["model"]: payload["model"] = payload["model"].split(".", 1)[1] # --- Apply Cache Control Logic --- if self.valves.ENABLE_CACHE_CONTROL and "messages" in payload: try: cache_applied = False messages = payload["messages"] # 1. Try applying to System Message for i, msg in enumerate(messages): if msg.get("role") == "system" and isinstance( msg.get("content"), list ): longest_index, max_len = -1, -1 for j, part in enumerate(msg["content"]): if part.get("type") == "text": text_len = len(part.get("text", "")) if text_len > max_len: max_len, longest_index = text_len, j if longest_index != -1: msg["content"][longest_index]["cache_control"] = { "type": "ephemeral" } cache_applied = True break # 2. Fallback to Last User Message if not cache_applied: for msg in reversed(messages): if msg.get("role") == "user" and isinstance( msg.get("content"), list ): longest_index, max_len = -1, -1 for j, part in enumerate(msg["content"]): if part.get("type") == "text": text_len = len(part.get("text", "")) if text_len > max_len: max_len, longest_index = text_len, j if longest_index != -1: msg["content"][longest_index]["cache_control"] = { "type": "ephemeral" } break except Exception as cache_err: print(f"Warning: Error applying cache_control logic: {cache_err}") traceback.print_exc() # --- End Cache Control Logic --- # if self.valves.INCLUDE_REASONING: # payload["include_reasoning"] = True if self.valves.REASONING_EFFORT and self.valves.REASONING_EFFORT in [ "high", "medium", "low", ]: payload["reasoning"] = {"effort": self.valves.REASONING_EFFORT} headers = { "Authorization": f"Bearer {self.valves.OPENROUTER_API_KEY}", "Content-Type": "application/json", "HTTP-Referer": body.get("http_referer", "https://openwebui.com/"), "X-Title": body.get("x_title", "Open WebUI via Pipe"), } url = "https://openrouter.ai/api/v1/chat/completions" is_streaming = body.get("stream", False) if is_streaming: async for chunk in self.stream_response( url, headers, payload, _insert_citations, _format_citation_list, self.valves.REQUEST_TIMEOUT, __event_emitter__, ): yield chunk else: async for chunk in self.non_stream_response( url, headers, payload, _insert_citations, _format_citation_list, self.valves.REQUEST_TIMEOUT, __event_emitter__, ): yield chunk # This tells the WebUI frontend that communications are complete. yield "data: [DONE]\n\n" except Exception as e: print(f"Error in pipe method: {e}") traceback.print_exc() # Send a final error status if an exception occurs if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": "An error occurred", "done": True}, } ) error_chunk = self._format_sse_chunk(f"Pipe Error: {e}") yield error_chunk yield "data: [DONE]\n\n" def _format_sse_chunk(self, content: str) -> str: """Formats a string into a Server-Sent Event (SSE) chunk.""" payload = { "choices": [ { "delta": {"content": content}, "index": 0, "finish_reason": None, } ] } json_payload = json.dumps(payload) return f"data: {json_payload}\n\n" async def non_stream_response( self, url: str, headers: dict, payload: dict, citation_inserter, citation_formatter, timeout: float, event_emitter=None, ) -> AsyncGenerator[str, None]: """ Executes a non-streaming API call, sends status updates with a running timer, and yields periodic keep-alive chunks so the frontend sees the updates immediately. """ # 1. Tell the UI we have started if event_emitter: await event_emitter( { "type": "status", "data": {"description": "Processing request…", "done": False}, } ) # Yield a keep-alive to flush the first status yield ":\n\n" # 2. Launch the OpenRouter request in the background async with httpx.AsyncClient(timeout=timeout) as client: api_task = asyncio.create_task( client.post(url, json=payload, headers=headers) ) elapsed = 0 # 3. Poll until the HTTP call is finished while not api_task.done(): await asyncio.sleep(1) elapsed += 1 if event_emitter: await event_emitter( { "type": "status", "data": { "description": f"Processing request… ({elapsed}s)", "done": False, }, } ) # Force Starlette to flush immediately yield ":\n\n" # 4. Get the response object (or raise if the task failed) response = await api_task # 5. Finalising status if event_emitter: await event_emitter( { "type": "status", "data": {"description": "Finalising response…", "done": False}, } ) yield ":\n\n" # flush again # 6. Normal result processing (unchanged) response.raise_for_status() data = response.json() choices = data.get("choices", []) if not choices: raise ValueError("API returned no choices in the response.") msg = choices[0].get("message", {}) content = msg.get("content", "") citations = data.get("citations", []) final_content = citation_inserter(content, citations) footer = citation_formatter(citations) # 7. Tell the UI we are done if event_emitter: await event_emitter( {"type": "status", "data": {"description": "Done", "done": True}} ) # 8. Yield the real answer yield self._format_sse_chunk(f"{final_content}{footer}") async def stream_response( self, url, headers, payload, citation_inserter, citation_formatter, timeout, event_emitter=None, ) -> AsyncGenerator[str, None]: """Handles streaming API requests using a generator.""" try: if event_emitter: await event_emitter( { "type": "status", "data": {"description": "Processing request...", "done": False}, } ) async with httpx.AsyncClient(timeout=timeout) as client: async with client.stream( "POST", url, headers=headers, json=payload ) as response: # Manually check for errors, as raise_for_status() must be after reading content if response.status_code != 200: # Eagerly read the error message error_body = await response.aread() yield f"Pipe Error: API returned HTTP {response.status_code}: {error_body.decode()}" return in_think = False latest_citations: List[str] = [] # Use aiter_lines() for async iteration async for line in response.aiter_lines(): if not line or not line.startswith("data: "): continue data = line[len("data: ") :] if data == "[DONE]": break try: chunk = json.loads(data) except json.JSONDecodeError: continue if "choices" in chunk: choice = chunk["choices"][0] citations = chunk.get("citations") if citations is not None: latest_citations = citations delta = choice.get("delta", {}) content = delta.get("content", "") reasoning = delta.get("reasoning", "") # reasoning if reasoning: if not in_think: yield "<think>\n" in_think = True yield citation_inserter(reasoning, latest_citations) # content if content: if in_think: yield "\n</think>\n\n" in_think = False yield citation_inserter(content, latest_citations) # Close thinking block if still open if in_think: yield "\n</think>\n\n" # Add citations at the end yield citation_formatter(latest_citations) # <--- CHANGE 5: Add final done status if event_emitter: await event_emitter( {"type": "status", "data": {"description": "Done", "done": True}} ) except httpx.TimeoutException: yield f"Pipe Error: Request timed out ({timeout}s)" except httpx.HTTPStatusError as e: yield f"Pipe Error: API returned HTTP {e.response.status_code}" except Exception as e: yield f"Pipe Error: Unexpected error during streaming: {e}" traceback.print_exc()