Tools

any_agent.tools

FinalOutputTool

A serializable final output tool that avoids closure issues.

Source code in src/any_agent/tools/final_output.py

class FinalOutputTool:
    """A serializable final output tool that avoids closure issues."""

    def __init__(self, output_type: type[BaseModel]):
        """Create the function that will be used as a tool."""
        self.output_type = output_type
        # Set docstring for the callable object
        self.__doc__ = f"""You must call this tool in order to return the final answer.

        Args:
            answer: The final output that can be loaded as a Pydantic model. This must be a JSON compatible string that matches the following schema:
                {output_type.model_json_schema()}

        Returns:
            A dictionary with the following keys:
                - success: True if the output is valid, False otherwise.
                - result: The final output if success is True, otherwise an error message.

        """
        # Set function name for tool frameworks that expect it
        self.__name__ = "final_output"

    def __call__(self, answer: str) -> dict:  # type: ignore[type-arg]
        """Validate the final output."""
        # First check if it's valid JSON
        try:
            json.loads(answer)
        except json.JSONDecodeError as json_err:
            return {
                "success": False,
                "result": f"Invalid JSON format: {json_err}. Please fix the 'answer' parameter so that it is a valid JSON string and call this tool again.",
            }
        # Then validate against the Pydantic model
        try:
            self.output_type.model_validate_json(answer)
        except ValidationError as e:
            return {
                "success": False,
                "result": f"Please fix this validation error: {e}. The format must conform to {self.output_type.model_json_schema()}",
            }
        else:
            return {"success": True, "result": answer}

__call__(answer)

Validate the final output.

Source code in src/any_agent/tools/final_output.py

def __call__(self, answer: str) -> dict:  # type: ignore[type-arg]
    """Validate the final output."""
    # First check if it's valid JSON
    try:
        json.loads(answer)
    except json.JSONDecodeError as json_err:
        return {
            "success": False,
            "result": f"Invalid JSON format: {json_err}. Please fix the 'answer' parameter so that it is a valid JSON string and call this tool again.",
        }
    # Then validate against the Pydantic model
    try:
        self.output_type.model_validate_json(answer)
    except ValidationError as e:
        return {
            "success": False,
            "result": f"Please fix this validation error: {e}. The format must conform to {self.output_type.model_json_schema()}",
        }
    else:
        return {"success": True, "result": answer}

__init__(output_type)

Create the function that will be used as a tool.

Source code in src/any_agent/tools/final_output.py

def __init__(self, output_type: type[BaseModel]):
    """Create the function that will be used as a tool."""
    self.output_type = output_type
    # Set docstring for the callable object
    self.__doc__ = f"""You must call this tool in order to return the final answer.

    Args:
        answer: The final output that can be loaded as a Pydantic model. This must be a JSON compatible string that matches the following schema:
            {output_type.model_json_schema()}

    Returns:
        A dictionary with the following keys:
            - success: True if the output is valid, False otherwise.
            - result: The final output if success is True, otherwise an error message.

    """
    # Set function name for tool frameworks that expect it
    self.__name__ = "final_output"

a2a_tool(url, toolname=None, http_kwargs=None)

Perform a query using A2A to another agent (synchronous version).

Parameters:

Name	Type	Description	Default
url	str	The url in which the A2A agent is located.	required
toolname	str	The name for the created tool. Defaults to call_{agent name in card}. Leading and trailing whitespace are removed. Whitespace in the middle is replaced by _.	None
http_kwargs	dict	Additional kwargs to pass to the httpx client.	None

Returns:

Type	Description
Callable[[str], str]	A sync Callable that takes a query and returns the agent response.

Source code in src/any_agent/tools/a2a.py

def a2a_tool(
    url: str, toolname: str | None = None, http_kwargs: dict[str, Any] | None = None
) -> Callable[[str], str]:
    """Perform a query using A2A to another agent (synchronous version).

    Args:
        url (str): The url in which the A2A agent is located.
        toolname (str): The name for the created tool. Defaults to `call_{agent name in card}`.
            Leading and trailing whitespace are removed. Whitespace in the middle is replaced by `_`.
        http_kwargs (dict): Additional kwargs to pass to the httpx client.

    Returns:
        A sync `Callable` that takes a query and returns the agent response.

    """
    if not a2a_tool_available:
        msg = "You need to `pip install 'any-agent[a2a]'` to use this tool"
        raise ImportError(msg)

    # Fetch the async tool upfront to get proper name and documentation (otherwise the tool doesn't have the right name and documentation)
    async_tool = run_async_in_sync(a2a_tool_async(url, toolname, http_kwargs))

    def sync_wrapper(query: str) -> Any:
        """Execute the A2A tool query synchronously."""
        return run_async_in_sync(async_tool(query))

    # Copy essential metadata from the async tool
    sync_wrapper.__name__ = async_tool.__name__
    sync_wrapper.__doc__ = async_tool.__doc__

    return sync_wrapper

