agent_search_gateway.providers.web.exa

  1"""Exa search and contents adapter."""
  2
  3from ...errors import ExecutionFailure
  4from ...observability import SecretValue
  5from ...providers.contracts import KeywordSearchHit, URLFetchCandidate
  6from ...url_normalization import NormalizedURL
  7from .common import (
  8    JsonRequester,
  9    endpoint,
 10    failure,
 11    non_empty_string,
 12    normalized_match,
 13    optional_string,
 14    require_list,
 15    require_object,
 16)
 17
 18
 19class ExaAdapter:
 20    def __init__(
 21        self,
 22        *,
 23        name: str,
 24        api_url: str,
 25        secret: SecretValue,
 26        http_executor: JsonRequester,
 27    ) -> None:
 28        self.name = name
 29        self._api_url = api_url
 30        self._secret = secret
 31        self._http = http_executor
 32
 33    @property
 34    def _headers(self) -> dict[str, str]:
 35        return {"x-api-key": self._secret.reveal()}
 36
 37    async def search(self, query: str) -> list[KeywordSearchHit]:
 38        payload = await self._http.request_json(
 39            "POST",
 40            endpoint(self._api_url, "/search"),
 41            stage="search",
 42            headers=self._headers,
 43            json_body={"query": query, "contents": {"text": True, "highlights": True}},
 44        )
 45        root = require_object(payload, self.name, "search", "response")
 46        results = require_list(root.get("results"), self.name, "search", "results")
 47        hits: list[KeywordSearchHit] = []
 48        for item in results:
 49            try:
 50                result = require_object(item, self.name, "search", "result")
 51                url = non_empty_string(result.get("url"), self.name, "search", "result.url")
 52                title = optional_string(result.get("title"), self.name, "search", "result.title")
 53                text = optional_string(result.get("text"), self.name, "search", "result.text")
 54                snippet = self._snippet(result, title)
 55                hits.append(
 56                    KeywordSearchHit(
 57                        url=url,
 58                        title=title,
 59                        snippet=snippet,
 60                        raw_content=text,
 61                        content=text,
 62                    )
 63                )
 64            except ExecutionFailure:
 65                continue
 66        return hits
 67
 68    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
 69        payload = await self._http.request_json(
 70            "POST",
 71            endpoint(self._api_url, "/contents"),
 72            stage="fetch",
 73            headers=self._headers,
 74            json_body={"urls": [str(url)], "text": True},
 75        )
 76        root = require_object(payload, self.name, "fetch", "response")
 77        self._require_success_status(root, url)
 78        results = require_list(root.get("results"), self.name, "fetch", "results")
 79        for item in results:
 80            result = require_object(item, self.name, "fetch", "result")
 81            if normalized_match(result.get("url"), url, self.name, "fetch"):
 82                text = non_empty_string(result.get("text"), self.name, "fetch", "result.text")
 83                return URLFetchCandidate(raw_content=text, content=text)
 84        raise failure(self.name, "fetch", "matching contents result was not returned")
 85
 86    def _snippet(self, result: dict[str, object], title: str) -> str:
 87        highlights_value = result.get("highlights")
 88        if highlights_value is not None:
 89            highlights = require_list(highlights_value, self.name, "search", "result.highlights")
 90            if highlights:
 91                first = optional_string(highlights[0], self.name, "search", "result.highlights[0]")
 92                if first.strip():
 93                    return first
 94        summary = optional_string(result.get("summary"), self.name, "search", "result.summary")
 95        return summary.strip() or title
 96
 97    def _require_success_status(self, root: dict[str, object], url: NormalizedURL) -> None:
 98        statuses = require_list(root.get("statuses"), self.name, "fetch", "statuses")
 99        for item in statuses:
100            status = require_object(item, self.name, "fetch", "status")
101            if normalized_match(status.get("id"), url, self.name, "fetch"):
102                state = non_empty_string(status.get("status"), self.name, "fetch", "status.status")
103                if state != "success":
104                    raise failure(self.name, "fetch", "provider reported per-URL failure")
105                return
106        raise failure(self.name, "fetch", "matching per-URL status was not returned")