datarobot_genai.drtools.core.clients.tavily

tavily

Tavily API Client and utilities for API key authentication.

TavilySearchResults

Bases: BaseModel

Source code in datarobot_genai/drtools/core/clients/tavily.py

class TavilySearchResults(BaseModel):
    results: list[_TavilySearchResult]
    response_time: float
    answer: str | None = None
    images: list[_TavilyImage] = Field(default_factory=list)

    model_config = ConfigDict(populate_by_name=True)

    @classmethod
    def from_tavily_sdk(cls, result: dict[str, Any]) -> "TavilySearchResults":
        """Create a TavilySearchResult from Tavily SDK response data."""
        return cls(
            results=[_TavilySearchResult.from_tavily_sdk(r) for r in result.get("results", [])],
            response_time=result.get("response_time", 0.0),
            answer=result.get("answer"),
            images=[_TavilyImage.from_tavily_sdk(img) for img in result.get("images", [])],
        )

    def as_flat_dict(
        self, include_images: bool = False, include_answer: bool = False
    ) -> dict[str, Any]:
        """Return a flat dictionary representation of the search result."""
        response: dict[str, Any] = {
            "results": [r.as_flat_dict() for r in self.results],
            "resultCount": len(self.results),
            "responseTime": self.response_time,
        }

        if include_answer:
            response["answer"] = self.answer
        if include_images:
            response["images"] = [img.as_flat_dict() for img in self.images]
        return response

from_tavily_sdk classmethod

from_tavily_sdk(result: dict[str, Any]) -> TavilySearchResults

Create a TavilySearchResult from Tavily SDK response data.

Source code in datarobot_genai/drtools/core/clients/tavily.py

@classmethod
def from_tavily_sdk(cls, result: dict[str, Any]) -> "TavilySearchResults":
    """Create a TavilySearchResult from Tavily SDK response data."""
    return cls(
        results=[_TavilySearchResult.from_tavily_sdk(r) for r in result.get("results", [])],
        response_time=result.get("response_time", 0.0),
        answer=result.get("answer"),
        images=[_TavilyImage.from_tavily_sdk(img) for img in result.get("images", [])],
    )

as_flat_dict

as_flat_dict(include_images: bool = False, include_answer: bool = False) -> dict[str, Any]

Return a flat dictionary representation of the search result.

Source code in datarobot_genai/drtools/core/clients/tavily.py

def as_flat_dict(
    self, include_images: bool = False, include_answer: bool = False
) -> dict[str, Any]:
    """Return a flat dictionary representation of the search result."""
    response: dict[str, Any] = {
        "results": [r.as_flat_dict() for r in self.results],
        "resultCount": len(self.results),
        "responseTime": self.response_time,
    }

    if include_answer:
        response["answer"] = self.answer
    if include_images:
        response["images"] = [img.as_flat_dict() for img in self.images]
    return response

TavilyExtractResults

Bases: BaseModel

Source code in datarobot_genai/drtools/core/clients/tavily.py

class TavilyExtractResults(BaseModel):
    results: list[_TavilyExtractResult]
    response_time: float = 0.0

    model_config = ConfigDict(populate_by_name=True)

    @classmethod
    def from_tavily_sdk(cls, result: dict[str, Any]) -> "TavilyExtractResults":
        """Create a TavilyExtractResults from Tavily SDK response data."""
        return cls(
            results=[_TavilyExtractResult.from_tavily_sdk(r) for r in result.get("results", [])],
            response_time=result.get("response_time", 0.0),
        )

    def as_flat_dict(self) -> dict[str, Any]:
        """Return a flat dictionary representation of the extract result."""
        return {
            "results": [r.as_flat_dict() for r in self.results],
            "resultCount": len(self.results),
            "responseTime": self.response_time,
        }

from_tavily_sdk classmethod

from_tavily_sdk(result: dict[str, Any]) -> TavilyExtractResults

Create a TavilyExtractResults from Tavily SDK response data.

Source code in datarobot_genai/drtools/core/clients/tavily.py

