Merge dc3ce1ffa7 into e9b1c11098

litellm/llms/scavio/__init__.py

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

litellm/llms/scavio/search/__init__.py

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

litellm/llms/scavio/search/transformation.py

@@ -0,0 +1,209 @@
+"""
+Calls Scavio's /api/v2/google endpoint to search Google.
+
+Scavio API Reference: https://scavio.dev/docs/search-api
+"""
+
+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, StrictInt, 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
+
+_SCAVIO_DOCS_URL: Final = "https://scavio.dev/docs/search-api"
+_SEARCH_PATH: Final = "/api/v2/google"
+
+
+class _ScavioOrganicResult(LiteLLMBaseModel):
+    model_config = ConfigDict(extra="ignore", frozen=True)
+
+    title: str | None = None
+    link: str | None = None
+    snippet: str | None = None
+
+
+class _ScavioSearchResponse(LiteLLMBaseModel):
+    model_config = ConfigDict(extra="ignore", frozen=True)
+
+    organic_results: tuple[_ScavioOrganicResult, ...]
+
+
+class _ErrorEnvelope(LiteLLMBaseModel):
+    model_config = ConfigDict(extra="ignore", frozen=True)
+
+    error: str | None = None
+
+
+class _UnifiedParams(LiteLLMBaseModel):
+    model_config = ConfigDict(extra="ignore", frozen=True)
+
+    max_results: StrictInt | None = None
+
+
+_DomainListAdapter: Final = TypeAdapter(tuple[str, ...])
+
+_NOTHING: Final[Mapping[str, object]] = MappingProxyType({})
+
+
+def _optional(key: str, value: object) -> Mapping[str, object]:
+    return MappingProxyType({key: value}) if value is not None else _NOTHING
+
+
+class ScavioSearchConfig(BaseSearchConfig):
+    SCAVIO_API_BASE = "https://api.scavio.dev"
+
+    @staticmethod
+    def ui_friendly_name() -> str:
+        return "Scavio"
+
+    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
+        resolved_api_key: Final = self.resolve_server_api_key(
+            caller_api_key=api_key,
+            caller_api_base=api_base,
+            key_env_vars=("SCAVIO_API_KEY",),
+            base_env_var=None,
+            default_api_base=self.SCAVIO_API_BASE,
+        )
+        if not resolved_api_key:
+            raise ValueError("SCAVIO_API_KEY is not set. Set `SCAVIO_API_KEY` environment variable.")
+        return {
+            **headers,
+            "Authorization": f"Bearer {resolved_api_key}",
+            "Content-Type": "application/json",
+        }
+
+    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:
+        resolved_base: Final = (api_base or self.SCAVIO_API_BASE).rstrip("/")
+        if resolved_base.endswith(_SEARCH_PATH):
+            return resolved_base
+        return f"{resolved_base}{_SEARCH_PATH}"
+
+    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
+        """
+        Scavio returns one Google page per call and has no result-count param, so `max_results` is
+        applied to the parsed response instead. `resolve_ai_overview` defaults to false because only
+        organic results are mapped and resolving a deferred AI Overview costs an extra upstream round trip
+        """
+        unified_params: Final = self.get_supported_perplexity_optional_params()
+        country: Final = optional_params.get("country")
+        joined_query: Final = " ".join(query) if isinstance(query, list) else query
+
+        passthrough: Final = MappingProxyType(
+            {param: value for param, value in optional_params.items() if param not in unified_params}
+        )
+
+        return {
+            "resolve_ai_overview": False,
+            **_optional("gl", country.lower() if isinstance(country, str) else None),
+            **passthrough,
+            "query": _with_domain_filter(joined_query, optional_params.get("search_domain_filter")),
+        }
+
+    def transform_search_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        **kwargs: object,  # kwargs-ok: BaseSearchConfig.transform_search_response signature
+    ) -> SearchResponse:
+        try:
+            parsed: Final = _ScavioSearchResponse.model_validate_json(raw_response.content)
+        except ValidationError as e:
+            raise self.get_error_class(
+                error_message=f"response does not match the documented /api/v2/google schema: {e}",
+                status_code=raw_response.status_code,
+                headers=dict(raw_response.headers),
+            )
+
+        max_results: Final = _requested_max_results(kwargs.get("optional_params"))
+        organic: Final = parsed.organic_results[:max_results] if max_results is not None else parsed.organic_results
+
+        return SearchResponse(
+            results=[
+                SearchResult(
+                    title=result.title or "",
+                    url=result.link or "",
+                    snippet=result.snippet or "",
+                    date=None,
+                    last_updated=None,
+                )
+                for result in organic
+            ],
+            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(". ")
+        return BaseLLMException(
+            status_code=status_code,
+            message=f"Scavio Search: {detail}. See {_SCAVIO_DOCS_URL} for details.",
+            headers=headers,
+        )
+
+
+def _unwrap_error_detail(error_message: str) -> str:
+    try:
+        body: Final = _ErrorEnvelope.model_validate_json(error_message)
+    except ValidationError:
+        return error_message
+    return body.error or error_message
+
+
+def _with_domain_filter(query: str, search_domain_filter: object) -> str:
+    """
+    Google ignores the `site:` filter on `(query) (site:a)` and on an ungrouped `site:a OR site:b`,
+    so the query stays bare, one include is `site:a`, and two or more are grouped as `(site:a OR site:b)`
+    """
+    try:
+        domains: Final = _DomainListAdapter.validate_python(search_domain_filter)
+    except ValidationError:
+        return query
+    included: Final = tuple(d for d in domains if d and not d.startswith("-"))
+    excluded: Final = tuple(d[1:] for d in domains if d.startswith("-") and len(d) > 1)
+    include_sites: Final = tuple(f"site:{d}" for d in included)
+    include_clause: Final = f"({' OR '.join(include_sites)})" if len(include_sites) > 1 else "".join(include_sites)
+    operators: Final = tuple(clause for clause in (include_clause, *(f"-site:{d}" for d in excluded)) if clause)
+    return " ".join((query, *operators))
+
+
+def _requested_max_results(optional_params: object) -> int | None:
+    try:
+        max_results: Final = _UnifiedParams.model_validate(optional_params).max_results
+    except ValidationError:
+        return None
+    return max_results if max_results is not None and max_results > 0 else None

litellm/model_prices_and_context_window_backup.json

@@ -25496,6 +25496,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."
         }
     },
