agent_search_gateway.providers.web.firecrawl

  1"""Firecrawl v2 search and scrape 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    optional_string,
 13    require_list,
 14    require_object,
 15)
 16
 17
 18def _v2_endpoint(api_url: str, suffix: str) -> str:
 19    base = api_url.rstrip("/")
 20    if base.endswith("/v2"):
 21        return endpoint(base, suffix)
 22    return endpoint(base, f"/v2/{suffix.lstrip('/')}")
 23
 24
 25def _page_body(
 26    result: dict[str, object],
 27    provider: str,
 28    stage: str,
 29) -> URLFetchCandidate:
 30    markdown = optional_string(result.get("markdown"), provider, stage, "markdown")
 31    raw_html = optional_string(result.get("rawHtml"), provider, stage, "rawHtml")
 32    raw = raw_html or markdown
 33    if not raw.strip():
 34        raise failure(provider, stage, "page body is empty")
 35    return URLFetchCandidate(raw_content=raw, content=markdown)
 36
 37
 38class FirecrawlAdapter:
 39    def __init__(
 40        self,
 41        *,
 42        name: str,
 43        api_url: str,
 44        secret: SecretValue,
 45        http_executor: JsonRequester,
 46    ) -> None:
 47        self.name = name
 48        self._api_url = api_url
 49        self._secret = secret
 50        self._http = http_executor
 51
 52    @property
 53    def _headers(self) -> dict[str, str]:
 54        return {"Authorization": f"Bearer {self._secret.reveal()}"}
 55
 56    async def search(self, query: str) -> list[KeywordSearchHit]:
 57        payload = await self._http.request_json(
 58            "POST",
 59            _v2_endpoint(self._api_url, "search"),
 60            stage="search",
 61            headers=self._headers,
 62            json_body={
 63                "query": query,
 64                "sources": ["web"],
 65                "scrapeOptions": {"formats": ["markdown", "rawHtml"]},
 66            },
 67        )
 68        root = self._successful_root(payload, "search")
 69        data = require_object(root.get("data"), self.name, "search", "data")
 70        web = require_list(data.get("web"), self.name, "search", "data.web")
 71        hits: list[KeywordSearchHit] = []
 72        for item in web:
 73            try:
 74                result = require_object(item, self.name, "search", "result")
 75                url = non_empty_string(result.get("url"), self.name, "search", "result.url")
 76                markdown = optional_string(
 77                    result.get("markdown"), self.name, "search", "result.markdown"
 78                )
 79                raw_html = optional_string(
 80                    result.get("rawHtml"), self.name, "search", "result.rawHtml"
 81                )
 82                hits.append(
 83                    KeywordSearchHit(
 84                        url=url,
 85                        title=optional_string(
 86                            result.get("title"), self.name, "search", "result.title"
 87                        ),
 88                        snippet=optional_string(
 89                            result.get("description"),
 90                            self.name,
 91                            "search",
 92                            "result.description",
 93                        ),
 94                        raw_content=raw_html or markdown,
 95                        content=markdown,
 96                    )
 97                )
 98            except ExecutionFailure:
 99                continue
100        return hits
101
102    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
103        payload = await self._http.request_json(
104            "POST",
105            _v2_endpoint(self._api_url, "scrape"),
106            stage="fetch",
107            headers=self._headers,
108            json_body={"url": str(url), "formats": ["markdown", "rawHtml"]},
109        )
110        root = self._successful_root(payload, "fetch")
111        data = require_object(root.get("data"), self.name, "fetch", "data")
112        return _page_body(data, self.name, "fetch")
113
114    def _successful_root(self, payload: object, stage: str) -> dict[str, object]:
115        root = require_object(payload, self.name, stage, "response")
116        if root.get("success") is not True:
117            raise failure(self.name, stage, "provider reported failure")
118        return root