@classmethod
def from_tavily_sdk(cls, result: dict[str, Any]) -> "TavilyExtractResults":
    """Create a TavilyExtractResults from Tavily SDK response data."""
    return cls(
        results=[_TavilyExtractResult.from_tavily_sdk(r) for r in result.get("results", [])],
        response_time=result.get("response_time", 0.0),
    )

as_flat_dict

as_flat_dict() -> dict[str, Any]

Return a flat dictionary representation of the extract result.

Source code in datarobot_genai/drtools/core/clients/tavily.py

def as_flat_dict(self) -> dict[str, Any]:
    """Return a flat dictionary representation of the extract result."""
    return {
        "results": [r.as_flat_dict() for r in self.results],
        "resultCount": len(self.results),
        "responseTime": self.response_time,
    }

TavilyMapResults

Bases: BaseModel

Source code in datarobot_genai/drtools/core/clients/tavily.py

class TavilyMapResults(BaseModel):
    results: list[str]  # List of sub-pages
    usage: _TavilyUsage | None = None

    model_config = ConfigDict(populate_by_name=True)

    @classmethod
    def from_tavily_sdk(cls, result: dict[str, Any]) -> "TavilyMapResults":
        """Create a TavilyMapResults from Tavily SDK response data."""
        tavily_usage = None
        if usage := result.get("usage"):
            tavily_usage = _TavilyUsage(credits=usage.get("credits", 0))

        return cls(results=result.get("results", []), usage=tavily_usage)

    def as_flat_dict(self) -> dict[str, Any]:
        """Return a flat dictionary representation of the map result."""
        return {
            "results": self.results,
            "usageCredits": self.usage.credits if self.usage else None,
            "count": len(self.results),
        }

from_tavily_sdk classmethod

from_tavily_sdk(result: dict[str, Any]) -> TavilyMapResults

Create a TavilyMapResults from Tavily SDK response data.

Source code in datarobot_genai/drtools/core/clients/tavily.py

@classmethod
def from_tavily_sdk(cls, result: dict[str, Any]) -> "TavilyMapResults":
    """Create a TavilyMapResults from Tavily SDK response data."""
    tavily_usage = None
    if usage := result.get("usage"):
        tavily_usage = _TavilyUsage(credits=usage.get("credits", 0))

    return cls(results=result.get("results", []), usage=tavily_usage)

as_flat_dict

as_flat_dict() -> dict[str, Any]

Return a flat dictionary representation of the map result.

Source code in datarobot_genai/drtools/core/clients/tavily.py

def as_flat_dict(self) -> dict[str, Any]:
    """Return a flat dictionary representation of the map result."""
    return {
        "results": self.results,
        "usageCredits": self.usage.credits if self.usage else None,
        "count": len(self.results),
    }

TavilyCrawlResults

Bases: BaseModel

Results from Tavily Crawl API.

Source code in datarobot_genai/drtools/core/clients/tavily.py

class TavilyCrawlResults(BaseModel):
    """Results from Tavily Crawl API."""

    base_url: str
    results: list[_TavilyExtractResult]
    response_time: float = 0.0

    model_config = ConfigDict(populate_by_name=True)

    @classmethod
    def from_tavily_sdk(cls, result: dict[str, Any]) -> "TavilyCrawlResults":
        """Create a TavilyCrawlResults from Tavily SDK response data."""
        return cls(
            base_url=result.get("base_url", ""),
            results=[_TavilyExtractResult.from_tavily_sdk(r) for r in result.get("results", [])],
            response_time=result.get("response_time", 0.0),
        )

    def as_flat_dict(self) -> dict[str, Any]:
        """Return a flat dictionary representation of the crawl result."""
        return {
            "baseUrl": self.base_url,
            "results": [r.as_flat_dict() for r in self.results],
            "resultCount": len(self.results),
            "responseTime": self.response_time,
        }

from_tavily_sdk classmethod

from_tavily_sdk(result: dict[str, Any]) -> TavilyCrawlResults

Create a TavilyCrawlResults from Tavily SDK response data.

Source code in datarobot_genai/drtools/core/clients/tavily.py

