ExaToolkit

def search(
    self,
    *,
    num_results: int | Variable | None = None,
    max_characters: int | Variable | None = None,
    search_type: SearchType | Variable | None = None,
    category: Category | Variable | None = None,
    include_domains: Sequence[str] | Variable | None = None,
    exclude_domains: Sequence[str] | Variable | None = None,
    start_published_date: str | Variable | None = None,
    end_published_date: str | Variable | None = None,
    start_crawl_date: str | Variable | None = None,
    end_crawl_date: str | Variable | None = None,
    livecrawl: Livecrawl | Variable | None = None,
    user_location: str | Variable | None = None,
    moderation: bool | Variable | None = None,
    name: str = "exa_search",
    description: str = (
        "Search the web using Exa's neural search engine. "
        "Returns ranked results with titles, URLs, relevance scores, and optional text content."
    ),
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool:
    api_key = self._api_key

    @tool(name=name, description=description, middleware=middleware)
    async def exa_search(
        query: Annotated[str, Field(description="The search query string.")],
        ctx: Context,
    ) -> ToolResult:
        """Search the web using Exa and return ranked results."""
        params: dict[str, Any] = {
            "num_results": resolve_variable(num_results, ctx, param_name="num_results"),
            "type": resolve_variable(search_type, ctx, param_name="search_type"),
            "category": resolve_variable(category, ctx, param_name="category"),
            "include_domains": resolve_variable(include_domains, ctx, param_name="include_domains"),
            "exclude_domains": resolve_variable(exclude_domains, ctx, param_name="exclude_domains"),
            "start_published_date": resolve_variable(start_published_date, ctx, param_name="start_published_date"),
            "end_published_date": resolve_variable(end_published_date, ctx, param_name="end_published_date"),
            "start_crawl_date": resolve_variable(start_crawl_date, ctx, param_name="start_crawl_date"),
            "end_crawl_date": resolve_variable(end_crawl_date, ctx, param_name="end_crawl_date"),
        }
        kwargs = {k: v for k, v in params.items() if v is not None}

        contents: dict[str, Any] = {}
        resolved_max_chars = resolve_variable(max_characters, ctx, param_name="max_characters")
        if resolved_max_chars is not None:
            contents["text"] = {"maxCharacters": resolved_max_chars}
        resolved_livecrawl = resolve_variable(livecrawl, ctx, param_name="livecrawl")
        if resolved_livecrawl is not None:
            contents["livecrawl"] = resolved_livecrawl
        if contents:
            kwargs["contents"] = contents

        c = AsyncExa(api_key=api_key)
        c.headers["x-exa-integration"] = "ag2"
        try:
            raw = await c.search(query, **kwargs)
        finally:
            await c.client.aclose()

        return ToolResult(
            ExaSearchResponse(
                query=query,
                results=[
                    ExaSearchResult(
                        title=r.title or "",
                        url=r.url,
                        score=r.score,
                        published_date=r.published_date,
                        author=r.author,
                        text=r.text,
                    )
                    for r in raw.results
                ],
            )
        )

    return exa_search

def find_similar(
    self,
    *,
    num_results: int | Variable | None = None,
    include_domains: Sequence[str] | Variable | None = None,
    exclude_domains: Sequence[str] | Variable | None = None,
    exclude_source_domain: bool | Variable | None = None,
    category: Category | Variable | None = None,
    name: str = "exa_find_similar",
    description: str = (
        "Find web pages similar to a given URL. Useful for discovering "
        "related content, competitors, or alternative sources."
    ),
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool:
    api_key = self._api_key

    @tool(name=name, description=description, middleware=middleware)
    async def exa_find_similar(
        url: Annotated[str, Field(description="The URL to find similar pages for.")],
        ctx: Context,
    ) -> ToolResult:
        """Find pages similar to a given URL."""
        params: dict[str, Any] = {
            "num_results": resolve_variable(num_results, ctx, param_name="num_results"),
            "include_domains": resolve_variable(include_domains, ctx, param_name="include_domains"),
            "exclude_domains": resolve_variable(exclude_domains, ctx, param_name="exclude_domains"),
            "exclude_source_domain": resolve_variable(
                exclude_source_domain, ctx, param_name="exclude_source_domain"
            ),
        }
        kwargs = {k: v for k, v in params.items() if v is not None}
        c = AsyncExa(api_key=api_key)
        c.headers["x-exa-integration"] = "ag2"
        try:
            raw = await c.find_similar(url, **kwargs)
        finally:
            await c.client.aclose()
        results = [
            ExaSearchResult(
                title=r.title or "",
                url=r.url,
                score=r.score,
                published_date=r.published_date,
                author=r.author,
                text=r.text,
            )
            for r in raw.results
        ]
        return ToolResult(results)

    return exa_find_similar

def get_contents(
    self,
    *,
    name: str = "exa_get_contents",
    description: str = (
        "Get the full text content of specific URLs. Useful when you already "
        "know which pages you need and want to read them."
    ),
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool:
    api_key = self._api_key

    @tool(name=name, description=description, middleware=middleware)
    async def exa_get_contents(
        urls: Annotated[list[str], Field(description="URLs to fetch content for.")],
        ctx: Context,
    ) -> ToolResult:
        """Fetch the full text content of specific URLs."""
        c = AsyncExa(api_key=api_key)
        c.headers["x-exa-integration"] = "ag2"
        try:
            raw = await c.get_contents(urls, text=True)
        finally:
            await c.client.aclose()
        results = [
            ExaContentResult(
                url=r.url,
                title=r.title or "",
                text=r.text or "",
                author=r.author,
                published_date=r.published_date,
            )
            for r in raw.results
        ]
        return ToolResult(results)

    return exa_get_contents

def answer(
    self,
    *,
    name: str = "exa_answer",
    description: str = ("Generate an AI-powered answer to a question with citations from web sources."),
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool:
    api_key = self._api_key

    @tool(name=name, description=description, middleware=middleware)
    async def exa_answer(
        query: Annotated[str, Field(description="The question to answer.")],
        ctx: Context,
    ) -> ToolResult:
        """Generate an AI answer with citations."""
        c = AsyncExa(api_key=api_key)
        c.headers["x-exa-integration"] = "ag2"
        try:
            raw = await c.answer(query, text=True)
        finally:
            await c.client.aclose()
        return ToolResult(
            ExaAnswerResult(
                answer=raw.answer,
                citations=[
                    ExaAnswerCitation(url=c.url, title=c.title or "", text=c.text or "")
                    for c in (raw.citations or [])
                ],
            )
        )

    return exa_answer

def set_provider(self, provider: Provider) -> None:
    for t in self.tools:
        t.set_provider(provider)

async def schemas(self, context: "Context") -> Iterable[ToolSchema]:
    schemas: list[ToolSchema] = []
    for t in self.tools:
        schemas.extend(await t.schemas(context))
    return schemas

def register(
    self,
    stack: "ExitStack",
    context: "Context",
    *,
    middleware: Iterable["BaseMiddleware"] = (),
) -> None:
    for t in self.tools:
        t.register(stack, context, middleware=middleware)

def tool(
    self,
    function: Callable[..., Any] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    schema: FunctionParameters | None = None,
    sync_to_thread: bool = True,
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool | Callable[[Callable[..., Any]], FunctionTool]:
    def make_tool(f: Callable[..., Any]) -> FunctionTool:
        t = tool(
            f,
            name=name,
            description=description,
            schema=schema,
            sync_to_thread=sync_to_thread,
            middleware=middleware,
        )
        self._add_tool(t)
        return t

    if function:
        return make_tool(function)

    return make_tool