Merge pull request #90313 from NousResearch/feat/keyless-web-search-fallback

feat: web search works keyless on fresh installs (Parallel + Exa free tiers)

agent/web_search_provider.py

@@ -126,6 +126,22 @@ class WebSearchProvider(abc.ABC):
         """Return True if this provider implements :meth:`search`."""
         return True
 
+    def is_keyless_available(self) -> bool:
+        """Return True when this provider can serve calls WITHOUT credentials.
+
+        A separate, weaker tier than :meth:`is_available`: providers with a
+        public anonymous free tier (Exa / Parallel MCP endpoints) return
+        True here so the registry can fall back to them when NO provider is
+        configured or keyed — and only then. Keyless availability must never
+        make :meth:`is_available` return True, or the legacy preference walk
+        would route users with real credentials for a lower-priority backend
+        onto the free tier of a higher-priority one.
+
+        Like :meth:`is_available`, this must be cheap and must NOT make
+        network calls. Default: False.
+        """
+        return False
+
     def supports_extract(self) -> bool:
         """Return True if this provider implements :meth:`extract`.
 

agent/web_search_registry.py

@@ -166,6 +166,40 @@ _LEGACY_PREFERENCE = (
     "ddgs",
 )
 
+# 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.
+# Disable the tier with ``web.keyless_fallback: false``.
+_KEYLESS_PREFERENCE = (
+    "exa",
+    "parallel",
+)
+
+
+def _keyless_preference() -> tuple:
+    """Return the keyless walk order, split 50/50 per process.
+
+    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.
+    """
+    try:
+        from plugins.web.keyless_mcp import _SESSION_ID
+
+        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)
+    return _KEYLESS_PREFERENCE
+
 
 def _resolve(configured: Optional[str], *, capability: str) -> Optional[WebSearchProvider]:
     """Resolve the active provider for a capability ("search" | "extract").
@@ -254,9 +288,39 @@ def _resolve(configured: Optional[str], *, capability: str) -> Optional[WebSearc
         ):
             return provider
 
+    # 4. Keyless free-tier walk — the user has NO credentialed/importable
+    #    backend at all. Fall back to providers that can serve anonymously
+    #    (public MCP free tiers), unless disabled via
+    #    ``web.keyless_fallback: false``. This tier never pre-empts a keyed
+    #    setup: it is only reachable when the legacy walk found nothing.
+    if _keyless_tier_enabled():
+        for name in _keyless_preference():
+            provider = snapshot.get(name)
+            if provider is None or not _capable(provider):
+                continue
+            try:
+                if provider.is_keyless_available():
+                    return provider
+            except Exception as exc:  # noqa: BLE001 — buggy provider skipped
+                logger.debug(
+                    "provider %s.is_keyless_available() raised %s", name, exc
+                )
+
     return None
 
 