a2a_tool_async(url, toolname=None, http_kwargs=None) async

Perform a query using A2A to another agent.

Parameters:

Name	Type	Description	Default
url	str	The url in which the A2A agent is located.	required
toolname	str	The name for the created tool. Defaults to call_{agent name in card}. Leading and trailing whitespace are removed. Whitespace in the middle is replaced by _.	None
http_kwargs	dict	Additional kwargs to pass to the httpx client.	None

Returns:

Type	Description
Callable[[str], Coroutine[Any, Any, str]]	An async Callable that takes a query and returns the agent response.

Source code in src/any_agent/tools/a2a.py

async def a2a_tool_async(
    url: str, toolname: str | None = None, http_kwargs: dict[str, Any] | None = None
) -> Callable[[str], Coroutine[Any, Any, str]]:
    """Perform a query using A2A to another agent.

    Args:
        url (str): The url in which the A2A agent is located.
        toolname (str): The name for the created tool. Defaults to `call_{agent name in card}`.
            Leading and trailing whitespace are removed. Whitespace in the middle is replaced by `_`.
        http_kwargs (dict): Additional kwargs to pass to the httpx client.

    Returns:
        An async `Callable` that takes a query and returns the agent response.

    """
    if not a2a_tool_available:
        msg = "You need to `pip install 'any-agent[a2a]'` to use this tool"
        raise ImportError(msg)

    async with httpx.AsyncClient(follow_redirects=True) as resolver_client:
        a2a_agent_card: AgentCard = await (
            A2ACardResolver(httpx_client=resolver_client, base_url=url)
        ).get_agent_card()

    async def _send_query(query: str, task_id: str = str(uuid4())) -> str:
        async with httpx.AsyncClient(follow_redirects=True) as query_client:
            client = A2AClient(httpx_client=query_client, agent_card=a2a_agent_card)
            send_message_payload = SendMessageRequest(
                id=str(uuid4),
                params=MessageSendParams(
                    message=Message(
                        role=Role.user,
                        parts=[Part(root=TextPart(text=query))],
                        # the id is not currently tracked
                        messageId=str(uuid4().hex),
                        taskId=task_id,
                    )
                ),
            )
            # TODO check how to capture exceptions and pass them on to the enclosing framework
            response = await client.send_message(
                send_message_payload, http_kwargs=http_kwargs
            )

            if not response.root:
                msg = (
                    "The A2A agent did not return a root. Are you using an A2A agent not managed by any-agent? "
                    "Please file an issue at https://github.com/mozilla-ai/any-agent/issues so we can help."
                )
                raise ValueError(msg)

            if hasattr(response.root, "error"):
                response_str = ""
                response_str += f"Error: {response.root.error.message}\n\n"
                response_str += f"Code: {response.root.error.code}\n\n"
                response_str += f"Data: {response.root.error.data}\n\n"
            elif hasattr(response.root, "result"):
                response_str = ""
                response_str += f"Status: {response.root.result.status.state}\n\n"
                response_str += f"""Message: {" ".join([part.root.text for part in response.root.result.status.message.parts if part.root.kind == "text"])}\n\n"""
                response_str += (
                    f"TaskId: {response.root.result.status.message.taskId}\n\n"
                )
                response_str += (
                    f"Timestamp: {response.root.result.status.timestamp}\n\n"
                )
            else:
                msg = (
                    "The A2A agent did not return a error or a result. Are you using an A2A agent not managed by any-agent? "
                    "Please file an issue at https://github.com/mozilla-ai/any-agent/issues so we can help."
                )
                raise ValueError(msg)

            return response_str

    new_name = toolname or a2a_agent_card.name
    new_name = re.sub(r"\s+", "_", new_name.strip())
    _send_query.__name__ = f"call_{new_name}"
    _send_query.__doc__ = f"""{a2a_agent_card.description}
        Send a query to the A2A hosted agent named {a2a_agent_card.name}.

        Agent description: {a2a_agent_card.description}

        Args:
            query (str): The query to perform.
            task_id (str): The task id to use for the conversation. Defaults to a new uuid. If you want to continue the conversation, you should provide the same task id that you received in a previous response.

        Returns:
            The result from the A2A agent, encoded as a json string.
    """
    return _send_query

ask_user_verification(query)

Asks user to verify the given query.

Parameters:

Name	Type	Description	Default
query	str	The question that requires verification.	required

Source code in src/any_agent/tools/user_interaction.py