@classmethod
def from_tavily_sdk(cls, result: dict[str, Any]) -> "TavilyCrawlResults":
    """Create a TavilyCrawlResults from Tavily SDK response data."""
    return cls(
        base_url=result.get("base_url", ""),
        results=[_TavilyExtractResult.from_tavily_sdk(r) for r in result.get("results", [])],
        response_time=result.get("response_time", 0.0),
    )

as_flat_dict

as_flat_dict() -> dict[str, Any]

Return a flat dictionary representation of the crawl result.

Source code in datarobot_genai/drtools/core/clients/tavily.py

def as_flat_dict(self) -> dict[str, Any]:
    """Return a flat dictionary representation of the crawl result."""
    return {
        "baseUrl": self.base_url,
        "results": [r.as_flat_dict() for r in self.results],
        "resultCount": len(self.results),
        "responseTime": self.response_time,
    }

TavilyClient

Client for interacting with Tavily API.

This is a wrapper around the official tavily-python SDK.

Source code in datarobot_genai/drtools/core/clients/tavily.py

class TavilyClient:
    """Client for interacting with Tavily API.

    This is a wrapper around the official tavily-python SDK.
    """

    def __init__(self, api_key: str) -> None:
        self._client = AsyncTavilyClient(api_key=api_key)

    async def search(
        self,
        query: str,
        *,
        topic: Literal["general", "news", "finance"] = "general",
        search_depth: Literal["basic", "advanced"] = "basic",
        max_results: int = MAX_RESULTS_DEFAULT,
        time_range: Literal["day", "week", "month", "year"] | None = None,
        include_images: bool = False,
        include_image_descriptions: bool = False,
        chunks_per_source: int = CHUNKS_PER_SOURCE_DEFAULT,
        include_answer: bool = False,
    ) -> TavilySearchResults:
        """
        Perform a web search using Tavily API.

        Args:
            query: The search query to execute.
            topic: The category of search ("general", "news", or "finance").
            search_depth: The depth of search ("basic" or "advanced").
            max_results: Maximum number of results to return (1-20).
            time_range: Time range filter ("day", "week", "month", "year").
            include_images: Whether to include images in results.
            include_image_descriptions: Whether to include image descriptions.
            chunks_per_source: Maximum content snippets per URL (1-3).
            include_answer: Whether to include an AI-generated answer.

        Returns
        -------
            TavilySearchResults

        Raises
        ------
            ValueError: If validation fails.
            TavilyInvalidAPIKeyError: If the API key is invalid.
            TavilyUsageLimitExceededError: If usage limit is exceeded.
            TavilyForbiddenError: If access is forbidden.
            TavilyBadRequestError: If the request is malformed.
        """
        # Validate inputs
        if not query:
            raise ValueError("query cannot be empty.")
        if isinstance(query, str) and not query.strip():
            raise ValueError("query cannot be empty.")
        if max_results <= 0:
            raise ValueError("max_results must be greater than 0.")
        if max_results > MAX_RESULTS:
            raise ValueError(f"max_results must be smaller than or equal to {MAX_RESULTS}.")
        if chunks_per_source <= 0:
            raise ValueError("chunks_per_source must be greater than 0.")
        if chunks_per_source > MAX_CHUNKS_PER_SOURCE:
            raise ValueError(
                f"chunks_per_source must be smaller than or equal to {MAX_CHUNKS_PER_SOURCE}."
            )

        # Clamp values to valid ranges
        max_results = min(max_results, MAX_RESULTS)
        chunks_per_source = min(chunks_per_source, MAX_CHUNKS_PER_SOURCE)

        # Build search parameters
        search_kwargs: dict[str, Any] = {
            "query": query,
            "topic": topic,
            "search_depth": search_depth,
            "max_results": max_results,
            "include_images": include_images,
            "include_image_descriptions": include_image_descriptions,
            "chunks_per_source": chunks_per_source,
            "include_answer": include_answer,
        }

        if time_range:
            search_kwargs["time_range"] = time_range

        result = await self._client.search(**search_kwargs)
        return TavilySearchResults.from_tavily_sdk(result)

    async def extract(
        self,
        urls: str | list[str],
        *,
        query: str | None = None,
        chunks_per_source: int = 3,
        extract_depth: Literal["basic", "advanced"] = "basic",
        format: Literal["markdown", "text"] = "markdown",
        include_images: bool = False,
        include_favicon: bool = False,
        timeout: float = 30.0,
    ) -> TavilyExtractResults:
        """
        Extract content from URLs using Tavily Extract API.

        Args:
            urls: The URL or list of URLs to extract content from (max 20).
            query: User intent for reranking extracted content chunks based on relevance.
            chunks_per_source: Maximum number of 500-char snippets per URL (1-5, default 3).
            extract_depth: Depth of extraction ("basic" or "advanced").
            format: Format of extracted content ("markdown" or "text").
            include_images: Whether to include images in results.
            include_favicon: Whether to include favicon in results.
            timeout: Request timeout in seconds (1.0-60.0, default 30.0).

        Returns
        -------
            TavilyExtractResults

        Raises
        ------
            ValueError: If validation fails.
            TavilyInvalidAPIKeyError: If the API key is invalid.
            TavilyUsageLimitExceededError: If usage limit is exceeded.
            TavilyForbiddenError: If access is forbidden.
            TavilyBadRequestError: If the request is malformed.
        """
        url_list = [urls] if isinstance(urls, str) else urls

        # Validate inputs
        if not url_list:
            raise ValueError("urls cannot be empty.")
        if len(url_list) > 20:
            raise ValueError("Maximum number of URLs is 20.")
        if chunks_per_source <= 0:
            raise ValueError("chunks_per_source must be greater than 0.")
        if chunks_per_source > 5:
            raise ValueError("chunks_per_source must be smaller than or equal to 5.")
        if timeout < 1.0 or timeout > 60.0:  # noqa: PLR2004
            raise ValueError("timeout must be between 1.0 and 60.0 seconds.")

        chunks_per_source = min(chunks_per_source, 5)

        extract_kwargs: dict[str, Any] = {
            "urls": url_list,
            "extract_depth": extract_depth,
            "format": format,
            "include_images": include_images,
            "include_favicon": include_favicon,
            "timeout": timeout,
        }

        if query:
            extract_kwargs["query"] = query
            extract_kwargs["chunks_per_source"] = chunks_per_source

        result = await self._client.extract(**extract_kwargs)
        return TavilyExtractResults.from_tavily_sdk(result)

    async def map_(
        self,
        url: str,
        instructions: str | None = None,
        limit: int = 50,
        include_usage: bool = False,
    ) -> TavilyMapResults:
        """
        Generate a structured map of a website to discover relevant sub-pages.

        Args:
            url: The root URL to begin mapping.
            instructions: Instructions to guide the mapper toward specific paths.
            limit: Total links to process.
            include_usage: Whether to include credit usage information in the response.

        Returns
        -------
            TavilyMapResults

        Raises
        ------
            ValueError: If validation fails.
            TavilyInvalidAPIKeyError: If the API key is invalid.
            TavilyUsageLimitExceededError: If usage limit is exceeded.
            TavilyForbiddenError: If access is forbidden.
            TavilyBadRequestError: If the request is malformed.
        """
        if not url or not url.strip():
            raise ValueError("Argument validation error: url cannot be empty.")
        if limit <= 0:
            raise ValueError("Argument validation error: limit must be greater than 0.")
        if limit > 200:
            raise ValueError("Argument validation error: limit must be less than 200.")

        result = await self._client.map(
            url=url,
            instructions=instructions,
            limit=limit,
            include_usage=include_usage,
        )
        return TavilyMapResults.from_tavily_sdk(result)

    async def crawl(
        self,
        url: str,
        *,
        instructions: str | None = None,
        limit: int = 20,
        max_depth: int = 1,
        exclude_paths: list[str] | None = None,
        include_images: bool = False,
    ) -> TavilyCrawlResults:
        """
        Crawl a website using Tavily Crawl API.

        Args:
            url: The root URL to begin the traversal.
            instructions: Natural language instructions for the crawler to filter relevant content.
            limit: Total number of links to process (1-500, default 20).
            max_depth: Maximum depth from base URL (1-5, default 1).
            exclude_paths: Regex patterns to exclude URL paths.
            include_images: Whether to include images found during crawl.

        Returns
        -------
            TavilyCrawlResults

        Raises
        ------
            ValueError: If validation fails.
            TavilyInvalidAPIKeyError: If the API key is invalid.
            TavilyUsageLimitExceededError: If usage limit is exceeded.
            TavilyForbiddenError: If access is forbidden.
            TavilyBadRequestError: If the request is malformed.
        """
        # Validate inputs
        if not url:
            raise ValueError("url cannot be empty.")
        if isinstance(url, str) and not url.strip():
            raise ValueError("url cannot be empty.")
        if limit < 1:
            raise ValueError("limit must be at least 1.")
        if limit > 500:  # noqa: PLR2004
            raise ValueError("limit must be at most 500.")
        if max_depth < 1:
            raise ValueError("max_depth must be at least 1.")
        if max_depth > 5:  # noqa: PLR2004
            raise ValueError("max_depth must be at most 5.")

        crawl_kwargs: dict[str, Any] = {
            "url": url,
            "limit": limit,
            "max_depth": max_depth,
            "include_images": include_images,
        }

        if instructions:
            crawl_kwargs["instructions"] = instructions

        if exclude_paths:
            crawl_kwargs["exclude_paths"] = exclude_paths

        result = await self._client.crawl(**crawl_kwargs)
        return TavilyCrawlResults.from_tavily_sdk(result)

    async def __aenter__(self) -> "TavilyClient":
        """Async context manager entry."""
        return self

    async def __aexit__(
        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any
    ) -> None:
        """Async context manager exit."""
        # AsyncTavilyClient doesn't have a close method, but we keep the context manager
        # pattern for consistency with other clients
        pass

