agent_search_gateway.providers.web.linkup
Linkup search and fetch adapter.
1"""Linkup search and fetch 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 18class LinkupAdapter: 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 @property 33 def _headers(self) -> dict[str, str]: 34 return {"Authorization": f"Bearer {self._secret.reveal()}"} 35 36 async def search(self, query: str) -> list[KeywordSearchHit]: 37 payload = await self._http.request_json( 38 "POST", 39 endpoint(self._api_url, "/v1/search"), 40 stage="search", 41 headers=self._headers, 42 json_body={"q": query, "depth": "standard", "outputType": "searchResults"}, 43 ) 44 root = require_object(payload, self.name, "search", "response") 45 results = require_list(root.get("results"), self.name, "search", "results") 46 hits: list[KeywordSearchHit] = [] 47 for item in results: 48 try: 49 result = require_object(item, self.name, "search", "result") 50 hits.append( 51 KeywordSearchHit( 52 url=non_empty_string(result.get("url"), self.name, "search", "result.url"), 53 title=non_empty_string( 54 result.get("name"), self.name, "search", "result.name" 55 ), 56 snippet=non_empty_string( 57 result.get("content"), self.name, "search", "result.content" 58 ), 59 ) 60 ) 61 except ExecutionFailure: 62 continue 63 return hits 64 65 async def fetch(self, url: NormalizedURL) -> URLFetchCandidate: 66 payload = await self._http.request_json( 67 "POST", 68 endpoint(self._api_url, "/v1/fetch"), 69 stage="fetch", 70 headers=self._headers, 71 json_body={ 72 "url": str(url), 73 "includeRawContent": True, 74 "extractImages": False, 75 }, 76 ) 77 root = require_object(payload, self.name, "fetch", "response") 78 markdown = optional_string(root.get("markdown"), self.name, "fetch", "markdown") 79 raw_content = optional_string(root.get("rawContent"), self.name, "fetch", "rawContent") 80 raw = raw_content or markdown 81 if not raw.strip(): 82 raise failure(self.name, "fetch", "page body is empty") 83 return URLFetchCandidate(raw_content=raw, content=markdown)
class
LinkupAdapter:
19class LinkupAdapter: 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 {"Authorization": f"Bearer {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, "/v1/search"), 41 stage="search", 42 headers=self._headers, 43 json_body={"q": query, "depth": "standard", "outputType": "searchResults"}, 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 hits.append( 52 KeywordSearchHit( 53 url=non_empty_string(result.get("url"), self.name, "search", "result.url"), 54 title=non_empty_string( 55 result.get("name"), self.name, "search", "result.name" 56 ), 57 snippet=non_empty_string( 58 result.get("content"), self.name, "search", "result.content" 59 ), 60 ) 61 ) 62 except ExecutionFailure: 63 continue 64 return hits 65 66 async def fetch(self, url: NormalizedURL) -> URLFetchCandidate: 67 payload = await self._http.request_json( 68 "POST", 69 endpoint(self._api_url, "/v1/fetch"), 70 stage="fetch", 71 headers=self._headers, 72 json_body={ 73 "url": str(url), 74 "includeRawContent": True, 75 "extractImages": False, 76 }, 77 ) 78 root = require_object(payload, self.name, "fetch", "response") 79 markdown = optional_string(root.get("markdown"), self.name, "fetch", "markdown") 80 raw_content = optional_string(root.get("rawContent"), self.name, "fetch", "rawContent") 81 raw = raw_content or markdown 82 if not raw.strip(): 83 raise failure(self.name, "fetch", "page body is empty") 84 return URLFetchCandidate(raw_content=raw, content=markdown)
LinkupAdapter( *, name: str, api_url: str, secret: agent_search_gateway.observability.SecretValue, http_executor: agent_search_gateway.providers.web.common.JsonRequester)
async def
search( self, query: str) -> list[agent_search_gateway.providers.contracts.KeywordSearchHit]:
37 async def search(self, query: str) -> list[KeywordSearchHit]: 38 payload = await self._http.request_json( 39 "POST", 40 endpoint(self._api_url, "/v1/search"), 41 stage="search", 42 headers=self._headers, 43 json_body={"q": query, "depth": "standard", "outputType": "searchResults"}, 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 hits.append( 52 KeywordSearchHit( 53 url=non_empty_string(result.get("url"), self.name, "search", "result.url"), 54 title=non_empty_string( 55 result.get("name"), self.name, "search", "result.name" 56 ), 57 snippet=non_empty_string( 58 result.get("content"), self.name, "search", "result.content" 59 ), 60 ) 61 ) 62 except ExecutionFailure: 63 continue 64 return hits
async def
fetch( self, url: agent_search_gateway.url_normalization.NormalizedURL) -> agent_search_gateway.providers.contracts.URLFetchCandidate:
66 async def fetch(self, url: NormalizedURL) -> URLFetchCandidate: 67 payload = await self._http.request_json( 68 "POST", 69 endpoint(self._api_url, "/v1/fetch"), 70 stage="fetch", 71 headers=self._headers, 72 json_body={ 73 "url": str(url), 74 "includeRawContent": True, 75 "extractImages": False, 76 }, 77 ) 78 root = require_object(payload, self.name, "fetch", "response") 79 markdown = optional_string(root.get("markdown"), self.name, "fetch", "markdown") 80 raw_content = optional_string(root.get("rawContent"), self.name, "fetch", "rawContent") 81 raw = raw_content or markdown 82 if not raw.strip(): 83 raise failure(self.name, "fetch", "page body is empty") 84 return URLFetchCandidate(raw_content=raw, content=markdown)