""" title: OpenRouter Adapter Pipe author: Haervwe author_url: https://github.com/Haervwe funding_url: https://github.com/Haervwe/open-webui-tools version: 0.1.0 description: OpenRouter Adapter Pipe for Open WebUI Tools. Features: Settings/valves: API key, allowed models, use websearch. Async streaming inference. Built-in websearch and image generation support. Citations and image event emission. Images are saved to the owui backend and URLs are emitted (not base64). """ import aiohttp import mimetypes import io import time from fastapi import Request, UploadFile from open_webui.routers.files import upload_file_handler # type: ignore import json import base64 import traceback from typing import List, Dict, Any, Callable, Awaitable, Optional from pydantic import BaseModel, Field import logging from open_webui.models.users import User, Users import uuid class Pipe: class Valves(BaseModel): API_KEY: str = Field(default="", description="OpenRouter API key") ALLOWED_MODELS: List[str] = Field( default_factory=list, description="Allowed model slugs" ) USE_WEBSEARCH: bool = Field( default=False, description="Enable web search plugin for all models, you can also add :online to a model ID to enable it per-model basis", ) USE_IMAGE_EMBEDDING: bool = Field( default=False, description="Enable image embedding as html for images generated by the model, false to emit as message content only in markdown format", ) def __init__(self) -> None: self.valves = self.Valves() self.content_buffer = "" self.assistant_content = "" self.reasoning_start = None self.reasoning_text = "" def pipes(self) -> List[Dict[str, str]]: return [ {"id": f"openrouter-{model}-pipe", "name": f"OpenRouter {model} Pipe"} for model in self.valves.ALLOWED_MODELS ] async def pipe( self, body: Dict[str, Any], __user__: Dict[str, Any] = {}, __event_emitter__: Optional[Callable[[Dict[str, Any]], Awaitable[None]]] = None, __request__: Optional[Request] = None, ) -> None: self.__request__ = __request__ self.__user__ = Users.get_user_by_id(__user__["id"]) pipe_id = body.get("model") model = None if pipe_id: if "-" in pipe_id and pipe_id.endswith("-pipe"): model = pipe_id.split("-", 1)[1][: -len("-pipe")] elif pipe_id.endswith("-pipe"): model = pipe_id[: -len("-pipe")] else: model = pipe_id messages = body.get("messages", []) modalities = body.get("modalities") def get_slug(model_id: str) -> str: return model_id.split("/")[-1] allowed_slugs = [get_slug(m) for m in self.valves.ALLOWED_MODELS] model_slug = get_slug(model) if model else None if model_slug not in allowed_slugs: if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "type": "error", "description": f"Model '{model}' not allowed", "done": True, }, } ) return if not model: return payload: Dict[str, Any] = { "model": model, "messages": messages, "stream": True, "include_reasoning": True, } async def get_model_capabilities(model_id: str) -> List[str]: url = "https://openrouter.ai/api/v1/models" headers = {"Authorization": f"Bearer {self.valves.API_KEY}"} async with aiohttp.ClientSession() as session: async with session.get(url, headers=headers) as resp: if resp.status != 200: logging.error(f"Failed to fetch models: {resp.status}") return ["text"] data = await resp.json() for m in data.get("data", []): if m.get("id") == model_id or get_slug( m.get("id", "") ) == get_slug(model_id): return m.get("modalities", ["text"]) return ["text"] modalities = await get_model_capabilities(model) payload["modalities"] = modalities if getattr(self.valves, "USE_WEBSEARCH", False): payload["plugins"] = [{"id": "web"}] headers = { "Authorization": f"Bearer {self.valves.API_KEY}", "Content-Type": "application/json", } async def stream_response( payload: Dict[str, Any], headers: Dict[str, str] ) -> str: self.assistant_content = "" async with aiohttp.ClientSession() as session: async with session.post( "https://openrouter.ai/api/v1/chat/completions", json=payload, headers=headers, ) as resp: if resp.status != 200: error_text = await resp.text() logging.error(f"OpenRouter API error: {resp.status}") if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "type": "error", "description": f"OpenRouter API error: {resp.status} {error_text}", "done": True, }, } ) raise RuntimeError( f"OpenRouter API error: {resp.status} {error_text}" ) buffer = b"" async for chunk in resp.content.iter_any(): buffer += chunk while b"\n" in buffer: line, buffer = buffer.split(b"\n", 1) try: line = line.decode().strip() if len(line) > 1000: logging.info( f"Raw line from OpenRouter: {line[:1000]}..." ) else: logging.info(f"Raw line from OpenRouter: {line}") if not line.startswith("data: "): continue data = line[6:] if data == "[DONE]": break try: event = json.loads(data) except Exception as e: logging.error( f"Failed to parse JSON: {e}, data: {data[:1000]}" ) continue delta = event.get("choices", [{}])[0].get("delta", {}) content = delta.get("content") reasoning = delta.get("reasoning") # Handle reasoning_details first if present in delta if "reasoning_details" in delta: pass # Not used for emission # Handle reasoning - accumulate and emit incremental if reasoning and reasoning.strip(): if self.reasoning_start is None: self.reasoning_start = time.time() self.reasoning_text += reasoning formatted_reasoning = f"<details>\n<summary>Thinking...</summary>\n{self.reasoning_text}\n\n---\n\n</details>" if __event_emitter__: await __event_emitter__( { "type": "chat:message", "data": { "content": formatted_reasoning, "attributes": {"done": "false"}, }, } ) if content and content.strip(): logging.info( f"Emitting content: {content[:1000]}..." ) self.assistant_content += content if __event_emitter__: await __event_emitter__( { "type": "chat:message:delta", "data": { "role": "assistant", "content": content, }, } ) if "images" in delta: # Close reasoning if still active if ( self.reasoning_start is not None and self.reasoning_text ): duration = time.time() - self.reasoning_start formatted_reasoning = f"<details>\n<summary>Thought for {duration:.2f} seconds</summary>\n{self.reasoning_text}\n\n---\n\n</details>" if __event_emitter__: await __event_emitter__( { "type": "chat:message", "data": { "content": formatted_reasoning, "attributes": {"done": "true"}, }, } ) embeds: List[str] = [] for image in delta["images"]: data_url = image.get("image_url", {}).get("url") if data_url and data_url.startswith( "data:image/" ): public_url = await self._save_image_and_get_public_url( self.__request__, data_url, self.__user__, ) embeds.append(public_url) if embeds and __event_emitter__: if self.valves.USE_IMAGE_EMBEDDING: html_embeds = [ f'<img src="{url}" alt="Generated Image" style="max-width:400px;max-height:400px;" />' for url in embeds ] await __event_emitter__( { "type": "embeds", "data": {"embeds": html_embeds}, } ) else: markdown_links = "\n".join( [ f"" for url in embeds ] ) await __event_emitter__( { "type": "message", "data": {"content": markdown_links}, } ) if "annotations" in delta: for annotation in delta["annotations"]: citation_data: Dict[str, Any] = {} url: Optional[str] = None title: Optional[str] = None document: List[str] = [] metadata: List[Dict[str, Any]] = [] if annotation.get("type") == "url_citation": url_info = annotation.get( "url_citation", {} ) url = url_info.get("url") title = url_info.get("title") if title: document.append(title) meta = {"source": url, "title": title} if "start_index" in url_info: meta["start_index"] = url_info[ "start_index" ] if "end_index" in url_info: meta["end_index"] = url_info[ "end_index" ] metadata.append(meta) if not document and annotation: document.append(str(annotation)) citation_data = { "document": document, "metadata": metadata, "source": { "name": title or url or "Citation", "url": url or "", }, } if __event_emitter__: await __event_emitter__( { "type": "citation", "data": citation_data, } ) except Exception as e: if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "type": "error", "description": f"Streaming error: {e}\n{traceback.format_exc()}", "done": True, }, } ) raise # Clear accumulators self.content_buffer = "" self.reasoning_start = None self.reasoning_text = "" return self.assistant_content await stream_response(payload, headers) # Emit final done event if __event_emitter__: await __event_emitter__({"type": "chat:completion", "data": {"done": True}}) async def _save_image_and_get_public_url( self, request: Optional[Request], data_url: str, user: User ) -> str: if not request: return "" try: header, b64data = data_url.split(",", 1) img_bytes = base64.b64decode(b64data) content_type = header.split(":")[1].split(";")[0] image_format = mimetypes.guess_extension(content_type) or ".png" file = UploadFile( file=io.BytesIO(img_bytes), filename=f"generated-image{image_format}" ) file_item: Any = upload_file_handler( # type: ignore request=request, file=file, metadata={}, process=False, user=user ) if not file_item: logging.error("Failed to save image to OpenWebUI") return "" file_id = str(getattr(file_item, "id", "")) # type: ignore base_url = str(request.base_url).rstrip("/") relative_path = request.app.url_path_for( "get_file_content_by_id", id=file_id ) timestamp = int(time.time() * 1000) url_with_cache_bust = f"{base_url}{relative_path}?t={timestamp}" return url_with_cache_bust except Exception as e: logging.error(f"Error saving image to OpenWebUI: {e}") return ""