search async

search(query: str, *, topic: Literal['general', 'news', 'finance'] = 'general', search_depth: Literal['basic', 'advanced'] = 'basic', max_results: int = MAX_RESULTS_DEFAULT, time_range: Literal['day', 'week', 'month', 'year'] | None = None, include_images: bool = False, include_image_descriptions: bool = False, chunks_per_source: int = CHUNKS_PER_SOURCE_DEFAULT, include_answer: bool = False) -> TavilySearchResults

Perform a web search using Tavily API.

Parameters:

Name	Type	Description	Default
query	str	The search query to execute.	required
topic	Literal['general', 'news', 'finance']	The category of search ("general", "news", or "finance").	'general'
search_depth	Literal['basic', 'advanced']	The depth of search ("basic" or "advanced").	'basic'
max_results	int	Maximum number of results to return (1-20).	MAX_RESULTS_DEFAULT
time_range	Literal['day', 'week', 'month', 'year'] | None	Time range filter ("day", "week", "month", "year").	None
include_images	bool	Whether to include images in results.	False
include_image_descriptions	bool	Whether to include image descriptions.	False
chunks_per_source	int	Maximum content snippets per URL (1-3).	CHUNKS_PER_SOURCE_DEFAULT
include_answer	bool	Whether to include an AI-generated answer.	False