def ask_user_verification(query: str) -> str:
    """Asks user to verify the given `query`.

    Args:
        query: The question that requires verification.

    """
    return input(f"{query} => Type your answer here:")

search_tavily(query, include_images=False)

Perform a Tavily web search based on your query and return the top search results.

See https://blog.tavily.com/getting-started-with-the-tavily-search-api for more information.

Parameters:

Name	Type	Description	Default
query	str	The search query to perform.	required
include_images	bool	Whether to include images in the results.	False

Returns:

Type	Description
str	The top search results as a formatted string.

Source code in src/any_agent/tools/web_browsing.py

def search_tavily(query: str, include_images: bool = False) -> str:
    """Perform a Tavily web search based on your query and return the top search results.

    See https://blog.tavily.com/getting-started-with-the-tavily-search-api for more information.

    Args:
        query (str): The search query to perform.
        include_images (bool): Whether to include images in the results.

    Returns:
        The top search results as a formatted string.

    """
    if not tavily_available:
        msg = "You need to `pip install 'tavily-python'` to use this tool"
        raise ImportError(msg)
    api_key = os.getenv("TAVILY_API_KEY")
    if not api_key:
        return "TAVILY_API_KEY environment variable not set."
    try:
        client = TavilyClient(api_key)
        response = client.search(query, include_images=include_images)
        results = response.get("results", [])
        output = []
        for result in results:
            output.append(
                f"[{result.get('title', 'No Title')}]({result.get('url', '#')})\n{result.get('content', '')}"
            )
        if include_images and "images" in response:
            output.append("\nImages:")
            for image in response["images"]:
                output.append(image)
        return "\n\n".join(output) if output else "No results found."
    except Exception as e:
        return f"Error performing Tavily search: {e!s}"

search_web(query)

Perform a duckduckgo web search based on your query (think a Google search) then returns the top search results.

Parameters:

Name	Type	Description	Default
query	str	The search query to perform.	required

Returns:

Type	Description
str	The top search results.

Source code in src/any_agent/tools/web_browsing.py

def search_web(query: str) -> str:
    """Perform a duckduckgo web search based on your query (think a Google search) then returns the top search results.

    Args:
        query (str): The search query to perform.

    Returns:
        The top search results.

    """
    ddgs = DDGS()
    results = ddgs.text(query, max_results=10)
    return "\n".join(
        f"[{result['title']}]({result['href']})\n{result['body']}" for result in results
    )

send_console_message(user, query)

Send the specified user a message via console and returns their response.

Parameters:

Name	Type	Description	Default
query	str	The question to ask the user.	required
user	str	The user to ask the question to.	required

Returns:

Name	Type	Description
str	str	The user's response.

Source code in src/any_agent/tools/user_interaction.py

def send_console_message(user: str, query: str) -> str:
    """Send the specified user a message via console and returns their response.

    Args:
        query: The question to ask the user.
        user: The user to ask the question to.

    Returns:
        str: The user's response.

    """
    return Prompt.ask(f"{query}\n{user}")

show_final_output(answer)

Show the final answer to the user.

Parameters:

Name	Type	Description	Default
answer	str	The final answer.	required

Source code in src/any_agent/tools/user_interaction.py

def show_final_output(answer: str) -> str:
    """Show the final answer to the user.

    Args:
        answer: The final answer.

    """
    logger.info(f"Final output: {answer}")
    return answer

show_plan(plan)

Show the current plan to the user.

Parameters:

Name	Type	Description	Default
plan	str	The current plan.	required

Source code in src/any_agent/tools/user_interaction.py

def show_plan(plan: str) -> str:
    """Show the current plan to the user.

    Args:
        plan: The current plan.

    """
    logger.info(f"Current plan: {plan}")
    return plan

visit_webpage(url, timeout=30)

Visits a webpage at the given url and reads its content as a markdown string. Use this to browse webpages.

Parameters:

Name	Type	Description	Default
url	str	The url of the webpage to visit.	required
timeout	int	The timeout in seconds for the request.	30

Source code in src/any_agent/tools/web_browsing.py

def visit_webpage(url: str, timeout: int = 30) -> str:
    """Visits a webpage at the given url and reads its content as a markdown string. Use this to browse webpages.

    Args:
        url: The url of the webpage to visit.
        timeout: The timeout in seconds for the request.

    """
    try:
        response = requests.get(url, timeout=timeout)
        response.raise_for_status()

        markdown_content = markdownify(response.text).strip()  # type: ignore[no-untyped-call]

        markdown_content = re.sub(r"\n{2,}", "\n", markdown_content)

        return _truncate_content(markdown_content, 10000)
    except RequestException as e:
        return f"Error fetching the webpage: {e!s}"
    except Exception as e:
        return f"An unexpected error occurred: {e!s}"