Merge pull request #90572 from NousResearch/feat/keyless-tavily-firecrawl-failover

feat: keyless web tier is now a 5-vendor free rotation (Exa/Parallel/Tavily/Firecrawl/Keenable) with ring failover + honest doctor readiness

agent/web_search_registry.py

@@ -168,36 +168,40 @@ _LEGACY_PREFERENCE = (
 
 # Keyless free-tier walk — strictly LAST-resort, tried only after the
 # availability-filtered legacy walk finds nothing (i.e. the user has zero
-# web credentials and no importable ddgs). These providers expose public
-# anonymous MCP endpoints (see plugins/web/keyless_mcp.py). Like opencode,
-# unpinned keyless traffic is split 50/50 between Exa and Parallel per
-# process (see _keyless_preference()); an explicit `hermes tools` pick
-# (web.backend / web.<capability>_backend) bypasses this walk entirely.
+# web credentials and no importable ddgs). All five vendors expose public
+# anonymous free tiers (see plugins/web/keyless_mcp.py). Unpinned keyless
+# traffic round-robins across the ring per request (the ring cursor lives
+# in keyless_mcp; an explicit `hermes tools` pick bypasses this walk
+# entirely, and rate-limited requests fail over to the next ring vendor).
 # Disable the tier with ``web.keyless_fallback: false``.
 _KEYLESS_PREFERENCE = (
     "exa",
     "parallel",
+    "tavily",
+    "firecrawl",
+    "keenable",
 )
 
 
 def _keyless_preference() -> tuple:
-    """Return the keyless walk order, split 50/50 per process.
+    """Return the keyless walk order for resolution.
 
-    Mirrors opencode's session-checksum A/B split between Exa and
-    Parallel: the per-process random session id (also used as Parallel's
-    free-tier rate-limit token) picks which vendor goes first, so keyless
-    load spreads evenly across both free tiers fleet-wide while staying
-    stable within one process. The runner-up stays in the walk as a
-    fallback if the first isn't registered. Explicit user selection never
-    reaches this function — configured names resolve in step 1.
+    Delegates the entry-vendor choice to the ring cursor in
+    :mod:`plugins.web.keyless_mcp` (round-robin per request, seeded by the
+    per-process random session id) so resolution and dispatch agree on
+    which vendor a fresh install starts at. The remaining vendors follow
+    in ring order as fallbacks for registration gaps.
     """
     try:
-        from plugins.web.keyless_mcp import _SESSION_ID
+        from plugins.web.keyless_mcp import _KEYLESS_RING, _ring_cursor
 
-        if int(_SESSION_ID, 16) % 2:
-            return ("parallel", "exa")
-    except Exception as exc:  # noqa: BLE001 — split is best-effort
-        logger.debug("keyless 50/50 split unavailable: %s", exc)
+        start = _ring_cursor % len(_KEYLESS_RING)
+        return tuple(
+            _KEYLESS_RING[(start + i) % len(_KEYLESS_RING)]
+            for i in range(len(_KEYLESS_RING))
+        )
+    except Exception as exc:  # noqa: BLE001 — ring optional in stripped envs
+        logger.debug("keyless ring order unavailable: %s", exc)
     return _KEYLESS_PREFERENCE
 
 

contributors/emails/lakshya.agarwal@tavily.com

@@ -0,0 +1 @@
+lakshyaag-tavily

hermes_cli/config_defaults.py

@@ -492,17 +492,20 @@ DEFAULT_CONFIG = {
         "search_backend": "",    # per-capability override for web_search (e.g. "searxng")
         "extract_backend": "",   # per-capability override for web_extract (e.g. "native")
         "extract_char_limit": 15000,  # per-page char budget for web_extract; larger pages truncate + store full text in cache/web
-        # Keyless free-tier fallback: with NO web backend configured or keyed,
-        # web_search/web_extract fall back to Parallel's / Exa's public
-        # anonymous MCP endpoints (rate-limited free tiers). Never pre-empts
-        # a configured or keyed backend. Set false to disable entirely.
+        # Keyless free-tier ring: with NO web backend configured or keyed,
+        # web_search/web_extract rotate round-robin across five vendors'
+        # public free tiers (exa, parallel, tavily, firecrawl, keenable),
+        # failing over to the next ring vendor on rate limits. Never
+        # pre-empts a configured or keyed backend. Set false to disable.
         "keyless_fallback": True,
-        # Per-provider tier selection for providers with both a keyless free
-        # endpoint and a keyed paid SDK path (exa, parallel). Set by the
-        # `hermes tools` picker's "Free (keyless)" / "Paid (API key)" rows.
+        # Per-provider tier selection for ring vendors with both a keyless
+        # free endpoint and a keyed paid path (exa, parallel, tavily,
+        # firecrawl, keenable). Set by the `hermes tools` picker's
+        # "Free (keyless)" / "Paid (API key)" rows.
         #   free  — always use the anonymous free endpoint (even with a key)
-        #   paid  — always use the keyed SDK path (missing key = error)
-        #   unset — auto: keyed when the API key is present, else keyless
+        #   paid  — always use the keyed path (missing key = error; vendor
+        #           is also excluded from the keyless ring)
+        #   unset — auto: keyed when the API key is present, else the ring
         "provider_tier": {},
     },
 
@@ -4117,13 +4120,21 @@ OPTIONAL_ENV_VARS = {
         "advanced": True,
     },
     "TAVILY_API_KEY": {
-        "description": "Tavily API key for AI-native web search and extract",
+        "description": "Tavily API key for AI-native web search and extract (optional — keyless works without it)",
         "prompt": "Tavily API key",
         "url": "https://app.tavily.com/home",
         "tools": ["web_search", "web_extract"],
         "password": True,
         "category": "tool",
     },
+    "KEENABLE_API_KEY": {
+        "description": "Keenable API key for fast independent-index web search and page fetch (optional — keyless free tier works without it)",
+        "prompt": "Keenable API key",
+        "url": "https://keenable.ai",
+        "tools": ["web_search", "web_extract"],
+        "password": True,
+        "category": "tool",
+    },
     "SEARXNG_URL": {
         "description": "URL of your SearXNG instance for free self-hosted web search",
         "prompt": "SearXNG URL (e.g. http://localhost:8080)",

hermes_cli/doctor.py

@@ -295,6 +295,60 @@ def _doctor_tool_availability_detail(toolset: str) -> str:
     return ""
 
 
+def _doctor_web_capability_rows() -> list[tuple[str, str, str]]:
+    """Return doctor rows for web search/extract provider readiness (#78412).
+
+    Each row is ``(status, label, detail)`` where *status* is ``ok`` or ``warn``.
+    Uses the same active-provider resolvers as the tools, but reports readiness
+    from ``is_available()`` so an explicitly selected but unconfigured backend
+    does not look healthy.
+    """
+    rows: list[tuple[str, str, str]] = []
+    try:
+        from agent.web_search_registry import (
+            get_active_extract_provider,
+            get_active_search_provider,
+        )
+        from tools.web_tools import _ensure_web_plugins_loaded, _provider_is_ready
+
+        # Doctor runs in a fresh process — bundled web providers register
+        # during plugin discovery, which nothing has triggered yet here.
+        # Without this the registry is empty and every row reads
+        # "no provider selected or registered" (idempotent, cheap on rerun).
+        _ensure_web_plugins_loaded()
+    except Exception:
+        return rows
+
+    for capability, getter in (
+        ("web search", get_active_search_provider),
+        ("web extract", get_active_extract_provider),
+    ):
+        try:
+            provider = getter()
+        except Exception:
+            provider = None
+        if provider is None:
+            rows.append(
+                (
+                    "warn",
+                    capability,
+                    "(no provider selected or registered)",
+                )
+            )
+            continue
+        name = getattr(provider, "name", None) or type(provider).__name__
+        if _provider_is_ready(provider):
+            rows.append(("ok", capability, f"({name})"))
+        else:
+            rows.append(
+                (
+                    "warn",
+                    capability,
+                    f"({name} selected; provider not configured)",
+                )
+            )
+    return rows
+
 def _apply_doctor_tool_availability_overrides(available: list[str], unavailable: list[dict]) -> tuple[list[str], list[dict]]:
     """Adjust runtime-gated tool availability for doctor diagnostics."""
     updated_available = list(available)
@@ -2835,11 +2889,26 @@ def run_doctor(args):
         
         available, unavailable = check_tool_availability()
         available, unavailable = _apply_doctor_tool_availability_overrides(available, unavailable)
-        
+
+        # Web is split into search/extract readiness rows so an explicitly
+        # selected but unconfigured backend cannot look healthy (#78412).
+        web_rows = []
+        if "web" in available or any(item.get("name") == "web" for item in unavailable):
+            web_rows = _doctor_web_capability_rows()
+            if web_rows:
+                available = [tid for tid in available if tid != "web"]
+                unavailable = [item for item in unavailable if item.get("name") != "web"]
+
         for tid in available:
             info = TOOLSET_REQUIREMENTS.get(tid, {})
             check_ok(info.get("name", tid), _doctor_tool_availability_detail(tid))
-        
+
+        for status, label, detail in web_rows:
+            if status == "ok":
+                check_ok(label, detail)
+            else:
+                check_warn(label, detail)
+
         for item in unavailable:
             env_vars = item.get("missing_vars") or item.get("env_vars") or []
             if env_vars:
@@ -2852,7 +2921,8 @@ def run_doctor(args):
         # current CLI platform. Default-off or explicitly disabled toolsets may
         # still show warnings above, but should not pollute the final summary.
         api_disabled = _missing_api_key_toolsets_for_summary(unavailable)
-        if api_disabled:
+        web_not_ready = any(status != "ok" for status, _, _ in web_rows)
+        if api_disabled or web_not_ready:
             issues.append("Run 'hermes setup' to configure missing API keys for full tool access")
     except Exception as e:
         check_warn("Could not check tool availability", f"({e})")

hermes_cli/nous_subscription.py

@@ -446,6 +446,7 @@ def get_nous_subscription_features(
     # Per-capability overrides: if set, they determine which backend is active for
     # search/extract independently of web.backend.
     web_search_backend = str(web_cfg.get("search_backend") or "").strip().lower()
+    web_extract_backend = str(web_cfg.get("extract_backend") or "").strip().lower()
     tts_provider = str(tts_cfg.get("provider") or "edge").strip().lower()
     # STT default is "local" (faster-whisper) per DEFAULT_CONFIG, which
     # requires `pip install faster-whisper`. For Nous subscribers we'd
@@ -505,6 +506,9 @@ def get_nous_subscription_features(
     direct_firecrawl = bool(get_env_value("FIRECRAWL_API_KEY") or get_env_value("FIRECRAWL_API_URL"))
     direct_parallel = bool(get_env_value("PARALLEL_API_KEY"))
     direct_tavily = bool(get_env_value("TAVILY_API_KEY"))
+    # Keyless Tavily is opt-in: selecting it in `hermes tools` / setup writes
+    # web.backend (or a per-capability override) without requiring a key.
+    tavily_selected = "tavily" in {web_backend, web_search_backend, web_extract_backend}
     direct_searxng = bool(get_env_value("SEARXNG_URL"))
     direct_fal = fal_key_is_configured()
     direct_fal_video = direct_fal  # same FAL_KEY; separate var so use_gateway is independent
@@ -537,6 +541,7 @@ def get_nous_subscription_features(
         direct_exa = False
         direct_parallel = False
         direct_tavily = False
+        tavily_selected = False
     if image_use_gateway:
         direct_fal = False
     if video_use_gateway:
@@ -624,6 +629,8 @@ def get_nous_subscription_features(
         # different browser choice wins over the env var.
         direct_camofox = False
 
+
+    tavily_ready = direct_tavily or tavily_selected
     web_managed = web_backend == "firecrawl" and managed_web_available and not direct_firecrawl
     web_active = bool(
         web_tool_enabled
@@ -632,7 +639,7 @@ def get_nous_subscription_features(
             or (web_backend == "exa" and direct_exa)
             or (web_backend == "firecrawl" and direct_firecrawl)
             or (web_backend == "parallel" and direct_parallel)
-            or (web_backend == "tavily" and direct_tavily)
+            or (web_backend == "tavily" and tavily_ready)
             or (web_backend == "searxng" and direct_searxng)
             # Per-capability overrides: search_backend or extract_backend may be set
             # without web.backend (using the new split config from #20061)
@@ -640,11 +647,17 @@ def get_nous_subscription_features(
             or (web_search_backend == "exa" and direct_exa)
             or (web_search_backend == "firecrawl" and direct_firecrawl)
             or (web_search_backend == "parallel" and direct_parallel)
-            or (web_search_backend == "tavily" and direct_tavily)
+            or (web_search_backend == "tavily" and tavily_ready)
+            or (web_extract_backend == "tavily" and tavily_ready)
         )
     )
     web_available = bool(
-        managed_web_available or direct_exa or direct_firecrawl or direct_parallel or direct_tavily or direct_searxng
+        managed_web_available
+        or direct_exa
+        or direct_firecrawl
+        or direct_parallel
+        or tavily_ready
+        or direct_searxng
     )
 
     image_managed = image_tool_enabled and managed_image_available and not direct_fal
@@ -754,8 +767,8 @@ def get_nous_subscription_features(
             managed_by_nous=web_managed,
             direct_override=web_active and not web_managed,
             toolset_enabled=web_tool_enabled,
-            current_provider=web_backend or web_search_backend or "",
-            explicit_configured=bool(web_backend or web_search_backend),
+            current_provider=web_backend or web_search_backend or web_extract_backend or "",
+            explicit_configured=bool(web_backend or web_search_backend or web_extract_backend),
         ),
         "image_gen": NousFeatureState(
             key="image_gen",

plugins/web/exa/provider.py

@@ -143,14 +143,14 @@ class ExaWebSearchProvider(WebSearchProvider):
 
             from agent.web_search_provider import get_provider_env
 
-            from plugins.web.keyless_mcp import exa_search_keyless, use_keyless
+            from plugins.web.keyless_mcp import search_with_failover, use_keyless
 
             if use_keyless("exa", get_provider_env("EXA_API_KEY")):
                 # Keyless free tier — public MCP endpoint, no SDK needed.
                 logger.info(
                     "Exa keyless search: '%s' (limit=%d)", query, limit
                 )
-                return exa_search_keyless(query, limit)
+                return search_with_failover("exa", query, limit)
 
             logger.info("Exa search: '%s' (limit=%d)", query, limit)
             response = _get_exa_client().search(
@@ -198,12 +198,12 @@ class ExaWebSearchProvider(WebSearchProvider):
 
             from agent.web_search_provider import get_provider_env
 
-            from plugins.web.keyless_mcp import exa_extract_keyless, use_keyless
+            from plugins.web.keyless_mcp import extract_with_failover, use_keyless
 
             if use_keyless("exa", get_provider_env("EXA_API_KEY")):
                 # Keyless free tier — public MCP endpoint, no SDK needed.
                 logger.info("Exa keyless extract: %d URL(s)", len(urls))
-                return exa_extract_keyless(list(urls))
+                return extract_with_failover("exa", list(urls))
 
             logger.info("Exa extract: %d URL(s)", len(urls))
             response = _get_exa_client().get_contents(urls, text=True)

plugins/web/firecrawl/plugin.yaml

@@ -1,6 +1,6 @@
 name: web-firecrawl
 version: 1.0.0
-description: "Firecrawl web search + content extraction. Supports direct API and Nous-hosted tool-gateway routing for subscribers. Requires FIRECRAWL_API_KEY (or FIRECRAWL_API_URL for self-hosted), or an active Nous subscription with FIRECRAWL_GATEWAY_URL."
+description: "Firecrawl web search + content extraction. Supports keyless cloud, direct API, and Nous-hosted tool-gateway routing for subscribers."
 author: NousResearch
 kind: backend
 provides_web_providers:

plugins/web/firecrawl/provider.py

@@ -50,12 +50,16 @@ import logging
 import os
 from typing import Any, Dict, List, NoReturn, Optional, TYPE_CHECKING
 
+import httpx
+
 from agent.web_search_provider import WebSearchProvider
 from tools.url_safety import is_safe_url
 from tools.website_policy import check_website_access
 
 logger = logging.getLogger(__name__)
 
+_FIRECRAWL_CLOUD_API_URL = "https://api.firecrawl.dev"
+
 
 # ---------------------------------------------------------------------------
 # Lazy Firecrawl SDK proxy
@@ -121,13 +125,26 @@ Firecrawl = _FirecrawlProxy()
 
 
 def _get_direct_firecrawl_config() -> Optional[tuple]:
-    """Return explicit direct Firecrawl kwargs + cache key, or None when unset."""
+    """Return direct Firecrawl (mode, kwargs, cache key), or None when unavailable.
+
+    ``mode`` is ``"sdk"`` (keyed / self-hosted via the Firecrawl SDK) or
+    ``"keyless"`` (explicit Firecrawl selection with no credentials — served
+    by :class:`_KeylessFirecrawlClient` against the public cloud API, which
+    accepts anonymous rate-limited requests). Keyless requires the explicit
+    selection so an unconfigured install never silently routes to it.
+    """
     from hermes_cli.config import get_env_value
 
     api_key = (get_env_value("FIRECRAWL_API_KEY") or "").strip()
     api_url = (get_env_value("FIRECRAWL_API_URL") or "").strip().rstrip("/")
 
     if not api_key and not api_url:
+        if _is_explicit_firecrawl_selection():
+            return (
+                "keyless",
+                {"api_url": _FIRECRAWL_CLOUD_API_URL},
+                ("direct-keyless", _FIRECRAWL_CLOUD_API_URL, None),
+            )
         return None
 
     kwargs: Dict[str, str] = {}
@@ -136,7 +153,78 @@ def _get_direct_firecrawl_config() -> Optional[tuple]:
     if api_url:
         kwargs["api_url"] = api_url
 
-    return kwargs, ("direct", api_url or None, api_key or None)
+    return "sdk", kwargs, ("direct", api_url or None, api_key or None)
+
+
+def _is_explicit_firecrawl_selection() -> bool:
+    """Return True when config explicitly selects Firecrawl for web tools."""
+    import tools.web_tools as _wt
+
+    cfg = _wt._load_web_config()
+    return any(
+        (cfg.get(key) or "").lower().strip() == "firecrawl"
+        for key in ("backend", "search_backend", "extract_backend")
+    )
+
+
+def _use_keyless_ring() -> bool:
+    """True when Firecrawl calls should route via the keyless ring.
+
+    Ring dispatch applies when there are no direct credentials, the
+    managed Nous gateway isn't the selected path, and the keyless tier
+    isn't disabled or pinned paid. Keyed/self-hosted/gateway setups never
+    reach the ring.
+    """
+    from hermes_cli.config import get_env_value
+
+    if (get_env_value("FIRECRAWL_API_KEY") or "").strip():
+        return False
+    if (get_env_value("FIRECRAWL_API_URL") or "").strip():
+        return False
+    import tools.web_tools as _wt
+    from tools.tool_backend_helpers import NOUS_MANAGED_PROVIDER, read_selection
+
+    try:
+        if read_selection("web") == NOUS_MANAGED_PROVIDER:
+            return False
+    except Exception:  # noqa: BLE001 — selection helpers optional
+        pass
+    try:
+        if _wt._is_tool_gateway_ready() and not _is_explicit_firecrawl_selection():
+            return False
+    except Exception:  # noqa: BLE001 — probe optional
+        pass
+    from plugins.web.keyless_mcp import use_keyless
+
+    return use_keyless("firecrawl", "")
+
+
+class _KeylessFirecrawlClient:
+    """Minimal REST client for Firecrawl's keyless cloud mode.
+
+    Duck-types the two SDK methods the provider calls (``search`` /
+    ``scrape``) so the rest of the pipeline (result normalizers, caching)
+    is unchanged. No Authorization header is ever sent.
+    """
+
+    def __init__(self, api_url: str = _FIRECRAWL_CLOUD_API_URL):
+        self.api_url = api_url.rstrip("/")
+
+    def _post(self, path: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+        response = httpx.post(
+            f"{self.api_url}{path}",
+            json=payload,
+            headers={"Content-Type": "application/json"},
+            timeout=60.0,
+        )
+        response.raise_for_status()
+        return response.json()
+
+    def search(self, *, query: str, limit: int = 5) -> Dict[str, Any]:
+        return self._post("/v2/search", {"query": query, "limit": limit})
+
+    def scrape(self, *, url: str, formats: List[str]) -> Dict[str, Any]:
+        return self._post("/v2/scrape", {"url": url, "formats": formats})
 
 
 def _get_firecrawl_gateway_url() -> str:
@@ -286,9 +374,11 @@ def _get_firecrawl_client() -> Any:
                 "unreachable)",
             ))
         kwargs, client_config = managed
+        client_mode = "sdk"
     elif selected is not None or selection_exists("web"):
         # Stored vendor selection (or per-capability web keys routing to
-        # firecrawl): direct Firecrawl only.
+        # firecrawl): direct Firecrawl only. With no credentials, the
+        # explicit selection unlocks keyless cloud mode instead of erroring.
         if direct_config is None:
             logger.error(
                 "Firecrawl client initialization failed: direct Firecrawl "
@@ -299,9 +389,9 @@ def _get_firecrawl_client() -> Any:
                 selected or "firecrawl",
                 "neither FIRECRAWL_API_KEY nor FIRECRAWL_API_URL is set",
             ))
-        kwargs, client_config = direct_config
+        client_mode, kwargs, client_config = direct_config
     elif direct_config is not None:
-        kwargs, client_config = direct_config
+        client_mode, kwargs, client_config = direct_config
     else:
         # Never-configured web section: legacy managed fallback.
         managed = _managed_kwargs()
@@ -312,6 +402,7 @@ def _get_firecrawl_client() -> Any:
             )
             _raise_web_backend_configuration_error()
         kwargs, client_config = managed
+        client_mode = "sdk"
 
     cached = getattr(_wt, "_firecrawl_client", None)
     cached_config = getattr(_wt, "_firecrawl_client_config", None)
@@ -320,7 +411,10 @@ def _get_firecrawl_client() -> Any:
 
     # Construct via the re-exported Firecrawl proxy on tools.web_tools so
     # unit tests patching ``tools.web_tools.Firecrawl`` see their mock.
-    _wt._firecrawl_client = _wt.Firecrawl(**kwargs)
+    if client_mode == "keyless":
+        _wt._firecrawl_client = _KeylessFirecrawlClient(api_url=kwargs["api_url"])
+    else:
+        _wt._firecrawl_client = _wt.Firecrawl(**kwargs)
     _wt._firecrawl_client_config = client_config
     return _wt._firecrawl_client
 
@@ -442,6 +536,17 @@ class FirecrawlWebSearchProvider(WebSearchProvider):
         """Return True when direct Firecrawl OR managed-gateway path is configured."""
         return check_firecrawl_api_key()
 
+    def is_keyless_available(self) -> bool:
+        """Firecrawl serves keyless cloud requests (public API, no auth).
+
+        Default-on ring member of the keyless free tier: fresh installs
+        rotate across Exa/Parallel/Tavily/Firecrawl/Keenable. False when
+        the user pinned ``web.provider_tier.firecrawl: paid``.
+        """
+        from plugins.web.keyless_mcp import keyless_enabled, provider_tier
+
+        return keyless_enabled() and provider_tier("firecrawl") != "paid"
+
     def supports_search(self) -> bool:
         return True
 
@@ -467,6 +572,16 @@ class FirecrawlWebSearchProvider(WebSearchProvider):
         if is_interrupted():
             return {"success": False, "error": "Interrupted"}
 
+        if _use_keyless_ring():
+            # No credentials and no managed gateway: ring dispatch with
+            # next-in-line failover on rate limits (default-on free tier).
+            from plugins.web.keyless_mcp import search_with_failover
+
+            logger.info(
+                "Firecrawl keyless search: '%s' (limit=%d)", query, limit
+            )
+            return search_with_failover("firecrawl", query, limit)
+
         logger.info("Firecrawl search: '%s' (limit=%d)", query, limit)
         # _get_firecrawl_client() raises ValueError on unconfigured systems —
         # let it propagate so the dispatcher emits the legacy envelope shape.
@@ -500,6 +615,18 @@ class FirecrawlWebSearchProvider(WebSearchProvider):
         if _is_interrupted():
             return [{"url": u, "error": "Interrupted", "title": ""} for u in urls]
 
+        if _use_keyless_ring():
+            # No credentials and no managed gateway: ring dispatch with
+            # next-in-line failover on rate limits (default-on free tier).
+            import asyncio as _asyncio
+
+            from plugins.web.keyless_mcp import extract_with_failover
+
+            logger.info("Firecrawl keyless extract: %d URL(s)", len(urls))
+            return await _asyncio.to_thread(
+                extract_with_failover, "firecrawl", list(urls)
+            )
+
         format = kwargs.get("format")
         formats: List[str] = []
         if format == "markdown":
@@ -662,15 +789,15 @@ class FirecrawlWebSearchProvider(WebSearchProvider):
     def get_setup_schema(self) -> Dict[str, Any]:
         return {
             "name": "Firecrawl",
-            "badge": "paid · optional gateway",
+            "badge": "keyless/paid · optional gateway",
             "tag": (
-                "Full search + extract; supports direct API and "
-                "Nous tool-gateway routing."
+                "Full search + extract; supports keyless cloud, direct API, "
+                "and Nous tool-gateway routing."
             ),
             "env_vars": [
                 {
                     "key": "FIRECRAWL_API_KEY",
-                    "prompt": "Firecrawl API key (or leave blank for self-hosted)",
+                    "prompt": "Firecrawl API key (optional; blank = keyless cloud or self-hosted)",
                     "url": "https://docs.firecrawl.dev/introduction",
                 },
             ],

plugins/web/keenable/__init__.py

@@ -0,0 +1,13 @@
+"""Keenable web search + extract plugin — bundled, auto-loaded.
+
+Keyless-ring member (keyed via KEENABLE_API_KEY for higher limits).
+"""
+
+from __future__ import annotations
+
+from plugins.web.keenable.provider import KeenableWebSearchProvider
+
+
+def register(ctx) -> None:
+    """Register the Keenable provider with the plugin context."""
+    ctx.register_web_search_provider(KeenableWebSearchProvider())

plugins/web/keenable/plugin.yaml

@@ -0,0 +1,7 @@
+name: web-keenable
+version: 1.0.0
+description: "Keenable web search + page fetch (independent web index for AI apps). Works keyless on Keenable's free tier as part of the default rotation; set KEENABLE_API_KEY for higher limits — https://keenable.ai."
+author: NousResearch
+kind: backend
+provides_web_providers:
+  - keenable

plugins/web/keenable/provider.py

@@ -0,0 +1,233 @@
+"""Keenable web search + content extraction — bundled plugin.
+
+Keenable (https://keenable.ai) operates an independent web index for AI
+apps with public keyless endpoints (rate-limited free tier; keyed access
+via KEENABLE_API_KEY for higher limits). Integrated as a keyless-ring
+member following the Exa/Parallel/Tavily/Firecrawl pattern: fresh installs
+with zero web credentials rotate across all five vendors' free tiers.
+
+Credit: Keenable integration originally proposed by Ilya Gusev (Keenable)
+in PR #49758; the native provider form follows the salvage of that work
+plus the keyless-ring design.
+
+Config keys this provider responds to::
+
+    web:
+      search_backend: "keenable"      # explicit per-capability
+      extract_backend: "keenable"     # explicit per-capability
+      backend: "keenable"             # shared fallback
+      provider_tier:
+        keenable: free|paid           # pin the tier (unset = auto)
+
+Env var::
+
+    KEENABLE_API_KEY=...   # optional — keyless free tier works without it
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List
+
+from agent.web_search_provider import WebSearchProvider
+
+logger = logging.getLogger(__name__)
+
+_KEENABLE_API_URL = "https://api.keenable.ai"
+
+
+def _keenable_headers(api_key: str) -> Dict[str, str]:
+    """Build Keenable request headers for keyed or keyless access.
+
+    Their keyless tier structurally requires an app-identifier header
+    (X-Keenable-Title); no user identifiers are sent.
+    """
+    headers = {"X-Keenable-Title": "hermes-agent"}
+    if api_key:
+        headers["Authorization"] = f"Bearer {api_key}"
+    return headers
+
+
+class KeenableWebSearchProvider(WebSearchProvider):
+    """Keenable search + extract provider (keyed or keyless)."""
+
+    @property
+    def name(self) -> str:
+        return "keenable"
+
+    @property
+    def display_name(self) -> str:
+        return "Keenable"
+
+    def is_available(self) -> bool:
+        """Return True when ``KEENABLE_API_KEY`` is set to a non-empty value."""
+        from agent.web_search_provider import get_provider_env
+
+        return bool(get_provider_env("KEENABLE_API_KEY"))
+
+    def is_keyless_available(self) -> bool:
+        """Keenable serves anonymous free-tier calls via its public endpoints.
+
+        Default-on ring member of the keyless free tier. False when the
+        user pinned ``web.provider_tier.keenable: paid``.
+        """
+        from plugins.web.keyless_mcp import keyless_enabled, provider_tier
+
+        return keyless_enabled() and provider_tier("keenable") != "paid"
+
+    def supports_search(self) -> bool:
+        return True
+
+    def supports_extract(self) -> bool:
+        return True
+
+    def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
+        """Execute a Keenable search (keyed path or keyless ring)."""
+        try:
+            from tools.interrupt import is_interrupted
+
+            if is_interrupted():
+                return {"success": False, "error": "Interrupted"}
+
+            from agent.web_search_provider import get_provider_env
+
+            from plugins.web.keyless_mcp import search_with_failover, use_keyless
+
+            api_key = get_provider_env("KEENABLE_API_KEY")
+            if use_keyless("keenable", api_key):
+                logger.info(
+                    "Keenable keyless search: '%s' (limit=%d)", query, limit
+                )
+                return search_with_failover("keenable", query, limit)
+
+            import requests
+
+            logger.info("Keenable search: '%s' (limit=%d)", query, limit)
+            response = requests.post(
+                f"{_KEENABLE_API_URL}/v1/search",
+                json={"query": query, "max_results": min(max(1, int(limit)), 20)},
+                headers=_keenable_headers(api_key),
+                timeout=30,
+            )
+            if response.status_code >= 400:
+                detail = (response.text or "").strip() or f"HTTP {response.status_code}"
+                return {"success": False, "error": f"Keenable search failed: {detail}"}
+            data = response.json()
+
+            web_results = []
+            for i, result in enumerate(data.get("results") or []):
+                web_results.append(
+                    {
+                        "url": result.get("url") or "",
+                        "title": result.get("title") or "",
+                        "description": result.get("snippet")
+                        or result.get("description")
+                        or "",
+                        "position": i + 1,
+                    }
+                )
+            return {"success": True, "data": {"web": web_results}}
+        except Exception as exc:  # noqa: BLE001 — surface as failure
+            logger.warning("Keenable search error: %s", exc)
+            return {"success": False, "error": f"Keenable search failed: {exc}"}
+
+    def extract(self, urls: List[str], **kwargs: Any) -> List[Dict[str, Any]]:
+        """Extract content via Keenable's fetch endpoint (per-URL).
+
+        Sync — the dispatcher wraps in a thread when the caller is async.
+        Returns the legacy list-of-results shape; per-URL failures become
+        items with an ``error`` field.
+        """
+        try:
+            from tools.interrupt import is_interrupted
+
+            if is_interrupted():
+                return [
+                    {"url": u, "error": "Interrupted", "title": ""} for u in urls
+                ]
+
+            from agent.web_search_provider import get_provider_env
+
+            from plugins.web.keyless_mcp import extract_with_failover, use_keyless
+
+            api_key = get_provider_env("KEENABLE_API_KEY")
+            if use_keyless("keenable", api_key):
+                logger.info("Keenable keyless extract: %d URL(s)", len(urls))
+                return extract_with_failover("keenable", list(urls))
+
+            import requests
+
+            logger.info("Keenable extract: %d URL(s)", len(urls))
+            results: List[Dict[str, Any]] = []
+            for url in urls:
+                try:
+                    response = requests.get(
+                        f"{_KEENABLE_API_URL}/v1/fetch",
+                        params={"url": url},
+                        headers=_keenable_headers(api_key),
+                        timeout=30,
+                    )
+                    if response.status_code >= 400:
+                        raise ValueError(
+                            (response.text or "").strip()
+                            or f"HTTP {response.status_code}"
+                        )
+                    data = response.json()
+                    content = data.get("content") or ""
+                    title = data.get("title") or ""
+                    results.append(
+                        {
+                            "url": data.get("url") or url,
+                            "title": title,
+                            "content": content,
+                            "raw_content": content,
+                            "metadata": {"sourceURL": url, "title": title},
+                        }
+                    )
+                except Exception as exc:  # noqa: BLE001 — per-URL error entry
+                    results.append(
+                        {
+                            "url": url,
+                            "title": "",
+                            "content": "",
+                            "error": f"Keenable extract failed: {exc}",
+                        }
+                    )
+            return results
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("Keenable extract error: %s", exc)
+            return [
+                {"url": u, "title": "", "content": "",
+                 "error": f"Keenable extract failed: {exc}"}
+                for u in urls
+            ]
+
+    def get_setup_schema(self) -> Dict[str, Any]:
+        return {
+            "name": "Keenable · Free (keyless)",
+            "badge": "free · no key",
+            "tag": (
+                "Independent web index for AI apps — fast search + page "
+                "fetch on Keenable's anonymous free tier."
+            ),
+            "env_vars": [],
+            "web_tier": "free",
+            "variants": [
+                {
+                    "name": "Keenable · Paid (API key)",
+                    "badge": "paid",
+                    "tag": (
+                        "Independent web index for AI apps. Keyed access "
+                        "with higher limits and guaranteed service."
+                    ),
+                    "env_vars": [
+                        {
+                            "key": "KEENABLE_API_KEY",
+                            "prompt": "Keenable API key",
+                            "url": "https://keenable.ai",
+                        },
+                    ],
+                    "web_tier": "paid",
+                },
+            ],
+        }

plugins/web/keyless_mcp.py

@@ -46,6 +46,23 @@ class KeylessMCPError(RuntimeError):
     """A keyless MCP call failed (transport, rate limit, or tool error)."""
 
 
+_RATE_LIMIT_MARKERS = (
+    "rate limit",
+    "rate-limit",
+    "ratelimit",
+    "too many requests",
+    "429",
+    "quota exceeded",
+    "slow down",
+)
+
+
+def _is_rate_limitish(message: str) -> bool:
+    """Heuristic: does an error message look like free-tier throttling?"""
+    lowered = (message or "").lower()
+    return any(marker in lowered for marker in _RATE_LIMIT_MARKERS)
+
+
 def keyless_enabled() -> bool:
     """Return True when the keyless fallback tier is enabled.
 
@@ -418,3 +435,437 @@ def exa_extract_keyless(urls: List[str]) -> List[Dict[str, Any]]:
             }
         )
     return results
+
+
+
+# ---------------------------------------------------------------------------
+# Tavily keyless (api.tavily.com — X-Tavily-Access-Mode: keyless)
+# ---------------------------------------------------------------------------
+
+
+TAVILY_API_URL = "https://api.tavily.com"
+
+
+def _tavily_keyless_post(endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+    """POST to Tavily with keyless headers; raise KeylessMCPError on failure."""
+    import requests
+
+    try:
+        response = requests.post(
+            f"{TAVILY_API_URL}/{endpoint.lstrip('/')}",
+            json=payload,
+            headers={
+                "Content-Type": "application/json",
+                "X-Client-Name": "hermes-agent",
+                "X-Tavily-Access-Mode": "keyless",
+            },
+            timeout=_TIMEOUT_SECONDS,
+        )
+    except requests.RequestException as exc:
+        raise KeylessMCPError(f"request failed: {exc}") from exc
+    if response.status_code >= 400:
+        raise KeylessMCPError(
+            (response.text or "").strip() or f"HTTP {response.status_code}"
+        )
+    return response.json()
+
+
+def tavily_search_keyless(query: str, limit: int = 5) -> Dict[str, Any]:
+    """Keyless Tavily search → legacy search response shape."""
+    try:
+        data = _tavily_keyless_post(
+            "search", {"query": query, "max_results": max(1, int(limit))}
+        )
+    except KeylessMCPError as exc:
+        return {
+            "success": False,
+            "error": (
+                f"Keyless Tavily search failed: {exc}. "
+                "Set TAVILY_API_KEY (https://app.tavily.com) or another web "
+                "backend via `hermes tools` for reliable service."
+            ),
+        }
+    web_results = []
+    for i, result in enumerate(data.get("results") or []):
+        web_results.append(
+            {
+                "url": result.get("url") or "",
+                "title": result.get("title") or "",
+                "description": result.get("content") or "",
+                "position": i + 1,
+            }
+        )
+    return {"success": True, "data": {"web": web_results}}
+
+
+def tavily_extract_keyless(urls: List[str]) -> List[Dict[str, Any]]:
+    """Keyless Tavily extract → legacy extract result list."""
+    try:
+        data = _tavily_keyless_post("extract", {"urls": list(urls)})
+    except KeylessMCPError as exc:
+        message = (
+            f"Keyless Tavily extract failed: {exc}. "
+            "Set TAVILY_API_KEY (https://app.tavily.com) or another web "
+            "backend via `hermes tools` for reliable service."
+        )
+        return [
+            {"url": u, "title": "", "content": "", "error": message}
+            for u in urls
+        ]
+    results: List[Dict[str, Any]] = []
+    seen = set()
+    for result in data.get("results") or []:
+        url = result.get("url") or ""
+        raw = result.get("raw_content") or result.get("content") or ""
+        seen.add(url)
+        results.append(
+            {
+                "url": url,
+                "title": result.get("title") or "",
+                "content": raw,
+                "raw_content": raw,
+                "metadata": {"sourceURL": url, "title": result.get("title") or ""},
+            }
+        )
+    for fail in data.get("failed_results") or []:
+        url = (fail.get("url") if isinstance(fail, dict) else str(fail)) or ""
+        seen.add(url)
+        results.append(
+            {
+                "url": url,
+                "title": "",
+                "content": "",
+                "error": (fail.get("error") if isinstance(fail, dict) else None)
+                or "extraction failed",
+            }
+        )
+    for u in urls:
+        if u not in seen:
+            results.append(
+                {"url": u, "title": "", "content": "", "error": "no content returned"}
+            )
+    return results
+
+
+# ---------------------------------------------------------------------------
+# Firecrawl keyless (public cloud API, no auth header)
+# ---------------------------------------------------------------------------
+
+
+def firecrawl_search_keyless(query: str, limit: int = 5) -> Dict[str, Any]:
+    """Keyless Firecrawl cloud search → legacy search response shape."""
+    from plugins.web.firecrawl.provider import (
+        _KeylessFirecrawlClient,
+        _extract_web_search_results,
+    )
+
+    try:
+        response = _KeylessFirecrawlClient().search(query=query, limit=limit)
+        return {"success": True, "data": {"web": _extract_web_search_results(response)}}
+    except Exception as exc:  # noqa: BLE001 — normalized below
+        return {
+            "success": False,
+            "error": (
+                f"Keyless Firecrawl search failed: {exc}. "
+                "Set FIRECRAWL_API_KEY (https://firecrawl.dev) or another web "
+                "backend via `hermes tools` for reliable service."
+            ),
+        }
+
+
+def firecrawl_extract_keyless(urls: List[str]) -> List[Dict[str, Any]]:
+    """Keyless Firecrawl cloud scrape → legacy extract result list."""
+    from plugins.web.firecrawl.provider import (
+        _KeylessFirecrawlClient,
+        _extract_scrape_payload,
+    )
+
+    client = _KeylessFirecrawlClient()
+    results: List[Dict[str, Any]] = []
+    for url in urls:
+        try:
+            response = client.scrape(url=url, formats=["markdown"])
+            payload = _extract_scrape_payload(response) or {}
+            metadata = payload.get("metadata") or {}
+            if not isinstance(metadata, dict):
+                metadata = {}
+            content = payload.get("markdown") or payload.get("html") or ""
+            title = metadata.get("title") or ""
+            results.append(
+                {
+                    "url": url,
+                    "title": title,
+                    "content": content,
+                    "raw_content": content,
+                    "metadata": {"sourceURL": url, "title": title},
+                }
+            )
+        except Exception as exc:  # noqa: BLE001 — per-URL error entry
+            results.append(
+                {
+                    "url": url,
+                    "title": "",
+                    "content": "",
+                    "error": (
+                        f"Keyless Firecrawl extract failed: {exc}. "
+                        "Set FIRECRAWL_API_KEY (https://firecrawl.dev) for "
+                        "reliable service."
+                    ),
+                }
+            )
+    return results
+
+
+# ---------------------------------------------------------------------------
+# Keenable keyless (api.keenable.ai public endpoints)
+# ---------------------------------------------------------------------------
+
+
+KEENABLE_API_URL = "https://api.keenable.ai"
+_KEENABLE_TITLE = "hermes-agent"
+
+
+def keenable_search_keyless(query: str, limit: int = 5) -> Dict[str, Any]:
+    """Keyless Keenable search → legacy search response shape.
+
+    POST /v1/search/public with the mandatory X-Keenable-Title app
+    identifier (their keyless tier requires an app name; no user
+    identifiers are sent). Response: {results: [{title, url, snippet}]}.
+    """
+    import requests
+
+    try:
+        response = requests.post(
+            f"{KEENABLE_API_URL}/v1/search/public",
+            json={"query": query, "max_results": max(1, int(limit))},
+            headers={
+                "Content-Type": "application/json",
+                "X-Keenable-Title": _KEENABLE_TITLE,
+            },
+            timeout=_TIMEOUT_SECONDS,
+        )
+        if response.status_code >= 400:
+            raise KeylessMCPError(
+                (response.text or "").strip() or f"HTTP {response.status_code}"
+            )
+        data = response.json()
+    except KeylessMCPError as exc:
+        return {
+            "success": False,
+            "error": (
+                f"Keyless Keenable search failed: {exc}. "
+                "Set KEENABLE_API_KEY (https://keenable.ai) or another web "
+                "backend via `hermes tools` for reliable service."
+            ),
+        }
+    except Exception as exc:  # noqa: BLE001 — transport/JSON errors
+        return {
+            "success": False,
+            "error": f"Keyless Keenable search failed: {exc}.",
+        }
+    web_results = []
+    for i, result in enumerate(data.get("results") or []):
+        web_results.append(
+            {
+                "url": result.get("url") or "",
+                "title": result.get("title") or "",
+                "description": result.get("snippet")
+                or result.get("description")
+                or "",
+                "position": i + 1,
+            }
+        )
+    return {"success": True, "data": {"web": web_results}}
+
+
+def keenable_extract_keyless(urls: List[str]) -> List[Dict[str, Any]]:
+    """Keyless Keenable page fetch → legacy extract result list.
+
+    GET /v1/fetch/public?url=... returns {url, title, content} (markdown).
+    Called per-URL; failures become per-URL error entries.
+    """
+    import requests
+
+    results: List[Dict[str, Any]] = []
+    for url in urls:
+        try:
+            response = requests.get(
+                f"{KEENABLE_API_URL}/v1/fetch/public",
+                params={"url": url},
+                headers={"X-Keenable-Title": _KEENABLE_TITLE},
+                timeout=_TIMEOUT_SECONDS,
+            )
+            if response.status_code >= 400:
+                raise KeylessMCPError(
+                    (response.text or "").strip() or f"HTTP {response.status_code}"
+                )
+            data = response.json()
+            content = data.get("content") or ""
+            title = data.get("title") or ""
+            results.append(
+                {
+                    "url": data.get("url") or url,
+                    "title": title,
+                    "content": content,
+                    "raw_content": content,
+                    "metadata": {"sourceURL": url, "title": title},
+                }
+            )
+        except Exception as exc:  # noqa: BLE001 — per-URL error entry
+            results.append(
+                {
+                    "url": url,
+                    "title": "",
+                    "content": "",
+                    "error": (
+                        f"Keyless Keenable extract failed: {exc}. "
+                        "Set KEENABLE_API_KEY (https://keenable.ai) for "
+                        "reliable service."
+                    ),
+                }
+            )
+    return results
+
+
+# ---------------------------------------------------------------------------
+# Round-robin ring + next-in-line failover (rate-limited free tiers)
+# ---------------------------------------------------------------------------
+
+_KEYLESS_RING = ("exa", "parallel", "tavily", "firecrawl", "keenable")
+
+_KEYLESS_SEARCHERS = {
+    "exa": lambda query, limit: exa_search_keyless(query, limit),
+    "parallel": lambda query, limit: parallel_search_keyless(query, limit),
+    "tavily": lambda query, limit: tavily_search_keyless(query, limit),
+    "firecrawl": lambda query, limit: firecrawl_search_keyless(query, limit),
+    "keenable": lambda query, limit: keenable_search_keyless(query, limit),
+}
+
+_KEYLESS_EXTRACTORS = {
+    "exa": lambda urls: exa_extract_keyless(urls),
+    "parallel": lambda urls: parallel_extract_keyless(urls),
+    "tavily": lambda urls: tavily_extract_keyless(urls),
+    "firecrawl": lambda urls: firecrawl_extract_keyless(urls),
+    "keenable": lambda urls: keenable_extract_keyless(urls),
+}
+
+# Per-process round-robin cursor, seeded by the random session id so the
+# fleet spreads evenly across all five free tiers; advances once per
+# unpinned keyless request so a single process also rotates.
+_ring_lock = __import__("threading").Lock()
+_ring_cursor = int(_SESSION_ID, 16) % len(_KEYLESS_RING)
+
+
+def _vendor_pinned(name: str) -> bool:
+    """True when config explicitly routes web traffic to *name*.
+
+    A pinned vendor starts every keyless request (rotation off); the ring
+    is only walked past it on throttle. Pin signals: web.backend /
+    web.search_backend / web.extract_backend naming the vendor, or a
+    free-tier pin in web.provider_tier.
+    """
+    if provider_tier(name) == "free":
+        return True
+    try:
+        import tools.web_tools as _wt
+
+        web_cfg = _wt._load_web_config()
+        return any(
+            (web_cfg.get(key) or "").lower().strip() == name
+            for key in ("backend", "search_backend", "extract_backend")
+        )
+    except Exception as exc:  # noqa: BLE001 — config layer optional
+        logger.debug("_vendor_pinned(%r) config read failed: %s", name, exc)
+        return False
+
+
+def _ring_order(name: str) -> List[str]:
+    """Return the vendor walk order for a request entering via *name*.
+
+    Pinned vendor → start at it (its position in the ring determines the
+    failover succession). Unpinned → true round-robin: start at the next
+    cursor position, advancing the cursor per request. Vendors whose tier
+    is pinned ``paid`` are excluded entirely (an explicit paid selection
+    opts that vendor's free endpoint out).
+    """
+    global _ring_cursor
+    if _vendor_pinned(name):
+        start = _KEYLESS_RING.index(name) if name in _KEYLESS_RING else 0
+    else:
+        with _ring_lock:
+            start = _ring_cursor
+            _ring_cursor = (_ring_cursor + 1) % len(_KEYLESS_RING)
+    ordered = [
+        _KEYLESS_RING[(start + i) % len(_KEYLESS_RING)]
+        for i in range(len(_KEYLESS_RING))
+    ]
+    return [v for v in ordered if provider_tier(v) != "paid"]
+
+
+def search_with_failover(name: str, query: str, limit: int = 5) -> Dict[str, Any]:
+    """Keyless search across the vendor ring with next-in-line failover.
+
+    Starts at *name* when the user pinned it, otherwise at the round-robin
+    cursor. Rate-limit-shaped errors advance to the next ring vendor;
+    non-throttle errors stop the walk (a malformed query fails everywhere).
+    The result notes the serving vendor via ``data.served_by`` whenever it
+    differs from *name*.
+    """
+    order = _ring_order(name)
+    if not order:
+        return {
+            "success": False,
+            "error": "All keyless web providers are pinned to paid tiers.",
+        }
+    last: Dict[str, Any] = {}
+    for i, vendor in enumerate(order):
+        result = _KEYLESS_SEARCHERS[vendor](query, limit)
+        if result.get("success"):
+            if vendor != name:
+                result.setdefault("data", {})["served_by"] = vendor
+            return result
+        last = result
+        if not _is_rate_limitish(result.get("error", "")):
+            return result
+        nxt = order[i + 1] if i + 1 < len(order) else None
+        if nxt:
+            logger.info(
+                "keyless %s search throttled; failing over to %s", vendor, nxt
+            )
+    last["error"] = (
+        f"{last.get('error', '')} (all keyless vendors throttled: "
+        f"{', '.join(order)})"
+    )
+    return last
+
+
+def extract_with_failover(name: str, urls: List[str]) -> List[Dict[str, Any]]:
+    """Keyless extract across the vendor ring, failing over per-batch.
+
+    Advances to the next ring vendor only when EVERY url in a batch comes
+    back with a rate-limit-shaped error — partial failures are page
+    problems, not throttling, and return as-is.
+    """
+    order = _ring_order(name)
+    if not order:
+        return [
+            {"url": u, "title": "", "content": "",
+             "error": "All keyless web providers are pinned to paid tiers."}
+            for u in urls
+        ]
+    last: List[Dict[str, Any]] = []
+    for i, vendor in enumerate(order):
+        results = _KEYLESS_EXTRACTORS[vendor](list(urls))
+        errors = [r.get("error", "") for r in results]
+        all_throttled = bool(results) and all(
+            e and _is_rate_limitish(e) for e in errors
+        )
+        if not all_throttled:
+            return results
+        last = results
+        nxt = order[i + 1] if i + 1 < len(order) else None
+        if nxt:
+            logger.info(
+                "keyless %s extract throttled; failing over to %s", vendor, nxt
+            )
+    return last

plugins/web/parallel/provider.py

@@ -198,14 +198,14 @@ class ParallelWebSearchProvider(WebSearchProvider):
 
             from agent.web_search_provider import get_provider_env
 
-            from plugins.web.keyless_mcp import parallel_search_keyless, use_keyless
+            from plugins.web.keyless_mcp import search_with_failover, use_keyless
 
             if use_keyless("parallel", get_provider_env("PARALLEL_API_KEY")):
                 # Keyless free tier — public MCP endpoint, no SDK needed.
                 logger.info(
                     "Parallel keyless search: '%s' (limit=%d)", query, limit
                 )
-                return parallel_search_keyless(query, limit)
+                return search_with_failover("parallel", query, limit)
 
             mode = _resolve_search_mode()
             logger.info(
@@ -262,7 +262,7 @@ class ParallelWebSearchProvider(WebSearchProvider):
 
             from agent.web_search_provider import get_provider_env
 
-            from plugins.web.keyless_mcp import parallel_extract_keyless, use_keyless
+            from plugins.web.keyless_mcp import extract_with_failover, use_keyless
 
             if use_keyless("parallel", get_provider_env("PARALLEL_API_KEY")):
                 # Keyless free tier — blocking HTTP, so hop off the loop.
@@ -270,7 +270,7 @@ class ParallelWebSearchProvider(WebSearchProvider):
 
                 logger.info("Parallel keyless extract: %d URL(s)", len(urls))
                 return await asyncio.to_thread(
-                    parallel_extract_keyless, list(urls)
+                    extract_with_failover, "parallel", list(urls)
                 )
 
             logger.info("Parallel extract: %d URL(s)", len(urls))

plugins/web/tavily/plugin.yaml

@@ -1,6 +1,6 @@
 name: web-tavily
 version: 1.0.0
-description: "Tavily web search + content extraction + crawl. Search + extract are mainstream; crawl is unique to Tavily among built-in providers. Requires TAVILY_API_KEY — sign up at https://app.tavily.com/home."
+description: "Tavily web search + content extraction. Works keyless (rate-limited); set TAVILY_API_KEY for higher limits — https://app.tavily.com/home."
 author: NousResearch
 kind: backend
 provides_web_providers:

plugins/web/tavily/provider.py

@@ -17,47 +17,62 @@ Config keys this provider responds to::
 
 Env vars::
 
-    TAVILY_API_KEY=...           # https://app.tavily.com/home (required)
+    TAVILY_API_KEY=...           # https://app.tavily.com/home (optional)
     TAVILY_BASE_URL=...          # optional override of https://api.tavily.com
+
+Auth is header-based. A key uses ``Authorization: Bearer``; without a key
+the request is keyless (``X-Tavily-Access-Mode: keyless``). Both paths
+send ``X-Client-Name: hermes-agent``.
 """
 
 from __future__ import annotations
 
 import logging
-import os
 from typing import Any, Dict, List
 
+import httpx
+
 from agent.web_search_provider import WebSearchProvider
 
 logger = logging.getLogger(__name__)
 
+_CLIENT_NAME = "hermes-agent"
+
+
+def _tavily_headers(api_key: str) -> Dict[str, str]:
+    """Build Tavily request headers for keyed or keyless access."""
+    headers = {"X-Client-Name": _CLIENT_NAME}
+    if api_key:
+        headers["Authorization"] = f"Bearer {api_key}"
+    else:
+        headers["X-Tavily-Access-Mode"] = "keyless"
+    return headers
+
 
 def _tavily_request(endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
     """POST to the Tavily API and return the parsed JSON response.
 
-    Mirrors :func:`tools.web_tools._tavily_request`. Raises ``ValueError``
-    when ``TAVILY_API_KEY`` is unset; the caller catches and surfaces as
-    a typed error response.
+    Keyed when ``TAVILY_API_KEY`` is set (Bearer auth); otherwise keyless.
+    Non-2xx responses raise ``ValueError`` with the response body so Tavily's
+    keyless rate-limit / upgrade text reaches the model.
     """
-    import httpx
-
     from agent.web_search_provider import get_provider_env
 
     api_key = get_provider_env("TAVILY_API_KEY")
-    if not api_key:
-        raise ValueError(
-            "TAVILY_API_KEY environment variable not set. "
-            "Get your API key at https://app.tavily.com/home"
-        )
-
     base_url = get_provider_env("TAVILY_BASE_URL") or "https://api.tavily.com"
-    payload = dict(payload)  # don't mutate caller's dict
-    payload["api_key"] = api_key
     url = f"{base_url}/{endpoint.lstrip('/')}"
     logger.info("Tavily %s request to %s", endpoint, url)
 
-    response = httpx.post(url, json=payload, timeout=60)
-    response.raise_for_status()
+    response = httpx.post(
+        url,
+        json=payload,
+        timeout=60,
+        headers=_tavily_headers(api_key),
+    )
+    if response.status_code >= 400:
+        body = (response.text or "").strip()
+        detail = body or f"HTTP {response.status_code}"
+        raise ValueError(detail)
     return response.json()
 
 
@@ -144,6 +159,18 @@ class TavilyWebSearchProvider(WebSearchProvider):
 
         return bool(get_provider_env("TAVILY_API_KEY"))
 
+    def is_keyless_available(self) -> bool:
+        """Tavily serves anonymous keyless requests (X-Tavily-Access-Mode).
+
+        Default-on ring member of the keyless free tier: fresh installs
+        rotate across Exa/Parallel/Tavily/Firecrawl/Keenable. False when
+        the user pinned ``web.provider_tier.tavily: paid`` — an explicit
+        paid selection opts the free endpoint out.
+        """
+        from plugins.web.keyless_mcp import keyless_enabled, provider_tier
+
+        return keyless_enabled() and provider_tier("tavily") != "paid"
+
     def supports_search(self) -> bool:
         return True
 
@@ -158,6 +185,18 @@ class TavilyWebSearchProvider(WebSearchProvider):
             if is_interrupted():
                 return {"success": False, "error": "Interrupted"}
 
+            from agent.web_search_provider import get_provider_env
+
+            from plugins.web.keyless_mcp import search_with_failover, use_keyless
+
+            if use_keyless("tavily", get_provider_env("TAVILY_API_KEY")):
+                # Keyless free tier — ring dispatch with next-in-line
+                # failover on rate limits.
+                logger.info(
+                    "Tavily keyless search: '%s' (limit=%d)", query, limit
+                )
+                return search_with_failover("tavily", query, limit)
+
             logger.info("Tavily search: '%s' (limit=%d)", query, limit)
             raw = _tavily_request(
                 "search",
@@ -189,6 +228,16 @@ class TavilyWebSearchProvider(WebSearchProvider):
                     {"url": u, "error": "Interrupted", "title": ""} for u in urls
                 ]
 
+            from agent.web_search_provider import get_provider_env
+
+            from plugins.web.keyless_mcp import extract_with_failover, use_keyless
+
+            if use_keyless("tavily", get_provider_env("TAVILY_API_KEY")):
+                # Keyless free tier — ring dispatch with next-in-line
+                # failover on rate limits.
+                logger.info("Tavily keyless extract: %d URL(s)", len(urls))
+                return extract_with_failover("tavily", list(urls))
+
             logger.info("Tavily extract: %d URL(s)", len(urls))
             raw = _tavily_request(
                 "extract",
@@ -212,12 +261,12 @@ class TavilyWebSearchProvider(WebSearchProvider):
     def get_setup_schema(self) -> Dict[str, Any]:
         return {
             "name": "Tavily",
-            "badge": "paid",
-            "tag": "Search + extract in one provider.",
+            "badge": "free · key optional",
+            "tag": "Search + extract. Works keyless; set TAVILY_API_KEY for higher limits.",
             "env_vars": [
                 {
                     "key": "TAVILY_API_KEY",
-                    "prompt": "Tavily API key",
+                    "prompt": "Tavily API key (optional — keyless works without it)",
                     "url": "https://app.tavily.com/home",
                 },
             ],

tests/hermes_cli/test_doctor.py

@@ -77,6 +77,52 @@ class TestDoctorToolAvailabilitySummary:
 
         assert [item["name"] for item in filtered] == ["web"]
 
+    def test_web_capability_rows_warn_when_selected_provider_not_ready(self, monkeypatch):
+        """#78412: selected firecrawl with is_available=False must warn."""
+        class _Unavailable:
+            name = "firecrawl"
+
+            def is_available(self):
+                return False
+
+        unavailable = _Unavailable()
+        monkeypatch.setattr(
+            "agent.web_search_registry.get_active_search_provider",
+            lambda: unavailable,
+        )
+        monkeypatch.setattr(
+            "agent.web_search_registry.get_active_extract_provider",
+            lambda: unavailable,
+        )
+
+        rows = doctor._doctor_web_capability_rows()
+        assert rows
+        assert all(status == "warn" for status, _, _ in rows)
+        assert any("firecrawl selected; provider not configured" in detail for _, _, detail in rows)
+
+    def test_web_capability_rows_ok_when_provider_ready(self, monkeypatch):
+        class _Ready:
+            name = "ddgs"
+
+            def is_available(self):
+                return True
+
+        ready = _Ready()
+        monkeypatch.setattr(
+            "agent.web_search_registry.get_active_search_provider",
+            lambda: ready,
+        )
+        monkeypatch.setattr(
+            "agent.web_search_registry.get_active_extract_provider",
+            lambda: ready,
+        )
+
+        rows = doctor._doctor_web_capability_rows()
+        assert rows == [
+            ("ok", "web search", "(ddgs)"),
+            ("ok", "web extract", "(ddgs)"),
+        ]
+
 
 class TestDoctorEnvFileEncoding:
     """Regression for #18637 (bug 3): `hermes doctor` crashed on Windows

tests/hermes_cli/test_nous_subscription.py

@@ -58,8 +58,67 @@ def test_get_nous_subscription_features_recognizes_direct_exa_backend(monkeypatc
     assert features.web.current_provider == "exa"
 
 
+def test_get_nous_subscription_features_recognizes_keyless_tavily_backend(monkeypatch):
+    """Selecting Tavily in setup/tools counts as available with no API key.
+
+    Mirrors tools.web_tools._is_backend_available('tavily'): keyless is
+    opt-in via web.backend / search_backend / extract_backend, not a
+    silent empty-install default. The setup summary previously required
+    TAVILY_API_KEY and printed a false 'missing' after a skipped key prompt.
+    """
+    monkeypatch.setattr(ns, "get_env_value", lambda name: "")
+    monkeypatch.setattr(
+        ns, "get_nous_portal_account_info", lambda: _account(logged_in=False)
+    )
+    monkeypatch.setattr(ns, "_toolset_enabled", lambda config, key: key == "web")
+    monkeypatch.setattr(ns, "_has_agent_browser", lambda: False)
+    monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
+    monkeypatch.setattr(ns, "has_direct_modal_credentials", lambda: False)
+
+    features = ns.get_nous_subscription_features({"web": {"backend": "tavily"}})
+
+    assert features.web.available is True
+    assert features.web.active is True
+    assert features.web.managed_by_nous is False
+    assert features.web.direct_override is True
+    assert features.web.current_provider == "tavily"
+    assert features.web.explicit_configured is True
 
 
+def test_keyless_tavily_search_backend_without_shared_backend(monkeypatch):
+    monkeypatch.setattr(ns, "get_env_value", lambda name: "")
+    monkeypatch.setattr(
+        ns, "get_nous_portal_account_info", lambda: _account(logged_in=False)
+    )
+    monkeypatch.setattr(ns, "_toolset_enabled", lambda config, key: key == "web")
+    monkeypatch.setattr(ns, "_has_agent_browser", lambda: False)
+    monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
+    monkeypatch.setattr(ns, "has_direct_modal_credentials", lambda: False)
+
+    features = ns.get_nous_subscription_features(
+        {"web": {"search_backend": "tavily"}}
+    )
+
+    assert features.web.available is True
+    assert features.web.active is True
+    assert features.web.current_provider == "tavily"
+
+
+def test_unconfigured_web_without_keys_is_unavailable(monkeypatch):
+    monkeypatch.setattr(ns, "get_env_value", lambda name: "")
+    monkeypatch.setattr(
+        ns, "get_nous_portal_account_info", lambda: _account(logged_in=False)
+    )
+    monkeypatch.setattr(ns, "_toolset_enabled", lambda config, key: key == "web")
+    monkeypatch.setattr(ns, "_has_agent_browser", lambda: False)
+    monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
+    monkeypatch.setattr(ns, "has_direct_modal_credentials", lambda: False)
+
+    features = ns.get_nous_subscription_features({})
+
+    assert features.web.available is False
+    assert features.web.active is False
+    assert features.web.explicit_configured is False
 def _stub_browser_probes(monkeypatch, *, has_agent_browser, chromium, lightpanda=False):
     """Common monkeypatches for local-browser readiness scenarios.
 

tests/plugins/web/test_web_search_provider_plugins.py

@@ -70,7 +70,7 @@ def _isolate_env(monkeypatch: pytest.MonkeyPatch) -> None:
 class TestBundledPluginsRegister:
     """All eight bundled web plugins discover and register correctly."""
 
-    def test_all_seven_plugins_present_in_registry(self) -> None:
+    def test_all_bundled_plugins_present_in_registry(self) -> None:
         _ensure_plugins_loaded()
         from agent.web_search_registry import list_providers
 
@@ -80,6 +80,7 @@ class TestBundledPluginsRegister:
             "ddgs",
             "exa",
             "firecrawl",
+            "keenable",
             "parallel",
             "searxng",
             "tavily",
@@ -203,6 +204,23 @@ class TestIsAvailable:
         monkeypatch.setenv("FIRECRAWL_API_URL", "http://localhost:3002")
         assert p.is_available() is True
 
+    def test_firecrawl_explicit_config_allows_keyless_cloud(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _ensure_plugins_loaded()
+        from agent.web_search_registry import get_provider
+
+        p = get_provider("firecrawl")
+        assert p is not None
+        assert p.is_available() is False
+
+        monkeypatch.setattr(
+            "tools.web_tools._load_web_config",
+            lambda: {"backend": "firecrawl"},
+            raising=False,
+        )
+        assert p.is_available() is True
+
     def test_ddgs_always_available_when_package_importable(self) -> None:
         """DDGS is the always-on fallback — no API key required.
 

tests/tools/test_web_keyless_fallback.py

@@ -172,25 +172,26 @@ class TestKeylessCalls:
 
 
 class TestProviderRouting:
-    def test_parallel_keyless_path_when_no_key(self):
+    def test_parallel_keyless_path_when_no_key(self, monkeypatch):
+        # Pin parallel so the ring deterministically starts there.
+        monkeypatch.setattr(keyless_mcp, "_vendor_pinned", lambda n: n == "parallel")
         provider = ParallelWebSearchProvider()
-        with patch.object(
-            keyless_mcp, "parallel_search_keyless",
-            return_value={"success": True, "data": {"web": []}},
-        ) as keyless:
+        with patch.dict(
+            keyless_mcp._KEYLESS_SEARCHERS,
+            {"parallel": lambda q, l: {"success": True, "data": {"web": []}}},
+        ):
             out = provider.search("q", limit=3)
         assert out["success"] is True
-        keyless.assert_called_once_with("q", 3)
 
-    def test_exa_keyless_path_when_no_key(self):
+    def test_exa_keyless_path_when_no_key(self, monkeypatch):
+        monkeypatch.setattr(keyless_mcp, "_vendor_pinned", lambda n: n == "exa")
         provider = ExaWebSearchProvider()
-        with patch.object(
-            keyless_mcp, "exa_search_keyless",
-            return_value={"success": True, "data": {"web": []}},
-        ) as keyless:
+        with patch.dict(
+            keyless_mcp._KEYLESS_SEARCHERS,
+            {"exa": lambda q, l: {"success": True, "data": {"web": []}}},
+        ):
             out = provider.search("q", limit=3)
         assert out["success"] is True
-        keyless.assert_called_once_with("q", 3)
 
     def test_parallel_keyed_path_skips_keyless(self, monkeypatch):
         monkeypatch.setattr(
@@ -258,15 +259,15 @@ class TestProviderRouting:
         assert keyless_mcp.provider_tier("tavily") == "auto"    # unset → auto
 
     @pytest.mark.asyncio
-    async def test_parallel_keyless_extract(self):
+    async def test_parallel_keyless_extract(self, monkeypatch):
+        monkeypatch.setattr(keyless_mcp, "_vendor_pinned", lambda n: n == "parallel")
         provider = ParallelWebSearchProvider()
-        with patch.object(
-            keyless_mcp, "parallel_extract_keyless",
-            return_value=[{"url": "https://a", "title": "", "content": "c"}],
-        ) as keyless:
+        with patch.dict(
+            keyless_mcp._KEYLESS_EXTRACTORS,
+            {"parallel": lambda urls: [{"url": "https://a", "title": "", "content": "c"}]},
+        ):
             out = await provider.extract(["https://a"])
         assert out[0]["content"] == "c"
-        keyless.assert_called_once_with(["https://a"])
 
 
 # ---------------------------------------------------------------------------
@@ -279,23 +280,28 @@ class TestResolutionOrder:
         monkeypatch.setattr(registry, "_read_config_key", lambda *p: None)
         provider = registry.get_active_search_provider()
         assert provider is not None
-        # 50/50 split: either keyless vendor is valid; it must match the
-        # process-stable preference order.
-        assert provider.name == registry._keyless_preference()[0]
-        assert provider.name in ("exa", "parallel")
-
-    def test_keyless_split_is_process_stable_and_covers_both(self, fresh_registry, monkeypatch):
-        monkeypatch.setattr(registry, "_read_config_key", lambda *p: None)
-        # Stable within a process: repeated resolution never flip-flops.
-        first = registry.get_active_search_provider().name
-        assert all(
-            registry.get_active_search_provider().name == first for _ in range(5)
+        # Ring: resolution picks the first REGISTERED vendor in ring order
+        # (only exa/parallel are registered in this fixture).
+        expected = next(
+            v for v in registry._keyless_preference() if v in ("exa", "parallel")
         )
-        # Both split outcomes route correctly (simulate the two parities).
-        monkeypatch.setattr(keyless_mcp, "_SESSION_ID", "0" * 32)  # even
-        assert registry._keyless_preference() == ("exa", "parallel")
-        monkeypatch.setattr(keyless_mcp, "_SESSION_ID", "1" * 32)  # odd
-        assert registry._keyless_preference() == ("parallel", "exa")
+        assert provider.name == expected
+
+    def test_keyless_ring_rotates_and_covers_all_vendors(self, fresh_registry, monkeypatch):
+        monkeypatch.setattr(registry, "_read_config_key", lambda *p: None)
+        # The ring order always contains all five vendors, starting at the
+        # current cursor and wrapping.
+        order = registry._keyless_preference()
+        assert sorted(order) == sorted(keyless_mcp._KEYLESS_RING)
+        # Unpinned dispatch rotates: consecutive _ring_order calls start at
+        # successive vendors (round-robin cursor advances per request).
+        monkeypatch.setattr(keyless_mcp, "_vendor_pinned", lambda name: False)
+        starts = [keyless_mcp._ring_order("exa")[0] for _ in range(len(keyless_mcp._KEYLESS_RING))]
+        assert sorted(starts) == sorted(keyless_mcp._KEYLESS_RING)  # full cycle
+        # Pinned dispatch starts at the pinned vendor every time.
+        monkeypatch.setattr(keyless_mcp, "_vendor_pinned", lambda name: name == "tavily")
+        assert keyless_mcp._ring_order("tavily")[0] == "tavily"
+        assert keyless_mcp._ring_order("tavily")[0] == "tavily"
 
     def test_registry_keyless_disabled_returns_none(self, fresh_registry, monkeypatch):
         monkeypatch.setattr(registry, "_read_config_key", lambda *p: None)
@@ -322,7 +328,10 @@ class TestResolutionOrder:
         )
         monkeypatch.setattr(web_tools, "_list_registered_web_providers", list)
         from agent.web_search_registry import _keyless_preference
-        assert web_tools._get_backend() == _keyless_preference()[0]
+        expected = next(
+            v for v in _keyless_preference() if v in ("exa", "parallel")
+        )
+        assert web_tools._get_backend() == expected
 
     def test_get_backend_key_beats_keyless(self, monkeypatch):
         monkeypatch.setattr(
@@ -414,3 +423,123 @@ class TestPickerTierRows:
         cfg_auto = {"web": {"backend": "parallel"}}
         assert _web_tier_matches(free_row, cfg_auto) is True
         assert _web_tier_matches(paid_row, cfg_auto) is False
+
+
+# ---------------------------------------------------------------------------
+# Cross-vendor keyless failover
+# ---------------------------------------------------------------------------
+
+
+class TestKeylessFailover:
+    def _ok(self, vendor):
+        return {"success": True, "data": {"web": [{"url": f"https://{vendor}.example"}]}}
+
+    def _throttled(self, vendor):
+        return {"success": False, "error": f"Keyless {vendor} search failed: free MCP rate limit."}
+
+    def _pin(self, monkeypatch, name):
+        """Pin *name* so the ring starts there deterministically."""
+        monkeypatch.setattr(keyless_mcp, "_vendor_pinned", lambda n: n == name)
+
+    def test_search_fails_over_on_rate_limit(self, monkeypatch):
+        self._pin(monkeypatch, "exa")
+        monkeypatch.setitem(keyless_mcp._KEYLESS_SEARCHERS, "exa", lambda q, l: self._throttled("Exa"))
+        monkeypatch.setitem(keyless_mcp._KEYLESS_SEARCHERS, "parallel", lambda q, l: self._ok("parallel"))
+        out = keyless_mcp.search_with_failover("exa", "q", 3)
+        assert out["success"] is True
+        assert out["data"]["served_by"] == "parallel"
+
+    def test_search_no_failover_on_non_throttle_error(self, monkeypatch):
+        self._pin(monkeypatch, "exa")
+        monkeypatch.setitem(
+            keyless_mcp._KEYLESS_SEARCHERS, "exa",
+            lambda q, l: {"success": False, "error": "Unrecognized MCP response shape"},
+        )
+        called = []
+        monkeypatch.setitem(
+            keyless_mcp._KEYLESS_SEARCHERS, "parallel",
+            lambda q, l: called.append(1) or self._ok("parallel"),
+        )
+        out = keyless_mcp.search_with_failover("exa", "q")
+        assert out["success"] is False
+        assert not called  # peer never tried
+
+    def test_search_all_throttled_reports_ring(self, monkeypatch):
+        self._pin(monkeypatch, "exa")
+        for vendor in keyless_mcp._KEYLESS_RING:
+            monkeypatch.setitem(
+                keyless_mcp._KEYLESS_SEARCHERS, vendor,
+                lambda q, l, v=vendor: self._throttled(v),
+            )
+        out = keyless_mcp.search_with_failover("exa", "q")
+        assert out["success"] is False
+        assert "all keyless vendors throttled" in out["error"]
+
+    def test_search_walks_ring_past_multiple_throttles(self, monkeypatch):
+        # exa -> parallel -> tavily all throttled; firecrawl serves.
+        self._pin(monkeypatch, "exa")
+        for vendor in ("exa", "parallel", "tavily"):
+            monkeypatch.setitem(
+                keyless_mcp._KEYLESS_SEARCHERS, vendor,
+                lambda q, l, v=vendor: self._throttled(v),
+            )
+        monkeypatch.setitem(
+            keyless_mcp._KEYLESS_SEARCHERS, "firecrawl",
+            lambda q, l: self._ok("firecrawl"),
+        )
+        out = keyless_mcp.search_with_failover("exa", "q")
+        assert out["success"] is True
+        assert out["data"]["served_by"] == "firecrawl"
+
+    def test_failover_respects_peer_paid_pin(self, monkeypatch):
+        # Every vendor except exa throttles; exa is pinned paid so its free
+        # endpoint must never be used.
+        monkeypatch.setattr(
+            keyless_mcp, "provider_tier",
+            lambda name: "paid" if name == "exa" else "auto",
+        )
+        monkeypatch.setattr(keyless_mcp, "_vendor_pinned", lambda n: n == "parallel")
+        called = []
+        monkeypatch.setitem(
+            keyless_mcp._KEYLESS_SEARCHERS, "exa",
+            lambda q, l: called.append(1) or self._ok("exa"),
+        )
+        for vendor in ("parallel", "tavily", "firecrawl", "keenable"):
+            monkeypatch.setitem(
+                keyless_mcp._KEYLESS_SEARCHERS, vendor,
+                lambda q, l, v=vendor: self._throttled(v),
+            )
+        out = keyless_mcp.search_with_failover("parallel", "q")
+        assert out["success"] is False
+        assert not called  # exa pinned paid: its free tier is opted out
+
+    def test_extract_fails_over_when_all_urls_throttled(self, monkeypatch):
+        self._pin(monkeypatch, "exa")
+        throttled = [
+            {"url": "https://a", "title": "", "content": "", "error": "rate limit hit"},
+            {"url": "https://b", "title": "", "content": "", "error": "429 too many requests"},
+        ]
+        good = [
+            {"url": "https://a", "title": "A", "content": "x"},
+            {"url": "https://b", "title": "B", "content": "y"},
+        ]
+        monkeypatch.setitem(keyless_mcp._KEYLESS_EXTRACTORS, "exa", lambda urls: throttled)
+        monkeypatch.setitem(keyless_mcp._KEYLESS_EXTRACTORS, "parallel", lambda urls: good)
+        out = keyless_mcp.extract_with_failover("exa", ["https://a", "https://b"])
+        assert out == good
+
+    def test_extract_partial_failure_stays_on_primary(self, monkeypatch):
+        self._pin(monkeypatch, "exa")
+        partial = [
+            {"url": "https://a", "title": "A", "content": "x"},
+            {"url": "https://b", "title": "", "content": "", "error": "rate limit"},
+        ]
+        called = []
+        monkeypatch.setitem(keyless_mcp._KEYLESS_EXTRACTORS, "exa", lambda urls: partial)
+        monkeypatch.setitem(
+            keyless_mcp._KEYLESS_EXTRACTORS, "parallel",
+            lambda urls: called.append(1) or [],
+        )
+        out = keyless_mcp.extract_with_failover("exa", ["https://a", "https://b"])
+        assert out == partial
+        assert not called

tests/tools/test_web_providers.py

@@ -202,22 +202,71 @@ class TestUnconfiguredErrorEnvelopeParity:
         from agent import web_search_registry
 
         self._clear_web_creds(monkeypatch)
-        # Reset firecrawl client cache so the unconfigured state is re-evaluated
         monkeypatch.setattr(web_tools, "_firecrawl_client", None, raising=False)
         monkeypatch.setattr(web_tools, "_firecrawl_client_config", None, raising=False)
         monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: False)
         monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
         monkeypatch.setattr(web_search_registry, "_keyless_tier_enabled", lambda: False)
+        monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
 
         result = json.loads(web_tools.web_search_tool("hello world", limit=3))
         assert "error" in result, f"expected top-level 'error' key, got {result}"
-        # ``Error searching web:`` prefix comes from web_tools' top-level except handler
         assert "Error searching web:" in result["error"]
         assert "FIRECRAWL_API_KEY" in result["error"]
-        # No per-result burying
         assert "results" not in result
 
 
+    def test_explicit_firecrawl_unconfigured_uses_firecrawl_keyless(self, monkeypatch):
+        """``web.backend: firecrawl`` with no creds routes through Firecrawl's
+        keyless cloud client (PR #50659 salvage) — keyless Tavily must not
+        silently take over, and the request must hit api.firecrawl.dev.
+        """
+        from tools import web_tools
+        from plugins.web.firecrawl import provider as fc
+
+        self._clear_web_creds(monkeypatch)
+        monkeypatch.setattr(web_tools, "_firecrawl_client", None, raising=False)
+        monkeypatch.setattr(web_tools, "_firecrawl_client_config", None, raising=False)
+        monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: False)
+        monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "firecrawl"})
+        monkeypatch.setattr(fc, "_load_web_config", lambda: {"backend": "firecrawl"}, raising=False)
+        monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
+        monkeypatch.setattr(web_tools, "check_firecrawl_api_key", lambda: False)
+        # Developer machines may carry FIRECRAWL_* in ~/.hermes/.env — the
+        # config-aware lookup must see a truly keyless environment here.
+        monkeypatch.setattr(
+            "hermes_cli.config.get_env_value", lambda name: None, raising=True
+        )
+
+        calls = {}
+
+        class _FakeResponse:
+            status_code = 200
+
+            def raise_for_status(self):
+                return None
+
+            def json(self):
+                return {
+                    "success": True,
+                    "data": [
+                        {"url": "https://example.com", "title": "Example",
+                         "description": "desc"},
+                    ],
+                }
+
+        def _fake_post(url, **kwargs):
+            calls["url"] = url
+            return _FakeResponse()
+
+        monkeypatch.setattr(fc.httpx, "post", _fake_post)
+
+        result = json.loads(web_tools.web_search_tool("hello world", limit=3))
+        assert result.get("success") is True, result
+        assert calls["url"].startswith("https://api.firecrawl.dev"), calls
+        assert result["data"]["web"], result
+
+
 class TestDispatchersTriggerPluginDiscovery:
     """Regression tests for #27580: each web_*_tool dispatcher must
     idempotently call ``_ensure_web_plugins_loaded()`` before consulting
@@ -317,6 +366,10 @@ class TestDispatchersTriggerPluginDiscovery:
                 web_tools, "_load_web_config",
                 lambda: {"extract_backend": "firecrawl"},
             )
+            monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-test")
+            async def _allow_ssrf(_url: str) -> bool:
+                return True
+            monkeypatch.setattr(web_tools, "async_is_safe_url", _allow_ssrf)
             # Sanity: registry IS empty before the tool call.
             assert web_search_registry.get_provider("firecrawl") is None
 

tests/tools/test_web_tools_config.py

@@ -115,6 +115,82 @@ class TestFirecrawlClientConfig:
                     with pytest.raises(ValueError):
                         _get_firecrawl_client()
 
+    def test_explicit_firecrawl_config_without_creds_uses_keyless_client(self):
+        """Explicit Firecrawl config should build the keyless cloud client."""
+        from plugins.web.firecrawl import provider as firecrawl_provider
+
+        with patch("tools.web_tools._load_web_config", return_value={"backend": "firecrawl"}):
+            with patch("tools.web_tools._read_nous_access_token", return_value=None):
+                with patch("tools.web_tools.Firecrawl", side_effect=AssertionError("SDK path should not run")):
+                    from tools.web_tools import _get_firecrawl_client
+
+                    result = _get_firecrawl_client()
+
+        assert isinstance(result, firecrawl_provider._KeylessFirecrawlClient)
+        assert result.api_url == "https://api.firecrawl.dev"
+
+    def test_keyless_firecrawl_search_omits_authorization_header(self, monkeypatch):
+        """Keyless Firecrawl search must not send a bearer header."""
+        from plugins.web.firecrawl import provider as firecrawl_provider
+
+        captured = {}
+
+        class _Response:
+            def raise_for_status(self):
+                return None
+
+            def json(self):
+                return {"success": True, "data": {"web": []}}
+
+        def _fake_post(url, *, json, headers, timeout):
+            captured["url"] = url
+            captured["json"] = json
+            captured["headers"] = headers
+            captured["timeout"] = timeout
+            return _Response()
+
+        monkeypatch.setattr(firecrawl_provider.httpx, "post", _fake_post)
+
+        client = firecrawl_provider._KeylessFirecrawlClient()
+        result = client.search(query="firecrawl", limit=1)
+
+        assert result["success"] is True
+        assert captured["url"] == "https://api.firecrawl.dev/v2/search"
+        assert captured["json"] == {"query": "firecrawl", "limit": 1}
+        assert captured["headers"] == {"Content-Type": "application/json"}
+        assert "Authorization" not in captured["headers"]
+
+    def test_keyless_firecrawl_scrape_omits_authorization_header(self, monkeypatch):
+        """Keyless Firecrawl scrape must not send a bearer header."""
+        from plugins.web.firecrawl import provider as firecrawl_provider
+
+        captured = {}
+
+        class _Response:
+            def raise_for_status(self):
+                return None
+
+            def json(self):
+                return {"success": True, "data": {"markdown": "# ok"}}
+
+        def _fake_post(url, *, json, headers, timeout):
+            captured["url"] = url
+            captured["json"] = json
+            captured["headers"] = headers
+            captured["timeout"] = timeout
+            return _Response()
+
+        monkeypatch.setattr(firecrawl_provider.httpx, "post", _fake_post)
+
+        client = firecrawl_provider._KeylessFirecrawlClient()
+        result = client.scrape(url="https://example.com", formats=["markdown"])
+
+        assert result["success"] is True
+        assert captured["url"] == "https://api.firecrawl.dev/v2/scrape"
+        assert captured["json"] == {"url": "https://example.com", "formats": ["markdown"]}
+        assert captured["headers"] == {"Content-Type": "application/json"}
+        assert "Authorization" not in captured["headers"]
+
 
 class TestBackendSelection:
     """Test suite for _get_backend() backend selection logic.
@@ -224,7 +300,9 @@ class TestBackendSelection:
         """
         from tools.web_tools import _get_backend
         with patch("tools.web_tools._load_web_config", return_value={}), \
+             patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \
              patch("tools.web_tools._ddgs_package_importable", return_value=False), \
+             patch("tools.web_tools._list_registered_web_providers", return_value=[]), \
              patch("agent.web_search_registry._keyless_tier_enabled", return_value=False):
             assert _get_backend() == "firecrawl"
 
@@ -467,6 +545,54 @@ class TestCheckWebApiKey:
                     from tools.web_tools import check_web_api_key
                     assert check_web_api_key() is True
 
+    def test_explicit_unavailable_active_provider_is_not_ready(self):
+        """#78412: get_active_* may return a configured backend whose
+        is_available() is False. check_web_api_key must still report False so
+        doctor does not paint a green check for a backend that cannot run.
+        """
+        class _UnavailableProvider:
+            name = "firecrawl"
+
+            def is_available(self):
+                return False
+
+        unavailable = _UnavailableProvider()
+        with patch("tools.web_tools._load_web_config", return_value={"backend": "firecrawl"}), \
+             patch("tools.web_tools._is_backend_available", return_value=False), \
+             patch(
+                 "agent.web_search_registry.get_active_search_provider",
+                 return_value=unavailable,
+             ), \
+             patch(
+                 "agent.web_search_registry.get_active_extract_provider",
+                 return_value=unavailable,
+             ):
+            from tools.web_tools import check_web_api_key, _provider_is_ready
+            assert _provider_is_ready(unavailable) is False
+            assert check_web_api_key() is False
+
+    def test_explicit_available_active_provider_is_ready(self):
+        """Registry-selected available provider still lights the gate."""
+        class _AvailableProvider:
+            name = "custom-ok"
+
+            def is_available(self):
+                return True
+
+        available = _AvailableProvider()
+        with patch("tools.web_tools._load_web_config", return_value={"backend": "custom-ok"}), \
+             patch("tools.web_tools._is_backend_available", return_value=False), \
+             patch(
+                 "agent.web_search_registry.get_active_search_provider",
+                 return_value=available,
+             ), \
+             patch(
+                 "agent.web_search_registry.get_active_extract_provider",
+                 return_value=None,
+             ):
+            from tools.web_tools import check_web_api_key
+            assert check_web_api_key() is True
+
 
 def test_web_requires_env_includes_exa_key():
     from tools.web_tools import _web_requires_env
@@ -606,7 +732,8 @@ class TestFirecrawlEnvResolution:
 
             result = _get_direct_firecrawl_config()
             assert result is not None, "get_env_value fallback should find the key"
-            kwargs, _cache_key = result
+            mode, kwargs, _cache_key = result
+            assert mode == "sdk"
             assert kwargs["api_key"] == fake_key
 
     def test_direct_config_reads_url_via_get_env_value(self, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -623,7 +750,8 @@ class TestFirecrawlEnvResolution:
 
             result = _get_direct_firecrawl_config()
             assert result is not None
-            kwargs, _cache_key = result
+            mode, kwargs, _cache_key = result
+            assert mode == "sdk"
             assert kwargs["api_url"] == fake_url.rstrip("/")
 
 
@@ -662,6 +790,28 @@ class TestSiblingProvidersEnvResolution:
                 "config-aware env layer (get_env_value)"
             )
 
+    def test_tavily_request_reads_key_via_get_env_value(self, monkeypatch):
+        """Keyed Tavily must Bearer-auth with a key that lives only in .env."""
+        monkeypatch.delenv("TAVILY_API_KEY", raising=False)
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {"results": []}
+        mock_response.text = "{}"
+
+        with patch(
+            "hermes_cli.config.get_env_value",
+            side_effect=lambda k: "tvly-from-dotenv" if k == "TAVILY_API_KEY" else None,
+        ), patch(
+            "plugins.web.tavily.provider.httpx.post", return_value=mock_response
+        ) as mock_post:
+            from plugins.web.tavily.provider import _tavily_request
+
+            _tavily_request("search", {"query": "q"})
+            headers = mock_post.call_args.kwargs["headers"]
+            assert headers["Authorization"] == "Bearer tvly-from-dotenv"
+            assert headers["X-Client-Name"] == "hermes-agent"
+            assert "X-Tavily-Access-Mode" not in headers
+
 
     def test_get_provider_env_unset_returns_empty(self, monkeypatch):
         monkeypatch.delenv("WSP_TEST_UNSET_KEY", raising=False)

tests/tools/test_web_tools_tavily.py

@@ -1,10 +1,11 @@
 """Tests for Tavily web backend integration.
 
 Coverage:
-  _tavily_request() — API key handling, endpoint construction, error propagation.
+  _tavily_request() — keyed Bearer vs keyless header, attribution, error bodies.
   _normalize_tavily_search_results() — search response normalization.
   _normalize_tavily_documents() — extract response normalization, failed_results.
   web_search_tool / web_extract_tool — Tavily dispatch paths.
+  auto-detect ranking — keyed paid-band; keyless only when Tavily is selected.
 """
 
 import json
@@ -16,49 +17,72 @@ from unittest.mock import patch, MagicMock
 from tests.tools.conftest import register_all_web_providers
 
 
+def _ok_response(payload=None):
+    mock_response = MagicMock()
+    mock_response.status_code = 200
+    mock_response.json.return_value = payload if payload is not None else {"results": []}
+    mock_response.text = json.dumps(mock_response.json.return_value)
+    return mock_response
+
+
 # ─── _tavily_request ─────────────────────────────────────────────────────────
 
 class TestTavilyRequest:
     """Test suite for the _tavily_request helper."""
 
-    def test_raises_without_api_key(self):
-        """No TAVILY_API_KEY → ValueError with guidance."""
+    def test_keyless_when_no_api_key(self):
+        """No TAVILY_API_KEY → keyless header, no Authorization, no body key."""
+        mock_response = _ok_response()
+
         with patch.dict(os.environ, {}, clear=False):
             os.environ.pop("TAVILY_API_KEY", None)
-            from tools.web_tools import _tavily_request
-            with pytest.raises(ValueError, match="TAVILY_API_KEY"):
+            with patch("plugins.web.tavily.provider.httpx.post", return_value=mock_response) as mock_post:
+                from plugins.web.tavily.provider import _tavily_request
                 _tavily_request("search", {"query": "test"})
 
-    def test_posts_with_api_key_in_body(self):
-        """api_key is injected into the JSON payload."""
-        mock_response = MagicMock()
-        mock_response.json.return_value = {"results": []}
-        mock_response.raise_for_status = MagicMock()
+                mock_post.assert_called_once()
+                headers = mock_post.call_args.kwargs["headers"]
+                payload = mock_post.call_args.kwargs["json"]
+                assert headers["X-Client-Name"] == "hermes-agent"
+                assert headers["X-Tavily-Access-Mode"] == "keyless"
+                assert "Authorization" not in headers
+                assert "api_key" not in payload
+                assert payload["query"] == "test"
+                assert "api.tavily.com/search" in mock_post.call_args.args[0]
+
+    def test_keyed_uses_bearer_not_body(self):
+        """TAVILY_API_KEY → Bearer auth, attribution, no body api_key."""
+        mock_response = _ok_response()
 
         with patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-test-key"}):
-            with patch("tools.web_tools.httpx.post", return_value=mock_response) as mock_post:
-                from tools.web_tools import _tavily_request
-                result = _tavily_request("search", {"query": "hello"})
+            with patch("plugins.web.tavily.provider.httpx.post", return_value=mock_response) as mock_post:
+                from plugins.web.tavily.provider import _tavily_request
+                _tavily_request("search", {"query": "hello"})
 
                 mock_post.assert_called_once()
-                call_kwargs = mock_post.call_args
-                payload = call_kwargs.kwargs.get("json") or call_kwargs[1].get("json")
-                assert payload["api_key"] == "tvly-test-key"
+                headers = mock_post.call_args.kwargs["headers"]
+                payload = mock_post.call_args.kwargs["json"]
+                assert headers == {
+                    "X-Client-Name": "hermes-agent",
+                    "Authorization": "Bearer tvly-test-key",
+                }
+                assert "X-Tavily-Access-Mode" not in headers
+                assert "api_key" not in payload
                 assert payload["query"] == "hello"
-                assert "api.tavily.com/search" in call_kwargs.args[0]
+                assert "api.tavily.com/search" in mock_post.call_args.args[0]
 
-    def test_raises_on_http_error(self):
-        """Non-2xx responses propagate as httpx.HTTPStatusError."""
-        import httpx as _httpx
+    def test_http_error_surfaces_response_body(self):
+        """Non-2xx responses raise ValueError with Tavily's response body."""
         mock_response = MagicMock()
-        mock_response.raise_for_status.side_effect = _httpx.HTTPStatusError(
-            "401 Unauthorized", request=MagicMock(), response=mock_response
-        )
+        mock_response.status_code = 429
+        mock_response.text = "Rate limit hit. Sign up for a free API key at https://app.tavily.com"
+        mock_response.json.return_value = {}
 
-        with patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-bad-key"}):
-            with patch("tools.web_tools.httpx.post", return_value=mock_response):
-                from tools.web_tools import _tavily_request
-                with pytest.raises(_httpx.HTTPStatusError):
+        with patch.dict(os.environ, {}, clear=False):
+            os.environ.pop("TAVILY_API_KEY", None)
+            with patch("plugins.web.tavily.provider.httpx.post", return_value=mock_response):
+                from plugins.web.tavily.provider import _tavily_request
+                with pytest.raises(ValueError, match="Rate limit hit"):
                     _tavily_request("search", {"query": "test"})
 
 
@@ -98,7 +122,7 @@ class TestNormalizeTavilySearchResults:
 # ─── _normalize_tavily_documents ──────────────────────────────────────────────
 
 class TestNormalizeTavilyDocuments:
-    """Test extract/crawl document normalization."""
+    """Test extract document normalization."""
 
     def test_basic_document(self):
         from tools.web_tools import _normalize_tavily_documents
@@ -125,6 +149,83 @@ class TestNormalizeTavilyDocuments:
         assert docs[0]["url"] == "https://fallback.com"
 
 
+# ─── availability / auto-detect ───────────────────────────────────────────────
+
+class TestTavilyAvailability:
+    """Keyed Tavily stays in the paid band; keyless only when selected."""
+
+    def test_is_available_without_key(self):
+        from plugins.web.tavily.provider import TavilyWebSearchProvider
+        with patch.dict(os.environ, {}, clear=False):
+            os.environ.pop("TAVILY_API_KEY", None)
+            assert TavilyWebSearchProvider().is_available() is False
+
+    def test_is_backend_available_without_key(self):
+        from tools.web_tools import _is_backend_available
+        with patch("tools.web_tools._load_web_config", return_value={}), \
+             patch.dict(os.environ, {}, clear=False):
+            os.environ.pop("TAVILY_API_KEY", None)
+            assert _is_backend_available("tavily") is False
+
+    def test_is_backend_available_when_configured_without_key(self):
+        from tools.web_tools import _is_backend_available
+        with patch("tools.web_tools._load_web_config", return_value={"backend": "tavily"}), \
+             patch.dict(os.environ, {}, clear=False):
+            os.environ.pop("TAVILY_API_KEY", None)
+            assert _is_backend_available("tavily") is True
+
+    def test_keyless_does_not_preempt_managed_firecrawl(self):
+        """No TAVILY_API_KEY + Nous gateway ready → firecrawl, not keyless tavily."""
+        from tools.web_tools import _get_backend
+        with patch("tools.web_tools._load_web_config", return_value={}), \
+             patch("tools.web_tools._is_tool_gateway_ready", return_value=True), \
+             patch("tools.web_tools._ddgs_package_importable", return_value=False):
+            os.environ.pop("TAVILY_API_KEY", None)
+            assert _get_backend() == "firecrawl"
+
+    def test_keyless_does_not_preempt_ddgs(self):
+        from tools.web_tools import _get_backend
+        with patch("tools.web_tools._load_web_config", return_value={}), \
+             patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \
+             patch("tools.web_tools._ddgs_package_importable", return_value=True):
+            os.environ.pop("TAVILY_API_KEY", None)
+            assert _get_backend() == "ddgs"
+
+    def test_no_keys_defaults_to_firecrawl(self):
+        """Keyless tier disabled: zero-credential resolve hits the legacy
+        firecrawl sentinel. (With the tier on — the default — it resolves
+        to the Exa/Parallel keyless split; see test_web_keyless_fallback.py.)
+        """
+        from tools.web_tools import _get_backend
+        with patch("tools.web_tools._load_web_config", return_value={}), \
+             patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \
+             patch("tools.web_tools._ddgs_package_importable", return_value=False), \
+             patch("tools.web_tools._list_registered_web_providers", return_value=[]), \
+             patch("agent.web_search_registry._keyless_tier_enabled", return_value=False):
+            os.environ.pop("TAVILY_API_KEY", None)
+            assert _get_backend() == "firecrawl"
+
+    def test_explicit_search_backend_tavily_without_key(self):
+        """web.search_backend=tavily sticks even with no TAVILY_API_KEY."""
+        from tools.web_tools import _get_search_backend
+        with patch("tools.web_tools._load_web_config",
+                   return_value={"backend": "firecrawl", "search_backend": "tavily"}), \
+             patch("tools.web_tools._is_tool_gateway_ready", return_value=True):
+            os.environ.pop("TAVILY_API_KEY", None)
+            assert _get_search_backend() == "tavily"
+
+    def test_check_web_api_key_when_tavily_configured_without_key(self):
+        from tools.web_tools import check_web_api_key
+        with patch("tools.web_tools._load_web_config", return_value={"backend": "tavily"}), \
+             patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \
+             patch("tools.web_tools.check_firecrawl_api_key", return_value=False), \
+             patch("tools.web_tools._ddgs_package_importable", return_value=False), \
+             patch("agent.web_search_registry.get_active_search_provider", return_value=None), \
+             patch("agent.web_search_registry.get_active_extract_provider", return_value=None):
+            os.environ.pop("TAVILY_API_KEY", None)
+            assert check_web_api_key() is True
+
+
 # ─── web_search_tool (Tavily dispatch) ────────────────────────────────────────
 
 class TestWebSearchTavily:
@@ -140,15 +241,13 @@ class TestWebSearchTavily:
         _reset_for_tests()
 
     def test_search_dispatches_to_tavily(self):
-        mock_response = MagicMock()
-        mock_response.json.return_value = {
+        mock_response = _ok_response({
             "results": [{"title": "Result", "url": "https://r.com", "content": "desc", "score": 0.9}]
-        }
-        mock_response.raise_for_status = MagicMock()
+        })
 
         with patch("tools.web_tools._get_backend", return_value="tavily"), \
              patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-test"}), \
-             patch("tools.web_tools.httpx.post", return_value=mock_response), \
+             patch("plugins.web.tavily.provider.httpx.post", return_value=mock_response), \
              patch("tools.interrupt.is_interrupted", return_value=False):
             from tools.web_tools import web_search_tool
             result = json.loads(web_search_tool("test query", limit=3))
@@ -156,6 +255,27 @@ class TestWebSearchTavily:
             assert len(result["data"]["web"]) == 1
             assert result["data"]["web"][0]["title"] == "Result"
 
+    def test_search_keyless_dispatch(self):
+        """Keyless Tavily routes through the ring; pinned tavily starts at
+        tavily and the ring searcher sends the keyless headers."""
+        from plugins.web import keyless_mcp
+
+        mock_response = _ok_response({
+            "results": [{"title": "Result", "url": "https://r.com", "content": "desc"}]
+        })
+
+        with patch("tools.web_tools._get_backend", return_value="tavily"), \
+             patch.object(keyless_mcp, "_vendor_pinned", lambda n: n == "tavily"), \
+             patch("requests.post", return_value=mock_response) as mock_post, \
+             patch("tools.interrupt.is_interrupted", return_value=False):
+            os.environ.pop("TAVILY_API_KEY", None)
+            from tools.web_tools import web_search_tool
+            result = json.loads(web_search_tool("test query"))
+            assert result["success"] is True
+            headers = mock_post.call_args.kwargs["headers"]
+            assert headers["X-Tavily-Access-Mode"] == "keyless"
+            assert headers["X-Client-Name"] == "hermes-agent"
+
 
 # ─── web_extract_tool (Tavily dispatch) ───────────────────────────────────────
 
@@ -172,15 +292,17 @@ class TestWebExtractTavily:
         _reset_for_tests()
 
     def test_extract_dispatches_to_tavily(self):
-        mock_response = MagicMock()
-        mock_response.json.return_value = {
+        mock_response = _ok_response({
             "results": [{"url": "https://example.com", "raw_content": "Extracted content", "title": "Page"}]
-        }
-        mock_response.raise_for_status = MagicMock()
+        })
+
+        async def _allow_ssrf(_url: str) -> bool:
+            return True
 
         with patch("tools.web_tools._get_backend", return_value="tavily"), \
              patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-test"}), \
-             patch("tools.web_tools.httpx.post", return_value=mock_response):
+             patch("plugins.web.tavily.provider.httpx.post", return_value=mock_response), \
+             patch("tools.web_tools.async_is_safe_url", _allow_ssrf):
             from tools.web_tools import web_extract_tool
             result = json.loads(asyncio.get_event_loop().run_until_complete(
                 web_extract_tool(["https://example.com"])
@@ -189,4 +311,3 @@ class TestWebExtractTavily:
             assert len(result["results"]) == 1
             assert result["results"][0]["url"] == "https://example.com"
             assert "Extracted content" in result["results"][0]["content"]
-

tools/web_tools.py

@@ -169,7 +169,7 @@ def _load_web_config() -> dict:
 # WebSearchProvider. Keep the two sets aligned by hand: if xai ever ships as
 # a registered provider, drop it here so the registry path takes over.
 _LEGACY_WEB_BACKENDS = frozenset(
-    {"parallel", "firecrawl", "tavily", "exa", "searxng", "brave-free", "ddgs", "xai"}
+    {"parallel", "firecrawl", "tavily", "exa", "searxng", "brave-free", "ddgs", "xai", "keenable"}
 )
 
 
@@ -262,6 +262,7 @@ def _get_backend() -> str:
         ("tavily", _has_env("TAVILY_API_KEY")),
         ("exa", _has_env("EXA_API_KEY")),
         ("parallel", _has_env("PARALLEL_API_KEY")),
+        ("keenable", _has_env("KEENABLE_API_KEY")),
         ("firecrawl", _has_env("FIRECRAWL_API_KEY") or _has_env("FIRECRAWL_API_URL")),
         ("firecrawl", _is_tool_gateway_ready()),
         ("searxng", _has_env("SEARXNG_URL")),
@@ -358,6 +359,14 @@ def _get_capability_backend(capability: str) -> str:
     return _get_backend()
 
 
+def _tavily_explicitly_configured() -> bool:
+    cfg = _load_web_config()
+    return any(
+        (cfg.get(key) or "").lower().strip() == "tavily"
+        for key in ("backend", "search_backend", "extract_backend")
+    )
+
+
 def _is_backend_available(backend: str) -> bool:
     """Return True when the selected backend is currently usable.
 
@@ -379,10 +388,12 @@ def _is_backend_available(backend: str) -> bool:
         return _has_env("EXA_API_KEY")
     if backend == "parallel":
         return _has_env("PARALLEL_API_KEY")
+    if backend == "keenable":
+        return _has_env("KEENABLE_API_KEY")
     if backend == "firecrawl":
         return check_firecrawl_api_key()
     if backend == "tavily":
-        return _has_env("TAVILY_API_KEY")
+        return _has_env("TAVILY_API_KEY") or _tavily_explicitly_configured()
     if backend == "searxng":
         return _has_env("SEARXNG_URL")
     if backend == "brave-free":
@@ -442,6 +453,7 @@ def _web_requires_env() -> list[str]:
         "EXA_API_KEY",
         "PARALLEL_API_KEY",
         "TAVILY_API_KEY",
+        "KEENABLE_API_KEY",
         "FIRECRAWL_API_KEY",
         "FIRECRAWL_API_URL",
         "FIRECRAWL_GATEWAY_URL",
@@ -1158,6 +1170,42 @@ async def web_extract_tool(
 
 
 # Convenience function to check Firecrawl credentials
+def _provider_is_ready(provider) -> bool:
+    """Return True when *provider* reports readiness without raising.
+
+    ``get_active_*_provider()`` intentionally returns an explicitly configured
+    backend even when ``is_available()`` is False so the dispatcher can emit a
+    precise missing-credential error. Tool/doctor readiness gates must still
+    require a true availability probe — otherwise ``hermes doctor`` paints a
+    green ✓ for a backend that cannot run (issue #78412).
+
+    A provider that can serve anonymously (``is_keyless_available()`` — the
+    Exa/Parallel free tier) IS ready: keyless mode is a working state, not a
+    misconfiguration.
+    """
+    if provider is None:
+        return False
+    try:
+        if provider.is_available():
+            return True
+    except Exception as exc:  # noqa: BLE001 — broken provider == not ready
+        logger.debug(
+            "web provider %r.is_available() raised during readiness check: %s",
+            getattr(provider, "name", provider),
+            exc,
+        )
+        return False
+    try:
+        return bool(provider.is_keyless_available())
+    except Exception as exc:  # noqa: BLE001 — broken provider == not ready
+        logger.debug(
+            "web provider %r.is_keyless_available() raised during readiness check: %s",
+            getattr(provider, "name", provider),
+            exc,
+        )
+        return False
+
+
 def check_web_api_key() -> bool:
     """Check whether the configured web backend is available.
 
@@ -1177,15 +1225,14 @@ def check_web_api_key() -> bool:
     # unlike _get_backend() the probe order is irrelevant.
     if any(_is_backend_available(backend) for backend in _LEGACY_WEB_BACKENDS):
         return True
-    # Any plugin-registered provider the registry considers active for either
-    # capability. Delegating to the registry's own availability-filtered
-    # resolvers keeps a single authority for "is a custom provider usable"
-    # rather than re-implementing the walk here. This also covers the
-    # keyless free tier (Parallel/Exa anonymous MCP endpoints): the registry
-    # walk falls back to keyless-capable providers when nothing is keyed,
-    # so a zero-credential install still lights the web tools up. Discovery
-    # must run first — check_fn fires at tool-registration time, before any
-    # dispatch has populated the registry.
+    # Plugin-registered path: the active-provider resolvers return an explicit
+    # config hit even when credentials are missing (so the tool can print a
+    # precise "set FOO_API_KEY" error). Readiness still requires a true
+    # availability probe — keyed (is_available) OR keyless-capable
+    # (is_keyless_available; the Exa/Parallel anonymous free tier serves
+    # zero-credential installs, so those count as ready). Discovery must run
+    # first — check_fn fires at tool-registration time, before any dispatch
+    # has populated the registry.
     try:
         _ensure_web_plugins_loaded()
         from agent.web_search_registry import (
@@ -1194,14 +1241,13 @@ def check_web_api_key() -> bool:
         )
 
         return (
-            get_active_search_provider() is not None
-            or get_active_extract_provider() is not None
+            _provider_is_ready(get_active_search_provider())
+            or _provider_is_ready(get_active_extract_provider())
         )
     except Exception as exc:  # noqa: BLE001 — registry optional; never fatal
         logger.debug("web provider registry availability check failed: %s", exc)
         return False
 
-
 if __name__ == "__main__":
     """
     Simple test/demo when run directly
@@ -1224,7 +1270,10 @@ if __name__ == "__main__":
         elif backend == "parallel":
             print("   Using Parallel API (https://parallel.ai)")
         elif backend == "tavily":
-            print("   Using Tavily API (https://tavily.com)")
+            if _has_env("TAVILY_API_KEY"):
+                print("   Using Tavily API (https://tavily.com)")
+            else:
+                print("   Using Tavily keyless (https://docs.tavily.com/documentation/keyless)")
         elif backend == "searxng":
             print(f"   Using SearXNG (search only): {_env_value('SEARXNG_URL')}")
         elif backend == "brave-free":

website/docs/integrations/index.md

@@ -34,7 +34,7 @@ The `web_search` and `web_extract` tools support eight backend providers, config
 | **SearXNG** | `SEARXNG_URL` | ✔ | — | — |
 | **Brave** (free tier) | `BRAVE_SEARCH_API_KEY` | ✔ | — | — |
 | **DuckDuckGo** (ddgs) | _(none)_ | ✔ | — | — |
-| **Tavily** | `TAVILY_API_KEY` | ✔ | ✔ | ✔ |
+| **Tavily** | `TAVILY_API_KEY` (optional) | ✔ | ✔ | — |
 | **Exa** | `EXA_API_KEY` | ✔ | ✔ | — |
 | **Parallel** | `PARALLEL_API_KEY` | ✔ | ✔ | — |
 | **xAI** | `XAI_API_KEY` | ✔ | — | — |
@@ -46,7 +46,7 @@ web:
   backend: firecrawl    # firecrawl | searxng | brave-free | ddgs | tavily | exa | parallel | xai
 ```
 
-If `web.backend` is not set, the backend is auto-detected from whichever API key is available. Self-hosted Firecrawl is also supported via `FIRECRAWL_API_URL`.
+If `web.backend` is not set, the backend is auto-detected from whichever API key is available. Self-hosted Firecrawl is also supported via `FIRECRAWL_API_URL`. Selecting Tavily in `hermes tools` works without a key.
 
 ## Browser Automation
 

website/docs/reference/environment-variables.md

@@ -140,7 +140,7 @@ For native Anthropic auth, Hermes prefers Claude Code's own credential files whe
 | `PARALLEL_API_KEY` | AI-native web search ([parallel.ai](https://parallel.ai/)) |
 | `FIRECRAWL_API_KEY` | Web scraping and cloud browser ([firecrawl.dev](https://firecrawl.dev/)) |
 | `FIRECRAWL_API_URL` | Custom Firecrawl API endpoint for self-hosted instances (optional) |
-| `TAVILY_API_KEY` | Tavily API key for AI-native web search, extract, and crawl ([app.tavily.com](https://app.tavily.com/home)) |
+| `TAVILY_API_KEY` | Optional Tavily API key for higher search/extract limits. After selecting Tavily as the web backend, keyless access works without it ([app.tavily.com](https://app.tavily.com/home), [keyless docs](https://docs.tavily.com/documentation/keyless)) |
 | `SEARXNG_URL` | SearXNG instance URL for free self-hosted web search — no API key required ([searxng.github.io](https://searxng.github.io/searxng/)) |
 | `TAVILY_BASE_URL` | Override the Tavily API endpoint. Useful for corporate proxies and self-hosted Tavily-compatible search backends. Same pattern as `GROQ_BASE_URL`. |
 | `EXA_API_KEY` | Exa API key for AI-native web search and contents ([exa.ai](https://exa.ai/)) |

website/docs/user-guide/configuration.md

@@ -2289,10 +2289,10 @@ web:
 | **Firecrawl** (default) | `FIRECRAWL_API_KEY` | ✔ | ✔ |
 | **SearXNG** | `SEARXNG_URL` | ✔ | — |
 | **Parallel** | `PARALLEL_API_KEY` (optional — keyless free tier) | ✔ | ✔ |
-| **Tavily** | `TAVILY_API_KEY` | ✔ | ✔ |
+| **Tavily** | `TAVILY_API_KEY` (optional — keyless when selected) | ✔ | ✔ |
 | **Exa** | `EXA_API_KEY` (optional — keyless free tier) | ✔ | ✔ |
 
-**Backend selection:** The runtime always uses the stored `web.backend` selection (set via `hermes tools`; `nous` routes through the managed Tool Gateway). Only if no web backend has ever been selected is one auto-detected from available API keys: if only `SEARXNG_URL` is set, SearXNG is used; if only `EXA_API_KEY` is set, Exa; if only `TAVILY_API_KEY` is set, Tavily; if only `PARALLEL_API_KEY` is set, Parallel. With **no selection and no credentials at all**, Hermes falls back to the Exa/Parallel keyless free tier (unpinned installs split 50/50 between the vendors) so web tools work on a fresh install — see the [Web Search guide](/user-guide/features/web-search) for details and limits. Once a selection exists, adding a key to `.env` does not change the route.
+**Backend selection:** The runtime always uses the stored `web.backend` selection (set via `hermes tools`; `nous` routes through the managed Tool Gateway). Only if no web backend has ever been selected is one auto-detected from available API keys: if only `SEARXNG_URL` is set, SearXNG is used; if only `EXA_API_KEY` is set, Exa; if only `TAVILY_API_KEY` is set, Tavily; if only `PARALLEL_API_KEY` is set, Parallel; if only `KEENABLE_API_KEY` is set, Keenable. With **no selection and no credentials at all**, requests rotate round-robin across the keyless free-tier ring (Exa / Parallel / Tavily / Firecrawl / Keenable) with automatic next-in-line failover on rate limits — see the [Web Search guide](/user-guide/features/web-search) for details. Once a selection exists, adding a key to `.env` does not change the route. Selecting Tavily, Firecrawl, or Keenable in `hermes tools` also works without a key.
 
 **SearXNG** is a free, self-hosted, privacy-respecting metasearch engine that queries 70+ search engines. No API key needed — just set `SEARXNG_URL` to your instance (e.g., `http://localhost:8080`). SearXNG is search-only; `web_extract` requires a separate extract provider (set `web.extract_backend`). See the [Web Search setup guide](/user-guide/features/web-search) for Docker setup instructions.
 

website/docs/user-guide/features/web-search.md

@@ -18,24 +18,25 @@ Both are configured through a single backend selection. Providers are chosen via
 
 | Provider | Env Var | Search | Extract | Free tier |
 |----------|---------|--------|---------|-----------|
-| **Firecrawl** (default) | `FIRECRAWL_API_KEY` | ✔ | ✔ | 500 credits/mo |
+| **Firecrawl** (default) | `FIRECRAWL_API_KEY` (optional — keyless when selected) | ✔ | ✔ | 500 credits/mo · keyless cloud when selected |
 | **SearXNG** | `SEARXNG_URL` | ✔ | — | ✔ Free (self-hosted) |
 | **Brave Search (free tier)** | `BRAVE_SEARCH_API_KEY` | ✔ | — | 2 000 queries/mo |
 | **DDGS (DuckDuckGo)** | — (no key) | ✔ | — | ✔ Free |
-| **Tavily** | `TAVILY_API_KEY` | ✔ | ✔ | 1 000 searches/mo |
-| **Exa** | `EXA_API_KEY` (optional) | ✔ | ✔ | ✔ Keyless free tier · 1 000 searches/mo with key |
-| **Parallel** | `PARALLEL_API_KEY` (optional) | ✔ | ✔ | ✔ Keyless free tier · paid with key |
+| **Tavily** | `TAVILY_API_KEY` (optional) | ✔ | ✔ | ✔ Keyless ring member · 1 000 searches/mo with a free key |
+| **Exa** | `EXA_API_KEY` (optional) | ✔ | ✔ | ✔ Keyless ring member · 1 000 searches/mo with key |
+| **Parallel** | `PARALLEL_API_KEY` (optional) | ✔ | ✔ | ✔ Keyless ring member · paid with key |
+| **Keenable** | `KEENABLE_API_KEY` (optional) | ✔ | ✔ | ✔ Keyless ring member · paid with key |
 | **xAI (Grok)** | `XAI_API_KEY` or `hermes auth add xai-oauth` | ✔ | — | Paid (SuperGrok or per-token) |
 
 Brave Search, DDGS, and xAI are **search-only** — pair any of them with Firecrawl/Tavily/Exa/Parallel when you also need `web_extract`. DDGS uses the [`ddgs` Python package](https://pypi.org/project/ddgs/) under the hood; if it isn't already installed, run `pip install ddgs` (or let Hermes lazy-install it on first use). xAI runs Grok's server-side `web_search` tool on the Responses API — results are LLM-generated rather than index-backed, so titles, descriptions, and URL choice are all model output (see the [trust-model caveat](#xai-grok) below).
 
 **Per-capability split:** you can use different providers for search and extract independently — for example SearXNG (free) for search and Firecrawl for extract. See [Per-capability configuration](#per-capability-configuration) below.
 
-:::info Works out of the box — keyless free tier
-A fresh install with **no web credentials at all** still gets working `web_search` and `web_extract`: Hermes falls back to Exa's and Parallel's public anonymous endpoints (rate-limited free tiers), splitting unpinned installs 50/50 between the two vendors — the pick is random per process and stable within it. No signup, no key. This tier is strictly last-resort — any configured backend or present API key always wins — and requests carry no user identifiers (only a random per-process session id, rotated on restart). For reliable, unthrottled service, set up a keyed provider. Disable the keyless tier entirely with `web.keyless_fallback: false`.
+:::info Works out of the box — keyless free-tier rotation
+A fresh install with **no web credentials at all** gets working `web_search` and `web_extract` out of the box: requests rotate round-robin across five vendors' public free tiers — **Exa, Parallel, Tavily, Firecrawl, and Keenable** — spreading load evenly, and a rate-limited request automatically retries on the next vendor in the ring (multi-hop, until one serves or all are throttled). No signup, no key. This tier is strictly last-resort — any configured backend or present API key always wins — and requests carry no user identifiers (only a random per-process session id, rotated on restart). For guaranteed, unthrottled service, set up a keyed provider. Disable the keyless tier entirely with `web.keyless_fallback: false`.
 :::
 
-**Choosing free vs paid explicitly:** in `hermes tools`, Exa and Parallel each appear as two rows — **Free (keyless)** and **Paid (API key)**. Picking Free pins the anonymous endpoint (even if you later add a key); picking Paid pins the keyed SDK path (a missing key then errors instead of silently downgrading to the free tier). The selection is stored as `web.provider_tier.<name>: free|paid`; leave it unset for auto (key present → paid, otherwise free).
+**Choosing free vs paid explicitly:** in `hermes tools`, Exa, Parallel, and Keenable each appear as two rows — **Free (keyless)** and **Paid (API key)**. Picking Free pins that vendor's anonymous endpoint (even if you later add a key); picking Paid pins the keyed path (a missing key then errors instead of silently downgrading to the free tier). The selection is stored as `web.provider_tier.<name>: free|paid`; leave it unset for auto (key present → paid, otherwise the keyless ring).
 
 :::tip Nous Subscribers
 If you have a paid [Nous Portal](https://portal.nousresearch.com) subscription, web search and extract are available through the **[Tool Gateway](tool-gateway.md)** via managed Firecrawl — no API key needed. New installs can run `hermes setup --portal` to log in and turn on all gateway tools at once; existing installs can flip just web via `hermes tools`.
@@ -239,14 +240,17 @@ With this config, Hermes uses SearXNG for all search queries and Firecrawl for U
 
 ### Tavily
 
-AI-optimised search and extract with a generous free tier.
+AI-optimised search and extract. Select Tavily in `hermes tools` (or set `web.backend: tavily`) to use it **keyless** with no account (rate-limited). Set an API key when you want higher limits.
 
 ```bash
+# optional — skip this for keyless access after selecting Tavily
 # ~/.hermes/.env
 TAVILY_API_KEY=tvly-your-key-here
 ```
 
-Get a key at [app.tavily.com](https://app.tavily.com/home). The free tier includes 1 000 searches/month.
+Get a key at [app.tavily.com](https://app.tavily.com/home). See [Tavily keyless](https://docs.tavily.com/documentation/keyless).
+
+Empty installs keep Firecrawl as the named default. Keyless Tavily is not auto-selected.
 
 ---
 
@@ -366,9 +370,9 @@ If no backend has **ever** been selected (no `web.backend` / per-capability key
 | `SEARXNG_URL` | searxng |
 | `BRAVE_SEARCH_API_KEY` | brave-free |
 | `ddgs` package importable | ddgs |
-| *(nothing set at all)* | exa / parallel keyless free tier (50/50 split) |
+| *(nothing set at all)* | keyless ring: exa / parallel / tavily / firecrawl / keenable (round-robin) |
 
-**Keyless free tier:** when *no* credential above is present, Hermes falls back to Exa's and Parallel's public anonymous endpoints so web tools work on a fresh install with zero setup — unpinned installs split 50/50 between the two vendors (random per process, stable within it); pick one explicitly in `hermes tools` to pin it. Both free tiers are rate-limited by the vendors under burst load; in practice sustained normal usage goes through fine. On throttling, the tool returns an error suggesting the matching API key. Set `web.keyless_fallback: false` to turn this tier off — with it off and no credentials, web tools are unavailable until a provider is configured.
+**Keyless free-tier ring:** when *no* credential above is present, requests rotate across five vendors' public free tiers (Exa, Parallel, Tavily, Firecrawl, Keenable) so web tools work on a fresh install with zero setup — and a rate-limited request fails over to the next vendor in the ring automatically. Pin one vendor in `hermes tools` to stop the rotation (the ring is then only used as failover succession on throttles). All free tiers are vendor-rate-limited under burst load; sustained normal usage goes through fine. Set `web.keyless_fallback: false` to turn the tier off — with it off and no credentials, web tools are unavailable until a provider is configured.
 
 xAI Web Search is **not** in the auto-detection chain — having `XAI_API_KEY` set (or being signed in via xAI Grok OAuth) does not automatically route web traffic through xAI, since those credentials are also used for inference / TTS / image gen and the user may want a different backend for web. Opt in explicitly with `web.backend: "xai"`.