feat(search): add Keenable as a search provider

This commit is contained in:
Ilya Bogin 2026-10-06 09:07:46 +03:00
parent c13746c951
commit d2b55428fd
15 changed files with 555 additions and 30 deletions

View file

@ -0,0 +1,3 @@
from litellm.llms.keenable.search.transformation import KeenableSearchConfig
__all__ = ("KeenableSearchConfig",)

View file

@ -0,0 +1,3 @@
from litellm.llms.keenable.search.transformation import KeenableSearchConfig
__all__ = ("KeenableSearchConfig",)

View file

@ -0,0 +1,270 @@
"""
Calls Keenable's /v1/search endpoint to search the web.
Keenable works without an API key: keyless requests go to /v1/search/public, which is rate
limited per IP and identifies the calling app by the X-Keenable-Title header. With an API key
the same body goes to /v1/search, which has higher limits.
Keenable API Reference: https://docs.keenable.ai/api-reference/search
"""
from __future__ import annotations
from collections.abc import Mapping
from types import MappingProxyType
from typing import TYPE_CHECKING, Final
import httpx
from pydantic import ConfigDict, TypeAdapter, ValidationError
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.search.transformation import (
BaseSearchConfig,
SearchResponse,
SearchResult,
)
from litellm.types.llms.base import LiteLLMBaseModel
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
_KEENABLE_DOCS_URL: Final = "https://docs.keenable.ai/api-reference/search"
# Sent on every request. The keyless endpoint rejects a request without it (400), and it
# names the calling software only, never the user.
_APP_TITLE: Final = "litellm"
class _KeenableResult(LiteLLMBaseModel):
"""One entry of Keenable's `results` array. Every field is optional so a single degraded
result degrades to empty strings instead of failing the whole call."""
model_config = ConfigDict(extra="ignore", frozen=True)
title: str | None = None
url: str | None = None
# Usually empty. The page text is in `snippet`.
description: str | None = None
snippet: str | None = None
published_at: str | None = None
acquired_at: str | None = None
class _KeenableSearchResponse(LiteLLMBaseModel):
"""Keenable's /v1/search response envelope."""
model_config = ConfigDict(extra="ignore", frozen=True)
# Required: a search with no hits returns `[]`, so a null or absent `results` means the
# body is not a search response and must not be reported as a successful empty search.
results: tuple[_KeenableResult, ...]
class _ErrorEnvelope(LiteLLMBaseModel):
"""Keenable reports errors as `{"error": "Invalid parameter", "message": "..."}`."""
model_config = ConfigDict(extra="ignore", frozen=True)
error: str | None = None
message: str | None = None
_DomainListAdapter: Final = TypeAdapter(tuple[str, ...])
_NOTHING: Final[Mapping[str, object]] = MappingProxyType({})
def _optional(key: str, value: object) -> Mapping[str, object]:
"""A one-entry mapping to spread into a payload, or nothing when the value is absent."""
return MappingProxyType({key: value}) if value is not None else _NOTHING
class KeenableSearchConfig(BaseSearchConfig):
KEENABLE_API_BASE = "https://api.keenable.ai/v1"
@staticmethod
def ui_friendly_name() -> str:
return "Keenable"
def _resolve_api_key(self, api_key: str | None, api_base: str | None) -> str | None:
"""The caller's key, else KEENABLE_API_KEY, else None for a keyless call."""
return self.resolve_server_api_key(
caller_api_key=api_key,
caller_api_base=api_base,
key_env_vars=("KEENABLE_API_KEY",),
base_env_var=None,
default_api_base=self.KEENABLE_API_BASE,
)
def validate_environment(
self,
headers: dict[str, str], # mutable-ok: BaseSearchConfig.validate_environment signature
api_key: str | None = None,
api_base: str | None = None,
**kwargs: object, # kwargs-ok: BaseSearchConfig.validate_environment signature
) -> dict[str, str]: # mutable-ok: the http handler passes this straight to httpx as headers
"""
Validate environment and return headers. No API key is required.
Returns a new dict rather than mutating ``headers``: the http handler calls this
a second time after ``litellm/search/main.py`` already did, so it has to be idempotent.
"""
resolved_api_key: Final = self._resolve_api_key(api_key, api_base)
return {
**headers,
"Content-Type": "application/json",
"X-Keenable-Title": _APP_TITLE,
**_optional("Authorization", f"Bearer {resolved_api_key}" if resolved_api_key else None),
}
def get_complete_url(
self,
api_base: str | None,
optional_params: dict[str, object], # mutable-ok: BaseSearchConfig.get_complete_url signature
data: dict[str, object] | list[dict[str, object]] | None = None, # mutable-ok: base signature
**kwargs: object, # kwargs-ok: BaseSearchConfig.get_complete_url signature
) -> str:
"""
The keyed and keyless endpoints take the same body and differ only by path, so the key
decides it: /search with a key, /search/public without. ``api_base`` is the version
root (https://api.keenable.ai/v1); a base that already ends in either path is accepted.
"""
caller_api_key: Final = kwargs.get("api_key")
resolved_api_key: Final = self._resolve_api_key(
caller_api_key if isinstance(caller_api_key, str) else None, api_base
)
root: Final = (
(api_base or self.KEENABLE_API_BASE).rstrip("/").removesuffix("/search/public").removesuffix("/search")
)
return f"{root}/search" if resolved_api_key else f"{root}/search/public"
def transform_search_request(
self,
query: str | list[str], # mutable-ok: BaseSearchConfig.transform_search_request signature
optional_params: dict[str, object], # mutable-ok: base signature
**kwargs: object, # kwargs-ok: BaseSearchConfig.transform_search_request signature
) -> dict[str, object]: # mutable-ok: the http handler passes this straight to httpx as the JSON body
"""
Transform Search request to Keenable API format.
- query -> query (a list is joined with spaces; Keenable takes a single string)
- max_results -> max_results (sent unclamped so Keenable's own 1-50 validation reports the error)
- search_domain_filter -> `site` for a single domain, `site:` / `-site:` clauses in the query
otherwise, following the Perplexity unified spec where a `-` prefix excludes a domain
- country, max_tokens_per_page -> dropped (no Keenable equivalent)
Everything else is forwarded as-is, so the rest of Keenable's surface (date filters,
snippet_max_length) stays reachable without LiteLLM tracking it.
"""
unified_params: Final = self.get_supported_perplexity_optional_params()
include, exclude = _split_domains(optional_params.get("search_domain_filter"))
site: Final = include[0] if len(include) == 1 else None
# Spread after the derived `site` so an explicitly supplied one wins.
passthrough: Final = MappingProxyType(
{param: value for param, value in optional_params.items() if param not in unified_params}
)
return {
**_optional("site", site),
**passthrough,
"query": _query_with_domain_clauses(
" ".join(query) if isinstance(query, list) else query,
include=() if site else include,
exclude=exclude,
),
**_optional("max_results", optional_params.get("max_results")),
}
def transform_search_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
**kwargs: object, # kwargs-ok: BaseSearchConfig.transform_search_response signature
) -> SearchResponse:
"""
Transform Keenable API response to LiteLLM unified SearchResponse format.
`snippet` carries the page text and `description` is usually empty, so the snippet comes
first. `date` is the page's `published_at` and `last_updated` is `acquired_at`, when
Keenable last fetched the page. A body that does not match the documented schema raises
an attributed error rather than being reported as a successful empty search.
"""
try:
parsed: Final = _KeenableSearchResponse.model_validate_json(raw_response.content)
except ValidationError as e:
raise self.get_error_class(
error_message=f"response does not match the documented /v1/search schema: {e}",
status_code=raw_response.status_code,
headers=dict(raw_response.headers),
)
return SearchResponse(
results=[
SearchResult(
title=result.title or "",
url=result.url or "",
snippet=result.snippet or result.description or "",
date=result.published_at,
last_updated=result.acquired_at,
)
for result in parsed.results
],
object="search",
)
def get_error_class(
self,
error_message: str,
status_code: int,
headers: dict[str, str], # mutable-ok: BaseSearchConfig.get_error_class signature
) -> Exception:
detail: Final = _unwrap_error_detail(error_message).rstrip(". ")
# Keyless calls share one per-IP budget, which a proxy serving many users exhausts first.
hint: Final = (
" Without an API key, requests share a per-IP limit; set KEENABLE_API_KEY to raise it."
if status_code == 429
else ""
)
return BaseLLMException(
status_code=status_code,
message=f"Keenable Search: {detail}.{hint} See {_KEENABLE_DOCS_URL} for details.",
headers=headers,
)
def _unwrap_error_detail(error_message: str) -> str:
"""
Surface the human-readable message inside Keenable's error envelope.
Falls back to the raw body for anything else (CDN HTML pages, plain text, other shapes).
"""
try:
body: Final = _ErrorEnvelope.model_validate_json(error_message)
except ValidationError:
return error_message
return body.message or body.error or error_message
def _split_domains(search_domain_filter: object) -> tuple[tuple[str, ...], tuple[str, ...]]:
"""
Split the unified `search_domain_filter` into included and excluded domains.
Anything that is not a list of strings is ignored rather than raising, since it only
ever narrows a search that is otherwise valid.
"""
try:
domains: Final = _DomainListAdapter.validate_python(search_domain_filter)
except ValidationError:
return (), ()
return (
tuple(d for d in domains if d and not d.startswith("-")),
tuple(d[1:] for d in domains if d.startswith("-") and len(d) > 1),
)
def _query_with_domain_clauses(query: str, include: tuple[str, ...], exclude: tuple[str, ...]) -> str:
"""Append `(site:a OR site:b)` and `-site:c` clauses, which Keenable honors in the query."""
included: Final = (f"({' OR '.join(f'site:{d}' for d in include)})",) if include else ()
excluded: Final = tuple(f"-site:{d}" for d in exclude)
return " ".join((query, *included, *excluded))

