Edit on GitHub

agent_search_gateway.providers.web.tinyfish

TinyFish search and fetch adapter using separate public endpoints.

View Source

 1"""TinyFish search and fetch adapter using separate public endpoints."""
 2
 3from urllib.parse import urlencode
 4
 5from ...errors import ExecutionFailure
 6from ...observability import SecretValue
 7from ...providers.contracts import KeywordSearchHit, URLFetchCandidate
 8from ...url_normalization import NormalizedURL
 9from .common import (
10    JsonRequester,
11    failure,
12    non_empty_string,
13    normalized_match,
14    optional_string,
15    require_list,
16    require_object,
17)
18
19
20class TinyFishAdapter:
21    def __init__(
22        self,
23        *,
24        name: str,
25        search_api_url: str,
26        fetch_api_url: str,
27        secret: SecretValue,
28        http_executor: JsonRequester,
29    ) -> None:
30        self.name = name
31        self._search_api_url = search_api_url.rstrip("/")
32        self._fetch_api_url = fetch_api_url.rstrip("/")
33        self._secret = secret
34        self._http = http_executor
35
36    @property
37    def _headers(self) -> dict[str, str]:
38        return {"X-API-Key": self._secret.reveal()}
39
40    async def search(self, query: str) -> list[KeywordSearchHit]:
41        request_url = f"{self._search_api_url}?{urlencode({'query': query})}"
42        payload = await self._http.request_json(
43            "GET",
44            request_url,
45            stage="search",
46            headers=self._headers,
47        )
48        root = require_object(payload, self.name, "search", "response")
49        results = require_list(root.get("results"), self.name, "search", "results")
50        hits: list[KeywordSearchHit] = []
51        for item in results:
52            try:
53                result = require_object(item, self.name, "search", "result")
54                url = non_empty_string(result.get("url"), self.name, "search", "result.url")
55                hits.append(
56                    KeywordSearchHit(
57                        url=url,
58                        title=optional_string(
59                            result.get("title"), self.name, "search", "result.title"
60                        ),
61                        snippet=optional_string(
62                            result.get("snippet"), self.name, "search", "result.snippet"
63                        ),
64                    )
65                )
66            except ExecutionFailure:
67                continue
68        return hits
69
70    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
71        payload = await self._http.request_json(
72            "POST",
73            self._fetch_api_url,
74            stage="fetch",
75            headers=self._headers,
76            json_body={
77                "urls": [str(url)],
78                "format": "markdown",
79                "links": False,
80                "image_links": False,
81            },
82        )
83        root = require_object(payload, self.name, "fetch", "response")
84        results = require_list(root.get("results"), self.name, "fetch", "results")
85        for item in results:
86            result = require_object(item, self.name, "fetch", "result")
87            if normalized_match(result.get("url"), url, self.name, "fetch"):
88                text = non_empty_string(result.get("text"), self.name, "fetch", "result.text")
89                return URLFetchCandidate(raw_content=text, content=text)
90
91        errors_value = root.get("errors", [])
92        errors = require_list(errors_value, self.name, "fetch", "errors")
93        for item in errors:
94            error = require_object(item, self.name, "fetch", "error")
95            if normalized_match(error.get("url"), url, self.name, "fetch"):
96                raise failure(self.name, "fetch", "provider reported per-URL failure")
97        raise failure(self.name, "fetch", "matching fetch result was not returned")

class TinyFishAdapter: View Source

