function-file

"""
title: Multi Provider External
author: codemonkeying
version: 1.2.1
license: MIT
requirements: pydantic>=2.0.0, requests>=2.0.0, google-generativeai>=0.3.0
environment_variables:
    - OPENAI_API_KEY (optional)
    - ANTHROPIC_API_KEY (optional)
    - GOOGLE_API_KEY (optional)

Auto-discovers models from:
- OpenAI: /v1/models endpoint
- Anthropic: /v1/models endpoint
- Google: genai.list_models()
"""

import os
import json
import time
import logging
import requests
from datetime import datetime
from typing import List, Union, Generator, Iterator, Dict, Optional
from pydantic import BaseModel, Field
from open_webui.utils.misc import pop_system_message

try:
    import google.generativeai as genai
    from google.generativeai.types import GenerationConfig

    GOOGLE_AVAILABLE = True
except ImportError:
    GOOGLE_AVAILABLE = False
    genai = None


class Pipe:
    def __init__(self):
        logging.basicConfig(level=logging.INFO)
        self.type = "manifold"
        self.id = "multi_provider_external"
        self.name = ""
        self.valves = self.Valves()
        self.request_id = None
        self._cached_models = {}
        self._cache_timestamp = {}

    class Valves(BaseModel):
        # API Keys
        OPENAI_API_KEY: str = Field(
            default=os.getenv("OPENAI_API_KEY", ""), description="OpenAI API key"
        )
        ANTHROPIC_API_KEY: str = Field(
            default=os.getenv("ANTHROPIC_API_KEY", ""), description="Anthropic API key"
        )
        GOOGLE_API_KEY: str = Field(
            default=os.getenv("GOOGLE_API_KEY", ""), description="Google API key"
        )

        # API Base URLs
        OPENAI_API_BASE_URL: str = Field(
            default="https://api.openai.com/v1", description="OpenAI API base URL"
        )
        ANTHROPIC_API_BASE_URL: str = Field(
            default="https://api.anthropic.com/v1", description="Anthropic API base URL"
        )

        # Provider Settings
        ENABLE_OPENAI: bool = Field(default=True, description="Enable OpenAI models")
        ENABLE_ANTHROPIC: bool = Field(
            default=True, description="Enable Anthropic models"
        )
        ENABLE_GOOGLE: bool = Field(default=True, description="Enable Google models")

        # Display Settings
        MODEL_PREFIX: str = Field(
            default="MPE:",
            description="Prefix for model names (e.g., 'MPE:', 'External:', 'Multi:')",
        )

        # Auto-discovery Settings
        AUTO_DISCOVER_MODELS: bool = Field(
            default=True, description="Auto-discover models from APIs"
        )
        CACHE_DURATION_MINUTES: int = Field(
            default=60, description="Cache model list for N minutes"
        )

        # Safety Settings
        USE_PERMISSIVE_SAFETY: bool = Field(
            default=True, description="Use permissive safety settings for Google"
        )

    def _is_cache_valid(self, provider: str) -> bool:
        """Check if cached models are still valid"""
        if provider not in self._cache_timestamp:
            return False

        cache_age = time.time() - self._cache_timestamp[provider]
        return cache_age < (self.valves.CACHE_DURATION_MINUTES * 60)

    def _discover_openai_models(self) -> List[dict]:
        """Auto-discover OpenAI models from API"""
        if not self.valves.OPENAI_API_KEY:
            return []

        # Check cache first
        if self._is_cache_valid("openai") and "openai" in self._cached_models:
            return self._cached_models["openai"]

        try:
            headers = {
                "Authorization": f"Bearer {self.valves.OPENAI_API_KEY}",
                "Content-Type": "application/json",
            }

            response = requests.get(
                f"{self.valves.OPENAI_API_BASE_URL}/models", headers=headers, timeout=10
            )

            if response.status_code == 200:
                data = response.json()
                models = []

                for model in data.get("data", []):
                    model_id = model.get("id", "")

                    # Filter for chat models and estimate context/vision support
                    if any(prefix in model_id for prefix in ["gpt-", "o1-", "o3-"]):
                        # Estimate context window based on model name
                        context = self._estimate_openai_context(model_id)
                        vision = self._estimate_openai_vision(model_id)

                        models.append(
                            {
                                "id": model_id,
                                "name": self._format_openai_name(model_id),
                                "context": context,
                                "vision": vision,
                            }
                        )

                # Cache the results
                self._cached_models["openai"] = models
                self._cache_timestamp["openai"] = time.time()

                print(f"DEBUG: Discovered {len(models)} OpenAI models")
                return models

        except Exception as e:
            print(f"DEBUG: Failed to discover OpenAI models: {e}")

        # Fallback to hardcoded models
        return self._get_fallback_openai_models()

    def _discover_anthropic_models(self) -> List[dict]:
        """Auto-discover Anthropic models from API"""
        if not self.valves.ANTHROPIC_API_KEY:
            return []

        # Check cache first
        if self._is_cache_valid("anthropic") and "anthropic" in self._cached_models:
            return self._cached_models["anthropic"]

        try:
            headers = {
                "x-api-key": self.valves.ANTHROPIC_API_KEY,
                "anthropic-version": "2023-06-01",
                "content-type": "application/json",
            }

            response = requests.get(
                f"{self.valves.ANTHROPIC_API_BASE_URL}/models",
                headers=headers,
                timeout=10,
            )

            if response.status_code == 200:
                data = response.json()
                models = []

                for model in data.get("data", []):
                    model_id = model.get("id", "")
                    display_name = model.get("display_name", model_id)

                    # Only include Claude models
                    if model_id.startswith("claude"):
                        context = self._estimate_anthropic_context(model_id)
                        vision = self._estimate_anthropic_vision(model_id)

                        models.append(
                            {
                                "id": model_id,
                                "name": display_name,
                                "context": context,
                                "vision": vision,
                            }
                        )

                # Cache the results
                self._cached_models["anthropic"] = models
                self._cache_timestamp["anthropic"] = time.time()

                print(f"DEBUG: Discovered {len(models)} Anthropic models")
                return models

        except Exception as e:
            print(f"DEBUG: Failed to discover Anthropic models: {e}")

        # Fallback to hardcoded models
        return self._get_fallback_anthropic_models()

    def _discover_google_models(self) -> List[dict]:
        """Auto-discover Google models from API"""
        if not self.valves.GOOGLE_API_KEY or not GOOGLE_AVAILABLE:
            return []

        # Check cache first
        if self._is_cache_valid("google") and "google" in self._cached_models:
            return self._cached_models["google"]

        try:
            genai.configure(api_key=self.valves.GOOGLE_API_KEY)
            models = []

            for model in genai.list_models():
                if "generateContent" in model.supported_generation_methods:
                    model_id = model.name.replace("models/", "")

                    # Only include Gemini models
                    if model_id.startswith("gemini"):
                        context = self._estimate_google_context(model_id)
                        vision = self._estimate_google_vision(model_id)

                        models.append(
                            {
                                "id": model_id,
                                "name": self._format_google_name(model_id),
                                "context": context,
                                "vision": vision,
                            }
                        )

            # Cache the results
            self._cached_models["google"] = models
            self._cache_timestamp["google"] = time.time()

            print(f"DEBUG: Discovered {len(models)} Google models")
            return models

        except Exception as e:
            print(f"DEBUG: Failed to discover Google models: {e}")

        # Fallback to hardcoded models
        return self._get_fallback_google_models()

    def _estimate_openai_context(self, model_id: str) -> int:
        """Estimate context window for OpenAI models"""
        if "gpt-5" in model_id:
            return 200000
        elif "gpt-4o" in model_id:
            return 128000
        elif "gpt-4-turbo" in model_id:
            return 128000
        elif "gpt-4" in model_id:
            return 8192
        elif "gpt-3.5-turbo" in model_id:
            return 16385
        elif model_id.startswith(("o1-", "o3-")):
            return 200000
        else:
            return 4096  # Default

    def _estimate_openai_vision(self, model_id: str) -> bool:
        """Estimate vision support for OpenAI models"""
        vision_models = ["gpt-4o", "gpt-4-turbo", "gpt-4-vision"]
        return any(vm in model_id for vm in vision_models) or "gpt-5" in model_id

    def _estimate_anthropic_context(self, model_id: str) -> int:
        """Estimate context window for Anthropic models"""
        if "claude-3" in model_id or "claude-4" in model_id:
            return 200000
        else:
            return 100000  # Default

    def _estimate_anthropic_vision(self, model_id: str) -> bool:
        """Estimate vision support for Anthropic models"""
        # Most Claude 3+ models support vision except Haiku
        return (
            "claude-3" in model_id or "claude-4" in model_id and "haiku" not in model_id
        )

    def _estimate_google_context(self, model_id: str) -> int:
        """Estimate context window for Google models"""
        if "gemini-1.5-pro" in model_id or "gemini-2" in model_id:
            return 2000000
        elif "gemini-1.5" in model_id:
            return 1000000
        else:
            return 32000  # Default

    def _estimate_google_vision(self, model_id: str) -> bool:
        """Estimate vision support for Google models"""
        return "gemini" in model_id  # Most Gemini models support vision

    def _format_openai_name(self, model_id: str) -> str:
        """Format OpenAI model name for display"""
        name_map = {
            "gpt-4o": "GPT-4o",
            "gpt-4o-mini": "GPT-4o Mini",
            "gpt-4-turbo": "GPT-4 Turbo",
            "gpt-3.5-turbo": "GPT-3.5 Turbo",
            "o1-preview": "o1-preview",
            "o1-mini": "o1-mini",
        }
        return name_map.get(model_id, model_id.replace("-", " ").title())

    def _format_google_name(self, model_id: str) -> str:
        """Format Google model name for display"""
        return model_id.replace("gemini-", "Gemini ").replace("-", " ").title()

    def _get_fallback_openai_models(self) -> List[dict]:
        """Fallback OpenAI models if discovery fails"""
        return [
            {"id": "gpt-4o", "name": "GPT-4o", "context": 128000, "vision": True},
            {
                "id": "gpt-4o-mini",
                "name": "GPT-4o Mini",
                "context": 128000,
                "vision": True,
            },
            {
                "id": "gpt-4-turbo",
                "name": "GPT-4 Turbo",
                "context": 128000,
                "vision": True,
            },
            {
                "id": "gpt-3.5-turbo",
                "name": "GPT-3.5 Turbo",
                "context": 16000,
                "vision": False,
            },
            {
                "id": "o1-preview",
                "name": "o1-preview",
                "context": 128000,
                "vision": False,
            },
            {"id": "o1-mini", "name": "o1-mini", "context": 128000, "vision": False},
        ]

    def _get_fallback_anthropic_models(self) -> List[dict]:
        """Fallback Anthropic models if discovery fails"""
        return [
            {
                "id": "claude-3-5-sonnet-20241022",
                "name": "Claude 3.5 Sonnet",
                "context": 200000,
                "vision": True,
            },
            {
                "id": "claude-3-5-haiku-20241022",
                "name": "Claude 3.5 Haiku",
                "context": 200000,
                "vision": False,
            },
            {
                "id": "claude-3-opus-20240229",
                "name": "Claude 3 Opus",
                "context": 200000,
                "vision": True,
            },
            {
                "id": "claude-3-sonnet-20240229",
                "name": "Claude 3 Sonnet",
                "context": 200000,
                "vision": True,
            },
            {
                "id": "claude-3-haiku-20240307",
                "name": "Claude 3 Haiku",
                "context": 200000,
                "vision": True,
            },
        ]

    def _get_fallback_google_models(self) -> List[dict]:
        """Fallback Google models if discovery fails"""
        return [
            {
                "id": "gemini-1.5-pro",
                "name": "Gemini 1.5 Pro",
                "context": 2000000,
                "vision": True,
            },
            {
                "id": "gemini-1.5-flash",
                "name": "Gemini 1.5 Flash",
                "context": 1000000,
                "vision": True,
            },
            {
                "id": "gemini-pro",
                "name": "Gemini Pro",
                "context": 32000,
                "vision": True,
            },
        ]

    def _format_model_name(self, model_name: str) -> str:
        """Format model name with configurable prefix"""
        prefix = self.valves.MODEL_PREFIX.strip()
        if prefix and not prefix.endswith(":"):
            prefix += ":"

        if prefix:
            return f"{prefix} {model_name}"
        else:
            return model_name

    def pipes(self) -> List[dict]:
        """Return all available models from enabled providers"""
        models = []

        # OpenAI Models
        if self.valves.ENABLE_OPENAI and self.valves.OPENAI_API_KEY:
            if self.valves.AUTO_DISCOVER_MODELS:
                openai_models = self._discover_openai_models()
            else:
                openai_models = self._get_fallback_openai_models()

            for model in openai_models:
                models.append(
                    {
                        "id": f"openai.{model['id']}",
                        "name": self._format_model_name(model["name"]),
                        "context_length": model["context"],
                        "supports_vision": model["vision"],
                        "provider": "openai",
                    }
                )

        # Anthropic Models
        if self.valves.ENABLE_ANTHROPIC and self.valves.ANTHROPIC_API_KEY:
            if self.valves.AUTO_DISCOVER_MODELS:
                anthropic_models = self._discover_anthropic_models()
            else:
                anthropic_models = self._get_fallback_anthropic_models()

            for model in anthropic_models:
                models.append(
                    {
                        "id": f"anthropic.{model['id']}",
                        "name": self._format_model_name(model["name"]),
                        "context_length": model["context"],
                        "supports_vision": model["vision"],
                        "provider": "anthropic",
                    }
                )

        # Google Models
        if (
            self.valves.ENABLE_GOOGLE
            and self.valves.GOOGLE_API_KEY
            and GOOGLE_AVAILABLE
        ):
            if self.valves.AUTO_DISCOVER_MODELS:
                google_models = self._discover_google_models()
            else:
                google_models = self._get_fallback_google_models()

            for model in google_models:
                models.append(
                    {
                        "id": f"google.{model['id']}",
                        "name": self._format_model_name(model["name"]),
                        "context_length": model["context"],
                        "supports_vision": model["vision"],
                        "provider": "google",
                    }
                )

        if not models:
            return [
                {
                    "id": "error",
                    "name": "No API keys configured or providers enabled",
                }
            ]

        return models

    def pipe(
        self, body: Dict, __event_emitter__=None, __user__=None
    ) -> Union[str, Generator, Iterator]:
        """Route request to appropriate provider"""
        model_id = body["model"]

        print(f"DEBUG: Received model_id: {model_id}")

        # Extract provider and model from various formats
        provider = None
        actual_model = None

        if ".openai." in model_id:
            provider = "openai"
            actual_model = model_id.split(".openai.")[-1]
        elif ".anthropic." in model_id:
            provider = "anthropic"
            actual_model = model_id.split(".anthropic.")[-1]
        elif ".google." in model_id:
            provider = "google"
            actual_model = model_id.split(".google.")[-1]
        elif model_id.startswith("openai/"):
            provider = "openai"
            actual_model = model_id.replace("openai/", "")
        elif model_id.startswith("anthropic/"):
            provider = "anthropic"
            actual_model = model_id.replace("anthropic/", "")
        elif model_id.startswith("google/"):
            provider = "google"
            actual_model = model_id.replace("google/", "")

        if provider and actual_model:
            print(f"DEBUG: Provider: {provider}, Model: {actual_model}")

            if provider == "openai":
                return self._handle_openai(body, __event_emitter__, actual_model)
            elif provider == "anthropic":
                return self._handle_anthropic(body, __event_emitter__, actual_model)
            elif provider == "google":
                return self._handle_google(body, __event_emitter__, actual_model)

        return f"Error: Could not parse model format: {model_id}"

    def _handle_openai(self, body: Dict, __event_emitter__=None, model_id: str = None):
        """Handle OpenAI requests"""
        if not self.valves.OPENAI_API_KEY:
            return "Error: OpenAI API key not configured"

        if not model_id:
            model_id = body["model"].replace("openai.", "").replace("openai/", "")

        print(f"DEBUG: OpenAI model_id: {model_id}")

        headers = {
            "Authorization": f"Bearer {self.valves.OPENAI_API_KEY}",
            "Content-Type": "application/json",
        }

        payload = {**body, "model": model_id}

        # Handle o1 and o3 models (no system messages, no streaming)
        if model_id.startswith(("o1-", "o3-")):
            if "messages" in payload:
                payload["messages"] = [
                    msg for msg in payload["messages"] if msg.get("role") != "system"
                ]
            payload["stream"] = False

        try:
            if __event_emitter__:
                __event_emitter__(
                    {
                        "type": "status",
                        "data": {
                            "description": "Processing OpenAI request...",
                            "done": False,
                        },
                    }
                )

            response = requests.post(
                f"{self.valves.OPENAI_API_BASE_URL}/chat/completions",
                json=payload,
                headers=headers,
                stream=payload.get("stream", False),
                timeout=60,
            )

            if response.status_code != 200:
                return f"OpenAI Error: HTTP {response.status_code}: {response.text}"

            if payload.get("stream", False):
                return self._stream_openai_response(response)
            else:
                result = response.json()
                return (
                    result.get("choices", [{}])[0].get("message", {}).get("content", "")
                )

        except Exception as e:
            return f"OpenAI Error: {str(e)}"

    def _handle_anthropic(
        self, body: Dict, __event_emitter__=None, model_id: str = None
    ):
        """Handle Anthropic requests"""
        if not self.valves.ANTHROPIC_API_KEY:
            return "Error: Anthropic API key not configured"

        if not model_id:
            model_id = body["model"].replace("anthropic.", "").replace("anthropic/", "")

        print(f"DEBUG: Anthropic model_id: {model_id}")

        system_message, messages = pop_system_message(body["messages"])

        headers = {
            "x-api-key": self.valves.ANTHROPIC_API_KEY,
            "anthropic-version": "2023-06-01",
            "content-type": "application/json",
        }

        payload = {
            "model": model_id,
            "messages": self._process_anthropic_messages(messages),
            "max_tokens": body.get("max_tokens", 4096),
            "temperature": body.get("temperature"),
            "stream": body.get("stream", False),
        }

        if system_message:
            payload["system"] = str(system_message)

        payload = {k: v for k, v in payload.items() if v is not None}

        try:
            if __event_emitter__:
                __event_emitter__(
                    {
                        "type": "status",
                        "data": {
                            "description": "Processing Anthropic request...",
                            "done": False,
                        },
                    }
                )

            response = requests.post(
                f"{self.valves.ANTHROPIC_API_BASE_URL}/messages",
                json=payload,
                headers=headers,
                stream=payload.get("stream", False),
                timeout=60,
            )

            if response.status_code != 200:
                return f"Anthropic Error: HTTP {response.status_code}: {response.text}"

            if payload.get("stream", False):
                return self._stream_anthropic_response(response)
            else:
                result = response.json()
                return result.get("content", [{}])[0].get("text", "")

        except Exception as e:
            return f"Anthropic Error: {str(e)}"

    def _handle_google(self, body: Dict, __event_emitter__=None, model_id: str = None):
        """Handle Google requests"""
        if not self.valves.GOOGLE_API_KEY or not GOOGLE_AVAILABLE:
            return "Error: Google API key not configured or google-generativeai not installed"

        if not model_id:
            model_id = body["model"].replace("google.", "").replace("google/", "")

        print(f"DEBUG: Google model_id: {model_id}")

        try:
            if __event_emitter__:
                __event_emitter__(
                    {
                        "type": "status",
                        "data": {
                            "description": "Processing Google request...",
                            "done": False,
                        },
                    }
                )

            genai.configure(api_key=self.valves.GOOGLE_API_KEY)

            messages = body["messages"]
            system_message = next(
                (msg["content"] for msg in messages if msg["role"] == "system"), None
            )

            # Process messages for Google format
            contents = []
            for message in messages:
                if message["role"] != "system":
                    if isinstance(message.get("content"), list):
                        parts = []
                        for content in message["content"]:
                            if content["type"] == "text":
                                parts.append({"text": content["text"]})
                            elif content["type"] == "image_url":
                                image_url = content["image_url"]["url"]
                                if image_url.startswith("data:image"):
                                    image_data = image_url.split(",")[1]
                                    parts.append(
                                        {
                                            "inline_data": {
                                                "mime_type": "image/jpeg",
                                                "data": image_data,
                                            }
                                        }
                                    )
                        contents.append({"role": message["role"], "parts": parts})
                    else:
                        contents.append(
                            {
                                "role": (
                                    "user" if message["role"] == "user" else "model"
                                ),
                                "parts": [{"text": message["content"]}],
                            }
                        )

            # Create model
            if system_message and "gemini-1.5" in model_id:
                model = genai.GenerativeModel(
                    model_name=model_id, system_instruction=system_message
                )
            else:
                model = genai.GenerativeModel(model_name=model_id)

            generation_config = GenerationConfig(
                temperature=body.get("temperature", 0.7),
                top_p=body.get("top_p", 0.9),
                top_k=body.get("top_k", 40),
                max_output_tokens=body.get("max_tokens", 8192),
            )

            # Safety settings - Default to permissive (no filtering)
            safety_settings = None
            if self.valves.USE_PERMISSIVE_SAFETY:
                safety_settings = {
                    genai.types.HarmCategory.HARM_CATEGORY_HARASSMENT: genai.types.HarmBlockThreshold.BLOCK_NONE,
                    genai.types.HarmCategory.HARM_CATEGORY_HATE_SPEECH: genai.types.HarmBlockThreshold.BLOCK_NONE,
                    genai.types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: genai.types.HarmBlockThreshold.BLOCK_NONE,
                    genai.types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: genai.types.HarmBlockThreshold.BLOCK_NONE,
                }

            if body.get("stream", False):

                def stream_generator():
                    response = model.generate_content(
                        contents,
                        generation_config=generation_config,
                        safety_settings=safety_settings,
                        stream=True,
                    )
                    for chunk in response:
                        if chunk.text:
                            yield chunk.text

                return stream_generator()
            else:
                response = model.generate_content(
                    contents,
                    generation_config=generation_config,
                    safety_settings=safety_settings,
                    stream=False,
                )
                return response.text

        except Exception as e:
            return f"Google Error: {str(e)}"

    def _process_anthropic_messages(self, messages: List[dict]) -> List[dict]:
        """Process messages for Anthropic format"""
        processed_messages = []
        for message in messages:
            if isinstance(message["content"], str):
                content = [{"type": "text", "text": message["content"]}]
            else:
                content = []
                for item in message["content"]:
                    if item["type"] == "text":
                        content.append({"type": "text", "text": item["text"]})
                    elif item["type"] == "image_url":
                        if item["image_url"]["url"].startswith("data:image"):
                            mime_type, base64_data = item["image_url"]["url"].split(
                                ",", 1
                            )
                            media_type = mime_type.split(":")[1].split(";")[0]
                            content.append(
                                {
                                    "type": "image",
                                    "source": {
                                        "type": "base64",
                                        "media_type": media_type,
                                        "data": base64_data,
                                    },
                                }
                            )

            processed_messages.append({"role": message["role"], "content": content})

        return processed_messages

    def _stream_openai_response(self, response):
        """Stream OpenAI response"""
        for line in response.iter_lines():
            if line:
                line = line.decode("utf-8")
                if line.startswith("data: "):
                    data = line[6:]
                    if data.strip() == "[DONE]":
                        break
                    try:
                        json_data = json.loads(data)
                        if "choices" in json_data and json_data["choices"]:
                            delta = json_data["choices"][0].get("delta", {})
                            if "content" in delta:
                                yield delta["content"]
                    except json.JSONDecodeError:
                        continue

    def _stream_anthropic_response(self, response):
        """Stream Anthropic response"""
        for line in response.iter_lines():
            if line and line.startswith(b"data: "):
                try:
                    data = json.loads(line[6:])
                    if (
                        data["type"] == "content_block_delta"
                        and "text" in data["delta"]
                    ):
                        yield data["delta"]["text"]
                    elif data["type"] == "message_stop":
                        break
                except json.JSONDecodeError:
                    continue

## 🌟 Repository

#Find the latest version and contribute at: https://github.com/codemonkeying/multi-provider-external

## 📝 License

#MIT License - Feel free to modify and distribute!

#---

#**Built by codemonkeying** | Version 1.2.1