Edit on GitHub

agent_search_gateway.providers.web.brave

Brave web search adapter.

View Source

 1"""Brave web search adapter."""
 2
 3from urllib.parse import urlencode
 4
 5from ...errors import ExecutionFailure
 6from ...observability import SecretValue
 7from ...providers.contracts import KeywordSearchHit
 8from .common import (
 9    JsonRequester,
10    endpoint,
11    non_empty_string,
12    optional_string,
13    require_list,
14    require_object,
15)
16
17
18class BraveAdapter:
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        base = endpoint(self._api_url, "/res/v1/web/search")
34        request_url = f"{base}?{urlencode({'q': query, 'count': 10})}"
35        payload = await self._http.request_json(
36            "GET",
37            request_url,
38            stage="search",
39            headers={"X-Subscription-Token": self._secret.reveal()},
40        )
41        root = require_object(payload, self.name, "search", "response")
42        web_value = root.get("web")
43        if web_value is None:
44            return []
45        web = require_object(web_value, self.name, "search", "web")
46        results_value = web.get("results")
47        if results_value is None:
48            return []
49        results = require_list(results_value, self.name, "search", "web.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("description"),
63                            self.name,
64                            "search",
65                            "result.description",
66                        ),
67                    )
68                )
69            except ExecutionFailure:
70                continue
71        return hits

class BraveAdapter: View Source

19class BraveAdapter:
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    async def search(self, query: str) -> list[KeywordSearchHit]:
34        base = endpoint(self._api_url, "/res/v1/web/search")
35        request_url = f"{base}?{urlencode({'q': query, 'count': 10})}"
36        payload = await self._http.request_json(
37            "GET",
38            request_url,
39            stage="search",
40            headers={"X-Subscription-Token": self._secret.reveal()},
41        )
42        root = require_object(payload, self.name, "search", "response")
43        web_value = root.get("web")
44        if web_value is None:
45            return []
46        web = require_object(web_value, self.name, "search", "web")
47        results_value = web.get("results")
48        if results_value is None:
49            return []
50        results = require_list(results_value, self.name, "search", "web.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("description"),
64                            self.name,
65                            "search",
66                            "result.description",
67                        ),
68                    )
69                )
70            except ExecutionFailure:
71                continue
72        return hits

BraveAdapter( *, name: str, api_url: str, secret: agent_search_gateway.observability.SecretValue, http_executor: agent_search_gateway.providers.web.common.JsonRequester) View Source

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

name

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

33    async def search(self, query: str) -> list[KeywordSearchHit]:
34        base = endpoint(self._api_url, "/res/v1/web/search")
35        request_url = f"{base}?{urlencode({'q': query, 'count': 10})}"
36        payload = await self._http.request_json(
37            "GET",
38            request_url,
39            stage="search",
40            headers={"X-Subscription-Token": self._secret.reveal()},
41        )
42        root = require_object(payload, self.name, "search", "response")
43        web_value = root.get("web")
44        if web_value is None:
45            return []
46        web = require_object(web_value, self.name, "search", "web")
47        results_value = web.get("results")
48        if results_value is None:
49            return []
50        results = require_list(results_value, self.name, "search", "web.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("description"),
64                            self.name,
65                            "search",
66                            "result.description",
67                        ),
68                    )
69                )
70            except ExecutionFailure:
71                continue
72        return hits