Spaces:
Running
Running
| # ============================================================================= | |
| # app/providers.py | |
| # 09.03.2026 | updated 23.03.2026 | reworked 10.11.2026 | |
| # LLM + Search Provider Registry + Fallback Chain | |
| # Universal MCP Hub (Sandboxed) - based on PyFundaments Architecture | |
| # Copyright 2026 - Volkan Kücükbudak | |
| # Apache License V. 2 + ESOL 1.1 | |
| # Repo: https://github.com/VolkanSah/Universal-MCP-Hub-sandboxed | |
| # ============================================================================= | |
| # ARCHITECTURE NOTE: | |
| # This file lives exclusively in app/ and is ONLY started by app/app.py. | |
| # NO direct access to fundaments/*, .env, or Guardian (main.py). | |
| # All config comes from app/.pyfun via app/config.py. | |
| # | |
| # PROVIDER PRINCIPLE: | |
| # No key = no provider = no tool = no crash. | |
| # Server always starts, just with fewer providers. | |
| # | |
| # HOW TO ADD A NEW LLM PROVIDER: | |
| # OpenAI-compatible API (Mistral, OpenAI, xAI, Groq, Together, DeepSeek, ...): | |
| # → ONLY a [LLM_PROVIDER.xyz] block in app/.pyfun with type = "openai_compat" | |
| # + the key in .env. NO code needed. See examples in .pyfun. | |
| # Different API format (like Anthropic / Gemini): | |
| # 1. Add a class below (subclass BaseProvider, implement complete()) | |
| # 2. Register it in _TYPE_CLASSES | |
| # 3. Add the [LLM_PROVIDER.xyz] block in .pyfun with type = "<your type>" | |
| # | |
| # FALLBACK CHAIN: | |
| # Defined in .pyfun per provider via fallback_to field. | |
| # anthropic → fails → gemini → fails → openrouter → fails → RuntimeError | |
| # Visited set prevents infinite loops. | |
| # A model chosen explicitly applies to the FIRST provider only — every | |
| # fallback provider uses its own default_model (model IDs are provider-specific). | |
| # | |
| # LIMITS (from .pyfun [HUB_LIMITS]): | |
| # MAX_PARALLEL_REQUESTS → semaphore around every provider call | |
| # RETRY_COUNT / RETRY_DELAY_SEC → attempts per provider on 429/5xx/connect errors | |
| # (timeouts are NOT retried → straight to fallback) | |
| # MAX_TOKENS_CAP → hard ceiling for max_tokens per call (0 = off) | |
| # Per-model max_output_tokens (manual or auto from pricing) clamps further. | |
| # | |
| # USAGE / COST: | |
| # complete() returns (text, usage). llm_complete() logs usage + estimated cost | |
| # into db_sync (usage_log) after every successful call. Never breaks a call. | |
| # | |
| # SECURITY NOTE: | |
| # API keys are NEVER logged or included in exception messages. | |
| # All errors are sanitized before propagation — only HTTP status codes | |
| # and safe_url (query params stripped) are ever exposed in logs. | |
| # | |
| # CACHING NOTE: | |
| # Anthropic → prompt_caching (cache_control: ephemeral) for long system/user prompts. | |
| # Enable per provider in .pyfun: supports_cache = "true" | |
| # Gemini → implicit caching (automatic, server-side). supports_cache = log hint only. | |
| # | |
| # DEPENDENCY CHAIN (app/* only, no fundaments!): | |
| # config.py → parses app/.pyfun — single source of truth | |
| # providers.py → LLM + Search registry + fallback chain | |
| # tools.py → calls providers.llm_complete() / providers.search() | |
| # mcp.py → calls providers.list_active_llm() / list_active_search() | |
| # ============================================================================= | |
| import os | |
| import re | |
| import time | |
| import asyncio | |
| import logging | |
| import httpx | |
| from html.parser import HTMLParser | |
| from urllib.parse import urlparse, parse_qs | |
| from typing import Optional, Tuple, Dict, Any | |
| from . import config | |
| from . import models as models_mod | |
| from . import pricing | |
| from . import db_sync | |
| logging.getLogger("httpx").setLevel(logging.WARNING) | |
| logger = logging.getLogger("providers") | |
| # ============================================================================= | |
| # SECTION 0 — Errors, Limits, Retry | |
| # ============================================================================= | |
| class ProviderError(RuntimeError): | |
| """Sanitized provider error. RuntimeError subclass → app.py maps it to 502.""" | |
| def __init__(self, message: str, status: Optional[int] = None): | |
| super().__init__(message) | |
| self.status = status | |
| _TRANSIENT_STATUS = {408, 425, 429, 500, 502, 503, 504, 529} | |
| def _limit(key: str, default, cast=int): | |
| """Read one value from .pyfun [HUB_LIMITS] with safe fallback.""" | |
| try: | |
| return cast(config.get_limits().get(key, default)) | |
| except (TypeError, ValueError): | |
| return cast(default) | |
| _sem: Optional[asyncio.Semaphore] = None | |
| def _get_sem() -> asyncio.Semaphore: | |
| """One global semaphore (MAX_PARALLEL_REQUESTS) — created lazily inside the loop.""" | |
| global _sem | |
| if _sem is None: | |
| _sem = asyncio.Semaphore(max(1, _limit("MAX_PARALLEL_REQUESTS", 5))) | |
| return _sem | |
| def _is_transient(e: Exception) -> bool: | |
| """429/5xx and connect errors are worth a retry. Timeouts are NOT (60s x N = forever).""" | |
| if isinstance(e, ProviderError): | |
| return e.status in _TRANSIENT_STATUS | |
| return isinstance(e, httpx.ConnectError) | |
| async def _with_retry(name: str, call): | |
| """ | |
| Run `call()` (async callable) under the global semaphore. | |
| Attempts = RETRY_COUNT (incl. the first), flat RETRY_DELAY_SEC between them. | |
| """ | |
| attempts = max(1, _limit("RETRY_COUNT", 3)) | |
| delay = max(0.0, _limit("RETRY_DELAY_SEC", 2, float)) | |
| for i in range(1, attempts + 1): | |
| try: | |
| async with _get_sem(): | |
| return await call() | |
| except Exception as e: | |
| if i >= attempts or not _is_transient(e): | |
| raise | |
| logger.info(f"'{name}' transient error ({type(e).__name__}) — retry {i}/{attempts - 1} in {delay}s") | |
| await asyncio.sleep(delay) | |
| def _usage(tokens_in: Any, tokens_out: Any) -> Dict[str, Optional[int]]: | |
| """Normalize token counts to int|None.""" | |
| def _i(v): | |
| try: | |
| return int(v) | |
| except (TypeError, ValueError): | |
| return None | |
| return {"input": _i(tokens_in), "output": _i(tokens_out)} | |
| # ============================================================================= | |
| # SECTION 1 — Base Provider | |
| # Shared HTTP logic — implemented ONCE, reused by all providers. | |
| # ============================================================================= | |
| class BaseProvider: | |
| """ | |
| Base class for all LLM providers. | |
| Subclasses only implement complete() — HTTP logic lives here. | |
| complete() returns (text, usage) with usage = {"input": int|None, "output": int|None}. | |
| """ | |
| def __init__(self, name: str, cfg: dict): | |
| self.name = name | |
| self.cfg = cfg | |
| self.key = os.getenv(cfg.get("env_key", "")) | |
| self.base_url = cfg.get("base_url", "") | |
| self.fallback = cfg.get("fallback_to", "") | |
| self.timeout = int(config.get_limits().get("REQUEST_TIMEOUT_SEC", "60")) | |
| self.model = cfg.get("default_model", "") | |
| self.supports_cache = cfg.get("supports_cache", "false").lower() == "true" | |
| # Safe key hint for debug logs — never log the full key | |
| self._key_hint = ( | |
| f"{self.key[:4]}...{self.key[-4:]}" | |
| if self.key and len(self.key) > 8 | |
| else "***" | |
| ) | |
| async def complete(self, prompt: str, model: str = None, max_tokens: int = 1024, | |
| system: str = None) -> Tuple[str, Dict[str, Optional[int]]]: | |
| """Override in each provider subclass.""" | |
| raise NotImplementedError | |
| async def _post(self, url: str, headers: dict, payload: dict) -> dict: | |
| """ | |
| Shared HTTP POST — used by all providers. | |
| Raises ProviderError with sanitized message on non-2xx responses. | |
| API keys are never included in raised exceptions or log output. | |
| """ | |
| safe_url = url.split("?")[0] # strip query params (may contain API keys) | |
| logger.debug(f"POST → {safe_url}") | |
| async with httpx.AsyncClient() as client: | |
| r = await client.post( | |
| url, | |
| headers=headers, | |
| json=payload, | |
| timeout=self.timeout, | |
| ) | |
| try: | |
| r.raise_for_status() | |
| except httpx.HTTPStatusError as e: | |
| # Sanitize: only status code + safe_url, never headers or body | |
| raise ProviderError( | |
| f"HTTP {e.response.status_code} from {safe_url}", | |
| e.response.status_code, | |
| ) from None | |
| return r.json() | |
| # ============================================================================= | |
| # SECTION 2 — LLM Provider Implementations | |
| # Only the API-specific parsing logic differs per provider. | |
| # ============================================================================= | |
| # --- OpenAI-compatible (GENERIC — no code needed for new providers) ----------- | |
| class OpenAICompatProvider(BaseProvider): | |
| """ | |
| Generic OpenAI-compatible /chat/completions provider. | |
| Covers smollm, openrouter, huggingface, mistral, openai, xai, groq, ... | |
| Everything provider-specific lives in the .pyfun block: | |
| [LLM_PROVIDER.mistral] | |
| active = "true" | |
| type = "openai_compat" | |
| base_url = "https://api.mistral.ai/v1" | |
| env_key = "MISTRAL_API_KEY" | |
| default_model = "mistral-small-latest" | |
| fallback_to = "" | |
| # optional: | |
| # auth_header = "Authorization" (default) | |
| # auth_scheme = "Bearer" (default; "" = send the raw key) | |
| # chat_path = "/chat/completions" (default) | |
| # header.X-Foo = "{key}" extra header; {key} = API key, | |
| # {env:NAME|fallback} = ENV value | |
| [LLM_PROVIDER.mistral_END] | |
| """ | |
| _PLACEHOLDER = re.compile(r"\{(key|env:([A-Za-z0-9_]+)(?:\|([^}]*))?)\}") | |
| def __init__(self, name: str, cfg: dict): | |
| super().__init__(name, cfg) | |
| self.auth_header = cfg.get("auth_header", "Authorization") | |
| self.auth_scheme = cfg.get("auth_scheme", "Bearer") | |
| self.chat_path = cfg.get("chat_path", "/chat/completions") | |
| def _fill(self, value: str) -> str: | |
| def sub(m): | |
| if m.group(1) == "key": | |
| return self.key or "" | |
| return os.getenv(m.group(2)) or (m.group(3) or "") | |
| return self._PLACEHOLDER.sub(sub, value) | |
| def _headers(self) -> dict: | |
| headers = { | |
| self.auth_header: f"{self.auth_scheme} {self.key}".strip(), | |
| "content-type": "application/json", | |
| } | |
| for k, v in self.cfg.items(): | |
| if k.startswith("header."): | |
| headers[k[len("header."):]] = self._fill(v) | |
| return headers | |
| async def complete(self, prompt: str, model: str = None, max_tokens: int = 1024, | |
| system: str = None): | |
| messages = [] | |
| if system: | |
| messages.append({"role": "system", "content": system}) | |
| messages.append({"role": "user", "content": prompt}) | |
| data = await self._post( | |
| f"{self.base_url}{self.chat_path}", | |
| headers=self._headers(), | |
| payload={ | |
| "model": model or self.model, | |
| "max_tokens": max_tokens, | |
| "messages": messages, | |
| }, | |
| ) | |
| text = data["choices"][0]["message"]["content"] | |
| if text is None: | |
| raise ProviderError("empty content in response") | |
| u = data.get("usage") or {} | |
| return text, _usage(u.get("prompt_tokens"), u.get("completion_tokens")) | |
| # --- Anthropic ---------------------------------------------------------------- | |
| class AnthropicProvider(BaseProvider): | |
| """ | |
| Anthropic Claude API — Messages endpoint. | |
| Prompt Caching (supports_cache = "true" in .pyfun): | |
| Uses cache_control: ephemeral on system prompt and long user prompts. | |
| Cache TTL: 5 minutes, extended on each cache hit. | |
| Min tokens to cache: ~1024 (Anthropic requirement). | |
| Cost: cache write ~25% more, cache read ~90% less than normal input. | |
| Usage: input = input + cache_creation + cache_read tokens (upper bound for cost). | |
| .pyfun block: | |
| [LLM_PROVIDER.anthropic] | |
| active = "true" | |
| type = "anthropic" | |
| base_url = "https://api.anthropic.com/v1" | |
| env_key = "ANTHROPIC_API_KEY" | |
| api_version_header = "2023-06-01" | |
| default_model = "claude-haiku-5-5" | |
| supports_cache = "true" | |
| fallback_to = "gemini" | |
| [LLM_PROVIDER.anthropic_END] | |
| """ | |
| # Rough chars-per-token estimate — avoids importing tiktoken in sandbox | |
| _CHARS_PER_TOKEN = 4 | |
| _CACHE_MIN_TOKENS = 1024 | |
| def __init__(self, name: str, cfg: dict): | |
| super().__init__(name, cfg) | |
| self.api_version = cfg.get("api_version_header", "2023-06-01") | |
| def _is_cacheable(self, text: str) -> bool: | |
| """Estimate if text is long enough to benefit from caching.""" | |
| return len(text) >= self._CACHE_MIN_TOKENS * self._CHARS_PER_TOKEN | |
| async def complete(self, prompt: str, model: str = None, max_tokens: int = 1024, | |
| system: str = None): | |
| headers = { | |
| "x-api-key": self.key, | |
| "anthropic-version": self.api_version, | |
| "content-type": "application/json", | |
| } | |
| # --- Build user content (cache_control if enabled + long enough) --- | |
| if self.supports_cache and self._is_cacheable(prompt): | |
| user_content = [{"type": "text", "text": prompt, | |
| "cache_control": {"type": "ephemeral"}}] | |
| headers["anthropic-beta"] = "prompt-caching-2024-07-31" | |
| logger.debug("Anthropic: prompt cache_control applied to user message.") | |
| else: | |
| user_content = prompt # short prompt — plain string, no overhead | |
| payload = { | |
| "model": model or self.model, | |
| "max_tokens": max_tokens, | |
| "messages": [{"role": "user", "content": user_content}], | |
| } | |
| # --- Optional system prompt with cache_control --- | |
| if system: | |
| if self.supports_cache and self._is_cacheable(system): | |
| payload["system"] = [{"type": "text", "text": system, | |
| "cache_control": {"type": "ephemeral"}}] | |
| headers["anthropic-beta"] = "prompt-caching-2024-07-31" | |
| logger.debug("Anthropic: prompt cache_control applied to system prompt.") | |
| else: | |
| payload["system"] = system | |
| data = await self._post(f"{self.base_url}/messages", headers, payload) | |
| text = data["content"][0]["text"] | |
| u = data.get("usage") or {} | |
| tin = None | |
| if u: | |
| tin = sum(int(u.get(k) or 0) for k in | |
| ("input_tokens", "cache_creation_input_tokens", "cache_read_input_tokens")) | |
| return text, _usage(tin, u.get("output_tokens")) | |
| # --- Gemini ------------------------------------------------------------------ | |
| class GeminiProvider(BaseProvider): | |
| """ | |
| Google Gemini API — generateContent endpoint. | |
| Implicit Caching (always active on Gemini side, no code needed). | |
| Explicit Context Caching is NOT implemented (only worth it for 32k+ static contexts). | |
| .pyfun block: | |
| [LLM_PROVIDER.gemini] | |
| active = "true" | |
| type = "gemini" | |
| base_url = "https://generativelanguage.googleapis.com/v1beta" | |
| env_key = "GEMINI_API_KEY" | |
| default_model = "gemini-2.5-flash" | |
| supports_cache = "true" | |
| fallback_to = "openrouter" | |
| [LLM_PROVIDER.gemini_END] | |
| """ | |
| async def complete(self, prompt: str, model: str = None, max_tokens: int = 1024, | |
| system: str = None): | |
| m = model or self.model | |
| safe_url = f"{self.base_url}/models/{m}:generateContent" | |
| if self.supports_cache: | |
| logger.debug(f"Gemini: implicit caching active for model {m} (server-side, automatic).") | |
| body = { | |
| "contents": [{"parts": [{"text": prompt}]}], | |
| "generationConfig": {"maxOutputTokens": max_tokens}, | |
| } | |
| if system: | |
| body["systemInstruction"] = {"parts": [{"text": system}]} | |
| async with httpx.AsyncClient() as client: | |
| r = await client.post( | |
| safe_url, | |
| params={"key": self.key}, # key in query param, never in logs | |
| json=body, | |
| timeout=self.timeout, | |
| ) | |
| try: | |
| r.raise_for_status() | |
| except httpx.HTTPStatusError as e: | |
| raise ProviderError( | |
| f"HTTP {e.response.status_code} from {safe_url}", | |
| e.response.status_code, | |
| ) from None | |
| data = r.json() | |
| text = data["candidates"][0]["content"]["parts"][0]["text"] | |
| meta = data.get("usageMetadata") or {} | |
| return text, _usage(meta.get("promptTokenCount"), meta.get("candidatesTokenCount")) | |
| # ============================================================================= | |
| # SECTION 3 — Provider Registry | |
| # Built from .pyfun [LLM_PROVIDERS] at initialize(). | |
| # Handler is chosen by the `type` field of the block (default: the block name). | |
| # type = "openai_compat" → OpenAICompatProvider (generic, no code per provider) | |
| # type = "anthropic" → AnthropicProvider | |
| # type = "gemini" → GeminiProvider | |
| # New API format = new class above + one line here. | |
| # ============================================================================= | |
| _TYPE_CLASSES = { | |
| "openai_compat": OpenAICompatProvider, | |
| "anthropic": AnthropicProvider, | |
| "gemini": GeminiProvider, | |
| } | |
| _registry: dict = {} | |
| def initialize() -> None: | |
| """ | |
| Build provider registries from .pyfun [LLM_PROVIDERS] + [SEARCH_PROVIDERS]. | |
| Called once by mcp.py during startup sequence. | |
| Skips providers with missing ENV keys — no crash, just fewer tools. | |
| """ | |
| global _registry | |
| active = config.get_active_llm_providers() | |
| for name, cfg in active.items(): | |
| env_key = cfg.get("env_key", "") | |
| if not env_key or not os.getenv(env_key): | |
| logger.info(f"Provider '{name}' skipped — ENV key not set.") | |
| continue | |
| ptype = (cfg.get("type") or name).strip().lower() | |
| cls = _TYPE_CLASSES.get(ptype) | |
| if not cls: | |
| logger.info(f"Provider '{name}' has no handler (type '{ptype}') — skipped.") | |
| continue | |
| _registry[name] = cls(name, cfg) | |
| cache_hint = " [cache: ON]" if cfg.get("supports_cache", "false") == "true" else "" | |
| logger.info(f"Provider registered: {name} (type: {ptype}){cache_hint}") | |
| # Search providers — same key-gating principle | |
| _initialize_search() | |
| # ============================================================================= | |
| # SECTION 4 — LLM Execution + Fallback Chain | |
| # ============================================================================= | |
| def provider_for_model(model: str) -> Optional[str]: | |
| """ | |
| Provider that owns `model` according to .pyfun [MODELS] (provider field). | |
| None if the model is unknown → caller keeps its default provider. | |
| """ | |
| if not model: | |
| return None | |
| entry = models_mod.get(model) or {} | |
| return entry.get("provider") or None | |
| def _clamp_tokens(model: str, max_tokens: int) -> int: | |
| """Hard cap (MAX_TOKENS_CAP) + per-model max_output_tokens (manual or auto-pricing).""" | |
| try: | |
| n = max(1, int(max_tokens)) | |
| except (TypeError, ValueError): | |
| n = 1024 | |
| cap = _limit("MAX_TOKENS_CAP", 8192) | |
| if cap > 0: | |
| n = min(n, cap) | |
| lim = pricing.max_output_for(model) | |
| if lim: | |
| n = min(n, lim) | |
| return n | |
| async def _record_usage(tool: str, provider: str, model: str, usage: dict, latency_ms: int) -> None: | |
| """Log usage + estimated cost. NEVER raises — accounting must not break a call.""" | |
| try: | |
| tin, tout = usage.get("input"), usage.get("output") | |
| cost = None | |
| if tin is not None or tout is not None: | |
| cost = await pricing.cost_for(model, tin, tout) | |
| await db_sync.log_usage(tool, provider, model, tin, tout, cost, latency_ms) | |
| cost_txt = f"${cost:.6f}" if cost is not None else "n/a" | |
| logger.info(f"Usage: {provider}/{model} in={tin} out={tout} cost≈{cost_txt} ({latency_ms} ms)") | |
| except Exception as e: | |
| logger.debug(f"Usage logging skipped: {type(e).__name__}") | |
| async def llm_complete( | |
| prompt: str, | |
| provider_name: str = None, | |
| model: str = None, | |
| max_tokens: int = 1024, | |
| system: str = None, | |
| tool_name: str = "llm_complete", | |
| ) -> str: | |
| """ | |
| Send prompt to LLM provider with automatic fallback chain. | |
| Fallback order is defined in .pyfun via fallback_to field. | |
| Raises RuntimeError if all providers in the chain fail. | |
| Args: | |
| prompt: Input text to send to the model. | |
| provider_name: Provider override. Unknown / non-LLM names are ignored. | |
| If only `model` is given, the provider is derived from [MODELS]. | |
| model: Model override — applies to the FIRST provider only. | |
| max_tokens: Max tokens in response (clamped, see header). | |
| system: Optional system prompt — sent as real system prompt to every provider. | |
| tool_name: Only used for the usage log. | |
| Returns: | |
| "[provider] text" | |
| """ | |
| model = (model or "").strip() or None | |
| if provider_name and provider_name not in config.get_llm_providers(): | |
| logger.debug(f"'{provider_name}' is not an LLM provider — using default.") | |
| provider_name = None | |
| if not provider_name: | |
| provider_name = provider_for_model(model) | |
| if not provider_name: | |
| tools_cfg = config.get_active_tools() | |
| provider_name = tools_cfg.get("llm_complete", {}).get("default_provider", "anthropic") | |
| visited = set() | |
| current = provider_name | |
| first = True | |
| while current and current not in visited: | |
| visited.add(current) | |
| provider = _registry.get(current) | |
| use_model = model if first else None # explicit model belongs to the first provider only | |
| first = False | |
| if not provider: | |
| logger.warning(f"Provider '{current}' not in registry — trying fallback.") | |
| else: | |
| m = use_model or provider.model | |
| mt = _clamp_tokens(m, max_tokens) | |
| try: | |
| t0 = time.monotonic() | |
| text, usage = await _with_retry( | |
| current, | |
| lambda p=provider, um=use_model, n=mt: p.complete(prompt, um, n, system=system), | |
| ) | |
| latency_ms = int((time.monotonic() - t0) * 1000) | |
| logger.info(f"Response from provider: '{current}'") | |
| await _record_usage(tool_name, current, m, usage, latency_ms) | |
| return f"[{current}] {text}" | |
| except Exception as e: | |
| # Log only exception type + sanitized message — never raw headers/keys/bodies | |
| logger.warning( | |
| f"Provider '{current}' failed: {type(e).__name__}: {e} — trying fallback." | |
| ) | |
| cfg = config.get_active_llm_providers().get(current, {}) | |
| current = cfg.get("fallback_to", "") | |
| raise RuntimeError("All providers failed — no fallback available.") | |
| # Alias — used internally by tools.py | |
| complete = llm_complete | |
| # ============================================================================= | |
| # SECTION 5 — Search Execution | |
| # Same pattern as LLM providers: class + _SEARCH_PROVIDER_CLASSES + fallback. | |
| # Adding a search provider = add class + register + .pyfun block. Nothing else. | |
| # ============================================================================= | |
| class BaseSearchProvider: | |
| """ | |
| Base class for all search providers. | |
| Subclasses only implement search() — HTTP logic lives here. | |
| """ | |
| def __init__(self, name: str, cfg: dict): | |
| self.name = name | |
| self.key = os.getenv(cfg.get("env_key", "")) | |
| self.base_url = cfg.get("base_url", "") | |
| self.fallback = cfg.get("fallback_to", "") | |
| self.timeout = int(config.get_limits().get("SEARCH_TIMEOUT_SEC", "30")) | |
| self.default_results = int(cfg.get("default_results", "5")) | |
| self.max_results = int(cfg.get("max_results", "20")) | |
| self._key_hint = ( | |
| f"{self.key[:4]}...{self.key[-4:]}" | |
| if self.key and len(self.key) > 8 | |
| else "***" | |
| ) | |
| async def search(self, query: str, max_results: int) -> str: | |
| """Override in each search provider subclass.""" | |
| raise NotImplementedError | |
| async def _get(self, url: str, headers: dict, params: dict) -> dict: | |
| """Shared HTTP GET with sanitized errors (never leak keys/bodies).""" | |
| safe_url = url.split("?")[0] | |
| async with httpx.AsyncClient() as client: | |
| r = await client.get(url, headers=headers, params=params, timeout=self.timeout) | |
| try: | |
| r.raise_for_status() | |
| except httpx.HTTPStatusError as e: | |
| raise ProviderError(f"HTTP {e.response.status_code} from {safe_url}", e.response.status_code) from None | |
| return r.json() | |
| async def _get_text(self, url: str, headers: dict, params: dict) -> str: | |
| """Shared HTTP GET returning the raw body (HTML) — sanitized errors.""" | |
| safe_url = url.split("?")[0] | |
| async with httpx.AsyncClient() as client: | |
| r = await client.get(url, headers=headers, params=params, timeout=self.timeout) | |
| try: | |
| r.raise_for_status() | |
| except httpx.HTTPStatusError as e: | |
| raise ProviderError(f"HTTP {e.response.status_code} from {safe_url}", e.response.status_code) from None | |
| return r.text | |
| async def _post(self, url: str, headers: dict, payload: dict) -> dict: | |
| """Shared HTTP POST with sanitized errors.""" | |
| safe_url = url.split("?")[0] | |
| async with httpx.AsyncClient() as client: | |
| r = await client.post(url, headers=headers, json=payload, timeout=self.timeout) | |
| try: | |
| r.raise_for_status() | |
| except httpx.HTTPStatusError as e: | |
| raise ProviderError(f"HTTP {e.response.status_code} from {safe_url}", e.response.status_code) from None | |
| return r.json() | |
| def _clip(self, max_results: int) -> int: | |
| """Clamp requested results into [1, max_results].""" | |
| try: | |
| n = int(max_results) | |
| except (TypeError, ValueError): | |
| n = self.default_results | |
| return max(1, min(n, self.max_results)) | |
| # --- Brave Search --------------------------------------------------------------- | |
| class BraveSearchProvider(BaseSearchProvider): | |
| """ | |
| Brave Search API — web search endpoint. | |
| Auth via X-Subscription-Token header (NOT Bearer!). | |
| .pyfun block: | |
| [SEARCH_PROVIDER.brave] | |
| active = "true" | |
| base_url = "https://api.search.brave.com/res/v1/web/search" | |
| env_key = "BRAVE_API_KEY" | |
| default_results = "5" | |
| max_results = "20" | |
| fallback_to = "tavily" | |
| [SEARCH_PROVIDER.brave_END] | |
| """ | |
| async def search(self, query: str, max_results: int = 5) -> str: | |
| data = await self._get( | |
| self.base_url, | |
| headers={ | |
| "X-Subscription-Token": self.key, | |
| "Accept": "application/json", | |
| "Accept-Encoding": "gzip", | |
| }, | |
| params={ | |
| "q": query, | |
| "count": self._clip(max_results), | |
| }, | |
| ) | |
| results = data.get("web", {}).get("results", []) | |
| if not results: | |
| raise RuntimeError("Brave returned no results") | |
| lines = [f"[{self.name}] Search results for: '{query}'"] | |
| for i, r in enumerate(results, 1): | |
| lines.append(f"{i}. {r.get('title', '')}\n {r.get('url', '')}\n {r.get('description', '')}") | |
| return "\n\n".join(lines) | |
| # --- Tavily --------------------------------------------------------------------- | |
| class TavilySearchProvider(BaseSearchProvider): | |
| """ | |
| Tavily Search API — AI-optimized web search. | |
| include_answer = "true" in .pyfun → adds AI-synthesized answer on top. | |
| .pyfun block: | |
| [SEARCH_PROVIDER.tavily] | |
| active = "true" | |
| base_url = "https://api.tavily.com/search" | |
| env_key = "TAVILY_API_KEY" | |
| default_results = "5" | |
| max_results = "10" | |
| include_answer = "true" | |
| fallback_to = "" | |
| [SEARCH_PROVIDER.tavily_END] | |
| """ | |
| def __init__(self, name: str, cfg: dict): | |
| super().__init__(name, cfg) | |
| self.include_answer = cfg.get("include_answer", "false").lower() == "true" | |
| async def search(self, query: str, max_results: int = 5) -> str: | |
| data = await self._post( | |
| self.base_url, | |
| headers={"Content-Type": "application/json"}, | |
| payload={ | |
| "api_key": self.key, # Tavily takes the key in the payload | |
| "query": query, | |
| "max_results": self._clip(max_results), | |
| "include_answer": self.include_answer, | |
| }, | |
| ) | |
| results = data.get("results", []) | |
| if not results and not data.get("answer"): | |
| raise RuntimeError("Tavily returned no results") | |
| lines = [f"[{self.name}] Search results for: '{query}'"] | |
| if self.include_answer and data.get("answer"): | |
| lines.append(f"AI answer: {data['answer']}") | |
| for i, r in enumerate(results, 1): | |
| lines.append(f"{i}. {r.get('title', '')}\n {r.get('url', '')}\n {r.get('content', '')}") | |
| return "\n\n".join(lines) | |
| # --- DuckDuckGo (keyless, gentle last-resort fallback) -------------------------- | |
| class _DDGParser(HTMLParser): | |
| """Minimal parser for html.duckduckgo.com result pages (stdlib only, no BeautifulSoup).""" | |
| def __init__(self): | |
| super().__init__() | |
| self.results: list = [] | |
| self._mode = None | |
| self._cur = None | |
| self._buf: list = [] | |
| def _unwrap(href: str) -> str: | |
| """DDG wraps targets as //duckduckgo.com/l/?uddg=<urlencoded>&rut=… → real URL.""" | |
| target = (parse_qs(urlparse(href).query).get("uddg") or [href])[0] | |
| return "https:" + target if target.startswith("//") else target | |
| def handle_starttag(self, tag, attrs): | |
| if tag != "a": | |
| return | |
| a = dict(attrs) | |
| cls = (a.get("class") or "").split() | |
| if "result__a" in cls: | |
| self._cur = {"title": "", "url": self._unwrap(a.get("href") or ""), "snippet": ""} | |
| self.results.append(self._cur) | |
| self._mode, self._buf = "title", [] | |
| elif "result__snippet" in cls and self.results: | |
| self._cur = self.results[-1] | |
| self._mode, self._buf = "snippet", [] | |
| def handle_data(self, data): | |
| if self._mode: | |
| self._buf.append(data) | |
| def handle_endtag(self, tag): | |
| if tag == "a" and self._mode: | |
| self._cur[self._mode] = " ".join("".join(self._buf).split()) | |
| self._mode = None | |
| class DuckDuckGoSearchProvider(BaseSearchProvider): | |
| """ | |
| Keyless search fallback (idea: "GentleSearchProvider"). | |
| DuckDuckGo has NO official search API — this reads the public HTML endpoint, so it is | |
| fragile by nature and gentle by design: | |
| - global cooldown (min_interval_sec, default 300): called too early it RAISES, so the | |
| fallback chain reports an honest failure instead of a fake "result" string | |
| - the attempt counts for the cooldown even if it fails (no hammering a blocked endpoint) | |
| - honest User-Agent, no evasion tricks, no retries | |
| - captcha / layout change = no parsable results = clear error | |
| .pyfun block (no key needed → requires_key = "false"): | |
| [SEARCH_PROVIDER.duckduckgo] | |
| active = "false" requires_key = "false" | |
| base_url = "https://html.duckduckgo.com/html/" | |
| min_interval_sec = "300" # optional user_agent = "..." # optional | |
| """ | |
| def __init__(self, name: str, cfg: dict): | |
| super().__init__(name, cfg) | |
| self.base_url = cfg.get("base_url") or "https://html.duckduckgo.com/html/" | |
| self.min_interval = float(cfg.get("min_interval_sec", "300")) | |
| self.user_agent = cfg.get("user_agent") or "UniversalMCPHub/1.0 (+https://github.com/VolkanSah)" | |
| self._last_call: Optional[float] = None | |
| async def search(self, query: str, max_results: int) -> str: | |
| now = time.monotonic() | |
| if self._last_call is not None and now - self._last_call < self.min_interval: | |
| wait = int(self.min_interval - (now - self._last_call)) | |
| raise ProviderError(f"DuckDuckGo cooldown — wait {wait}s (min_interval_sec={int(self.min_interval)})") | |
| self._last_call = now # counts even if the request fails | |
| try: | |
| html = await self._get_text( | |
| self.base_url, | |
| headers={ | |
| "User-Agent": self.user_agent, | |
| "Accept": "text/html,application/xhtml+xml", | |
| "Accept-Language": "de,en-US;q=0.7,en;q=0.3", | |
| "DNT": "1", | |
| }, | |
| params={"q": query}, | |
| ) | |
| except ProviderError as e: | |
| raise ProviderError(str(e)) from None # drop status → never retried (gentle) | |
| parser = _DDGParser() | |
| parser.feed(html) | |
| hits = [r for r in parser.results if r["url"]][: self._clip(max_results)] | |
| if not hits: | |
| raise ProviderError("DuckDuckGo: no parsable results (blocked/captcha or layout changed)") | |
| lines = [f"[{self.name}] Search results for: '{query}'"] | |
| for i, r in enumerate(hits, 1): | |
| lines.append(f"{i}. {r['title']}\n {r['url']}\n {r['snippet']}") | |
| return "\n\n".join(lines) | |
| # ============================================================================= | |
| # Search Registry — built from .pyfun [SEARCH_PROVIDERS] at initialize() | |
| # ============================================================================= | |
| _SEARCH_PROVIDER_CLASSES = { | |
| "brave": BraveSearchProvider, | |
| "tavily": TavilySearchProvider, | |
| "duckduckgo": DuckDuckGoSearchProvider, # keyless (requires_key = "false" in .pyfun) | |
| # "serper": SerperSearchProvider, # ← uncomment to activate | |
| } | |
| _search_registry: dict = {} | |
| def _initialize_search() -> None: | |
| """ | |
| Build search provider registry from .pyfun [SEARCH_PROVIDERS]. | |
| Called inside initialize() — same key-gating as LLM providers: | |
| no ENV key = provider skipped, no crash. | |
| """ | |
| global _search_registry | |
| for name, cfg in config.get_active_search_providers().items(): | |
| env_key = cfg.get("env_key", "") | |
| needs_key = str(cfg.get("requires_key", "true")).lower() != "false" | |
| if needs_key and (not env_key or not os.getenv(env_key)): | |
| logger.info(f"Search provider '{name}' skipped — ENV key not set.") | |
| continue | |
| cls = _SEARCH_PROVIDER_CLASSES.get(name) | |
| if not cls: | |
| logger.info(f"Search provider '{name}' has no handler yet — skipped.") | |
| continue | |
| _search_registry[name] = cls(name, cfg) | |
| logger.info(f"Search provider registered: {name}") | |
| async def search( | |
| query: str, | |
| provider_name: str = None, | |
| max_results: int = 5, | |
| ) -> str: | |
| """ | |
| Search the web via configured search provider with automatic fallback chain. | |
| Fallback order is defined in .pyfun via fallback_to field. | |
| Raises RuntimeError if all providers in the chain fail. | |
| Args: | |
| query: Search query string. | |
| provider_name: Provider name override (e.g. 'brave', 'tavily'). | |
| Defaults to default_provider from .pyfun [TOOL.web_search]. | |
| max_results: Maximum number of results. Default: 5. | |
| Returns: | |
| Formatted search results as plain text string. | |
| """ | |
| if provider_name and provider_name not in config.get_search_providers(): | |
| logger.debug(f"'{provider_name}' is not a search provider — using default.") | |
| provider_name = None | |
| if not provider_name: | |
| tools_cfg = config.get_active_tools() | |
| provider_name = tools_cfg.get("web_search", {}).get("default_provider", "brave") | |
| visited = set() | |
| current = provider_name | |
| while current and current not in visited: | |
| visited.add(current) | |
| provider = _search_registry.get(current) | |
| if not provider: | |
| logger.warning(f"Search provider '{current}' not in registry — trying fallback.") | |
| else: | |
| try: | |
| result = await _with_retry(current, lambda p=provider: p.search(query, max_results)) | |
| logger.info(f"Search response from provider: '{current}'") | |
| return result | |
| except Exception as e: | |
| logger.warning( | |
| f"Search provider '{current}' failed: {type(e).__name__}: {e} — trying fallback." | |
| ) | |
| cfg = config.get_active_search_providers().get(current, {}) | |
| current = cfg.get("fallback_to", "") | |
| raise RuntimeError("All search providers failed — no fallback available.") | |
| # ============================================================================= | |
| # SECTION 6 — Registry Helpers | |
| # Used by mcp.py for tool registration decisions. | |
| # ============================================================================= | |
| def list_active_llm() -> list: | |
| """Returns list of active LLM provider names.""" | |
| return list(_registry.keys()) | |
| def list_active_search() -> list: | |
| """Returns list of active search provider names.""" | |
| return list(_search_registry.keys()) | |
| def get(name: str) -> BaseProvider: | |
| """Get a specific provider instance by name.""" | |
| return _registry.get(name) | |
| # ============================================================================= | |
| # Direct execution guard | |
| # ============================================================================= | |
| if __name__ == "__main__": | |
| print("WARNING: Run via main.py → app.py, not directly.") |