Returns

TavilySearchResults

Raises

ValueError: If validation fails.
TavilyInvalidAPIKeyError: If the API key is invalid.
TavilyUsageLimitExceededError: If usage limit is exceeded.
TavilyForbiddenError: If access is forbidden.
TavilyBadRequestError: If the request is malformed.

Source code in datarobot_genai/drtools/core/clients/tavily.py

async def search(
    self,
    query: str,
    *,
    topic: Literal["general", "news", "finance"] = "general",
    search_depth: Literal["basic", "advanced"] = "basic",
    max_results: int = MAX_RESULTS_DEFAULT,
    time_range: Literal["day", "week", "month", "year"] | None = None,
    include_images: bool = False,
    include_image_descriptions: bool = False,
    chunks_per_source: int = CHUNKS_PER_SOURCE_DEFAULT,
    include_answer: bool = False,
) -> TavilySearchResults:
    """
    Perform a web search using Tavily API.

    Args:
        query: The search query to execute.
        topic: The category of search ("general", "news", or "finance").
        search_depth: The depth of search ("basic" or "advanced").
        max_results: Maximum number of results to return (1-20).
        time_range: Time range filter ("day", "week", "month", "year").
        include_images: Whether to include images in results.
        include_image_descriptions: Whether to include image descriptions.
        chunks_per_source: Maximum content snippets per URL (1-3).
        include_answer: Whether to include an AI-generated answer.

    Returns
    -------
        TavilySearchResults

    Raises
    ------
        ValueError: If validation fails.
        TavilyInvalidAPIKeyError: If the API key is invalid.
        TavilyUsageLimitExceededError: If usage limit is exceeded.
        TavilyForbiddenError: If access is forbidden.
        TavilyBadRequestError: If the request is malformed.
    """
    # Validate inputs
    if not query:
        raise ValueError("query cannot be empty.")
    if isinstance(query, str) and not query.strip():
        raise ValueError("query cannot be empty.")
    if max_results <= 0:
        raise ValueError("max_results must be greater than 0.")
    if max_results > MAX_RESULTS:
        raise ValueError(f"max_results must be smaller than or equal to {MAX_RESULTS}.")
    if chunks_per_source <= 0:
        raise ValueError("chunks_per_source must be greater than 0.")
    if chunks_per_source > MAX_CHUNKS_PER_SOURCE:
        raise ValueError(
            f"chunks_per_source must be smaller than or equal to {MAX_CHUNKS_PER_SOURCE}."
        )

    # Clamp values to valid ranges
    max_results = min(max_results, MAX_RESULTS)
    chunks_per_source = min(chunks_per_source, MAX_CHUNKS_PER_SOURCE)

    # Build search parameters
    search_kwargs: dict[str, Any] = {
        "query": query,
        "topic": topic,
        "search_depth": search_depth,
        "max_results": max_results,
        "include_images": include_images,
        "include_image_descriptions": include_image_descriptions,
        "chunks_per_source": chunks_per_source,
        "include_answer": include_answer,
    }

    if time_range:
        search_kwargs["time_range"] = time_range

    result = await self._client.search(**search_kwargs)
    return TavilySearchResults.from_tavily_sdk(result)

