Edit on GitHub

agent_search_gateway.providers.web.anysearch

AnySearch unified search adapter.

 1"""AnySearch unified search adapter."""
 2
 3from ...errors import ExecutionFailure
 4from ...observability import SecretValue
 5from ...providers.contracts import KeywordSearchHit
 6from .common import (
 7    JsonRequester,
 8    endpoint,
 9    failure,
10    non_empty_string,
11    optional_string,
12    require_list,
13    require_object,
14)
15
16
17class AnySearchAdapter:
18    def __init__(
19        self,
20        *,
21        name: str,
22        api_url: str,
23        secret: SecretValue,
24        http_executor: JsonRequester,
25    ) -> None:
26        self.name = name
27        self._api_url = api_url
28        self._secret = secret
29        self._http = http_executor
30
31    async def search(self, query: str) -> list[KeywordSearchHit]:
32        payload = await self._http.request_json(
33            "POST",
34            endpoint(self._api_url, "/v1/search"),
35            stage="search",
36            headers={"Authorization": f"Bearer {self._secret.reveal()}"},
37            json_body={"query": query, "format": "json", "max_results": 10},
38        )
39        root = require_object(payload, self.name, "search", "response")
40        code = root.get("code")
41        if not isinstance(code, int) or isinstance(code, bool) or code != 0:
42            raise failure(self.name, "search", "provider reported failure")
43        data = require_object(root.get("data"), self.name, "search", "data")
44        results = require_list(data.get("results"), self.name, "search", "data.results")
45        hits: list[KeywordSearchHit] = []
46        for item in results:
47            try:
48                result = require_object(item, self.name, "search", "result")
49                url = non_empty_string(result.get("url"), self.name, "search", "result.url")
50                hits.append(
51                    KeywordSearchHit(
52                        url=url,
53                        title=optional_string(
54                            result.get("title"), self.name, "search", "result.title"
55                        ),
56                        snippet=optional_string(
57                            result.get("snippet"), self.name, "search", "result.snippet"
58                        ),
59                    )
60                )
61            except ExecutionFailure:
62                continue
63        return hits
class AnySearchAdapter: 
18class AnySearchAdapter:
19    def __init__(
20        self,
21        *,
22        name: str,
23        api_url: str,
24        secret: SecretValue,
25        http_executor: JsonRequester,
26    ) -> None:
27        self.name = name
28        self._api_url = api_url
29        self._secret = secret
30        self._http = http_executor
31
32    async def search(self, query: str) -> list[KeywordSearchHit]:
33        payload = await self._http.request_json(
34            "POST",
35            endpoint(self._api_url, "/v1/search"),
36            stage="search",
37            headers={"Authorization": f"Bearer {self._secret.reveal()}"},
38            json_body={"query": query, "format": "json", "max_results": 10},
39        )
40        root = require_object(payload, self.name, "search", "response")
41        code = root.get("code")
42        if not isinstance(code, int) or isinstance(code, bool) or code != 0:
43            raise failure(self.name, "search", "provider reported failure")
44        data = require_object(root.get("data"), self.name, "search", "data")
45        results = require_list(data.get("results"), self.name, "search", "data.results")
46        hits: list[KeywordSearchHit] = []
47        for item in results:
48            try:
49                result = require_object(item, self.name, "search", "result")
50                url = non_empty_string(result.get("url"), self.name, "search", "result.url")
51                hits.append(
52                    KeywordSearchHit(
53                        url=url,
54                        title=optional_string(
55                            result.get("title"), self.name, "search", "result.title"
56                        ),
57                        snippet=optional_string(
58                            result.get("snippet"), self.name, "search", "result.snippet"
59                        ),
60                    )
61                )
62            except ExecutionFailure:
63                continue
64        return hits
AnySearchAdapter( *, name: str, api_url: str, secret: agent_search_gateway.observability.SecretValue, http_executor: agent_search_gateway.providers.web.common.JsonRequester) 
19    def __init__(
20        self,
21        *,
22        name: str,
23        api_url: str,
24        secret: SecretValue,
25        http_executor: JsonRequester,
26    ) -> None:
27        self.name = name
28        self._api_url = api_url
29        self._secret = secret
30        self._http = http_executor
name
async def search( self, query: str) -> list[agent_search_gateway.providers.contracts.KeywordSearchHit]: 
32    async def search(self, query: str) -> list[KeywordSearchHit]:
33        payload = await self._http.request_json(
34            "POST",
35            endpoint(self._api_url, "/v1/search"),
36            stage="search",
37            headers={"Authorization": f"Bearer {self._secret.reveal()}"},
38            json_body={"query": query, "format": "json", "max_results": 10},
39        )
40        root = require_object(payload, self.name, "search", "response")
41        code = root.get("code")
42        if not isinstance(code, int) or isinstance(code, bool) or code != 0:
43            raise failure(self.name, "search", "provider reported failure")
44        data = require_object(root.get("data"), self.name, "search", "data")
45        results = require_list(data.get("results"), self.name, "search", "data.results")
46        hits: list[KeywordSearchHit] = []
47        for item in results:
48            try:
49                result = require_object(item, self.name, "search", "result")
50                url = non_empty_string(result.get("url"), self.name, "search", "result.url")
51                hits.append(
52                    KeywordSearchHit(
53                        url=url,
54                        title=optional_string(
55                            result.get("title"), self.name, "search", "result.title"
56                        ),
57                        snippet=optional_string(
58                            result.get("snippet"), self.name, "search", "result.snippet"
59                        ),
60                    )
61                )
62            except ExecutionFailure:
63                continue
64        return hits