+    "scavio/search": {
+        "input_cost_per_query": 0.01,
+        "litellm_provider": "scavio",
+        "mode": "search",
+        "metadata": {
+            "notes": "Scavio Google Search API pay-as-you-go list price: $0.01 per search (1 credit). Subscription plans price lower per credit."
+        }
+    },
     "elevenlabs/scribe_v1": {
         "input_cost_per_second": 6.11e-05,
         "litellm_provider": "elevenlabs",

litellm/types/utils.py

@@ -4853,6 +4853,7 @@ class SearchProviders(str, Enum):
     TINYFISH = "tinyfish"
     AGENTCORE = "agentcore"
     NIMBLE = "nimble"
+    SCAVIO = "scavio"
     BING_GROUNDING = "bing_grounding"
     DEEPSEEK = "deepseek"
     XAI = "xai"

litellm/utils.py

@@ -11479,6 +11479,7 @@ class ProviderConfigManager:
             ParallelAISearchConfig,
         )
         from litellm.llms.perplexity.search.transformation import PerplexitySearchConfig
+        from litellm.llms.scavio.search.transformation import ScavioSearchConfig
         from litellm.llms.searchapi.search.transformation import SearchAPIConfig
         from litellm.llms.searxng.search.transformation import SearXNGSearchConfig
         from litellm.llms.serper.search.transformation import SerperSearchConfig
@@ -11507,6 +11508,7 @@ class ProviderConfigManager:
             SearchProviders.TINYFISH: TinyfishSearchConfig,
             SearchProviders.AGENTCORE: AgentCoreSearchConfig,
             SearchProviders.NIMBLE: NimbleSearchConfig,
+            SearchProviders.SCAVIO: ScavioSearchConfig,
             SearchProviders.BING_GROUNDING: BingGroundingSearchConfig,
             SearchProviders.DEEPSEEK: DeepSeekSearchConfig,
             SearchProviders.XAI: XAISearchConfig,

model_prices_and_context_window.json

@@ -25496,6 +25496,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."
         }
     },
