feat(web-search): add an opt-in Keenable search provider

.env.example

@@ -24,7 +24,11 @@ FRONTIER_AGENT_DATASETS_DIR=
 # Serper remains the default. Set WEB_SEARCH_PROVIDER=parallel to use the free,
 # keyless Parallel Search MCP for web_search (up to 10 results per query);
 # this does not change web_fetch.
+# Set WEB_SEARCH_PROVIDER=keenable to use the keyless Keenable search API for
+# web_search (up to 50 results per query); KEENABLE_API_KEY is optional and
+# only raises the rate limits. This does not change web_fetch either.
 WEB_SEARCH_PROVIDER=serper
+KEENABLE_API_KEY=
 # Support any Serper.dev-compatible endpoint (like litescrape.com, serpbase.dev,
 # and others): set SERPER_BASE_URL and a provider-issued SERPER_API_KEY.
 SERPER_API_KEY=

README.md

@@ -166,6 +166,12 @@ custom region, language, or time filters, or counts above 10 per query; choose
 Serper when those are needed.
 `web_fetch` continues to use its existing fetch provider.
 
+Set `WEB_SEARCH_PROVIDER=keenable` to use the keyless
+[Keenable search API](https://docs.keenable.ai/api-reference) for `web_search`
+instead. An optional `KEENABLE_API_KEY` only raises the rate limits. Keenable
+returns up to 50 results per query and does not support the tool's custom
+region, language, or time filters; choose Serper when those are needed.
+
 Start the TUI:
 
 ```bash

apodex/README.md

@@ -90,6 +90,8 @@ cp .env.example .env
 #   OPENAI_MODEL=gpt-4o
 # For keyless Parallel search set WEB_SEARCH_PROVIDER=parallel; it returns up
 # to 10 results per query. Serper remains the default and uses SERPER_API_KEY.
+# For keyless Keenable search set WEB_SEARCH_PROVIDER=keenable; it returns up
+# to 50 results per query, and an optional KEENABLE_API_KEY raises rate limits.
 # Page fetch uses JINA_API_KEY.
 
 # Interactive TUI, Stateful ReAct, against another repository

apodex/config.py

@@ -223,10 +223,10 @@ def inspect_runtime_config(
         env=env,
     )
     search_provider = (env.get("WEB_SEARCH_PROVIDER") or "serper").strip().lower()
-    if "web_search" in tool_names and search_provider not in {"serper", "parallel"}:
+    if "web_search" in tool_names and search_provider not in {"serper", "parallel", "keenable"}:
         issues.append(RuntimeConfigIssue(
             code="invalid_web_search_provider",
-            message="WEB_SEARCH_PROVIDER must be set to 'serper' or 'parallel'.",
+            message="WEB_SEARCH_PROVIDER must be set to 'serper', 'parallel', or 'keenable'.",
             env_var="WEB_SEARCH_PROVIDER",
         ))
     if (

apodex/tests/test_config_preflight.py

@@ -119,6 +119,14 @@ def test_search_credentials_are_checked_when_the_profile_binds_the_web_tools():
     assert parallel.ok
     assert [issue.code for issue in parallel.warnings] == ["missing_jina_api_key"]
 
+    keenable = inspect_runtime_config(
+        cfg,
+        profile=_profile(tool_names=_WEB_TOOLS),
+        environ={"WEB_SEARCH_PROVIDER": "keenable"},
+    )
+    assert keenable.ok
+    assert [issue.code for issue in keenable.warnings] == ["missing_jina_api_key"]
+
     invalid_provider = inspect_runtime_config(
         cfg,
         profile=_profile(tool_names=_WEB_TOOLS),

docs/install/global-install.md

@@ -74,6 +74,9 @@ Add `WEB_SEARCH_PROVIDER=parallel` to this file to use the keyless Parallel
 Search MCP for `web_search`; it supports up to 10 results per query. Serper
 remains the default. Exporting the setting also works for native runs.
 `web_fetch` keeps its existing provider.
+`WEB_SEARCH_PROVIDER=keenable` selects the keyless Keenable search API in the
+same way; it supports up to 50 results per query, and an optional
+`KEENABLE_API_KEY` in this file raises the rate limits.
 `APODEX_ENV_FILE=/path/to/file` points the CLI at a different file.
 
 Rules for the file:

plugins/tools/_keenable_search.py

@@ -0,0 +1,137 @@
+"""Keyless Keenable search transport for the native web-search tools."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+from typing import Any
+
+import httpx
+
+from frontier_agent.infra.usage_meter import record_api_request
+
+logger = logging.getLogger(__name__)
+
+_PUBLIC_URL = "https://api.keenable.ai/v1/search/public"
+_KEYED_URL = "https://api.keenable.ai/v1/search"
+_APP_TITLE = "FrontierAgent"
+_MIN_RESULTS = 10
+_MAX_RESULTS = 50
+_MAX_RETRIES = 3
+# Keenable returns page text rather than a short search-engine snippet; cap it
+# so ten results stay a compact tool observation.
+_SNIPPET_CHARS = 1000
+
+
+def _api_key() -> str:
+    """Optional key, read at call time because ``.env`` loads after import."""
+    return (os.getenv("KEENABLE_API_KEY") or "").strip()
+
+
+def _headers(api_key: str) -> dict[str, str]:
+    headers = {"Content-Type": "application/json", "X-Keenable-Title": _APP_TITLE}
+    if api_key:
+        headers["X-API-Key"] = api_key
+    return headers
+
+
+def _snippet(value: Any) -> str:
+    """Collapse page text into one line and cap its length."""
+    text = " ".join(str(value or "").split())
+    if len(text) > _SNIPPET_CHARS:
+        text = text[:_SNIPPET_CHARS].rstrip() + "..."
+    return text
+
+
+def _normalise_result(value: Any) -> dict[str, Any]:
+    """Convert a Keenable response into the native Serper-shaped result."""
+    rows = value.get("results") if isinstance(value, dict) else None
+    organic = [
+        {
+            "title": str(item.get("title") or ""),
+            "link": str(item["url"]),
+            "snippet": _snippet(item.get("snippet") or item.get("description")),
+        }
+        for item in rows or []
+        if isinstance(item, dict) and item.get("url")
+    ]
+    return {"organic": organic} if organic else {}
+
+
+async def _search(
+    client: httpx.AsyncClient,
+    url: str,
+    headers: dict[str, str],
+    query: str,
+    max_results: int,
+) -> dict[str, Any]:
+    """One Keenable query, retrying rate limits and server errors."""
+    payload = {"query": query, "max_results": max_results}
+    for attempt in range(_MAX_RETRIES):
+        response = await client.post(url, json=payload, headers=headers)
+        retryable = response.status_code == 429 or response.status_code >= 500
+        if retryable and attempt < _MAX_RETRIES - 1:
+            record_api_request("keenable", requests=0, retries=1)
+            await asyncio.sleep(2 ** attempt)
+            continue
+        response.raise_for_status()
+        record_api_request("keenable")
+        return _normalise_result(response.json())
+    return {}
+
+
+async def keenable_search_batch(
+    queries: list[str],
+    *,
+    num_results: int,
+    gl: str = "us",
+    hl: str = "en",
+    tbs: str = "",
+) -> list[dict[str, Any]] | str:
+    """Search Keenable for each query concurrently.
+
+    Works without a key; ``KEENABLE_API_KEY`` is optional and only raises the
+    rate limits. Locale and time controls are rejected when requested because
+    this backend does not expose them. Each query asks for at least ten and at
+    most fifty results, and all candidates are returned so callers can limit
+    results after display filtering and URL deduplication. The selected
+    provider never falls back to Serper on a transport or rate-limit error.
+    """
+    if gl != "us" or hl != "en" or tbs:
+        return (
+            "Keenable search does not support custom region, language, or "
+            "time filters. Select Serper to use those search options."
+        )
+    if not queries:
+        return []
+
+    api_key = _api_key()
+    url = _KEYED_URL if api_key else _PUBLIC_URL
+    headers = _headers(api_key)
+    max_results = max(_MIN_RESULTS, min(int(num_results), _MAX_RESULTS))
+    try:
+        async with httpx.AsyncClient(timeout=httpx.Timeout(30.0, connect=10.0)) as client:
+            tasks = [
+                asyncio.create_task(_search(client, url, headers, query, max_results))
+                for query in queries
+            ]
+            try:
+                return list(await asyncio.gather(*tasks))
+            except BaseException:
+                for task in tasks:
+                    task.cancel()
+                await asyncio.gather(*tasks, return_exceptions=True)
+                raise
+    except httpx.HTTPStatusError as error:
+        status = error.response.status_code
+        logger.warning("Keenable search request failed with HTTP %d", status)
+        record_api_request("keenable", requests=0, errors=1)
+        return f"Keenable search returned HTTP {status}."
+    except (httpx.HTTPError, ValueError, TypeError, KeyError):
+        logger.warning("Keenable search request failed")
+        record_api_request("keenable", requests=0, errors=1)
+        return "Keenable search failed. Check network access and try again."
+
+
+__all__ = ["keenable_search_batch"]

plugins/tools/_parallel_search.py

@@ -21,7 +21,7 @@
 _MCP_URL = "https://search.parallel.ai/mcp"
 _INITIAL_PROTOCOL_VERSION = "2025-03-26"
 _ANONYMOUS_MAX_RESULTS = 10
-_PROVIDER_NAMES = frozenset({"serper", "parallel"})
+_PROVIDER_NAMES = frozenset({"serper", "parallel", "keenable"})
 
 
 def selected_search_provider() -> str:

plugins/tools/web_search.py

@@ -14,6 +14,7 @@
 from frontier_agent.infra.config import get_config
 from frontier_agent.infra.usage_meter import record_api_request
 from plugins.tools._coerce import coerce_json_list
+from plugins.tools._keenable_search import keenable_search_batch
 from plugins.tools._parallel_search import (
     parallel_search_batch,
     selected_search_provider,
@@ -464,9 +465,10 @@ async def web_search(
 
     provider = selected_search_provider()
     if not valid_search_provider(provider):
-        return "Error: WEB_SEARCH_PROVIDER must be 'serper' or 'parallel'."
-    if provider == "parallel":
-        datas = await parallel_search_batch(
+        return "Error: WEB_SEARCH_PROVIDER must be 'serper', 'parallel', or 'keenable'."
+    if provider in ("parallel", "keenable"):
+        search_batch = keenable_search_batch if provider == "keenable" else parallel_search_batch
+        datas = await search_batch(
             queries,
             num_results=num_results,
             gl=gl,
@@ -507,7 +509,7 @@ async def web_search(
 def _format_parallel_query_results(
     queries: list[str], datas: list[dict], max_results: int,
 ) -> str:
-    """Keep the canonical query-labelled output for Parallel MCP batches."""
+    """Keep the canonical query-labelled output for Parallel MCP and Keenable batches."""
     seen_urls: set[str] = set()
     blocks: list[str] = []
     for query, data in zip(queries, datas, strict=False):

plugins/tools/web_search_aligned.py

@@ -24,6 +24,7 @@
 
 from frontier_agent.core.tool import tool
 from frontier_agent.infra.usage_meter import record_api_request
+from plugins.tools._keenable_search import keenable_search_batch
 from plugins.tools._parallel_search import (
     parallel_search_batch,
     selected_search_provider,
@@ -284,15 +285,17 @@ async def web_search_aligned(
 
     provider = selected_search_provider()
     if not valid_search_provider(provider):
-        return "[ERROR]: WEB_SEARCH_PROVIDER must be 'serper' or 'parallel'."
-    if provider == "parallel":
+        return "[ERROR]: WEB_SEARCH_PROVIDER must be 'serper', 'parallel', or 'keenable'."
+    if provider in ("parallel", "keenable"):
+        backend = "Keenable search" if provider == "keenable" else "Parallel Search MCP"
         if location is not None or page not in (None, 1) or autocorrect is not None:
             return (
-                "[ERROR]: Parallel Search MCP does not support location, page, or "
+                f"[ERROR]: {backend} does not support location, page, or "
                 "autocorrect options. Select Serper to use those search options."
             )
         result_limit = 10 if num is None else max(1, min(num, 100))
-        datas = await parallel_search_batch(
+        search_batch = keenable_search_batch if provider == "keenable" else parallel_search_batch
+        datas = await search_batch(
             queries,
             num_results=result_limit,
             gl=gl,