From 39a4d075c42dc8e81af3cafe0b8eee1dd1d3bb5a Mon Sep 17 00:00:00 2001 From: Ilya Bogin Date: Fri, 23 Oct 2026 12:46:29 +0300 Subject: [PATCH] feat(web-search): add an opt-in Keenable search provider WEB_SEARCH_PROVIDER=keenable routes both native web_search implementations to the Keenable search API. It works without a key; an optional KEENABLE_API_KEY only raises the rate limits. Serper stays the default and Parallel is unchanged. --- .env.example | 4 + README.md | 6 + apodex/README.md | 2 + apodex/config.py | 4 +- apodex/tests/test_config_preflight.py | 8 + docs/install/global-install.md | 3 + plugins/tools/_keenable_search.py | 137 ++++++++++++ plugins/tools/_parallel_search.py | 2 +- plugins/tools/web_search.py | 10 +- plugins/tools/web_search_aligned.py | 11 +- tests/test_keenable_search.py | 294 ++++++++++++++++++++++++++ 11 files changed, 470 insertions(+), 11 deletions(-) create mode 100644 plugins/tools/_keenable_search.py create mode 100644 tests/test_keenable_search.py diff --git a/.env.example b/.env.example index dcb7517..da6144a 100644 --- a/.env.example +++ b/.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= diff --git a/README.md b/README.md index 2097d40..cd74706 100644 --- a/README.md +++ b/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 diff --git a/apodex/README.md b/apodex/README.md index 66f0107..2c125fc 100644 --- a/apodex/README.md +++ b/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 diff --git a/apodex/config.py b/apodex/config.py index a77d759..1ea3d23 100644 --- a/apodex/config.py +++ b/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 ( diff --git a/apodex/tests/test_config_preflight.py b/apodex/tests/test_config_preflight.py index ba9452a..0f5bc54 100644 --- a/apodex/tests/test_config_preflight.py +++ b/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), diff --git a/docs/install/global-install.md b/docs/install/global-install.md index 3952aa6..d2593a0 100644 --- a/docs/install/global-install.md +++ b/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: diff --git a/plugins/tools/_keenable_search.py b/plugins/tools/_keenable_search.py new file mode 100644 index 0000000..46a4a21 --- /dev/null +++ b/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"] diff --git a/plugins/tools/_parallel_search.py b/plugins/tools/_parallel_search.py index 6cf6f8b..520a5b0 100644 --- a/plugins/tools/_parallel_search.py +++ b/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: diff --git a/plugins/tools/web_search.py b/plugins/tools/web_search.py index 5698f43..0aee105 100644 --- a/plugins/tools/web_search.py +++ b/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): diff --git a/plugins/tools/web_search_aligned.py b/plugins/tools/web_search_aligned.py index fbf1ea0..4e3178e 100644 --- a/plugins/tools/web_search_aligned.py +++ b/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, diff --git a/tests/test_keenable_search.py b/tests/test_keenable_search.py new file mode 100644 index 0000000..84e1d82 --- /dev/null +++ b/tests/test_keenable_search.py @@ -0,0 +1,294 @@ +"""Opt-in Keenable search transport for the native web-search tools.""" + +from __future__ import annotations + +import asyncio +import importlib +import json +from collections.abc import Callable +from typing import Any + +import httpx +import pytest + +from plugins.tools import _keenable_search as keenable +from plugins.tools import _parallel_search as parallel + + +def _row(title: str, url: str, snippet: str = "Useful text.") -> dict[str, Any]: + return { + "title": title, + "url": url, + "description": "", + "snippet": snippet, + "acquired_at": "2026-10-01T00:00:00Z", + } + + +def _install( + monkeypatch: pytest.MonkeyPatch, + handler: Callable[[httpx.Request], Any], +) -> list[httpx.Request]: + """Route the module's HTTP client through ``handler`` and record requests.""" + requests: list[httpx.Request] = [] + real_client = httpx.AsyncClient + + async def recording(request: httpx.Request) -> httpx.Response: + requests.append(request) + response = handler(request) + if not isinstance(response, httpx.Response): + response = await response + return response + + def client(**kwargs: Any) -> httpx.AsyncClient: + return real_client(transport=httpx.MockTransport(recording), **kwargs) + + monkeypatch.setattr(keenable.httpx, "AsyncClient", client) + return requests + + +def _results(*rows: dict[str, Any]) -> Callable[[httpx.Request], httpx.Response]: + return lambda _request: httpx.Response(200, json={"results": list(rows)}) + + +def test_keenable_is_an_opt_in_provider(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("WEB_SEARCH_PROVIDER", raising=False) + assert parallel.selected_search_provider() == "serper" + assert parallel.valid_search_provider("keenable") + monkeypatch.setenv("WEB_SEARCH_PROVIDER", "Keenable") + assert parallel.selected_search_provider() == "keenable" + assert parallel.valid_search_provider() + + +@pytest.mark.asyncio +async def test_keyless_search_uses_the_public_endpoint( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("KEENABLE_API_KEY", raising=False) + requests = _install(monkeypatch, _results( + _row("Example", "https://example.com/page", "First line.\n\n## Heading\nSecond."), + )) + + result = await keenable.keenable_search_batch(["FrontierAgent search"], num_results=5) + + assert result == [{ + "organic": [{ + "title": "Example", + "link": "https://example.com/page", + "snippet": "First line. ## Heading Second.", + }], + }] + assert len(requests) == 1 + request = requests[0] + assert str(request.url) == "https://api.keenable.ai/v1/search/public" + assert request.headers["X-Keenable-Title"] == "FrontierAgent" + assert "X-API-Key" not in request.headers + assert "Authorization" not in request.headers + assert json.loads(request.content) == {"query": "FrontierAgent search", "max_results": 10} + + +@pytest.mark.asyncio +async def test_optional_api_key_selects_the_keyed_endpoint( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("KEENABLE_API_KEY", "test-key") + requests = _install(monkeypatch, _results(_row("Example", "https://example.com"))) + + result = await keenable.keenable_search_batch(["query"], num_results=10) + + assert isinstance(result, list) + assert str(requests[0].url) == "https://api.keenable.ai/v1/search" + assert requests[0].headers["X-API-Key"] == "test-key" + assert requests[0].headers["X-Keenable-Title"] == "FrontierAgent" + + +@pytest.mark.parametrize(("requested", "sent"), [(1, 10), (25, 25), (100, 50)]) +@pytest.mark.asyncio +async def test_result_count_stays_within_the_supported_range( + monkeypatch: pytest.MonkeyPatch, requested: int, sent: int, +) -> None: + monkeypatch.delenv("KEENABLE_API_KEY", raising=False) + requests = _install(monkeypatch, _results(_row("Example", "https://example.com"))) + + await keenable.keenable_search_batch(["query"], num_results=requested) + + assert json.loads(requests[0].content)["max_results"] == sent + + +@pytest.mark.parametrize("options", [{"gl": "gb"}, {"hl": "de"}, {"tbs": "qdr:d"}]) +@pytest.mark.asyncio +async def test_rejects_filters_it_cannot_honor( + monkeypatch: pytest.MonkeyPatch, options: dict[str, str], +) -> None: + requests = _install(monkeypatch, _results()) + + result = await keenable.keenable_search_batch(["query"], num_results=5, **options) + + assert isinstance(result, str) + assert "custom region, language, or time filters" in result + assert "Select Serper" in result + assert requests == [] + + +def test_long_page_text_is_capped() -> None: + text = "word " * 1000 + result = keenable._normalise_result({"results": [_row("Long", "https://example.com", text)]}) + snippet = result["organic"][0]["snippet"] + assert len(snippet) <= keenable._SNIPPET_CHARS + 3 + assert snippet.endswith("...") + assert keenable._normalise_result({"results": [{"title": "No URL"}]}) == {} + + +@pytest.mark.asyncio +async def test_rate_limits_are_retried_then_reported( + monkeypatch: pytest.MonkeyPatch, +) -> None: + delays: list[float] = [] + + async def no_wait(delay: float) -> None: + delays.append(delay) + + monkeypatch.setattr(keenable.asyncio, "sleep", no_wait) + requests = _install(monkeypatch, lambda _request: httpx.Response(429)) + + result = await keenable.keenable_search_batch(["query"], num_results=5) + + assert result == "Keenable search returned HTTP 429." + assert len(requests) == 3 + assert delays == [1, 2] + + +@pytest.mark.asyncio +async def test_server_error_is_retried_before_results( + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def no_wait(_delay: float) -> None: + return None + + statuses = iter([503, 200]) + + def handler(_request: httpx.Request) -> httpx.Response: + status = next(statuses) + if status != 200: + return httpx.Response(status) + return httpx.Response(200, json={"results": [_row("Example", "https://example.com")]}) + + monkeypatch.setattr(keenable.asyncio, "sleep", no_wait) + requests = _install(monkeypatch, handler) + + result = await keenable.keenable_search_batch(["query"], num_results=5) + + assert isinstance(result, list) + assert result[0]["organic"][0]["link"] == "https://example.com" + assert len(requests) == 2 + + +@pytest.mark.asyncio +async def test_list_queries_are_dispatched_concurrently( + monkeypatch: pytest.MonkeyPatch, +) -> None: + active = 0 + peak_active = 0 + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal active, peak_active + active += 1 + peak_active = max(peak_active, active) + await asyncio.sleep(0.02) + active -= 1 + query = json.loads(request.content)["query"] + return httpx.Response(200, json={ + "results": [_row(f"{query} result", f"https://{query}.example.com")], + }) + + requests = _install(monkeypatch, handler) + + result = await keenable.keenable_search_batch(["first", "second"], num_results=3) + + assert isinstance(result, list) + assert [data["organic"][0]["title"] for data in result] == [ + "first result", "second result", + ] + assert len(requests) == 2 + assert peak_active == 2 + + +@pytest.mark.asyncio +async def test_keenable_errors_do_not_fall_back_to_serper( + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = importlib.import_module("plugins.tools.web_search") + monkeypatch.setenv("WEB_SEARCH_PROVIDER", "keenable") + monkeypatch.delenv("SERPER_API_KEY", raising=False) + + async def failed(*_args: Any, **_kwargs: Any) -> str: + return "Keenable search returned HTTP 429." + + async def no_serper(*_args: Any, **_kwargs: Any) -> dict[str, Any]: + raise AssertionError("explicit Keenable selection must not call Serper") + + monkeypatch.setattr(module, "keenable_search_batch", failed) + monkeypatch.setattr(module, "raw_web_search", no_serper) + output = await module.web_search.ainvoke({"q": "sample topic"}) + + assert "Keenable search returned HTTP 429" in output + + +@pytest.mark.asyncio +async def test_aligned_route_rejects_options_keenable_cannot_honor( + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = importlib.import_module("plugins.tools.web_search_aligned") + monkeypatch.setenv("WEB_SEARCH_PROVIDER", "keenable") + + output = await module.web_search_aligned.ainvoke({"q": "sample", "page": 2}) + + assert output.startswith("[ERROR]: Keenable search does not support location, page") + + +@pytest.mark.parametrize("implementation", ["original", "aligned"]) +@pytest.mark.parametrize("scenario", ["filtered", "duplicates", "multiple_queries"]) +@pytest.mark.asyncio +async def test_keenable_limits_displayed_results_after_filtering_and_deduplication( + monkeypatch: pytest.MonkeyPatch, implementation: str, scenario: str, +) -> None: + first = _row("First useful", "https://example.com/first") + second = _row("Second useful", "https://example.com/second") + extra = _row("Over limit", "https://example.com/extra") + blocked_url = ( + "https://youtube.com/watch/1" if implementation == "original" + else "https://huggingface.co/datasets/example" + ) + blocked = _row("Blocked result", blocked_url) + if scenario == "filtered": + queries, limit = ["first"], 1 + responses = {"first": [blocked, first, extra]} + expected = ["First useful"] + elif scenario == "duplicates": + queries, limit = ["first"], 2 + responses = {"first": [first, first, second, extra]} + expected = ["First useful", "Second useful"] + else: + queries, limit = ["first", "second"], 1 + responses = {"first": [first, second], "second": [first, blocked, second, extra]} + expected = ["First useful", "Second useful"] + + def handler(request: httpx.Request) -> httpx.Response: + query = json.loads(request.content)["query"] + return httpx.Response(200, json={"results": responses[query]}) + + monkeypatch.delenv("KEENABLE_API_KEY", raising=False) + monkeypatch.setenv("WEB_SEARCH_PROVIDER", "keenable") + _install(monkeypatch, handler) + if implementation == "original": + module = importlib.import_module("plugins.tools.web_search") + output = await module.web_search.ainvoke({"q": queries, "num_results": limit}) + else: + module = importlib.import_module("plugins.tools.web_search_aligned") + output = await module.web_search_aligned.ainvoke({"q": queries, "num": limit}) + + assert output.count("URL: https://") == len(expected) + for title in expected: + assert output.count(title) == 1 + assert "Blocked result" not in output + assert "Over limit" not in output