Edit on GitHub

agent_search_gateway.providers.web.tavily

Tavily search and extract adapter.

View Source

  1"""Tavily search and extract 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    normalized_match,
 13    optional_string,
 14    require_list,
 15    require_object,
 16)
 17
 18
 19class TavilyAdapter:
 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, "/search"),
 41            stage="search",
 42            headers=self._headers,
 43            json_body={"query": query, "include_raw_content": "markdown"},
 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                url = non_empty_string(result.get("url"), self.name, "search", "result.url")
 52                hits.append(
 53                    KeywordSearchHit(
 54                        url=url,
 55                        title=optional_string(
 56                            result.get("title"), self.name, "search", "result.title"
 57                        ),
 58                        snippet=optional_string(
 59                            result.get("content"), self.name, "search", "result.content"
 60                        ),
 61                        raw_content=optional_string(
 62                            result.get("raw_content"),
 63                            self.name,
 64                            "search",
 65                            "result.raw_content",
 66                        ),
 67                    )
 68                )
 69            except ExecutionFailure:
 70                continue
 71        return hits
 72
 73    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
 74        payload = await self._http.request_json(
 75            "POST",
 76            endpoint(self._api_url, "/extract"),
 77            stage="fetch",
 78            headers=self._headers,
 79            json_body={"urls": [str(url)]},
 80        )
 81        try:
 82            root = require_object(payload, self.name, "fetch", "response")
 83            results = require_list(root.get("results"), self.name, "fetch", "results")
 84        except ExecutionFailure as exc:
 85            reason = "invalid_results_envelope"
 86            raise failure(self.name, "fetch", reason, reason_code=reason) from exc
 87
 88        empty_match_seen = False
 89        for item in results:
 90            try:
 91                result = require_object(item, self.name, "fetch", "result")
 92                if not normalized_match(result.get("url"), url, self.name, "fetch"):
 93                    continue
 94            except ExecutionFailure:
 95                continue
 96            raw = result.get("raw_content")
 97            if not isinstance(raw, str) or not raw.strip():
 98                empty_match_seen = True
 99                continue
100            return URLFetchCandidate(raw_content=raw)
101
102        reason = "empty_raw_content" if empty_match_seen else "no_matching_result"
103        raise failure(self.name, "fetch", reason, reason_code=reason)

class TavilyAdapter: View Source

 20class TavilyAdapter:
 21    def __init__(
 22        self,
 23        *,
 24        name: str,
 25        api_url: str,
 26        secret: SecretValue,
 27        http_executor: JsonRequester,
 28    ) -> None:
 29        self.name = name
 30        self._api_url = api_url
 31        self._secret = secret
 32        self._http = http_executor
 33
 34    @property
 35    def _headers(self) -> dict[str, str]:
 36        return {"Authorization": f"Bearer {self._secret.reveal()}"}
 37
 38    async def search(self, query: str) -> list[KeywordSearchHit]:
 39        payload = await self._http.request_json(
 40            "POST",
 41            endpoint(self._api_url, "/search"),
 42            stage="search",
 43            headers=self._headers,
 44            json_body={"query": query, "include_raw_content": "markdown"},
 45        )
 46        root = require_object(payload, self.name, "search", "response")
 47        results = require_list(root.get("results"), self.name, "search", "results")
 48        hits: list[KeywordSearchHit] = []
 49        for item in results:
 50            try:
 51                result = require_object(item, self.name, "search", "result")
 52                url = non_empty_string(result.get("url"), self.name, "search", "result.url")
 53                hits.append(
 54                    KeywordSearchHit(
 55                        url=url,
 56                        title=optional_string(
 57                            result.get("title"), self.name, "search", "result.title"
 58                        ),
 59                        snippet=optional_string(
 60                            result.get("content"), self.name, "search", "result.content"
 61                        ),
 62                        raw_content=optional_string(
 63                            result.get("raw_content"),
 64                            self.name,
 65                            "search",
 66                            "result.raw_content",
 67                        ),
 68                    )
 69                )
 70            except ExecutionFailure:
 71                continue
 72        return hits
 73
 74    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
 75        payload = await self._http.request_json(
 76            "POST",
 77            endpoint(self._api_url, "/extract"),
 78            stage="fetch",
 79            headers=self._headers,
 80            json_body={"urls": [str(url)]},
 81        )
 82        try:
 83            root = require_object(payload, self.name, "fetch", "response")
 84            results = require_list(root.get("results"), self.name, "fetch", "results")
 85        except ExecutionFailure as exc:
 86            reason = "invalid_results_envelope"
 87            raise failure(self.name, "fetch", reason, reason_code=reason) from exc
 88
 89        empty_match_seen = False
 90        for item in results:
 91            try:
 92                result = require_object(item, self.name, "fetch", "result")
 93                if not normalized_match(result.get("url"), url, self.name, "fetch"):
 94                    continue
 95            except ExecutionFailure:
 96                continue
 97            raw = result.get("raw_content")
 98            if not isinstance(raw, str) or not raw.strip():
 99                empty_match_seen = True
100                continue
101            return URLFetchCandidate(raw_content=raw)
102
103        reason = "empty_raw_content" if empty_match_seen else "no_matching_result"
104        raise failure(self.name, "fetch", reason, reason_code=reason)

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

21    def __init__(
22        self,
23        *,
24        name: str,
25        api_url: str,
26        secret: SecretValue,
27        http_executor: JsonRequester,
28    ) -> None:
29        self.name = name
30        self._api_url = api_url
31        self._secret = secret
32        self._http = http_executor

name

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

38    async def search(self, query: str) -> list[KeywordSearchHit]:
39        payload = await self._http.request_json(
40            "POST",
41            endpoint(self._api_url, "/search"),
42            stage="search",
43            headers=self._headers,
44            json_body={"query": query, "include_raw_content": "markdown"},
45        )
46        root = require_object(payload, self.name, "search", "response")
47        results = require_list(root.get("results"), self.name, "search", "results")
48        hits: list[KeywordSearchHit] = []
49        for item in results:
50            try:
51                result = require_object(item, self.name, "search", "result")
52                url = non_empty_string(result.get("url"), self.name, "search", "result.url")
53                hits.append(
54                    KeywordSearchHit(
55                        url=url,
56                        title=optional_string(
57                            result.get("title"), self.name, "search", "result.title"
58                        ),
59                        snippet=optional_string(
60                            result.get("content"), self.name, "search", "result.content"
61                        ),
62                        raw_content=optional_string(
63                            result.get("raw_content"),
64                            self.name,
65                            "search",
66                            "result.raw_content",
67                        ),
68                    )
69                )
70            except ExecutionFailure:
71                continue
72        return hits

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

 74    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
 75        payload = await self._http.request_json(
 76            "POST",
 77            endpoint(self._api_url, "/extract"),
 78            stage="fetch",
 79            headers=self._headers,
 80            json_body={"urls": [str(url)]},
 81        )
 82        try:
 83            root = require_object(payload, self.name, "fetch", "response")
 84            results = require_list(root.get("results"), self.name, "fetch", "results")
 85        except ExecutionFailure as exc:
 86            reason = "invalid_results_envelope"
 87            raise failure(self.name, "fetch", reason, reason_code=reason) from exc
 88
 89        empty_match_seen = False
 90        for item in results:
 91            try:
 92                result = require_object(item, self.name, "fetch", "result")
 93                if not normalized_match(result.get("url"), url, self.name, "fetch"):
 94                    continue
 95            except ExecutionFailure:
 96                continue
 97            raw = result.get("raw_content")
 98            if not isinstance(raw, str) or not raw.strip():
 99                empty_match_seen = True
100                continue
101            return URLFetchCandidate(raw_content=raw)
102
103        reason = "empty_raw_content" if empty_match_seen else "no_matching_result"
104        raise failure(self.name, "fetch", reason, reason_code=reason)