21class TinyFishAdapter:
22    def __init__(
23        self,
24        *,
25        name: str,
26        search_api_url: str,
27        fetch_api_url: str,
28        secret: SecretValue,
29        http_executor: JsonRequester,
30    ) -> None:
31        self.name = name
32        self._search_api_url = search_api_url.rstrip("/")
33        self._fetch_api_url = fetch_api_url.rstrip("/")
34        self._secret = secret
35        self._http = http_executor
36
37    @property
38    def _headers(self) -> dict[str, str]:
39        return {"X-API-Key": self._secret.reveal()}
40
41    async def search(self, query: str) -> list[KeywordSearchHit]:
42        request_url = f"{self._search_api_url}?{urlencode({'query': query})}"
43        payload = await self._http.request_json(
44            "GET",
45            request_url,
46            stage="search",
47            headers=self._headers,
48        )
49        root = require_object(payload, self.name, "search", "response")
50        results = require_list(root.get("results"), self.name, "search", "results")
51        hits: list[KeywordSearchHit] = []
52        for item in results:
53            try:
54                result = require_object(item, self.name, "search", "result")
55                url = non_empty_string(result.get("url"), self.name, "search", "result.url")
56                hits.append(
57                    KeywordSearchHit(
58                        url=url,
59                        title=optional_string(
60                            result.get("title"), self.name, "search", "result.title"
61                        ),
62                        snippet=optional_string(
63                            result.get("snippet"), self.name, "search", "result.snippet"
64                        ),
65                    )
66                )
67            except ExecutionFailure:
68                continue
69        return hits
70
71    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
72        payload = await self._http.request_json(
73            "POST",
74            self._fetch_api_url,
75            stage="fetch",
76            headers=self._headers,
77            json_body={
78                "urls": [str(url)],
79                "format": "markdown",
80                "links": False,
81                "image_links": False,
82            },
83        )
84        root = require_object(payload, self.name, "fetch", "response")
85        results = require_list(root.get("results"), self.name, "fetch", "results")
86        for item in results:
87            result = require_object(item, self.name, "fetch", "result")
88            if normalized_match(result.get("url"), url, self.name, "fetch"):
89                text = non_empty_string(result.get("text"), self.name, "fetch", "result.text")
90                return URLFetchCandidate(raw_content=text, content=text)
91
92        errors_value = root.get("errors", [])
93        errors = require_list(errors_value, self.name, "fetch", "errors")
94        for item in errors:
95            error = require_object(item, self.name, "fetch", "error")
96            if normalized_match(error.get("url"), url, self.name, "fetch"):
97                raise failure(self.name, "fetch", "provider reported per-URL failure")
98        raise failure(self.name, "fetch", "matching fetch result was not returned")

TinyFishAdapter( *, name: str, search_api_url: str, fetch_api_url: str, secret: agent_search_gateway.observability.SecretValue, http_executor: agent_search_gateway.providers.web.common.JsonRequester) View Source

22    def __init__(
23        self,
24        *,
25        name: str,
26        search_api_url: str,
27        fetch_api_url: str,
28        secret: SecretValue,
29        http_executor: JsonRequester,
30    ) -> None:
31        self.name = name
32        self._search_api_url = search_api_url.rstrip("/")
33        self._fetch_api_url = fetch_api_url.rstrip("/")
34        self._secret = secret
35        self._http = http_executor

name

async def search( self, query: str) -> list[agent_search_gateway.providers.contracts.KeywordSearchHit]: View Source

41    async def search(self, query: str) -> list[KeywordSearchHit]:
42        request_url = f"{self._search_api_url}?{urlencode({'query': query})}"
43        payload = await self._http.request_json(
44            "GET",
45            request_url,
46            stage="search",
47            headers=self._headers,
48        )
49        root = require_object(payload, self.name, "search", "response")
50        results = require_list(root.get("results"), self.name, "search", "results")
51        hits: list[KeywordSearchHit] = []
52        for item in results:
53            try:
54                result = require_object(item, self.name, "search", "result")
55                url = non_empty_string(result.get("url"), self.name, "search", "result.url")
56                hits.append(
57                    KeywordSearchHit(
58                        url=url,
59                        title=optional_string(
60                            result.get("title"), self.name, "search", "result.title"
61                        ),
62                        snippet=optional_string(
63                            result.get("snippet"), self.name, "search", "result.snippet"
64                        ),
65                    )
66                )
67            except ExecutionFailure:
68                continue
69        return hits

async def fetch( self, url: agent_search_gateway.url_normalization.NormalizedURL) -> agent_search_gateway.providers.contracts.URLFetchCandidate: View Source

71    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
72        payload = await self._http.request_json(
73            "POST",
74            self._fetch_api_url,
75            stage="fetch",
76            headers=self._headers,
77            json_body={
78                "urls": [str(url)],
79                "format": "markdown",
80                "links": False,
81                "image_links": False,
82            },
83        )
84        root = require_object(payload, self.name, "fetch", "response")
85        results = require_list(root.get("results"), self.name, "fetch", "results")
86        for item in results:
87            result = require_object(item, self.name, "fetch", "result")
88            if normalized_match(result.get("url"), url, self.name, "fetch"):
89                text = non_empty_string(result.get("text"), self.name, "fetch", "result.text")
90                return URLFetchCandidate(raw_content=text, content=text)
91
92        errors_value = root.get("errors", [])
93        errors = require_list(errors_value, self.name, "fetch", "errors")
94        for item in errors:
95            error = require_object(item, self.name, "fetch", "error")
96            if normalized_match(error.get("url"), url, self.name, "fetch"):
97                raise failure(self.name, "fetch", "provider reported per-URL failure")
98        raise failure(self.name, "fetch", "matching fetch result was not returned")