extract async

extract(urls: str | list[str], *, query: str | None = None, chunks_per_source: int = 3, extract_depth: Literal['basic', 'advanced'] = 'basic', format: Literal['markdown', 'text'] = 'markdown', include_images: bool = False, include_favicon: bool = False, timeout: float = 30.0) -> TavilyExtractResults

Extract content from URLs using Tavily Extract API.

Parameters:

Name	Type	Description	Default
urls	str | list[str]	The URL or list of URLs to extract content from (max 20).	required
query	str | None	User intent for reranking extracted content chunks based on relevance.	None
chunks_per_source	int	Maximum number of 500-char snippets per URL (1-5, default 3).	3
extract_depth	Literal['basic', 'advanced']	Depth of extraction ("basic" or "advanced").	'basic'
format	Literal['markdown', 'text']	Format of extracted content ("markdown" or "text").	'markdown'
include_images	bool	Whether to include images in results.	False
include_favicon	bool	Whether to include favicon in results.	False
timeout	float	Request timeout in seconds (1.0-60.0, default 30.0).	30.0

Returns

TavilyExtractResults

Raises

ValueError: If validation fails.
TavilyInvalidAPIKeyError: If the API key is invalid.
TavilyUsageLimitExceededError: If usage limit is exceeded.
TavilyForbiddenError: If access is forbidden.
TavilyBadRequestError: If the request is malformed.

Source code in datarobot_genai/drtools/core/clients/tavily.py