View file

@ -22865,6 +22865,14 @@
"notes": "Nimble Search API pay-as-you-go list price: $5 per 1,000 searches, up to 100 results per search. Volume plans price differently."
}
},
"keenable/search": {
"input_cost_per_query": 0.004,
"litellm_provider": "keenable",
"mode": "search",
"metadata": {
"notes": "Keenable pay-as-you-go list price: $4 per 1,000 requests, after a free allowance of 100,000 requests a month per organization. Keyless requests to /v1/search/public are free and rate limited per IP."
}
},
"elevenlabs/scribe_v1": {
"input_cost_per_second": 6.11e-05,
"litellm_provider": "elevenlabs",

View file

@ -4224,6 +4224,7 @@ class SearchProviders(str, Enum):
DATAFORSEO = "dataforseo"
FIRECRAWL = "firecrawl"
FASTCRW = "fastcrw"
KEENABLE = "keenable"
SEARXNG = "searxng"
LINKUP = "linkup"
DUCKDUCKGO = "duckduckgo"

View file

@ -9823,6 +9823,7 @@ class ProviderConfigManager:
from litellm.llms.fastcrw.search.transformation import FastCRWSearchConfig
from litellm.llms.firecrawl.search.transformation import FirecrawlSearchConfig
from litellm.llms.google_pse.search.transformation import GooglePSESearchConfig
from litellm.llms.keenable.search.transformation import KeenableSearchConfig
from litellm.llms.linkup.search.transformation import LinkupSearchConfig
from litellm.llms.nimble.search.transformation import NimbleSearchConfig
from litellm.llms.parallel_ai.search.transformation import (
@ -9846,6 +9847,7 @@ class ProviderConfigManager:
SearchProviders.DATAFORSEO: DataForSEOSearchConfig,
SearchProviders.FIRECRAWL: FirecrawlSearchConfig,
SearchProviders.FASTCRW: FastCRWSearchConfig,
SearchProviders.KEENABLE: KeenableSearchConfig,
SearchProviders.SEARXNG: SearXNGSearchConfig,
SearchProviders.LINKUP: LinkupSearchConfig,
SearchProviders.DUCKDUCKGO: DuckDuckGoSearchConfig,

View file

@ -22865,6 +22865,14 @@
"notes": "Nimble Search API pay-as-you-go list price: $5 per 1,000 searches, up to 100 results per search. Volume plans price differently."
}
},
"keenable/search": {
"input_cost_per_query": 0.004,
"litellm_provider": "keenable",
"mode": "search",
"metadata": {
"notes": "Keenable pay-as-you-go list price: $4 per 1,000 requests, after a free allowance of 100,000 requests a month per organization. Keyless requests to /v1/search/public are free and rate limited per IP."
}
},
"elevenlabs/scribe_v1": {
"input_cost_per_second": 6.11e-05,
"litellm_provider": "elevenlabs",

View file

@ -2619,6 +2619,13 @@
"search": true
}
},
"keenable": {
"display_name": "Keenable (`keenable`)",
"url": "https://docs.keenable.ai/api-reference/search",
"endpoints": {
"search": true
}
},
"triton": {
"display_name": "Triton (`triton`)",
"url": "https://docs.litellm.ai/docs/providers/triton-inference-server",

View file

@ -13,6 +13,7 @@ SEARCH_PROVIDERS = [
"brave",
"firecrawl",
"fastcrw",
"keenable",
"searxng",
"linkup",
"duckduckgo",

View file

@ -25,6 +25,7 @@ from litellm.llms.brave.search.transformation import BraveSearchConfig
from litellm.llms.dataforseo.search.transformation import DataForSEOSearchConfig
from litellm.llms.exa_ai.search.transformation import ExaAISearchConfig
from litellm.llms.fastcrw.search.transformation import FastCRWSearchConfig
from litellm.llms.keenable.search.transformation import KeenableSearchConfig
from litellm.llms.firecrawl.search.transformation import FirecrawlSearchConfig
from litellm.llms.google_pse.search.transformation import GooglePSESearchConfig
from litellm.llms.linkup.search.transformation import LinkupSearchConfig
@ -100,6 +101,7 @@ PROVIDERS: Tuple[ProviderSpec, ...] = (
),
(TinyfishSearchConfig, {"TINYFISH_API_KEY": "srv"}, "caller-key", {}),
(FastCRWSearchConfig, {"CRW_API_KEY": "srv"}, "caller-key", {}),
(KeenableSearchConfig, {"KEENABLE_API_KEY": "srv"}, "caller-key", {}),
(NimbleSearchConfig, {"NIMBLE_API_KEY": "srv"}, "caller-key", {}),
(BingGroundingSearchConfig, {"BING_GROUNDING_TOKEN": "srv"}, "caller-key", {}),
)
@ -107,9 +109,7 @@ PROVIDERS: Tuple[ProviderSpec, ...] = (
_IDS = tuple(spec[0].__name__ for spec in PROVIDERS)
@pytest.mark.parametrize(
"config_cls, server_env, caller_key, extra_env", PROVIDERS, ids=_IDS
)
@pytest.mark.parametrize("config_cls, server_env, caller_key, extra_env", PROVIDERS, ids=_IDS)
def test_server_secret_refused_for_caller_api_base(
config_cls: Type[BaseSearchConfig],
server_env: Dict[str, str],
@ -124,9 +124,7 @@ def test_server_secret_refused_for_caller_api_base(
config_cls().validate_environment(headers={}, api_base=ATTACKER_BASE)
@pytest.mark.parametrize(
"config_cls, server_env, caller_key, extra_env", PROVIDERS, ids=_IDS
)
@pytest.mark.parametrize("config_cls, server_env, caller_key, extra_env", PROVIDERS, ids=_IDS)
def test_caller_supplied_key_is_honored_for_custom_api_base(
config_cls: Type[BaseSearchConfig],
server_env: Dict[str, str],
@ -139,14 +137,10 @@ def test_caller_supplied_key_is_honored_for_custom_api_base(
# An explicit caller key is the caller's own credential, so pointing it at
# the caller's own host must be allowed.
config_cls().validate_environment(
headers={}, api_key=caller_key, api_base=ATTACKER_BASE
)
config_cls().validate_environment(headers={}, api_key=caller_key, api_base=ATTACKER_BASE)
@pytest.mark.parametrize(
"config_cls, server_env, caller_key, extra_env", PROVIDERS, ids=_IDS
)
@pytest.mark.parametrize("config_cls, server_env, caller_key, extra_env", PROVIDERS, ids=_IDS)
def test_server_secret_used_without_caller_api_base(
config_cls: Type[BaseSearchConfig],
server_env: Dict[str, str],
@ -167,9 +161,7 @@ def test_keyless_provider_allows_caller_api_base(
) -> None:
monkeypatch.delenv("SEARXNG_API_KEY", raising=False)
headers = SearXNGSearchConfig().validate_environment(
headers={}, api_base="https://my-searxng.internal"
)
headers = SearXNGSearchConfig().validate_environment(headers={}, api_base="https://my-searxng.internal")
assert "Authorization" not in headers
@ -182,9 +174,7 @@ def test_operator_env_base_override_is_trusted(
# Mirrors the second validate_environment call in the search handler, which
# receives the already-resolved operator base as api_base.
headers = SerperSearchConfig().validate_environment(
headers={}, api_base="https://serper.internal.corp/search"
)
headers = SerperSearchConfig().validate_environment(headers={}, api_base="https://serper.internal.corp/search")
assert headers["X-API-KEY"] == "srv"
@ -200,9 +190,7 @@ class TestResolveServerApiKey:
)
assert result == "mine"
def test_returns_none_when_no_server_secret(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
def test_returns_none_when_no_server_secret(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("SEARXNG_API_KEY", raising=False)
result = BaseSearchConfig().resolve_server_api_key(
caller_api_key=None,
@ -228,14 +216,10 @@ class TestResolveServerApiKey:
class TestIsTrustedSearchApiBase:
def test_matches_default_host(self) -> None:
assert _is_trusted_search_api_base(
"https://google.serper.dev/search", "https://google.serper.dev", None
)
assert _is_trusted_search_api_base("https://google.serper.dev/search", "https://google.serper.dev", None)
def test_foreign_host_untrusted(self) -> None:
assert not _is_trusted_search_api_base(
ATTACKER_BASE, "https://google.serper.dev", None
)
assert not _is_trusted_search_api_base(ATTACKER_BASE, "https://google.serper.dev", None)
def test_env_override_host_trusted(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("SERPER_API_BASE", "https://serper.internal.corp")
@ -248,9 +232,7 @@ class TestIsTrustedSearchApiBase:
def test_schemeless_candidate_untrusted(self) -> None:
# Without a scheme urlsplit puts the value in the path, leaving an empty
# netloc; an unparseable host must never be treated as trusted.
assert not _is_trusted_search_api_base(
"attacker.example.com", "https://google.serper.dev", None
)
assert not _is_trusted_search_api_base("attacker.example.com", "https://google.serper.dev", None)
@pytest.mark.asyncio

View file

View file

@ -0,0 +1,238 @@
import json
from unittest.mock import Mock, patch
import pytest
import litellm
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.keenable.search.transformation import KeenableSearchConfig
DEFAULT_ROOT = "https://api.keenable.ai/v1"
@pytest.fixture(autouse=True)
def _no_server_key(monkeypatch: pytest.MonkeyPatch):
monkeypatch.delenv("KEENABLE_API_KEY", raising=False)
def _config() -> KeenableSearchConfig:
return KeenableSearchConfig()
def _resp(payload, status_code: int = 200):
r = Mock()
r.status_code = status_code
r.headers = {}
r.content = (payload if isinstance(payload, str) else json.dumps(payload)).encode()
return r
def _result(**overrides):
base = {
"title": "Test Title",
"url": "https://example.com",
"description": "",
"snippet": "Test page text",
"published_at": "2026-01-15T10:30:00Z",
"acquired_at": "2026-01-16T08:12:34Z",
}
return {**base, **overrides}
def test_ui_friendly_name():
assert _config().ui_friendly_name() == "Keenable"
def test_keyless_request_names_the_app_and_targets_the_public_endpoint():
headers = _config().validate_environment({})
assert "Authorization" not in headers
assert headers["X-Keenable-Title"] == "litellm"
assert headers["Content-Type"] == "application/json"
assert _config().get_complete_url(None, {}) == f"{DEFAULT_ROOT}/search/public"
def test_caller_key_authenticates_and_targets_the_keyed_endpoint():
headers = _config().validate_environment({}, api_key="caller-key")
assert headers["Authorization"] == "Bearer caller-key"
assert headers["X-Keenable-Title"] == "litellm"
assert _config().get_complete_url(None, {}, api_key="caller-key") == f"{DEFAULT_ROOT}/search"
def test_server_key_authenticates_and_targets_the_keyed_endpoint(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("KEENABLE_API_KEY", "env-key")
assert _config().validate_environment({})["Authorization"] == "Bearer env-key"
assert _config().get_complete_url(None, {}) == f"{DEFAULT_ROOT}/search"
def test_validate_environment_does_not_mutate_and_is_idempotent():
"""The http handler re-runs validate_environment after search/main.py already did."""
config = _config()
caller_headers = {"X-Custom": "keep-me"}
once = config.validate_environment(caller_headers, api_key="k")
twice = config.validate_environment(once, api_key="k")
assert caller_headers == {"X-Custom": "keep-me"}
assert once == twice
assert once["X-Custom"] == "keep-me"
@pytest.mark.parametrize(
"api_base",
[
"https://self-hosted.local/v1",
"https://self-hosted.local/v1/",
"https://self-hosted.local/v1/search",
"https://self-hosted.local/v1/search/public",
"https://self-hosted.local/v1/search/public/",
],
)
@pytest.mark.parametrize(
"api_key, path",
[(None, "/search/public"), ("caller-key", "/search")],
)
def test_get_complete_url_is_idempotent_and_the_key_picks_the_path(api_base: str, api_key, path: str):
# search/main.py builds the URL, then the http handler calls get_complete_url again with
# that URL as api_base, so feeding the result back in must not change it.
url = _config().get_complete_url(api_base, {}, api_key=api_key)
assert url == f"https://self-hosted.local/v1{path}"
assert _config().get_complete_url(url, {}, api_key=api_key) == url
def test_request_maps_query_and_max_results_and_forwards_provider_params():
body = _config().transform_search_request(
["latest", "rust release"],
{
"max_results": 7,
"country": "US",
"max_tokens_per_page": 512,
"published_after": "7d",
"snippet_max_length": 500,
},
)
assert body == {
"query": "latest rust release",
"max_results": 7,
"published_after": "7d",
"snippet_max_length": 500,
}
def test_single_domain_filter_uses_the_site_field():
body = _config().transform_search_request("asyncio", {"search_domain_filter": ["docs.python.org"]})
assert body == {"query": "asyncio", "site": "docs.python.org"}
def test_several_domains_and_exclusions_become_query_clauses():
body = _config().transform_search_request(
"asyncio",
{"search_domain_filter": ["realpython.com", "docs.python.org", "-medium.com", "-"]},
)
assert body == {"query": "asyncio (site:realpython.com OR site:docs.python.org) -site:medium.com"}
def test_single_domain_with_exclusion_combines_site_and_clause():
body = _config().transform_search_request("asyncio", {"search_domain_filter": ["docs.python.org", "-medium.com"]})
assert body == {"query": "asyncio -site:medium.com", "site": "docs.python.org"}
def test_explicit_site_wins_over_the_derived_one():
body = _config().transform_search_request(
"asyncio", {"search_domain_filter": ["docs.python.org"], "site": "peps.python.org"}
)
assert body["site"] == "peps.python.org"
def test_malformed_domain_filter_is_ignored():
body = _config().transform_search_request("asyncio", {"search_domain_filter": "docs.python.org"})
assert body == {"query": "asyncio"}
def test_response_maps_snippet_and_dates():
response = _config().transform_search_response(
_resp({"query": "q", "results": [_result(), _result(snippet="", description="Only a description")]}),
logging_obj=Mock(),
)
first, second = response.results
assert (first.title, first.url, first.snippet) == ("Test Title", "https://example.com", "Test page text")
assert first.date == "2026-01-15T10:30:00Z"
assert first.last_updated == "2026-01-16T08:12:34Z"
assert second.snippet == "Only a description"
def test_degraded_result_maps_to_empty_strings():
response = _config().transform_search_response(_resp({"results": [{}]}), logging_obj=Mock())
only = response.results[0]
assert (only.title, only.url, only.snippet, only.date, only.last_updated) == ("", "", "", None, None)
def test_empty_result_list_is_a_successful_search():
assert _config().transform_search_response(_resp({"results": []}), logging_obj=Mock()).results == []
@pytest.mark.parametrize("payload", [{}, {"results": None}, "<html>bad gateway</html>"])
def test_body_that_is_not_a_search_response_raises(payload):
with pytest.raises(BaseLLMException, match="Keenable Search: response does not match"):
_config().transform_search_response(_resp(payload, status_code=200), logging_obj=Mock())
def test_error_class_surfaces_the_api_message():
error = _config().get_error_class(
'{"error": "Invalid parameter", "message": "\\"max_results\\": must be an integer between 1 and 50"}',
400,
{},
)
assert error.status_code == 400
assert error.message.startswith('Keenable Search: "max_results": must be an integer between 1 and 50.')
assert "KEENABLE_API_KEY" not in error.message
def test_rate_limit_error_points_at_the_api_key():
error = _config().get_error_class(
'{"error": "Rate limit exceeded", "message": "Public API hourly limit reached"}', 429, {}
)
assert error.status_code == 429
assert "Public API hourly limit reached" in error.message
assert "set KEENABLE_API_KEY" in error.message
def test_non_json_error_body_is_kept():
assert "upstream timeout" in _config().get_error_class("upstream timeout", 502, {}).message
@pytest.mark.parametrize(
"api_key, path, auth",
[(None, "/search/public", None), ("caller-key", "/search", "Bearer caller-key")],
)
def test_litellm_search_end_to_end(api_key, path: str, auth):
with patch(
"litellm.llms.custom_httpx.http_handler.HTTPHandler.post",
return_value=_resp({"query": "q", "results": [_result()]}),
) as mock_post:
response = litellm.search(
query="rust release",
search_provider="keenable",
api_key=api_key,
max_results=3,
)
call = mock_post.call_args.kwargs
assert call["url"] == f"{DEFAULT_ROOT}{path}"
assert call["headers"]["X-Keenable-Title"] == "litellm"
assert call["headers"].get("Authorization") == auth
assert call["json"] == {"query": "rust release", "max_results": 3}
assert response.results[0].snippet == "Test page text"

Binary file not shown.

After

Width:  |  Height:  |  Size: 6.3 KiB

View file

@ -31,6 +31,7 @@ import bingLogo from "../../../../../public/assets/logos/bing.png";
import dataforseoLogo from "../../../../../public/assets/logos/dataforseo.png";
import exaAiLogo from "../../../../../public/assets/logos/exa_ai.png";
import googlePseLogo from "../../../../../public/assets/logos/google_pse.png";
import keenableLogo from "../../../../../public/assets/logos/keenable.png";
import nimbleLogo from "../../../../../public/assets/logos/nimble.png";
import parallelAiLogo from "../../../../../public/assets/logos/parallel_ai.png";
import perplexityLogo from "../../../../../public/assets/logos/perplexity.png";
@ -44,6 +45,7 @@ const searchProviderLogoMap: Record<string, string> = {
exa_ai: exaAiLogo.src,
google_pse: googlePseLogo.src,
dataforseo: dataforseoLogo.src,
keenable: keenableLogo.src,
nimble: nimbleLogo.src,
bing_grounding: bingLogo.src,
};