feat: add optional Tavily search provider

pyproject.toml

@@ -61,6 +61,9 @@ free-web-search-mcp = "zero_api_key_web_search_compat.mcp_server:run"
 pdf = [
   "pypdf>=4.0.0",
 ]
+tavily = [
+  "tavily-python>=0.5.0",
+]
 dev = [
   "build>=1.2.2",
   "twine>=6.1.0",

tests/test_search_web.py

@@ -11,6 +11,7 @@
 
 from zero_api_key_web_search.providers.base import ProviderResult
 from zero_api_key_web_search.providers.brightdata import BrightDataProvider
+from zero_api_key_web_search.providers.tavily import TavilyProvider
 from zero_api_key_web_search.search_web import Answer, Source, UltimateSearcher
 
 
@@ -265,12 +266,23 @@ def test_default_providers_detects_brightdata_env(self):
             providers = [provider.name for provider in self.searcher._default_providers()]
         self.assertEqual(providers, ["ddgs", "brightdata"])
 
+    def test_default_providers_detects_tavily_env(self):
+        with patch.dict(os.environ, {"ZERO_SEARCH_TAVILY_API_KEY": "test-key"}, clear=True):
+            providers = [provider.name for provider in self.searcher._default_providers()]
+        self.assertEqual(providers, ["ddgs", "tavily"])
+
     def test_provider_statuses_include_brightdata_signup(self):
         statuses = {item["name"]: item for item in self.searcher.provider_statuses()}
         self.assertIn("brightdata", statuses)
         self.assertEqual(statuses["brightdata"]["status"], "not_configured")
         self.assertEqual(statuses["brightdata"]["signup_url"], BrightDataProvider.SIGNUP_URL)
 
+    def test_provider_statuses_include_tavily_setup(self):
+        statuses = {item["name"]: item for item in self.searcher.provider_statuses()}
+        self.assertIn("tavily", statuses)
+        self.assertEqual(statuses["tavily"]["status"], "not_configured")
+        self.assertIn("ZERO_SEARCH_TAVILY_API_KEY", statuses["tavily"]["setup"])
+
     def test_search_metadata_includes_free_provider_guidance(self):
         searcher = UltimateSearcher(
             timeout=5,
@@ -291,10 +303,13 @@ def test_search_metadata_includes_free_provider_guidance(self):
         guidance = result.metadata["provider_guidance"]
         self.assertEqual(guidance["free_recommended_pair"], ["ddgs", "searxng"])
         self.assertEqual(guidance["production_provider"], "brightdata")
+        self.assertEqual(guidance["optional_provider"], "tavily")
         self.assertIn("ZERO_SEARCH_SEARXNG_URL", guidance["free_setup_hint"])
         self.assertIn("ZERO_SEARCH_BRIGHTDATA_API_KEY", guidance["production_setup_hint"])
+        self.assertIn("ZERO_SEARCH_TAVILY_API_KEY", guidance["tavily_setup_hint"])
         self.assertFalse(guidance["searxng_configured"])
         self.assertFalse(guidance["brightdata_configured"])
+        self.assertFalse(guidance["tavily_configured"])
 
     def test_search_metadata_marks_free_dual_provider_active(self):
         searcher = UltimateSearcher(
@@ -326,6 +341,51 @@ def test_explicit_unconfigured_brightdata_returns_configuration_error(self):
             any("ZERO_SEARCH_BRIGHTDATA_API_KEY" in error for error in result.metadata["errors"])
         )
 
+    def test_explicit_unconfigured_tavily_returns_configuration_error(self):
+        searcher = UltimateSearcher(timeout=5, providers=[FakeProvider("ddgs", [])])
+        with patch.dict(os.environ, {}, clear=True):
+            result = searcher.search("python release", providers=["tavily"])
+        self.assertEqual(result.sources, [])
+        self.assertTrue(any("ZERO_SEARCH_TAVILY_API_KEY" in error for error in result.metadata["errors"]))
+
+    def test_tavily_provider_normalizes_results(self):
+        captured = {}
+
+        class FakeTavilyClient:
+            def __init__(self, api_key):
+                captured["api_key"] = api_key
+
+            def search(self, **payload):
+                captured["payload"] = payload
+                return {
+                    "results": [
+                        {
+                            "url": "https://example.com/tavily",
+                            "title": "Tavily result",
+                            "content": "Tavily snippet.",
+                            "score": 0.91,
+                            "published_date": "2026-04-01",
+                        }
+                    ]
+                }
+
+        fake_module = type("FakeTavilyModule", (), {"TavilyClient": FakeTavilyClient})
+        with patch.dict(sys.modules, {"tavily": fake_module}):
+            provider = TavilyProvider(timeout=7, api_key="test-key")
+            results = provider.search("ai agents", "news", timelimit="w", max_results=3)
+
+        self.assertEqual(captured["api_key"], "test-key")
+        self.assertEqual(captured["payload"]["topic"], "news")
+        self.assertEqual(captured["payload"]["time_range"], "week")
+        self.assertEqual(captured["payload"]["max_results"], 3)
+        self.assertEqual(results[0].url, "https://example.com/tavily")
+        self.assertEqual(results[0].metadata["provider"], "tavily")
+        self.assertEqual(results[0].metadata["score"], 0.91)
+
+    def test_tavily_provider_skips_unsupported_search_types(self):
+        provider = TavilyProvider(api_key="test-key")
+        self.assertEqual(provider.search("ai agents", "images"), [])
+
     def test_brightdata_provider_normalizes_serp_results(self):
         captured = {}
 

zero_api_key_web_search/core.py

@@ -22,6 +22,7 @@
     ProviderConfigurationError,
     SearchProvider,
     SearxngProvider,
+    TavilyProvider,
     WebUnlockerProvider,
 )
 
@@ -280,7 +281,7 @@ class LlmContextResult:
     "free-verified": ["ddgs", "searxng"],
     "production": ["brightdata"],
     "production-unlock": ["brightdata"],
-    "max-evidence": ["ddgs", "searxng", "brightdata"],
+    "max-evidence": ["ddgs", "searxng", "brightdata", "tavily"],
 }
 
 GOGGLES_PRESETS = {
@@ -336,22 +337,30 @@ def _default_providers(self) -> list[SearchProvider]:
         brightdata = BrightDataProvider(timeout=self.timeout)
         if brightdata.is_configured():
             providers.append(brightdata)
+        tavily = TavilyProvider(timeout=self.timeout)
+        if tavily.is_configured():
+            providers.append(tavily)
         return providers
 
     def _provider_guidance(self) -> dict:
         searxng = SearxngProvider(timeout=self.timeout)
         brightdata = BrightDataProvider(timeout=self.timeout)
+        tavily = TavilyProvider(timeout=self.timeout)
         searxng_configured = searxng.is_configured()
         brightdata_configured = brightdata.is_configured()
+        tavily_configured = tavily.is_configured()
         return {
             "free_recommended_pair": ["ddgs", "searxng"],
             "production_provider": "brightdata",
+            "optional_provider": "tavily",
             "searxng_configured": searxng_configured,
             "searxng_env_vars": list(SearxngProvider.ENV_VARS),
             "brightdata_configured": brightdata_configured,
             "brightdata_api_key_env_vars": list(BrightDataProvider.API_KEY_ENV_VARS),
             "brightdata_zone_env_vars": list(BrightDataProvider.ZONE_ENV_VARS),
             "brightdata_signup_url": BrightDataProvider.SIGNUP_URL,
+            "tavily_configured": tavily_configured,
+            "tavily_api_key_env_vars": list(TavilyProvider.ENV_VARS),
             "free_setup_hint": (
                 "Self-host a SearXNG instance and point "
                 f"{SearxngProvider.ENV_VARS[0]} to it for a free dual-provider path."
@@ -362,9 +371,13 @@ def _provider_guidance(self) -> dict:
                 f"{BrightDataProvider.API_KEY_ENV_VARS[0]}. New users can sign up at "
                 f"{BrightDataProvider.SIGNUP_URL}."
             ),
+            "tavily_setup_hint": (
+                "For an optional Tavily evidence path, install the tavily extra and set "
+                f"{TavilyProvider.ENV_VARS[0]}."
+            ),
             "recommended_next_step": (
                 "Multi-provider evidence path is active."
-                if searxng_configured or brightdata_configured
+                if searxng_configured or brightdata_configured or tavily_configured
                 else (
                     "Configure a self-hosted SearXNG instance for a free second provider "
                     "or Bright Data for production-grade search."
@@ -398,7 +411,7 @@ def provider_profiles(self) -> dict:
                 ),
             },
             "max-evidence": {
-                "providers": ["ddgs", "searxng", "brightdata"],
+                "providers": ["ddgs", "searxng", "brightdata", "tavily"],
                 "description": "Maximum provider diversity across free and production backends.",
             },
         }
@@ -420,6 +433,7 @@ def provider_statuses(self) -> list[dict]:
         searxng = SearxngProvider(timeout=self.timeout)
         brightdata = BrightDataProvider(timeout=self.timeout)
         web_unlocker = WebUnlockerProvider(timeout=self.timeout)
+        tavily = TavilyProvider(timeout=self.timeout)
         return [
             {
                 "name": "ddgs",
@@ -462,6 +476,14 @@ def provider_statuses(self) -> list[dict]:
                 "setup": WebUnlockerProvider.configuration_hint(),
                 "signup_url": WebUnlockerProvider.SIGNUP_URL,
             },
+            {
+                "name": "tavily",
+                "status": "ready" if tavily.is_configured() else "not_configured",
+                "default": "tavily" in configured_names,
+                "kind": "optional api",
+                "description": "Optional Tavily search provider for text and news evidence.",
+                "setup": TavilyProvider.configuration_hint(),
+            },
         ]
 
     def _resolve_provider_profile(self, profile: str | None) -> list[str] | None:
@@ -480,11 +502,13 @@ def _known_optional_provider(self, provider_name: str) -> SearchProvider | None:
             return SearxngProvider(timeout=self.timeout)
         if provider_name == "brightdata":
             return BrightDataProvider(timeout=self.timeout)
+        if provider_name == "tavily":
+            return TavilyProvider(timeout=self.timeout)
         return None
 
     def _provider_registry(self) -> dict[str, SearchProvider]:
         registry = {provider.name: provider for provider in self.providers}
-        for provider_name in ("searxng", "brightdata"):
+        for provider_name in ("searxng", "brightdata", "tavily"):
             if provider_name not in registry:
                 provider = self._known_optional_provider(provider_name)
                 if provider is not None:

zero_api_key_web_search/providers/__init__.py

@@ -8,6 +8,7 @@
 from zero_api_key_web_search.providers.brightdata import BrightDataProvider
 from zero_api_key_web_search.providers.ddgs import DdgsProvider
 from zero_api_key_web_search.providers.searxng import SearxngProvider
+from zero_api_key_web_search.providers.tavily import TavilyProvider
 from zero_api_key_web_search.providers.web_unlocker import WebUnlockerProvider
 
 __all__ = [
@@ -17,5 +18,6 @@
     "ProviderResult",
     "SearchProvider",
     "SearxngProvider",
+    "TavilyProvider",
     "WebUnlockerProvider",
 ]

zero_api_key_web_search/providers/tavily.py

@@ -0,0 +1,118 @@
+"""Tavily-backed search provider."""
+
+from __future__ import annotations
+
+import os
+
+from zero_api_key_web_search.providers.base import ProviderConfigurationError, ProviderResult
+
+
+class TavilyProvider:
+    """Search provider backed by the optional tavily-python package."""
+
+    name = "tavily"
+    ENV_VARS = (
+        "ZERO_SEARCH_TAVILY_API_KEY",
+        "TAVILY_API_KEY",
+    )
+
+    def __init__(self, timeout: int = 15, api_key: str | None = None):
+        self.timeout = timeout
+        self.api_key = api_key or self._configured_api_key()
+
+    @classmethod
+    def _configured_api_key(cls) -> str:
+        for env_var in cls.ENV_VARS:
+            value = os.getenv(env_var, "").strip()
+            if value:
+                return value
+        return ""
+
+    @classmethod
+    def configuration_hint(cls) -> str:
+        primary = cls.ENV_VARS[0]
+        aliases = ", ".join(cls.ENV_VARS[1:])
+        return (
+            f"Configure Tavily by setting {primary}. "
+            f"Alias env vars also supported: {aliases}."
+        )
+
+    def is_configured(self) -> bool:
+        return bool(self.api_key)
+
+    def search(
+        self,
+        query: str,
+        search_type: str,
+        timelimit: str | None = None,
+        region: str = "wt-wt",
+        max_results: int = 15,
+        **kwargs,
+    ) -> list[ProviderResult]:
+        if search_type not in ("text", "news"):
+            return []
+        if not self.api_key:
+            raise ProviderConfigurationError(self.configuration_hint())
+
+        try:
+            from tavily import TavilyClient
+        except ImportError as exc:
+            raise ProviderConfigurationError(
+                "Install Tavily support with: pip install zero-api-key-web-search[tavily]"
+            ) from exc
+
+        time_range_map = {
+            "d": "day",
+            "w": "week",
+            "m": "month",
+            "y": "year",
+        }
+        payload = {
+            "query": query,
+            "search_depth": kwargs.get("search_depth", "advanced"),
+            "topic": "news" if search_type == "news" else "general",
+            "max_results": min(max_results, 20),
+        }
+        time_range = time_range_map.get(timelimit or "")
+        if time_range:
+            payload["time_range"] = time_range
+
+        client = TavilyClient(api_key=self.api_key)
+        response = client.search(**payload)
+        api_results = response.get("results", []) if isinstance(response, dict) else []
+
+        results: list[ProviderResult] = []
+        for item in api_results[:max_results]:
+            if not isinstance(item, dict):
+                continue
+            url = item.get("url", "")
+            if not isinstance(url, str) or not url:
+                continue
+            metadata: dict = {"provider": "tavily"}
+            if "score" in item:
+                metadata["score"] = item.get("score")
+            results.append(
+                ProviderResult(
+                    url=url,
+                    title=str(item.get("title", "")),
+                    snippet=str(item.get("content", item.get("snippet", ""))),
+                    date=str(item.get("published_date", item.get("publishedDate", ""))),
+                    metadata=metadata,
+                )
+            )
+        return results
+
+    async def asearch(
+        self,
+        query: str,
+        search_type: str,
+        timelimit: str | None = None,
+        region: str = "wt-wt",
+        max_results: int = 15,
+        **kwargs,
+    ) -> list[ProviderResult]:
+        import asyncio
+
+        return await asyncio.to_thread(
+            self.search, query, search_type, timelimit, region, max_results, **kwargs
+        )