async def extract(
    self,
    urls: str | list[str],
    *,
    query: str | None = None,
    chunks_per_source: int = 3,
    extract_depth: Literal["basic", "advanced"] = "basic",
    format: Literal["markdown", "text"] = "markdown",
    include_images: bool = False,
    include_favicon: bool = False,
    timeout: float = 30.0,
) -> TavilyExtractResults:
    """
    Extract content from URLs using Tavily Extract API.

    Args:
        urls: The URL or list of URLs to extract content from (max 20).
        query: User intent for reranking extracted content chunks based on relevance.
        chunks_per_source: Maximum number of 500-char snippets per URL (1-5, default 3).
        extract_depth: Depth of extraction ("basic" or "advanced").
        format: Format of extracted content ("markdown" or "text").
        include_images: Whether to include images in results.
        include_favicon: Whether to include favicon in results.
        timeout: Request timeout in seconds (1.0-60.0, default 30.0).

    Returns
    -------
        TavilyExtractResults

    Raises
    ------
        ValueError: If validation fails.
        TavilyInvalidAPIKeyError: If the API key is invalid.
        TavilyUsageLimitExceededError: If usage limit is exceeded.
        TavilyForbiddenError: If access is forbidden.
        TavilyBadRequestError: If the request is malformed.
    """
    url_list = [urls] if isinstance(urls, str) else urls

    # Validate inputs
    if not url_list:
        raise ValueError("urls cannot be empty.")
    if len(url_list) > 20:
        raise ValueError("Maximum number of URLs is 20.")
    if chunks_per_source <= 0:
        raise ValueError("chunks_per_source must be greater than 0.")
    if chunks_per_source > 5:
        raise ValueError("chunks_per_source must be smaller than or equal to 5.")
    if timeout < 1.0 or timeout > 60.0:  # noqa: PLR2004
        raise ValueError("timeout must be between 1.0 and 60.0 seconds.")

    chunks_per_source = min(chunks_per_source, 5)

    extract_kwargs: dict[str, Any] = {
        "urls": url_list,
        "extract_depth": extract_depth,
        "format": format,
        "include_images": include_images,
        "include_favicon": include_favicon,
        "timeout": timeout,
    }

    if query:
        extract_kwargs["query"] = query
        extract_kwargs["chunks_per_source"] = chunks_per_source

    result = await self._client.extract(**extract_kwargs)
    return TavilyExtractResults.from_tavily_sdk(result)

map_ async

map_(url: str, instructions: str | None = None, limit: int = 50, include_usage: bool = False) -> TavilyMapResults

Generate a structured map of a website to discover relevant sub-pages.

Parameters:

Name	Type	Description	Default
url	str	The root URL to begin mapping.	required
instructions	str | None	Instructions to guide the mapper toward specific paths.	None
limit	int	Total links to process.	50
include_usage	bool	Whether to include credit usage information in the response.	False

Returns

TavilyMapResults

Raises

ValueError: If validation fails.
TavilyInvalidAPIKeyError: If the API key is invalid.
TavilyUsageLimitExceededError: If usage limit is exceeded.
TavilyForbiddenError: If access is forbidden.
TavilyBadRequestError: If the request is malformed.

Source code in datarobot_genai/drtools/core/clients/tavily.py

async def map_(
    self,
    url: str,
    instructions: str | None = None,
    limit: int = 50,
    include_usage: bool = False,
) -> TavilyMapResults:
    """
    Generate a structured map of a website to discover relevant sub-pages.

    Args:
        url: The root URL to begin mapping.
        instructions: Instructions to guide the mapper toward specific paths.
        limit: Total links to process.
        include_usage: Whether to include credit usage information in the response.

    Returns
    -------
        TavilyMapResults

    Raises
    ------
        ValueError: If validation fails.
        TavilyInvalidAPIKeyError: If the API key is invalid.
        TavilyUsageLimitExceededError: If usage limit is exceeded.
        TavilyForbiddenError: If access is forbidden.
        TavilyBadRequestError: If the request is malformed.
    """
    if not url or not url.strip():
        raise ValueError("Argument validation error: url cannot be empty.")
    if limit <= 0:
        raise ValueError("Argument validation error: limit must be greater than 0.")
    if limit > 200:
        raise ValueError("Argument validation error: limit must be less than 200.")

    result = await self._client.map(
        url=url,
        instructions=instructions,
        limit=limit,
        include_usage=include_usage,
    )
    return TavilyMapResults.from_tavily_sdk(result)

crawl async

crawl(url: str, *, instructions: str | None = None, limit: int = 20, max_depth: int = 1, exclude_paths: list[str] | None = None, include_images: bool = False) -> TavilyCrawlResults

Crawl a website using Tavily Crawl API.

