feat: add Tavily as middle-tier search provider alongside Exa

pyproject.toml

@@ -47,6 +47,7 @@ dependencies = [
     "scrapling[fetchers]>=0.4",
     "html2text>=2024.2.26",
     "beautifulsoup4>=4.12.0",
+    "tavily-python>=0.5.0",
 ]
 
 [project.urls]

src/searxng_mcp/config.py

@@ -39,6 +39,7 @@ def _env_float(key: str, default: float) -> float:
 CRAWL_MAX_CHARS: Final[int] = _env_int("SEARXNG_CRAWL_MAX_CHARS", 8000)
 PIXABAY_API_KEY: Final[str] = _env_str("PIXABAY_API_KEY", "")
 EXA_API_KEY: Final[str] = _env_str("EXA_API_KEY", "")
+TAVILY_API_KEY: Final[str] = _env_str("TAVILY_API_KEY", "")
 
 # Search provider preference: "searxng", "exa", or "auto" (try exa first, fallback to searxng)
 SEARCH_PROVIDER: Final[str] = _env_str("SEARCH_PROVIDER", "auto")
@@ -89,6 +90,7 @@ def clamp_text(text: str, limit: int = MAX_RESPONSE_CHARS, *, suffix: str | None
     "CRAWL_MAX_CHARS",
     "PIXABAY_API_KEY",
     "EXA_API_KEY",
+    "TAVILY_API_KEY",
     "SEARCH_PROVIDER",
     "MAX_RETRIES",
     "RETRY_BASE_DELAY",

src/searxng_mcp/server.py

@@ -25,6 +25,7 @@
 from .domain_health import get_domain_health_tracker
 from .errors import ErrorParser
 from .exa import ExaSearcher
+from .tavily import TavilySearcher
 from .extractor import DataExtractor
 from .github import GitHubClient, RepoInfo
 from .images import PixabayClient
@@ -36,6 +37,7 @@
 mcp = FastMCP("web-research-assistant")
 searxng_searcher = SearxSearcher()
 exa_searcher = ExaSearcher()
+tavily_searcher = TavilySearcher()
 crawler_client = CrawlerClient()
 
 
@@ -46,12 +48,13 @@ async def unified_search(
     max_results: int = 5,
     time_range: str | None = None,
 ) -> list[SearchHit]:
-    """Unified search that uses Exa or SearXNG based on configuration.
+    """Unified search that uses Exa, Tavily, or SearXNG based on configuration.
 
     Provider selection:
     - "exa": Use Exa AI only
+    - "tavily": Use Tavily only
     - "searxng": Use SearXNG only
-    - "auto" (default): Try Exa first if API key is set, fallback to SearXNG
+    - "auto" (default): Try Exa first, then Tavily, fallback to SearXNG
 
     Args:
         query: Search query string
@@ -104,9 +107,29 @@ async def unified_search(
                 start_published_date=start_date,
             )
         except Exception as e:
-            # If Exa fails and we're in auto mode, try SearXNG
+            # If Exa fails and we're in auto mode, try Tavily then SearXNG
             if provider == "auto":
-                logging.warning(f"Exa search failed, falling back to SearXNG: {e}")
+                logging.warning(f"Exa search failed, falling back to next provider: {e}")
+            else:
+                raise
+
+    # Try Tavily as middle-tier fallback (auto mode or explicit tavily provider)
+    tavily_available = (
+        provider in ("tavily", "auto")
+        and tavily_searcher.has_api_key()
+    )
+    if tavily_available:
+        try:
+            tavily_topic = "news" if category == "news" else "general"
+            return await tavily_searcher.search(
+                query,
+                max_results=max_results,
+                topic=tavily_topic,
+                time_range=time_range,
+            )
+        except Exception as e:
+            if provider == "auto":
+                logging.warning(f"Tavily search failed, falling back to SearXNG: {e}")
             else:
                 raise
 

src/searxng_mcp/tavily.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import logging
+
+from tavily import AsyncTavilyClient
+
+from .config import MAX_SNIPPET_CHARS, TAVILY_API_KEY, clamp_text
+from .search import SearchHit
+
+logger = logging.getLogger(__name__)
+
+
+class TavilySearcher:
+    """Async client for the Tavily search API.
+
+    Tavily provides AI-optimised web search, acting as a middle-tier
+    fallback between Exa and SearXNG in the 'auto' provider chain.
+    """
+
+    def __init__(self, api_key: str | None = None) -> None:
+        self.api_key = api_key or TAVILY_API_KEY
+        self._client: AsyncTavilyClient | None = None
+
+    def has_api_key(self) -> bool:
+        """Check if an API key is configured."""
+        return bool(self.api_key)
+
+    def _get_client(self) -> AsyncTavilyClient:
+        if self._client is None:
+            self._client = AsyncTavilyClient(api_key=self.api_key)
+        return self._client
+
+    async def search(
+        self,
+        query: str,
+        *,
+        max_results: int = 5,
+        search_depth: str = "basic",
+        topic: str = "general",
+        time_range: str | None = None,
+        include_domains: list[str] | None = None,
+        exclude_domains: list[str] | None = None,
+    ) -> list[SearchHit]:
+        """Search using the Tavily API.
+
+        Args:
+            query: Search query string (max 400 chars recommended).
+            max_results: Number of results to return.
+            search_depth: "basic" or "advanced".
+            topic: "general", "news", or "finance".
+            time_range: Time filter - "day", "week", "month", "year".
+            include_domains: Limit results to these domains.
+            exclude_domains: Exclude results from these domains.
+
+        Returns:
+            List of SearchHit objects.
+
+        Raises:
+            ValueError: If no API key is configured.
+        """
+        if not self.api_key:
+            raise ValueError("Tavily API key not configured. Set TAVILY_API_KEY environment variable.")
+
+        client = self._get_client()
+
+        kwargs: dict = {
+            "query": query,
+            "max_results": max_results,
+            "search_depth": search_depth,
+            "topic": topic,
+        }
+        if time_range:
+            kwargs["time_range"] = time_range
+        if include_domains:
+            kwargs["include_domains"] = include_domains
+        if exclude_domains:
+            kwargs["exclude_domains"] = exclude_domains
+
+        response = await client.search(**kwargs)
+
+        hits: list[SearchHit] = []
+        for result in response.get("results", []):
+            title = (result.get("title") or "Untitled").strip()
+            url = result.get("url", "")
+            snippet = result.get("content", "")
+            if snippet:
+                snippet = clamp_text(snippet, MAX_SNIPPET_CHARS, suffix="...")
+
+            hits.append(SearchHit(title=title, url=url, snippet=snippet))
+
+        return hits
+
+
+__all__ = ["TavilySearcher"]