+def _keyless_tier_enabled() -> bool:
+    """Read ``web.keyless_fallback`` from config.yaml (default: enabled)."""
+    try:
+        from hermes_cli.config import load_config
+
+        web_cfg = load_config().get("web") or {}
+        return bool(web_cfg.get("keyless_fallback", True))
+    except Exception as exc:  # noqa: BLE001 — config layer optional
+        logger.debug("keyless_fallback config read failed: %s", exc)
+        return True
+
+
 def _disabled_web_plugin_for(configured: Optional[str] = None, *, capability: Optional[str] = None) -> Optional[str]:
     """Return the plugin key of a *disabled* bundled web plugin that would
     have provided the configured backend, or None.

hermes_cli/config_defaults.py

@@ -492,6 +492,18 @@ 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_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.
+        #   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
+        "provider_tier": {},
     },
 
     "browser": {

hermes_cli/tools_config.py

@@ -3129,18 +3129,29 @@ def _plugin_web_search_providers() -> list[dict]:
             continue
         if not isinstance(schema, dict):
             continue
-        row = {
-            "name": schema.get("name", provider.display_name),
-            "badge": schema.get("badge", ""),
-            "tag": schema.get("tag", ""),
-            "env_vars": schema.get("env_vars", []),
-            "web_backend": name,
-            "web_search_plugin_name": name,
-        }
-        # Optional pass-through fields the schema can opt into.
-        if schema.get("post_setup"):
-            row["post_setup"] = schema["post_setup"]
-        rows.append(row)
+        # A schema may expose tier ``variants`` (e.g. Exa/Parallel free
+        # keyless endpoint vs paid SDK) — flatten the base row plus each
+        # variant into separate picker rows sharing the same backend name,
+        # distinguished by ``web_tier`` (persisted to
+        # ``web.provider_tier.<name>`` on selection).
+        schemas = [schema] + [
+            v for v in (schema.get("variants") or []) if isinstance(v, dict)
+        ]
+        for entry in schemas:
+            row = {
+                "name": entry.get("name", provider.display_name),
+                "badge": entry.get("badge", ""),
+                "tag": entry.get("tag", ""),
+                "env_vars": entry.get("env_vars", []),
+                "web_backend": name,
+                "web_search_plugin_name": name,
+            }
+            if entry.get("web_tier"):
+                row["web_tier"] = entry["web_tier"]
+            # Optional pass-through fields the schema can opt into.
+            if entry.get("post_setup"):
+                row["post_setup"] = entry["post_setup"]
+            rows.append(row)
     return rows
 
 
@@ -3788,6 +3799,44 @@ def _configure_tool_category(
         _configure_provider(providers[provider_idx], config, force_fresh=force_fresh)
 
 
+def _web_tier_matches(provider: dict, config: dict) -> bool:
+    """Return True when a web picker row's tier matches the configured tier.
+
+    Tiered rows (Exa/Parallel Free vs Paid) share one ``web_backend`` name
+    and differ only in ``web_tier``. The configured tier lives at
+    ``web.provider_tier.<backend>`` (set on selection). Matching rules:
+
+    - row has no ``web_tier`` → tier-agnostic row, matches (legacy rows)
+    - configured tier set     → must equal the row's tier
+    - configured tier unset   → "auto": the effective tier is paid when the
+      row's env vars are all present, free otherwise — highlight the row
+      the runtime would actually use
+    """
+    row_tier = provider.get("web_tier")
+    if not row_tier:
+        return True
+    web_cfg = config.get("web")
+    if not isinstance(web_cfg, dict):
+        web_cfg = {}
+    tiers = web_cfg.get("provider_tier")
+    if not isinstance(tiers, dict):
+        tiers = {}
+    configured = str(tiers.get(provider["web_backend"], "") or "").lower().strip()
+    if configured in ("free", "paid"):
+        return configured == row_tier
+    # Auto: mirror plugins.web.keyless_mcp.use_keyless — key present → paid.
+    try:
+        from agent.web_search_provider import get_provider_env
+
+        key_var = {"exa": "EXA_API_KEY", "parallel": "PARALLEL_API_KEY"}.get(
+            provider["web_backend"]
+        )
+        has_key = bool(get_provider_env(key_var)) if key_var else False
+    except Exception:
+        has_key = False
+    return row_tier == ("paid" if has_key else "free")
+
+
 def _is_provider_active(
     provider: dict,
     config: dict,
@@ -3868,10 +3917,11 @@ def _is_provider_active(
             }
         if provider.get("web_backend"):
             current = cfg_get(config, "web", "backend")
-            return feature.managed_by_nous and current in {
-                provider["web_backend"],
-                NOUS_MANAGED_PROVIDER,
-            }
+            return (
+                feature.managed_by_nous
+                and current in {provider["web_backend"], NOUS_MANAGED_PROVIDER}
+                and _web_tier_matches(provider, config)
+            )
         return feature.managed_by_nous
 
     if provider.get("tts_provider"):
@@ -3919,7 +3969,9 @@ def _is_provider_active(
             return False
     if provider.get("web_backend"):
         current = cfg_get(config, "web", "backend")
-        return current == provider["web_backend"]
+        if current != provider["web_backend"]:
+            return False
+        return _web_tier_matches(provider, config)
     if provider.get("computer_use_backend"):
         current = cfg_get(config, "computer_use", "backend")
         return current == provider["computer_use_backend"]
@@ -4406,6 +4458,16 @@ def _write_provider_config(provider: dict, config: dict, *, managed_feature) ->
     # Set web search backend in config if applicable
     if provider.get("web_backend"):
         _set_selection("web", "backend", provider["web_backend"])
+        web_cfg = config.get("web")
+        if isinstance(web_cfg, dict):
+            if provider.get("web_tier"):
+                tiers = web_cfg.setdefault("provider_tier", {})
+                if isinstance(tiers, dict):
+                    tiers[provider["web_backend"]] = provider["web_tier"]
+            else:
+                stale_tiers = web_cfg.get("provider_tier")
+                if isinstance(stale_tiers, dict):
+                    stale_tiers.pop(provider["web_backend"], None)
 
     # Set computer_use backend in config if applicable
     if provider.get("computer_use_backend"):
@@ -5132,7 +5194,19 @@ def _reconfigure_provider(
             NOUS_MANAGED_PROVIDER if managed_feature else provider["web_backend"]
         )
         web_cfg.pop("use_gateway", None)
-        _print_success(f"  Web backend set to: {provider['web_backend']}")
+        if provider.get("web_tier"):
+            tiers = web_cfg.setdefault("provider_tier", {})
+            if isinstance(tiers, dict):
+                tiers[provider["web_backend"]] = provider["web_tier"]
+            _print_success(
+                f"  Web backend set to: {provider['web_backend']} "
+                f"({provider['web_tier']} tier)"
+            )
+        else:
+            stale_tiers = web_cfg.get("provider_tier")
+            if isinstance(stale_tiers, dict):
+                stale_tiers.pop(provider["web_backend"], None)
+            _print_success(f"  Web backend set to: {provider['web_backend']}")
 
     # Set computer_use backend in config if applicable
     if provider.get("computer_use_backend"):

plugins/web/exa/provider.py

@@ -101,11 +101,27 @@ class ExaWebSearchProvider(WebSearchProvider):
         return "Exa"
 
     def is_available(self) -> bool:
-        """Return True when ``EXA_API_KEY`` is set to a non-empty value."""
+        """Return True when ``EXA_API_KEY`` is set to a non-empty value.
+
+        Deliberately does NOT consider the keyless free tier — that would
+        let the legacy preference walk route keyed users of lower-priority
+        backends onto Exa's anonymous tier. Keyless availability is a
+        separate, last-resort signal (:meth:`is_keyless_available`).
+        """
         from agent.web_search_provider import get_provider_env
 
         return bool(get_provider_env("EXA_API_KEY"))
 
+    def is_keyless_available(self) -> bool:
+        """Exa serves anonymous free-tier calls via its public MCP endpoint.
+
+        False when the user forced ``web.provider_tier.exa: paid`` — an
+        explicit paid selection must never silently resolve keyless.
+        """
+        from plugins.web.keyless_mcp import keyless_enabled, provider_tier
+
+        return keyless_enabled() and provider_tier("exa") != "paid"
+
     def supports_search(self) -> bool:
         return True
 
@@ -125,6 +141,17 @@ class ExaWebSearchProvider(WebSearchProvider):
             if is_interrupted():
                 return {"success": False, "error": "Interrupted"}
 
+            from agent.web_search_provider import get_provider_env
+
+            from plugins.web.keyless_mcp import exa_search_keyless, 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)
+
             logger.info("Exa search: '%s' (limit=%d)", query, limit)
             response = _get_exa_client().search(
                 query,
@@ -169,6 +196,15 @@ class ExaWebSearchProvider(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 exa_extract_keyless, 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))
+
             logger.info("Exa extract: %d URL(s)", len(urls))
             response = _get_exa_client().get_contents(urls, text=True)
 
@@ -203,14 +239,30 @@ class ExaWebSearchProvider(WebSearchProvider):
 
     def get_setup_schema(self) -> Dict[str, Any]:
         return {
-            "name": "Exa",
-            "badge": "paid",
-            "tag": "Semantic + neural web search with content extraction.",
-            "env_vars": [
+            "name": "Exa · Free (keyless)",
+            "badge": "free · no key",
+            "tag": (
+                "Semantic + neural web search with content extraction on "
+                "Exa's anonymous free tier. Rate-limited under burst load."
+            ),
+            "env_vars": [],
+            "web_tier": "free",
+            "variants": [
                 {
-                    "key": "EXA_API_KEY",
-                    "prompt": "Exa API key",
-                    "url": "https://exa.ai",
+                    "name": "Exa · Paid (API key)",
+                    "badge": "paid",
+                    "tag": (
+                        "Semantic + neural web search with content extraction "
+                        "via the Exa SDK. Unthrottled, guaranteed service."
+                    ),
+                    "env_vars": [
+                        {
+                            "key": "EXA_API_KEY",
+                            "prompt": "Exa API key",
+                            "url": "https://exa.ai",
+                        },
+                    ],
+                    "web_tier": "paid",
                 },
             ],
         }

plugins/web/keyless_mcp.py

@@ -0,0 +1,420 @@
+"""Keyless web search/extract via public MCP endpoints.
+
+Exa and Parallel both operate public, anonymous MCP endpoints with a free
+tier (the same endpoints the opencode CLI ships as its default search
+path):
+
+- Exa:      https://mcp.exa.ai/mcp           (tools: web_search_exa, web_fetch_exa)
+- Parallel: https://search.parallel.ai/mcp   (tools: web_search, web_fetch)
+
+This module implements a minimal JSON-RPC ``tools/call`` client for those
+two endpoints so a fresh Hermes install with **zero web credentials** still
+gets working ``web_search`` / ``web_extract`` tools. The keyless tier is
+resolved strictly LAST — after every keyed backend, the managed tool
+gateway, ddgs, and custom plugin providers — so it never pre-empts a
+deliberate setup (see ``tools.web_tools._get_backend`` and the registry's
+``_KEYLESS_PREFERENCE`` walk).
+
+Privacy: requests carry no user identifiers. Parallel's free tier asks for
+a ``session_id`` used for rate limiting; we send a random per-process UUID
+(rotates every restart, never persisted). Their optional ``model_name``
+analytics field is deliberately omitted.
+
+Disable the whole tier with ``web.keyless_fallback: false`` in config.yaml.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import uuid
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+EXA_MCP_URL = "https://mcp.exa.ai/mcp"
+PARALLEL_MCP_URL = "https://search.parallel.ai/mcp"
+
+# Free-tier rate-limit correlation id for Parallel — random per process,
+# never persisted, not derived from any user/machine identifier.
+_SESSION_ID = uuid.uuid4().hex
+
+_TIMEOUT_SECONDS = 30
+
+
+class KeylessMCPError(RuntimeError):
+    """A keyless MCP call failed (transport, rate limit, or tool error)."""
+
+
+def keyless_enabled() -> bool:
+    """Return True when the keyless fallback tier is enabled.
+
+    Delegates to :func:`agent.web_search_registry._keyless_tier_enabled` so
+    the config chokepoint (``web.keyless_fallback``, default on) lives in
+    one place alongside the rest of backend resolution.
+    """
+    try:
+        from agent.web_search_registry import _keyless_tier_enabled
+
+        return _keyless_tier_enabled()
+    except Exception as exc:  # noqa: BLE001 — resolver optional in stripped envs
+        logger.debug("keyless_enabled(): registry helper unavailable: %s", exc)
+        return True
+
+
+def provider_tier(name: str) -> str:
+    """Return the user-selected tier for *name*: ``free``, ``paid``, or ``auto``.
+
+    Reads ``web.provider_tier.<name>`` from config.yaml (set by the
+    ``hermes tools`` picker's Free/Paid rows). ``free`` forces the keyless
+    public endpoint even when the vendor API key is present; ``paid``
+    forces the keyed SDK path (missing key surfaces the standard
+    "X_API_KEY not set" error instead of silently downgrading to the free
+    tier). Anything else — including unset — is ``auto``: key present →
+    keyed, otherwise keyless when the tier is enabled.
+    """
+    try:
+        from hermes_cli.config import load_config
+
+        web_cfg = load_config().get("web") or {}
+        tiers = web_cfg.get("provider_tier") or {}
+        value = str(tiers.get(name, "") or "").lower().strip()
+        return value if value in ("free", "paid") else "auto"
+    except Exception as exc:  # noqa: BLE001 — config layer optional
+        logger.debug("provider_tier(%r) config read failed: %s", name, exc)
+        return "auto"
+
+
+def use_keyless(name: str, api_key: str) -> bool:
+    """Decide whether provider *name* should route via the keyless endpoint.
+
+    Single chokepoint shared by the Exa/Parallel search + extract paths so
+    tier semantics can't drift between capabilities:
+
+    - tier ``free``  → keyless, even when *api_key* is set
+    - tier ``paid``  → keyed, even when *api_key* is missing (the keyed
+      path then raises its usual missing-key error)
+    - tier ``auto``  → keyed when *api_key* is set; otherwise keyless when
+      ``web.keyless_fallback`` is enabled
+    """
+    tier = provider_tier(name)
+    if tier == "free":
+        return True
+    if tier == "paid":
+        return False
+    return not api_key and keyless_enabled()
+
+
+def _parse_mcp_body(body: str) -> str:
+    """Extract the first text content item from an MCP tools/call response.
+
+    Handles both plain-JSON bodies and SSE (``data: {...}`` lines) — the
+    Exa endpoint answers as an event stream, Parallel as direct JSON.
+    Raises :class:`KeylessMCPError` for JSON-RPC errors and ``isError``
+    tool results (e.g. Exa's free-tier rate-limit message).
+    """
+
+    def _from_payload(payload: str) -> Optional[str]:
+        payload = payload.strip()
+        if not payload.startswith("{"):
+            return None
+        data = json.loads(payload)
+        err = data.get("error")
+        if err:
+            raise KeylessMCPError(str(err.get("message") or err))
+        result = data.get("result") or {}
+        content = result.get("content") or []
+        if result.get("isError"):
+            texts = [c.get("text", "") for c in content if isinstance(c, dict)]
+            raise KeylessMCPError(
+                " ".join(t for t in texts if t) or "MCP tool call failed"
+            )
+        for item in content:
+            if isinstance(item, dict) and item.get("text"):
+                return str(item["text"])
+        return None
+
+    stripped = body.strip()
+    if stripped.startswith("{"):
+        try:
+            text = _from_payload(stripped)
+            if text is not None:
+                return text
+        except json.JSONDecodeError:
+            pass
+
+    for line in body.splitlines():
+        if not line.startswith("data: "):
+            continue
+        try:
+            text = _from_payload(line[len("data: "):])
+        except json.JSONDecodeError:
+            continue
+        if text is not None:
+            return text
+
+    raise KeylessMCPError("Unrecognized MCP response shape")
+
+
+def mcp_call(
+    url: str,
+    tool: str,
+    arguments: Dict[str, Any],
+    timeout: int = _TIMEOUT_SECONDS,
+) -> str:
+    """POST a JSON-RPC ``tools/call`` to *url* and return the text payload.
+
+    Raises :class:`KeylessMCPError` on transport failures, non-2xx
+    statuses, JSON-RPC errors, and error-shaped tool results.
+    """
+    import requests
+
+    payload = {
+        "jsonrpc": "2.0",
+        "id": 1,
+        "method": "tools/call",
+        "params": {"name": tool, "arguments": arguments},
+    }
+    headers = {
+        "Content-Type": "application/json",
+        "Accept": "application/json, text/event-stream",
+        "User-Agent": "hermes-agent",
+    }
+    try:
+        response = requests.post(url, json=payload, headers=headers, timeout=timeout)
+    except requests.RequestException as exc:
+        raise KeylessMCPError(f"request failed: {exc}") from exc
+    if response.status_code >= 400:
+        raise KeylessMCPError(
+            f"HTTP {response.status_code}: {response.text[:300]}"
+        )
+    return _parse_mcp_body(response.text)
+
+
+# ---------------------------------------------------------------------------
+# Parallel (search.parallel.ai) — JSON text payloads
+# ---------------------------------------------------------------------------
+
+
+def parallel_search_keyless(query: str, limit: int = 5) -> Dict[str, Any]:
+    """Keyless Parallel web search → legacy search response shape."""
+    try:
+        text = mcp_call(
+            PARALLEL_MCP_URL,
+            "web_search",
+            {
+                "objective": query,
+                "search_queries": [query],
+                "session_id": _SESSION_ID,
+            },
+        )
+        data = json.loads(text)
+        web_results = []
+        for i, result in enumerate(data.get("results") or []):
+            if limit and i >= limit:
+                break
+            excerpts = result.get("excerpts") or []
+            web_results.append(
+                {
+                    "url": result.get("url") or "",
+                    "title": result.get("title") or "",
+                    "description": " ".join(excerpts) if excerpts else "",
+                    "position": i + 1,
+                }
+            )
+        return {"success": True, "data": {"web": web_results}}
+    except KeylessMCPError as exc:
+        return {
+            "success": False,
+            "error": (
+                f"Keyless Parallel search failed: {exc}. "
+                "Set PARALLEL_API_KEY (https://parallel.ai) or another web "
+                "backend via `hermes tools` for reliable service."
+            ),
+        }
+    except (json.JSONDecodeError, TypeError, KeyError) as exc:
+        return {"success": False, "error": f"Keyless Parallel search returned an unexpected payload: {exc}"}
+
+
+def parallel_extract_keyless(urls: List[str]) -> List[Dict[str, Any]]:
+    """Keyless Parallel web fetch → legacy extract result list."""
+    try:
+        text = mcp_call(
+            PARALLEL_MCP_URL,
+            "web_fetch",
+            {
+                "urls": list(urls),
+                "objective": "Full page content",
+                "session_id": _SESSION_ID,
+            },
+        )
+        data = json.loads(text)
+    except (KeylessMCPError, json.JSONDecodeError, TypeError) as exc:
+        message = (
+            f"Keyless Parallel extract failed: {exc}. "
+            "Set PARALLEL_API_KEY (https://parallel.ai) 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 ""
+        title = result.get("title") or ""
+        content = (
+            result.get("full_content")
+            or result.get("content")
+            or "\n\n".join(result.get("excerpts") or [])
+        )
+        seen.add(url)
+        results.append(
+            {
+                "url": url,
+                "title": title,
+                "content": content,
+                "raw_content": content,
+                "metadata": {"sourceURL": url, "title": title},
+            }
+        )
+    for error in data.get("errors") or []:
+        url = error.get("url") or ""
+        seen.add(url)
+        results.append(
+            {
+                "url": url,
+                "title": "",
+                "content": "",
+                "error": str(
+                    error.get("content") or error.get("error_type") or "extraction failed"
+                ),
+                "metadata": {"sourceURL": url},
+            }
+        )
+    # Any URL the endpoint silently dropped still gets an error entry so the
+    # caller's per-URL contract holds.
+    for u in urls:
+        if u not in seen:
+            results.append(
+                {"url": u, "title": "", "content": "", "error": "no content returned"}
+            )
+    return results
+
+
+# ---------------------------------------------------------------------------
+# Exa (mcp.exa.ai) — formatted plain-text payloads
+# ---------------------------------------------------------------------------
+
+
+def _parse_exa_search_text(text: str, limit: int) -> List[Dict[str, Any]]:
+    """Parse Exa's formatted search text into result dicts.
+
+    The payload is blocks separated by ``---`` lines, each shaped like::
+
+        Title: <title>
+        URL: <url>
+        Published: ...
+        Author: ...
+        Highlights:
+        <free text>
+    """
+    results: List[Dict[str, Any]] = []
+    for block in text.split("\n---\n"):
+        title = ""
+        url = ""
+        highlight_lines: List[str] = []
+        in_highlights = False
+        for line in block.splitlines():
+            stripped = line.strip()
+            if stripped.startswith("Title:"):
+                title = stripped[len("Title:"):].strip()
+                in_highlights = False
+            elif stripped.startswith("URL:"):
+                url = stripped[len("URL:"):].strip()
+                in_highlights = False
+            elif stripped.startswith("Highlights:"):
+                in_highlights = True
+            elif stripped.startswith(("Published:", "Author:")):
+                in_highlights = False
+            elif in_highlights and stripped:
+                highlight_lines.append(stripped)
+        if url:
+            results.append(
+                {
+                    "url": url,
+                    "title": title,
+                    "description": " ".join(highlight_lines),
+                    "position": len(results) + 1,
+                }
+            )
+        if limit and len(results) >= limit:
+            break
+    return results
+
+
+def exa_search_keyless(query: str, limit: int = 5) -> Dict[str, Any]:
+    """Keyless Exa web search → legacy search response shape."""
+    try:
+        text = mcp_call(
+            EXA_MCP_URL,
+            "web_search_exa",
+            {"query": query, "numResults": max(1, int(limit))},
+        )
+    except KeylessMCPError as exc:
+        return {
+            "success": False,
+            "error": (
+                f"Keyless Exa search failed: {exc}. "
+                "Set EXA_API_KEY (https://exa.ai) or another web backend "
+                "via `hermes tools` for reliable service."
+            ),
+        }
+    return {"success": True, "data": {"web": _parse_exa_search_text(text, limit)}}
+
+
+def exa_extract_keyless(urls: List[str]) -> List[Dict[str, Any]]:
+    """Keyless Exa web fetch → legacy extract result list.
+
+    ``web_fetch_exa`` takes a ``urls`` array but returns one combined text
+    payload; we call it per-URL so each result maps cleanly.
+    """
+    results: List[Dict[str, Any]] = []
+    for url in urls:
+        try:
+            text = mcp_call(EXA_MCP_URL, "web_fetch_exa", {"urls": [url]})
+        except KeylessMCPError as exc:
+            results.append(
+                {
+                    "url": url,
+                    "title": "",
+                    "content": "",
+                    "error": (
+                        f"Keyless Exa extract failed: {exc}. "
+                        "Set EXA_API_KEY (https://exa.ai) or another web "
+                        "backend via `hermes tools` for reliable service."
+                    ),
+                }
+            )
+            continue
+        title = ""
+        for line in text.splitlines():
+            stripped = line.strip()
+            if stripped.startswith("# "):
+                title = stripped[2:].strip()
+                break
+            if stripped.startswith("Title:"):
+                title = stripped[len("Title:"):].strip()
+                break
+        results.append(
+            {
+                "url": url,
+                "title": title,
+                "content": text,
+                "raw_content": text,
+                "metadata": {"sourceURL": url, "title": title},
+            }
+        )
+    return results

plugins/web/parallel/provider.py

@@ -156,11 +156,27 @@ class ParallelWebSearchProvider(WebSearchProvider):
         return "Parallel"
 
     def is_available(self) -> bool:
-        """Return True when ``PARALLEL_API_KEY`` is set to a non-empty value."""
+        """Return True when ``PARALLEL_API_KEY`` is set to a non-empty value.
+
+        Deliberately does NOT consider the keyless free tier — that would
+        let the legacy preference walk route keyed users of lower-priority
+        backends onto Parallel's anonymous tier. Keyless availability is a
+        separate, last-resort signal (:meth:`is_keyless_available`).
+        """
         from agent.web_search_provider import get_provider_env
 
         return bool(get_provider_env("PARALLEL_API_KEY"))
 
+    def is_keyless_available(self) -> bool:
+        """Parallel serves anonymous free-tier calls via its public MCP endpoint.
+
+        False when the user forced ``web.provider_tier.parallel: paid`` —
+        an explicit paid selection must never silently resolve keyless.
+        """
+        from plugins.web.keyless_mcp import keyless_enabled, provider_tier
+
+        return keyless_enabled() and provider_tier("parallel") != "paid"
+
     def supports_search(self) -> bool:
         return True
 
@@ -180,6 +196,17 @@ class ParallelWebSearchProvider(WebSearchProvider):
             if is_interrupted():
                 return {"success": False, "error": "Interrupted"}
 
+            from agent.web_search_provider import get_provider_env
+
+            from plugins.web.keyless_mcp import parallel_search_keyless, 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)
+
             mode = _resolve_search_mode()
             logger.info(
                 "Parallel search: '%s' (mode=%s, limit=%d)", query, mode, limit
@@ -233,6 +260,19 @@ class ParallelWebSearchProvider(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 parallel_extract_keyless, use_keyless
+
+            if use_keyless("parallel", get_provider_env("PARALLEL_API_KEY")):
+                # Keyless free tier — blocking HTTP, so hop off the loop.
+                import asyncio
+
+                logger.info("Parallel keyless extract: %d URL(s)", len(urls))
+                return await asyncio.to_thread(
+                    parallel_extract_keyless, list(urls)
+                )
+
             logger.info("Parallel extract: %d URL(s)", len(urls))
             response = await _get_async_client().beta.extract(
                 urls=urls,
@@ -284,14 +324,30 @@ class ParallelWebSearchProvider(WebSearchProvider):
 
     def get_setup_schema(self) -> Dict[str, Any]:
         return {
-            "name": "Parallel",
-            "badge": "paid",
-            "tag": "Objective-tuned search + parallel page extraction.",
-            "env_vars": [
+            "name": "Parallel · Free (keyless)",
+            "badge": "free · no key",
+            "tag": (
+                "Objective-tuned search + page extraction on Parallel's "
+                "anonymous free tier. Rate-limited under burst load."
+            ),
+            "env_vars": [],
+            "web_tier": "free",
+            "variants": [
                 {
-                    "key": "PARALLEL_API_KEY",
-                    "prompt": "Parallel API key",
-                    "url": "https://parallel.ai",
+                    "name": "Parallel · Paid (API key)",
+                    "badge": "paid",
+                    "tag": (
+                        "Objective-tuned search + parallel page extraction "
+                        "via the Parallel SDK. Unthrottled, guaranteed service."
+                    ),
+                    "env_vars": [
+                        {
+                            "key": "PARALLEL_API_KEY",
+                            "prompt": "Parallel API key",
+                            "url": "https://parallel.ai",
+                        },
+                    ],
+                    "web_tier": "paid",
                 },
             ],
         }

tests/plugins/web/test_web_search_provider_plugins.py

@@ -278,20 +278,21 @@ class TestRegistryResolution:
     def test_no_config_no_credentials_returns_none(
         self,
     ) -> None:
-        """No backend configured AND no available providers → typically None.
+        """No backend configured AND no credentials → keyless tier or ddgs.
 
-        ``ddgs`` is the no-credential fallback; if its ``ddgs`` Python
-        package is installed in the test env, ddgs will be picked.
-        Otherwise the resolver returns None. Either outcome is correct.
+        Resolution order with zero credentials: ddgs if its Python package
+        is importable, else the keyless free tier (Parallel/Exa public
+        endpoints — resolves with ``is_available() == False`` but
+        ``is_keyless_available() == True``), else None (keyless tier
+        disabled). All three outcomes are correct; a provider that is
+        neither keyed nor keyless-capable means an env var leaked in.
         """
         _ensure_plugins_loaded()
         from agent.web_search_registry import _resolve
 
         result = _resolve(None, capability="search")
         if result is not None:
-            # The only no-credential provider is ddgs; anything else
-            # means an env var leaked in.
-            assert result.is_available() is True
+            assert result.is_available() or result.is_keyless_available()
 
 
 # ---------------------------------------------------------------------------

tests/tools/test_web_keyless_fallback.py

@@ -0,0 +1,416 @@
+"""Keyless free-tier web search/extract fallback (Parallel + Exa MCP).
+
+Covers:
+- keyless_mcp response parsing (SSE + plain JSON, error shapes)
+- provider keyless routing: no key -> keyless path; key present -> SDK path
+- registry keyless walk: fires only when nothing is keyed; respects
+  web.keyless_fallback: false
+- _get_backend() keyless tier: strictly after every keyed candidate
+- check_web_api_key() lights up on a zero-credential install
+"""
+
+import json
+from unittest.mock import patch
+
+import pytest
+
+import tools.web_tools as web_tools
+from agent import web_search_registry as registry
+from plugins.web import keyless_mcp
+from plugins.web.exa.provider import ExaWebSearchProvider
+from plugins.web.parallel.provider import ParallelWebSearchProvider
+
+
+@pytest.fixture(autouse=True)
+def _no_web_env(monkeypatch):
+    """Blank every web credential and neutralize config lookups."""
+    for var in (
+        "EXA_API_KEY", "PARALLEL_API_KEY", "TAVILY_API_KEY",
+        "FIRECRAWL_API_KEY", "FIRECRAWL_API_URL", "BRAVE_SEARCH_API_KEY",
+        "SEARXNG_URL", "TOOL_GATEWAY_USER_TOKEN",
+    ):
+        monkeypatch.delenv(var, raising=False)
+    monkeypatch.setattr(
+        "agent.web_search_provider.get_provider_env", lambda name: "", raising=True
+    )
+    monkeypatch.setattr(web_tools, "_env_value", lambda name: "", raising=True)
+    monkeypatch.setattr(web_tools, "_load_web_config", dict, raising=True)
+    monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False, raising=True)
+    monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: False, raising=True)
+    yield
+
+
+@pytest.fixture()
+def fresh_registry():
+    """Isolated registry snapshot with real exa/parallel providers."""
+    with registry._lock:
+        saved = dict(registry._providers)
+        saved_scoped = {k: dict(v) for k, v in registry._scoped_providers.items()}
+        registry._providers.clear()
+        registry._scoped_providers.clear()
+    registry.register_provider(ParallelWebSearchProvider())
+    registry.register_provider(ExaWebSearchProvider())
+    yield registry
+    with registry._lock:
+        registry._providers.clear()
+        registry._providers.update(saved)
+        registry._scoped_providers.clear()
+        registry._scoped_providers.update(saved_scoped)
+
+
+# ---------------------------------------------------------------------------
+# keyless_mcp parsing
+# ---------------------------------------------------------------------------
+
+
+class TestParseMcpBody:
+    def test_sse_body(self):
+        payload = {"result": {"content": [{"type": "text", "text": "hello"}]}}
+        body = f"event: message\ndata: {json.dumps(payload)}\n\n"
+        assert keyless_mcp._parse_mcp_body(body) == "hello"
+
+    def test_plain_json_body(self):
+        payload = {"result": {"content": [{"type": "text", "text": "hi"}]}}
+        assert keyless_mcp._parse_mcp_body(json.dumps(payload)) == "hi"
+
+    def test_jsonrpc_error_raises(self):
+        body = json.dumps({"error": {"code": -32000, "message": "rate limit"}})
+        with pytest.raises(keyless_mcp.KeylessMCPError, match="rate limit"):
+            keyless_mcp._parse_mcp_body(body)
+
+    def test_is_error_result_raises(self):
+        body = json.dumps(
+            {"result": {"isError": True, "content": [{"type": "text", "text": "boom"}]}}
+        )
+        with pytest.raises(keyless_mcp.KeylessMCPError, match="boom"):
+            keyless_mcp._parse_mcp_body(body)
+
+    def test_garbage_raises(self):
+        with pytest.raises(keyless_mcp.KeylessMCPError):
+            keyless_mcp._parse_mcp_body("<html>nope</html>")
+
+
+class TestExaTextParsing:
+    def test_parses_blocks(self):
+        text = (
+            "Title: First\nURL: https://a.example\nPublished: N/A\n"
+            "Highlights:\nsome highlight\nmore\n"
+            "\n---\n"
+            "Title: Second\nURL: https://b.example\nHighlights:\nother\n"
+        )
+        results = keyless_mcp._parse_exa_search_text(text, limit=5)
+        assert [r["url"] for r in results] == ["https://a.example", "https://b.example"]
+        assert results[0]["description"] == "some highlight more"
+        assert results[0]["position"] == 1
+
+    def test_limit_respected(self):
+        text = "\n---\n".join(
+            f"Title: T{i}\nURL: https://x{i}.example" for i in range(6)
+        )
+        assert len(keyless_mcp._parse_exa_search_text(text, limit=2)) == 2
+
+
+class TestKeylessCalls:
+    def test_parallel_search_shapes_results(self):
+        payload = json.dumps(
+            {
+                "results": [
+                    {"url": "https://a", "title": "A", "excerpts": ["x", "y"]},
+                    {"url": "https://b", "title": "B", "excerpts": []},
+                ]
+            }
+        )
+        with patch.object(keyless_mcp, "mcp_call", return_value=payload) as call:
+            out = keyless_mcp.parallel_search_keyless("query", limit=5)
+        assert out["success"] is True
+        assert out["data"]["web"][0] == {
+            "url": "https://a", "title": "A", "description": "x y", "position": 1,
+        }
+        args = call.call_args[0]
+        assert args[0] == keyless_mcp.PARALLEL_MCP_URL
+        assert args[1] == "web_search"
+        assert "model_name" not in args[2]  # analytics field deliberately omitted
+
+    def test_parallel_search_failure_mentions_key_setup(self):
+        with patch.object(
+            keyless_mcp, "mcp_call", side_effect=keyless_mcp.KeylessMCPError("429")
+        ):
+            out = keyless_mcp.parallel_search_keyless("q")
+        assert out["success"] is False
+        assert "PARALLEL_API_KEY" in out["error"]
+
+    def test_parallel_extract_covers_missing_urls(self):
+        payload = json.dumps({"results": [{"url": "https://a", "title": "A", "excerpts": ["c"]}]})
+        with patch.object(keyless_mcp, "mcp_call", return_value=payload):
+            out = keyless_mcp.parallel_extract_keyless(["https://a", "https://gone"])
+        assert out[0]["content"] == "c"
+        assert out[1]["url"] == "https://gone"
+        assert "error" in out[1]
+
+    def test_exa_search_rate_limit_is_soft_error(self):
+        with patch.object(
+            keyless_mcp, "mcp_call",
+            side_effect=keyless_mcp.KeylessMCPError("free MCP rate limit"),
+        ):
+            out = keyless_mcp.exa_search_keyless("q")
+        assert out["success"] is False
+        assert "EXA_API_KEY" in out["error"]
+
+    def test_exa_extract_per_url(self):
+        with patch.object(
+            keyless_mcp, "mcp_call", return_value="# Page Title\nbody text"
+        ) as call:
+            out = keyless_mcp.exa_extract_keyless(["https://a", "https://b"])
+        assert call.call_count == 2
+        assert out[0]["title"] == "Page Title"
+        assert out[0]["content"].startswith("# Page Title")
+
+
+# ---------------------------------------------------------------------------
+# Provider routing: keyless vs keyed
+# ---------------------------------------------------------------------------
+
+
+class TestProviderRouting:
+    def test_parallel_keyless_path_when_no_key(self):
+        provider = ParallelWebSearchProvider()
+        with patch.object(
+            keyless_mcp, "parallel_search_keyless",
+            return_value={"success": True, "data": {"web": []}},
+        ) as keyless:
+            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):
+        provider = ExaWebSearchProvider()
+        with patch.object(
+            keyless_mcp, "exa_search_keyless",
+            return_value={"success": True, "data": {"web": []}},
+        ) as keyless:
+            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(
+            "agent.web_search_provider.get_provider_env",
+            lambda name: "sk-real" if name == "PARALLEL_API_KEY" else "",
+        )
+        provider = ParallelWebSearchProvider()
+        with patch.object(keyless_mcp, "parallel_search_keyless") as keyless, \
+                patch("plugins.web.parallel.provider._get_sync_client") as client:
+            client.return_value.beta.search.return_value.results = []
+            out = provider.search("q")
+        keyless.assert_not_called()
+        assert out["success"] is True
+
+    def test_keyless_disabled_falls_through_to_key_error(self, monkeypatch):
+        monkeypatch.setattr(registry, "_keyless_tier_enabled", lambda: False)
+        provider = ParallelWebSearchProvider()
+        out = provider.search("q")
+        assert out["success"] is False
+        assert "PARALLEL_API_KEY" in out["error"]
+
+    def test_is_available_stays_false_keyless(self):
+        # Keyless tier must NOT leak into is_available() (legacy walk order).
+        assert ParallelWebSearchProvider().is_available() is False
+        assert ExaWebSearchProvider().is_available() is False
+        assert ParallelWebSearchProvider().is_keyless_available() is True
+        assert ExaWebSearchProvider().is_keyless_available() is True
+
+    def test_tier_free_forces_keyless_even_with_key(self, monkeypatch):
+        monkeypatch.setattr(
+            "agent.web_search_provider.get_provider_env",
+            lambda name: "sk-real" if name == "PARALLEL_API_KEY" else "",
+        )
+        monkeypatch.setattr(keyless_mcp, "provider_tier", lambda name: "free")
+        provider = ParallelWebSearchProvider()
+        with patch.object(
+            keyless_mcp, "parallel_search_keyless",
+            return_value={"success": True, "data": {"web": []}},
+        ) as keyless:
+            out = provider.search("q")
+        keyless.assert_called_once()
+        assert out["success"] is True
+
+    def test_tier_paid_forces_keyed_without_key(self, monkeypatch):
+        monkeypatch.setattr(keyless_mcp, "provider_tier", lambda name: "paid")
+        provider = ParallelWebSearchProvider()
+        with patch.object(keyless_mcp, "parallel_search_keyless") as keyless:
+            out = provider.search("q")
+        keyless.assert_not_called()
+        assert out["success"] is False
+        assert "PARALLEL_API_KEY" in out["error"]
+
+    def test_tier_paid_disables_keyless_availability(self, monkeypatch):
+        monkeypatch.setattr(keyless_mcp, "provider_tier", lambda name: "paid")
+        assert ParallelWebSearchProvider().is_keyless_available() is False
+        assert ExaWebSearchProvider().is_keyless_available() is False
+
+    def test_provider_tier_reads_config(self, monkeypatch):
+        monkeypatch.setattr(
+            "hermes_cli.config.load_config",
+            lambda: {"web": {"provider_tier": {"exa": "FREE", "parallel": "bogus"}}},
+        )
+        assert keyless_mcp.provider_tier("exa") == "free"
+        assert keyless_mcp.provider_tier("parallel") == "auto"  # invalid → auto
+        assert keyless_mcp.provider_tier("tavily") == "auto"    # unset → auto
+
+    @pytest.mark.asyncio
+    async def test_parallel_keyless_extract(self):
+        provider = ParallelWebSearchProvider()
+        with patch.object(
+            keyless_mcp, "parallel_extract_keyless",
+            return_value=[{"url": "https://a", "title": "", "content": "c"}],
+        ) as keyless:
+            out = await provider.extract(["https://a"])
+        assert out[0]["content"] == "c"
+        keyless.assert_called_once_with(["https://a"])
+
+
+# ---------------------------------------------------------------------------
+# Registry + _get_backend resolution order
+# ---------------------------------------------------------------------------
+
+
+class TestResolutionOrder:
+    def test_registry_falls_back_to_keyless(self, fresh_registry, monkeypatch):
+        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)
+        )
+        # 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")
+
+    def test_registry_keyless_disabled_returns_none(self, fresh_registry, monkeypatch):
+        monkeypatch.setattr(registry, "_read_config_key", lambda *p: None)
+        monkeypatch.setattr(registry, "_keyless_tier_enabled", lambda: False)
+        assert registry.get_active_search_provider() is None
+
+    def test_keyed_provider_beats_keyless(self, fresh_registry, monkeypatch):
+        # Exa keyed, Parallel keyless: legacy walk must pick exa (keyed)
+        # even though parallel precedes exa in _KEYLESS_PREFERENCE.
+        monkeypatch.setattr(registry, "_read_config_key", lambda *p: None)
+        monkeypatch.setattr(
+            "agent.web_search_provider.get_provider_env",
+            lambda name: "sk-real" if name == "EXA_API_KEY" else "",
+        )
+        provider = registry.get_active_search_provider()
+        assert provider is not None and provider.name == "exa"
+
+    def test_get_backend_keyless_last(self, monkeypatch):
+        # No creds at all -> a keyless vendor per the process-stable split.
+        monkeypatch.setattr(
+            web_tools, "_registered_web_provider",
+            lambda name: {"parallel": ParallelWebSearchProvider(),
+                          "exa": ExaWebSearchProvider()}.get(name),
+        )
+        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]
+
+    def test_get_backend_key_beats_keyless(self, monkeypatch):
+        monkeypatch.setattr(
+            web_tools, "_env_value",
+            lambda name: "sk-x" if name == "TAVILY_API_KEY" else "",
+        )
+        assert web_tools._get_backend() == "tavily"
+
+    def test_get_backend_keyless_disabled(self, monkeypatch):
+        monkeypatch.setattr(
+            web_tools, "_registered_web_provider",
+            lambda name: {"parallel": ParallelWebSearchProvider(),
+                          "exa": ExaWebSearchProvider()}.get(name),
+        )
+        monkeypatch.setattr(web_tools, "_list_registered_web_providers", list)
+        monkeypatch.setattr(registry, "_keyless_tier_enabled", lambda: False)
+        assert web_tools._get_backend() == "firecrawl"  # legacy sentinel
+
+    def test_check_web_api_key_true_on_keyless_install(self, fresh_registry, monkeypatch):
+        monkeypatch.setattr(registry, "_read_config_key", lambda *p: None)
+        monkeypatch.setattr(web_tools, "_ensure_web_plugins_loaded", lambda: None)
+        monkeypatch.setattr(web_tools, "check_firecrawl_api_key", lambda: False)
+        assert web_tools.check_web_api_key() is True
+
+    def test_check_web_api_key_false_when_disabled(self, fresh_registry, monkeypatch):
+        monkeypatch.setattr(registry, "_read_config_key", lambda *p: None)
+        monkeypatch.setattr(registry, "_keyless_tier_enabled", lambda: False)
+        monkeypatch.setattr(web_tools, "_ensure_web_plugins_loaded", lambda: None)
+        monkeypatch.setattr(web_tools, "check_firecrawl_api_key", lambda: False)
+        assert web_tools.check_web_api_key() is False
+
+
+# ---------------------------------------------------------------------------
+# hermes tools picker: tier variant rows
+# ---------------------------------------------------------------------------
+
+
+class TestPickerTierRows:
+    def test_variant_schemas_flatten_to_tier_rows(self, fresh_registry, monkeypatch):
+        from hermes_cli import tools_config
+
+        monkeypatch.setattr(
+            "hermes_cli.plugins._ensure_plugins_discovered", lambda: None
+        )
+        rows = tools_config._plugin_web_search_providers()
+        by_backend_tier = {
+            (r["web_backend"], r.get("web_tier")): r["name"] for r in rows
+        }
+        assert ("parallel", "free") in by_backend_tier
+        assert ("parallel", "paid") in by_backend_tier
+        assert ("exa", "free") in by_backend_tier
+        assert ("exa", "paid") in by_backend_tier
+        # Free rows must not prompt for a key; paid rows must.
+        for r in rows:
+            if r.get("web_tier") == "free":
+                assert r["env_vars"] == []
+            if r.get("web_tier") == "paid":
+                assert r["env_vars"], r
+
+    def test_selection_persists_tier(self):
+        from hermes_cli.tools_config import _write_provider_config
+
+        config: dict = {}
+        _write_provider_config(
+            {"web_backend": "exa", "web_tier": "free", "env_vars": []},
+            config,
+            managed_feature=None,
+        )
+        assert config["web"]["backend"] == "exa"
+        assert config["web"]["provider_tier"]["exa"] == "free"
+        # Re-selecting a tier-agnostic row clears the stale tier.
+        _write_provider_config(
+            {"web_backend": "exa", "env_vars": []}, config, managed_feature=None
+        )
+        assert "exa" not in config["web"]["provider_tier"]
+
+    def test_tier_match_highlights_correct_row(self):
+        from hermes_cli.tools_config import _web_tier_matches
+
+        free_row = {"web_backend": "parallel", "web_tier": "free"}
+        paid_row = {"web_backend": "parallel", "web_tier": "paid"}
+        cfg_free = {"web": {"backend": "parallel", "provider_tier": {"parallel": "free"}}}
+        cfg_paid = {"web": {"backend": "parallel", "provider_tier": {"parallel": "paid"}}}
+        assert _web_tier_matches(free_row, cfg_free) is True
+        assert _web_tier_matches(paid_row, cfg_free) is False
+        assert _web_tier_matches(paid_row, cfg_paid) is True
+        assert _web_tier_matches(free_row, cfg_paid) is False
+        # Auto (unset tier, no key in the hermetic env): free row highlights.
+        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

tests/tools/test_web_providers.py

@@ -193,8 +193,13 @@ class TestUnconfiguredErrorEnvelopeParity:
     def test_unconfigured_search_emits_top_level_error(self, monkeypatch):
         """``web_search_tool`` with no creds returns ``{"error": "Error searching web: ..."}``
         — matching main's ``tool_error()`` envelope, not a per-result shape.
+
+        Keyless fallback (Parallel/Exa free tiers) is disabled here: with it
+        on, a zero-credential install routes to the keyless tier instead of
+        erroring (covered in test_web_keyless_fallback.py).
         """
         from tools import web_tools
+        from agent import web_search_registry
 
         self._clear_web_creds(monkeypatch)
         # Reset firecrawl client cache so the unconfigured state is re-evaluated
@@ -202,6 +207,7 @@ class TestUnconfiguredErrorEnvelopeParity:
         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)
 
         result = json.loads(web_tools.web_search_tool("hello world", limit=3))
         assert "error" in result, f"expected top-level 'error' key, got {result}"

tests/tools/test_web_providers_searxng.py

@@ -200,6 +200,7 @@ class TestCheckWebApiKey:
 
     def test_no_credentials_fails(self, monkeypatch):
         from tools import web_tools
+        from agent import web_search_registry
         monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
         monkeypatch.delenv("FIRECRAWL_API_KEY", raising=False)
         monkeypatch.delenv("FIRECRAWL_API_URL", raising=False)
@@ -210,6 +211,9 @@ class TestCheckWebApiKey:
         monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
         monkeypatch.setattr(web_tools, "check_firecrawl_api_key", lambda: False)
         monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: False)
+        # Disable the keyless free tier — with it on, zero credentials still
+        # resolves (Parallel/Exa anonymous MCP; see test_web_keyless_fallback.py).
+        monkeypatch.setattr(web_search_registry, "_keyless_tier_enabled", lambda: False)
         assert web_tools.check_web_api_key() is False
 
 

tests/tools/test_web_tools_config.py

@@ -216,10 +216,16 @@ class TestBackendSelection:
             assert _get_backend() == "firecrawl"
 
     def test_fallback_no_keys_defaults_to_firecrawl(self):
-        """No keys, no config → 'firecrawl' (will fail at client init)."""
+        """No keys, no config, keyless tier off → 'firecrawl' sentinel.
+
+        With the keyless tier on (default), zero credentials resolves to
+        the Parallel/Exa free tier instead — covered in
+        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._ddgs_package_importable", return_value=False):
+             patch("tools.web_tools._ddgs_package_importable", return_value=False), \
+             patch("agent.web_search_registry._keyless_tier_enabled", return_value=False):
             assert _get_backend() == "firecrawl"
 
     def test_invalid_config_is_returned_verbatim(self):

tools/web_tools.py

@@ -287,6 +287,32 @@ def _get_backend() -> str:
         except Exception as exc:  # noqa: BLE001 — a broken provider is skipped
             logger.debug("web provider %r.is_available() raised: %s", provider.name, exc)
 
+    # Keyless free-tier walk — zero credentials anywhere. Providers with a
+    # public anonymous endpoint (Parallel, Exa — see
+    # plugins/web/keyless_mcp.py) can still serve, unless the user disabled
+    # the tier via ``web.keyless_fallback: false``. Strictly last so it
+    # never pre-empts any keyed/importable backend above. Discovery must
+    # run first — this path is reachable from contexts that haven't loaded
+    # plugins yet (subprocess agent runs, delegate children, scripts).
+    try:
+        _ensure_web_plugins_loaded()
+        from agent.web_search_registry import _keyless_preference, _keyless_tier_enabled
+
+        if _keyless_tier_enabled():
+            for name in _keyless_preference():
+                provider = _registered_web_provider(name)
+                if provider is None:
+                    continue
+                try:
+                    if provider.is_keyless_available():
+                        return name
+                except Exception as exc:  # noqa: BLE001 — skip broken provider
+                    logger.debug(
+                        "web provider %r.is_keyless_available() raised: %s", name, exc
+                    )
+    except Exception as exc:  # noqa: BLE001 — registry optional; never fatal
+        logger.debug("keyless fallback walk failed: %s", exc)
+
     return "firecrawl"  # default (backward compat)
 
 
@@ -1154,8 +1180,14 @@ def check_web_api_key() -> bool:
     # 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.
+    # 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.
     try:
+        _ensure_web_plugins_loaded()
         from agent.web_search_registry import (
             get_active_search_provider,
             get_active_extract_provider,

website/docs/user-guide/configuration.md

@@ -2268,17 +2268,29 @@ web:
   # Or use per-capability keys to mix providers (e.g. free search + paid extract):
   search_backend: "searxng"
   extract_backend: "firecrawl"
+
+  # Keyless free-tier fallback (default: true). With no backend configured
+  # and no API keys present, web tools fall back to Parallel's / Exa's
+  # public anonymous endpoints (rate-limited). Set false to disable.
+  keyless_fallback: true
+
+  # Pin Exa/Parallel to a tier (set by the hermes tools Free/Paid rows).
+  # free = always the anonymous endpoint; paid = always the keyed SDK path;
+  # unset = auto (key present -> paid, otherwise free).
+  provider_tier:
+    parallel: free
+    exa: paid
 ```
 
 | Backend | Env Var | Search | Extract |
 |---------|---------|--------|---------|
 | **Firecrawl** (default) | `FIRECRAWL_API_KEY` | ✔ | ✔ |
 | **SearXNG** | `SEARXNG_URL` | ✔ | — |
-| **Parallel** | `PARALLEL_API_KEY` | ✔ | ✔ |
+| **Parallel** | `PARALLEL_API_KEY` (optional — keyless free tier) | ✔ | ✔ |
 | **Tavily** | `TAVILY_API_KEY` | ✔ | ✔ |
-| **Exa** | `EXA_API_KEY` | ✔ | ✔ |
+| **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. Otherwise Firecrawl is the default. 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. 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.
 
 **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

@@ -23,14 +23,20 @@ Both are configured through a single backend selection. Providers are chosen via
 | **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` | ✔ | ✔ | 1 000 searches/mo |
-| **Parallel** | `PARALLEL_API_KEY` | ✔ | ✔ | Paid |
+| **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 |
 | **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`.
+:::
+
+**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).
+
 :::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`.
 :::
@@ -360,6 +366,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) |
+
+**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.
 
 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"`.