Parameters:

Name	Type	Description	Default
url	str	The root URL to begin the traversal.	required
instructions	str | None	Natural language instructions for the crawler to filter relevant content.	None
limit	int	Total number of links to process (1-500, default 20).	20
max_depth	int	Maximum depth from base URL (1-5, default 1).	1
exclude_paths	list[str] | None	Regex patterns to exclude URL paths.	None
include_images	bool	Whether to include images found during crawl.	False

Returns

TavilyCrawlResults

Raises

ValueError: If validation fails.
TavilyInvalidAPIKeyError: If the API key is invalid.
TavilyUsageLimitExceededError: If usage limit is exceeded.
TavilyForbiddenError: If access is forbidden.
TavilyBadRequestError: If the request is malformed.

Source code in datarobot_genai/drtools/core/clients/tavily.py

async def crawl(
    self,
    url: str,
    *,
    instructions: str | None = None,
    limit: int = 20,
    max_depth: int = 1,
    exclude_paths: list[str] | None = None,
    include_images: bool = False,
) -> TavilyCrawlResults:
    """
    Crawl a website using Tavily Crawl API.

    Args:
        url: The root URL to begin the traversal.
        instructions: Natural language instructions for the crawler to filter relevant content.
        limit: Total number of links to process (1-500, default 20).
        max_depth: Maximum depth from base URL (1-5, default 1).
        exclude_paths: Regex patterns to exclude URL paths.
        include_images: Whether to include images found during crawl.

    Returns
    -------
        TavilyCrawlResults

    Raises
    ------
        ValueError: If validation fails.
        TavilyInvalidAPIKeyError: If the API key is invalid.
        TavilyUsageLimitExceededError: If usage limit is exceeded.
        TavilyForbiddenError: If access is forbidden.
        TavilyBadRequestError: If the request is malformed.
    """
    # Validate inputs
    if not url:
        raise ValueError("url cannot be empty.")
    if isinstance(url, str) and not url.strip():
        raise ValueError("url cannot be empty.")
    if limit < 1:
        raise ValueError("limit must be at least 1.")
    if limit > 500:  # noqa: PLR2004
        raise ValueError("limit must be at most 500.")
    if max_depth < 1:
        raise ValueError("max_depth must be at least 1.")
    if max_depth > 5:  # noqa: PLR2004
        raise ValueError("max_depth must be at most 5.")

    crawl_kwargs: dict[str, Any] = {
        "url": url,
        "limit": limit,
        "max_depth": max_depth,
        "include_images": include_images,
    }

    if instructions:
        crawl_kwargs["instructions"] = instructions

    if exclude_paths:
        crawl_kwargs["exclude_paths"] = exclude_paths

    result = await self._client.crawl(**crawl_kwargs)
    return TavilyCrawlResults.from_tavily_sdk(result)

__aenter__ async

__aenter__() -> TavilyClient

Async context manager entry.

Source code in datarobot_genai/drtools/core/clients/tavily.py

async def __aenter__(self) -> "TavilyClient":
    """Async context manager entry."""
    return self

__aexit__ async

__aexit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> None

Async context manager exit.

Source code in datarobot_genai/drtools/core/clients/tavily.py

async def __aexit__(
    self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any
) -> None:
    """Async context manager exit."""
    # AsyncTavilyClient doesn't have a close method, but we keep the context manager
    # pattern for consistency with other clients
    pass

get_tavily_access_token async

get_tavily_access_token() -> str

Get Tavily API key from HTTP headers.

Returns

API key string

Raises

ToolError: If API key is not found in headers

Source code in datarobot_genai/drtools/core/clients/tavily.py

async def get_tavily_access_token() -> str:
    """
    Get Tavily API key from HTTP headers.

    Returns
    -------
        API key string

    Raises
    ------
        ToolError: If API key is not found in headers
    """
    creds = get_credentials()
    if api_key := resolve_secret("x-tavily-api-key", creds.tavily_api_key):
        return api_key

    logger.warning("Tavily API key not found")
    raise ToolError(
        "Tavily API key not found.",
        kind=ToolErrorKind.AUTHENTICATION,
    )