+    "scavio/search": {
+        "input_cost_per_query": 0.01,
+        "litellm_provider": "scavio",
+        "mode": "search",
+        "metadata": {
+            "notes": "Scavio Google Search API pay-as-you-go list price: $0.01 per search (1 credit). Subscription plans price lower per credit."
+        }
+    },
     "elevenlabs/scribe_v1": {
         "input_cost_per_second": 6.11e-05,
         "litellm_provider": "elevenlabs",

provider_endpoints_support.json

@@ -4254,6 +4254,13 @@
         "generateContent": false
       }
     },
+    "scavio": {
+      "display_name": "Scavio (`scavio`)",
+      "url": "https://scavio.dev/docs/search-api",
+      "endpoints": {
+        "search": true
+      }
+    },
     "triton": {
       "display_name": "Triton (`triton`)",
       "url": "https://docs.litellm.ai/docs/providers/triton-inference-server",

tests/code_coverage_tests/enforce_llms_folder_style.py

@@ -21,6 +21,7 @@ SEARCH_PROVIDERS = [
     "apiserpent",
     "tinyfish",
     "nimble",
+    "scavio",
 ]
 
 ALLOWED_FILES_IN_LLMS_FOLDER = [

tests/unit/llms/base_llm/search/test_base_search_transformation.py

@@ -31,6 +31,7 @@ from litellm.llms.linkup.search.transformation import LinkupSearchConfig
 from litellm.llms.nimble.search.transformation import NimbleSearchConfig
 from litellm.llms.parallel_ai.search.transformation import ParallelAISearchConfig
 from litellm.llms.perplexity.search.transformation import PerplexitySearchConfig
+from litellm.llms.scavio.search.transformation import ScavioSearchConfig
 from litellm.llms.searchapi.search.transformation import SearchAPIConfig
 from litellm.llms.searxng.search.transformation import SearXNGSearchConfig
 from litellm.llms.serper.search.transformation import SerperSearchConfig
@@ -101,6 +102,7 @@ PROVIDERS: Tuple[ProviderSpec, ...] = (
     (TinyfishSearchConfig, {"TINYFISH_API_KEY": "srv"}, "caller-key", {}),
     (FastCRWSearchConfig, {"CRW_API_KEY": "srv"}, "caller-key", {}),
     (NimbleSearchConfig, {"NIMBLE_API_KEY": "srv"}, "caller-key", {}),
+    (ScavioSearchConfig, {"SCAVIO_API_KEY": "srv"}, "caller-key", {}),
     (BingGroundingSearchConfig, {"BING_GROUNDING_TOKEN": "srv"}, "caller-key", {}),
 )
 

tests/unit/llms/scavio/search/test_scavio_search_transformation.py

@@ -0,0 +1,361 @@
+import datetime
+import json
+from collections.abc import Mapping, Sequence
+from typing import Final
+
+import httpx
+import pytest
+
+import litellm
+from litellm.litellm_core_utils.litellm_logging import Logging
+from litellm.llms.base_llm.search.transformation import SearchResponse
+from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
+from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
+from litellm.llms.scavio.search.transformation import ScavioSearchConfig
+from litellm.search.cost_calculator import search_provider_cost_per_query
+from litellm.types.utils import SearchProviders
+from litellm.utils import ProviderConfigManager
+
+SCAVIO_URL: Final = "https://api.scavio.dev/api/v2/google"
+
+LIVE_RESPONSE: Final[Mapping[str, object]] = {
+    "search_parameters": {"q": "open source llm gateway", "hl": "en", "gl": "us", "device": "desktop", "start": 0},
+    "search_information": {"query_displayed": "open source llm gateway", "total_results": 2},
+    "organic_results": [
+        {
+            "position": 1,
+            "title": "LiteLLM - GitHub",
+            "link": "https://github.com/BerriAI/litellm",
+            "displayed_link": "https://github.com > BerriAI > litellm",
+            "snippet": "Python SDK, Proxy Server (AI Gateway) to call 100+ LLM APIs in OpenAI format.",
+            "snippet_highlighted_words": ["AI Gateway"],
+            "source": "github.com",
+        },
+        {
+            "position": 2,
+            "title": "Is there an open-source LLM gateway with runtime routing?",
+            "link": "https://www.reddit.com/r/SelfHostedAI/comments/example",
+            "source": "reddit.com",
+        },
+    ],
+    "related_searches": [{"query": "llm gateway comparison"}],
+    "cached": False,
+    "response_time": 3.1,
+    "credits_used": 1,
+    "credits_remaining": 99,
+}
+
+
+def _config() -> ScavioSearchConfig:
+    return ScavioSearchConfig()
+
+
+def _logging_obj() -> Logging:
+    return Logging(
+        model="scavio/search",
+        messages=[],
+        stream=False,
+        call_type="search",
+        start_time=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
+        litellm_call_id="test-call",
+        function_id="test-function",
+    )
+
+
+def _resp(payload: Mapping[str, object] | str, status_code: int = 200) -> httpx.Response:
+    body: Final = payload if isinstance(payload, str) else json.dumps(payload)
+    return httpx.Response(status_code, content=body.encode(), request=httpx.Request("POST", SCAVIO_URL))
+
+
+def _result(title: str = "Test Title", link: str = "https://example.com") -> Mapping[str, object]:
+    return {
+        "position": 1,
+        "title": title,
+        "link": link,
+        "displayed_link": f"{link} > docs",
+        "snippet": "Test snippet",
+        "source": "example.com",
+    }
+
+
+class _ScavioStub:
+    def __init__(self, status_code: int, body: Mapping[str, object]) -> None:
+        self.status_code: Final = status_code
+        self.body: Final = body
+        self.requests: tuple[httpx.Request, ...] = ()
+
+    def __call__(self, request: httpx.Request) -> httpx.Response:
+        self.requests = (*self.requests, request)
+        return httpx.Response(self.status_code, content=json.dumps(self.body).encode())
+
+
+def test_ui_friendly_name() -> None:
+    assert _config().ui_friendly_name() == "Scavio"
+
+
+def test_search_provider_resolves_to_scavio_config() -> None:
+    config: Final = ProviderConfigManager.get_provider_search_config(provider=SearchProviders("scavio"))
+    assert isinstance(config, ScavioSearchConfig)
+
+
+def test_validate_environment_with_explicit_key() -> None:
+    headers: Final = _config().validate_environment({}, api_key="explicit-key")
+    assert headers["Authorization"] == "Bearer explicit-key"
+    assert headers["Content-Type"] == "application/json"
+
+
+def test_validate_environment_reads_env_key(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("SCAVIO_API_KEY", "env-key")
+    assert _config().validate_environment({})["Authorization"] == "Bearer env-key"
+
+
+def test_validate_environment_missing_key_raises(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("SCAVIO_API_KEY", raising=False)
+    with pytest.raises(ValueError, match="SCAVIO_API_KEY"):
+        _config().validate_environment({})
+
+
+def test_validate_environment_does_not_mutate_and_is_idempotent() -> None:
+    config: Final = _config()
+    caller_headers: Final = {"X-Custom": "keep-me"}
+
+    once: Final = config.validate_environment(caller_headers, api_key="k")
+    twice: Final = config.validate_environment(once, api_key="k")
+
+    assert caller_headers == {"X-Custom": "keep-me"}
+    assert once == twice
+    assert once["X-Custom"] == "keep-me"
+
+
+def test_get_complete_url_default_base() -> None:
+    assert _config().get_complete_url(None, {}) == SCAVIO_URL
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://proxy.local",
+        "https://proxy.local/",
+        "https://proxy.local/api/v2/google",
+        "https://proxy.local/api/v2/google/",
+    ],
+)
+def test_get_complete_url_appends_path_exactly_once(api_base: str) -> None:
+    assert _config().get_complete_url(api_base, {}) == "https://proxy.local/api/v2/google"
+
+
+def test_transform_search_request_joins_list_query() -> None:
+    assert _config().transform_search_request(["foo", "bar"], {})["query"] == "foo bar"
+
+
+def test_transform_search_request_maps_country_to_lowercase_gl() -> None:
+    assert _config().transform_search_request("q", {"country": "GB"})["gl"] == "gb"
+
+
+def test_transform_search_request_explicit_gl_wins_over_country() -> None:
+    assert _config().transform_search_request("q", {"country": "GB", "gl": "de"})["gl"] == "de"
+
+
+@pytest.mark.parametrize("unified_param", ["max_results", "max_tokens_per_page", "country", "search_domain_filter"])
+def test_transform_search_request_does_not_leak_unified_params(unified_param: str) -> None:
+    data: Final = _config().transform_search_request(
+        "q",
+        {"max_results": 3, "max_tokens_per_page": 1024, "country": "us", "search_domain_filter": ["a.com"]},
+    )
+    assert unified_param not in data
+
+
+def test_transform_search_request_skips_ai_overview_resolution_by_default() -> None:
+    assert _config().transform_search_request("q", {})["resolve_ai_overview"] is False
+
+
+def test_transform_search_request_caller_can_enable_ai_overview_resolution() -> None:
+    assert _config().transform_search_request("q", {"resolve_ai_overview": True})["resolve_ai_overview"] is True
+
+
+@pytest.mark.parametrize(
+    "domains, expected",
+    [
+        (["arxiv.org"], "q site:arxiv.org"),
+        (["arxiv.org", "nature.com"], "q (site:arxiv.org OR site:nature.com)"),
+        (["arxiv.org", "nature.com", "science.org"], "q (site:arxiv.org OR site:nature.com OR site:science.org)"),
+        (["-spam.com"], "q -site:spam.com"),
+        (["arxiv.org", "-spam.com", "-junk.net"], "q site:arxiv.org -site:spam.com -site:junk.net"),
+        (["-spam.com", "arxiv.org", "nature.com"], "q (site:arxiv.org OR site:nature.com) -site:spam.com"),
+        ([], "q"),
+        (["", "-"], "q"),
+        ("arxiv.org", "q"),
+    ],
+)
+def test_transform_search_request_folds_domain_filter_into_query(domains: Sequence[str], expected: str) -> None:
+    assert _config().transform_search_request("q", {"search_domain_filter": domains})["query"] == expected
+
+
+def test_transform_search_request_without_domain_filter_leaves_query_unchanged() -> None:
+    assert _config().transform_search_request("latest AI developments", {})["query"] == "latest AI developments"
+
+
+def test_transform_search_request_never_wraps_user_query_in_parentheses() -> None:
+    data: Final = _config().transform_search_request(
+        ["latest", "AI developments"], {"search_domain_filter": ["github.com", "gitlab.com"]}
+    )
+    assert data["query"] == "latest AI developments (site:github.com OR site:gitlab.com)"
+
+
+def test_transform_search_request_forwards_scavio_params() -> None:
+    data: Final = _config().transform_search_request(
+        "q", {"hl": "fr", "location": "Paris,France", "device": "mobile", "start": 10, "time_period": "last_week"}
+    )
+    assert (data["hl"], data["location"], data["device"], data["start"], data["time_period"]) == (
+        "fr",
+        "Paris,France",
+        "mobile",
+        10,
+        "last_week",
+    )
+
+
+def test_transform_search_response_maps_organic_results_in_order() -> None:
+    resp: Final = _config().transform_search_response(
+        _resp({"organic_results": [_result(title=t, link=f"https://{t}.com") for t in ("first", "second", "third")]}),
+        logging_obj=_logging_obj(),
+    )
+    assert [(r.title, r.url) for r in resp.results] == [
+        ("first", "https://first.com"),
+        ("second", "https://second.com"),
+        ("third", "https://third.com"),
+    ]
+    assert resp.results[0].snippet == "Test snippet"
+    assert resp.results[0].date is None
+
+
+def test_transform_search_response_result_without_snippet_does_not_fail_the_call() -> None:
+    no_snippet: Final = {"position": 4, "title": "Forum thread", "link": "https://forum.example"}
+    resp: Final = _config().transform_search_response(
+        _resp({"organic_results": [no_snippet, _result()]}), logging_obj=_logging_obj()
+    )
+    assert [r.snippet for r in resp.results] == ["", "Test snippet"]
+
+
+@pytest.mark.parametrize("max_results, expected_count", [(2, 2), (10, 5), (0, 5), (True, 5), (None, 5)])
+def test_transform_search_response_caps_to_max_results(max_results: int | bool | None, expected_count: int) -> None:
+    resp: Final = _config().transform_search_response(
+        _resp({"organic_results": [_result(title=str(i)) for i in range(5)]}),
+        logging_obj=_logging_obj(),
+        optional_params={"max_results": max_results},
+    )
+    assert [r.title for r in resp.results] == [str(i) for i in range(expected_count)]
+
+
+def test_transform_search_response_zero_hits() -> None:
+    payload: Final[Mapping[str, object]] = {
+        "search_information": {"total_results": 0},
+        "organic_results": [],
+        "credits_used": 1,
+    }
+    assert _config().transform_search_response(_resp(payload), logging_obj=_logging_obj()).results == []
+
+
+@pytest.mark.parametrize(
+    "body",
+    [
+        "<html>502 Bad Gateway</html>",
+        '{"organic_results": ["garbage"]}',
+        '{"organic_results": {"unexpected": "shape"}}',
+        '{"organic_results": null}',
+        "{}",
+    ],
+)
+def test_transform_search_response_malformed_body_raises_instead_of_reporting_empty(body: str) -> None:
+    with pytest.raises(Exception, match="Scavio Search"):
+        _config().transform_search_response(_resp(body, status_code=502), logging_obj=_logging_obj())
+
+
+def test_get_error_class_unwraps_scavio_error_envelope() -> None:
+    error: Final = _config().get_error_class(error_message='{"error":"Invalid API key"}', status_code=401, headers={})
+    assert getattr(error, "status_code", None) == 401
+    assert str(error) == "Scavio Search: Invalid API key. See https://scavio.dev/docs/search-api for details."
+
+
+@pytest.mark.parametrize("body", ["<html>502 Bad Gateway</html>", '{"error": null}'])
+def test_get_error_class_falls_back_to_the_raw_body(body: str) -> None:
+    assert f"Scavio Search: {body}." in str(_config().get_error_class(body, status_code=502, headers={}))
+
+
+def test_search_handler_sends_mapped_request_and_parses_response() -> None:
+    stub: Final = _ScavioStub(200, LIVE_RESPONSE)
+    client: Final = HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(stub)))
+
+    response: Final = BaseLLMHTTPHandler().search(
+        query="open source llm gateway",
+        optional_params={"max_results": 1, "country": "US", "search_domain_filter": ["github.com"]},
+        timeout=10.0,
+        logging_obj=_logging_obj(),
+        api_key="test-api-key",
+        api_base=None,
+        custom_llm_provider="scavio",
+        client=client,
+        provider_config=_config(),
+    )
+
+    assert isinstance(response, SearchResponse)
+    assert [str(r.url) for r in stub.requests] == [SCAVIO_URL]
+    assert stub.requests[0].headers["Authorization"] == "Bearer test-api-key"
+    assert json.loads(stub.requests[0].content) == {
+        "resolve_ai_overview": False,
+        "gl": "us",
+        "query": "open source llm gateway site:github.com",
+    }
+    assert [(r.title, r.url, r.snippet) for r in response.results] == [
+        (
+            "LiteLLM - GitHub",
+            "https://github.com/BerriAI/litellm",
+            "Python SDK, Proxy Server (AI Gateway) to call 100+ LLM APIs in OpenAI format.",
+        )
+    ]
+
+
+def test_search_handler_surfaces_scavio_error_message() -> None:
+    stub: Final = _ScavioStub(401, {"error": "Invalid API key"})
+    client: Final = HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(stub)))
+
+    with pytest.raises(Exception, match="Scavio Search: Invalid API key"):
+        BaseLLMHTTPHandler().search(
+            query="q",
+            optional_params={},
+            timeout=10.0,
+            logging_obj=_logging_obj(),
+            api_key="wrong-key",
+            api_base=None,
+            custom_llm_provider="scavio",
+            client=client,
+            provider_config=_config(),
+        )
+    assert len(stub.requests) == 1
+
+
+@pytest.mark.asyncio
+async def test_async_search_handler_sends_mapped_request_and_parses_response() -> None:
+    stub: Final = _ScavioStub(200, LIVE_RESPONSE)
+    client: Final = AsyncHTTPHandler(transport=httpx.MockTransport(stub))
+
+    response: Final = await BaseLLMHTTPHandler().async_search(
+        query=["open source", "llm gateway"],
+        optional_params={"max_results": 1},
+        timeout=10.0,
+        logging_obj=_logging_obj(),
+        api_key="test-api-key",
+        api_base=None,
+        custom_llm_provider="scavio",
+        client=client,
+        provider_config=_config(),
+    )
+
+    assert json.loads(stub.requests[0].content)["query"] == "open source llm gateway"
+    assert [r.title for r in response.results] == ["LiteLLM - GitHub"]
+
+
+def test_search_cost_comes_from_the_cost_map() -> None:
+    expected: Final[float] = litellm.model_cost["scavio/search"]["input_cost_per_query"]
+    assert expected > 0
+    assert search_provider_cost_per_query(model="scavio/search", custom_llm_provider="scavio") == (expected, 0.0)