# any-agent Documentation
> Complete documentation for any-agent - A Python library providing a single interface to different agent frameworks.
This file contains all documentation pages concatenated for easy consumption by AI systems.
---
## index.md
# any-agent
`any-agent` is a Python library providing a single interface to different agent frameworks.
!!! warning
Compared to traditional code-defined workflows, agent frameworks introduce complexity,
additional security implications to consider, and demand much more computational power.
Before jumping to use one, carefully consider and evaluate how much value you
would get compared to manually defining a sequence of tools and LLM calls.
## Requirements
- Python 3.11 or newer
## Installation
You can install the bare bones library as follows (only [`TinyAgent`](#./agents/frameworks/tinyagent.md) will be available):
```bash
pip install any-agent
```
Or you can install it with the required dependencies for different frameworks:
```bash
pip install any-agent[agno,openai]
```
Refer to [pyproject.toml](https://github.com/mozilla-ai/any-agent/blob/main/pyproject.toml) for a list of the options available.
## For AI Systems
This documentation is available in two AI-friendly formats:
- **[llms.txt](/llms.txt)** - A structured overview with curated links to key documentation sections
- **[llms-full.txt](/llms-full.txt)** - Complete documentation content concatenated into a single file
---
## agents/index.md
# Defining and Running Agents
## Defining Agents
To define any agent system you will always use the same imports:
```python
from any_agent import AgentConfig, AnyAgent, AgentRunError
# In these examples, the built-in tools will be used
from any_agent.tools import search_web, visit_webpage
```
Check [`AgentConfig`][any_agent.config.AgentConfig] for more info on how to configure agents.
### Single Agent
```python
agent = AnyAgent.create(
"openai", # See other options under `Frameworks`
AgentConfig(
model_id="mistral/mistral-small-latest",
instructions="Use the tools to find an answer",
tools=[search_web, visit_webpage]
),
)
```
### Multi-Agent
!!! warning
A multi-agent system introduces even more complexity than a single agent.
As stated before, carefully consider whether you need to adopt this pattern to
solve the task.
Multi-Agent systems can be implemented [using Agent-As-Tools](./tools.md#using-agents-as-tools).
### Framework Specific Arguments
Sometimes, there may be a new feature in a framework that you want to use that isn't yet supported universally in any-agent. The `agent_args` parameter in `AgentConfig` allows you to pass arguments specific to the underlying framework that the agent instance is built on.
**Example-1**: To pass the `output_guardrails` parameter, when using the OpenAI Agents SDK:
```python
from pydantic import BaseModel
from any_agent import AgentConfig, AgentFramework, AnyAgent
from agents import (
Agent,
GuardrailFunctionOutput,
OutputGuardrailTripwireTriggered,
RunContextWrapper,
Runner,
output_guardrail,
)
class MessageOutput(BaseModel):
response: str
class MathOutput(BaseModel):
reasoning: str
is_math: bool
guardrail_agent = Agent(
name="Guardrail check",
instructions="Check if the output includes any math.",
output_type=MathOutput,
)
@output_guardrail
async def math_guardrail(
ctx: RunContextWrapper, agent: Agent, output: MessageOutput
) -> GuardrailFunctionOutput:
result = await Runner.run(guardrail_agent, output.response, context=ctx.context)
return GuardrailFunctionOutput(
output_info=result.final_output,
tripwire_triggered=result.final_output.is_math,
)
framework = AgentFramework.OPENAI
agent = AnyAgent.create(
framework,
AgentConfig(
model_id="mistral/mistral-small-latest",
instructions="Check if the output contains any math",
agent_args={
"output_guardrails": [math_guardrail]
}
)
)
```
## Running Agents
```python
try:
agent_trace = agent.run("Which Agent Framework is the best??")
print(agent_trace.final_output)
except AgentRunError as are:
agent_trace = are.trace
```
Check [`AgentTrace`][any_agent.tracing.agent_trace.AgentTrace] for more info on the return type.
Exceptions are wrapped in an [`AgentRunError`][any_agent.AgentRunError], that carries the original exception in the `__cause__` attribute. Additionally, its `trace` property holds the trace containing the spans collected so far.
### Async
If you are running in `async` context, you should use the equivalent [`create_async`][any_agent.AnyAgent.create_async] and [`run_async`][any_agent.AnyAgent.run_async] methods:
```python
import asyncio
async def main():
agent = await AnyAgent.create_async(
"openai",
AgentConfig(
model_id="mistral/mistral-small-latest",
instructions="Use the tools to find an answer",
tools=[search_web, visit_webpage]
)
)
agent_trace = await agent.run_async("Which Agent Framework is the best??")
print(agent_trace.final_output)
if __name__ == "__main__":
asyncio.run(main())
```
### Batch Processing
While any-agent doesn't provide a dedicated `.run_batch()` API, we recommend using `asyncio.gather` with the [`AnyAgent.run_async`][any_agent.AnyAgent.run_async] API for concurrent processing:
```python
import asyncio
from any_agent import AgentConfig, AnyAgent
async def process_batch():
agent = await AnyAgent.create_async("tinyagent", AgentConfig(...))
inputs = ["Input 1", "Input 2", "Input 3"]
tasks = [agent.run_async(input_text) for input_text in inputs]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
```
### Multi-Turn Conversations
For scenarios where you need to maintain conversation history across multiple agent interactions, you can leverage the [`spans_to_messages`][any_agent.tracing.agent_trace.AgentTrace.spans_to_messages] method built into the AgentTrace. This function converts agent traces into a standardized message format that can be used to provide context in subsequent conversations.
!!! tip "When to Use Each Approach"
- **Multi-turn with `spans_to_messages`**: When you need to maintain context across separate agent invocations or implement complex conversation management logic
- **User interaction tools**: When you want the agent to naturally interact with users during its execution, asking questions as needed to complete its task
- **Hybrid approach**: Combine both patterns for sophisticated agents that maintain long-term context while also gathering real-time user input
#### Basic Multi-Turn Example
```python
from any_agent import AgentConfig, AnyAgent
# Create your agent
agent = AnyAgent.create(
"tinyagent",
AgentConfig(
model_id="mistral/mistral-small-latest",
instructions="You are a helpful assistant. Use previous conversation context when available.",
)
)
response1 = agent.run("What's the capital of California?")
print(f"Agent: {response1.final_output}")
conversation_history = response1.spans_to_messages()
# Convert previous conversation to readable format
history_text = "\n".join([
f"{msg.role.capitalize()}: {msg.content}"
for msg in conversation_history
if msg.role != "system"
])
user_message = "What's the closest national park to that city"
full_prompt = f"""Previous conversation:
{history_text}
Current user message: {user_message}
Please respond taking into account the conversation history above."""
response2 = agent.run(full_prompt)
print(f"Agent: {response2.final_output}") # Agent will understand "that city" refers to Sacramento
```
#### Design Philosophy: Thoughtful Message History Management
You may notice that the `agent.run()` method doesn't accept a `messages` parameter directly. This is an intentional design choice to encourage thoughtful handling of conversation history by developers. Rather than automatically managing message history, any-agent empowers you to:
- **Choose your context strategy**: Decide what parts of conversation history are relevant
- **Manage token usage**: Control how much context you include to optimize costs and performance
- **Handle complex scenarios**: Implement custom logic for conversation branching, summarization, or context windowing
This approach ensures that conversation context is handled intentionally rather than automatically, leading to more efficient and purposeful agent interactions.
#### Using User Interaction Tools for Regular Conversations
For scenarios where you need regular, back-and-forth interaction with users, we recommend using or building your own **user interaction tools** rather than managing conversation history manually. This pattern allows the agent to naturally ask follow-up questions and gather information as needed. We provide a default `send_console_message` tool which uses console inputs and outputs, but you may need to use a more advanced tool (such as a Slack MCP Server) to handle user interaction.
```python
from any_agent import AgentConfig, AnyAgent
from any_agent.tools.user_interaction import send_console_message
# Create agent with user interaction capabilities
agent = AnyAgent.create(
"tinyagent",
AgentConfig(
model_id="mistral/mistral-small-latest",
instructions="You are a helpful travel assistant. Send console messages to ask more questions. Do not stop until you've answered the question.",
tools=[send_console_message]
)
)
# The agent can now naturally ask questions during its execution
prompt = """
I'm planning a trip and need help finding accommodations.
Please ask me some questions to understand my preferences, then provide recommendations.
"""
agent_trace = agent.run(prompt)
print(f"Final recommendations: {agent_trace.final_output}")
```
This approach is demonstrated in our [MCP Agent cookbook example](../cookbook/mcp_agent.ipynb), where an agent uses user interaction tools to gather trip planning information dynamically. The agent can ask clarifying questions, get user preferences, and provide personalized recommendations all within a single `run()` call.
---
## agents/callbacks.md
# Agent Callbacks
For greater control when running your agent, `any-agent` includes support for custom [`Callbacks`][any_agent.callbacks.base.Callback] that
will be called at different points of the [`AnyAgent.run`][any_agent.AnyAgent.run]:
```py
# pseudocode of an Agent run
history = [system_prompt, user_prompt]
context = Context()
while True:
for callback in agent.config.callbacks:
context = callback.before_llm_call(context)
response = CALL_LLM(history)
for callback in agent.config.callbacks:
context = callback.after_llm_call(context)
history.append(response)
if response.tool_executions:
for tool_execution in tool_executions:
for callback in agent.config.callbacks:
context = callback.before_tool_execution(context)
tool_response = EXECUTE_TOOL(tool_execution)
for callback in agent.config.callbacks:
context = callback.after_tool_execution(context)
history.append(tool_response)
else:
return response
```
Advanced designs such as safety guardrails or custom side-effects can be integrated into your agentic system using this functionality.
## Context
During the agent run ( [`agent.run_async`][any_agent.AnyAgent.run_async] or [`agent.run`][any_agent.AnyAgent.run] ), a unique [`Context`][any_agent.callbacks.context.Context] object is created and shared across all callbacks.
`any-agent` populates the [`Context.current_span`][any_agent.callbacks.context.Context.current_span]
property so that callbacks can access information in a framework-agnostic way.
You can see what attributes are available for LLM Calls and Tool Executions by examining the [`GenAI`][any_agent.tracing.attributes.GenAI] class.
## Implementing Callbacks
All callbacks must inherit from the base [`Callback`][any_agent.callbacks.base.Callback] class and can choose to implement any subset of the available callback methods.
You can use [`Context.shared`][any_agent.callbacks.context.Context.shared] to store information meant
to be reused across callbacks:
```python
from any_agent.callbacks import Callback, Context
from any_agent.tracing.attributes import GenAI
class CountSearchWeb(Callback):
def after_tool_execution(self, context: Context, *args, **kwargs) -> Context:
if "search_web_count" not in context.shared:
context.shared["search_web_count"] = 0
if context.current_span.attributes[GenAI.TOOL_NAME] == "search_web":
context.shared["search_web_count"] += 1
```
Callbacks can raise exceptions to stop agent execution. This is useful for implementing safety guardrails or validation logic:
```python
class LimitSearchWeb(Callback):
def __init__(self, max_calls: int):
self.max_calls = max_calls
def before_tool_execution(self, context: Context, *args, **kwargs) -> Context:
if context.shared["search_web_count"] > self.max_calls:
raise RuntimeError("Reached limit of `search_web` calls.")
```
!!! warning
Raising exceptions in callbacks will terminate the agent run immediately. Use this feature carefully to implement safety measures or validation logic.
## Default Callbacks
`any-agent` comes with a set of default callbacks that will be used by default (if you don't pass a value to `AgentConfig.callbacks`):
- [`AddCostInfo`][any_agent.callbacks.span_cost.AddCostInfo]
- [`ConsolePrintSpan`][any_agent.callbacks.span_print.ConsolePrintSpan]
If you want to disable these default callbacks, you can pass an empty list:
```python
from any_agent import AgentConfig, AnyAgent
from any_agent.tools import search_web, visit_webpage
agent = AnyAgent.create(
"tinyagent",
AgentConfig(
model_id="mistral/mistral-small-latest",
instructions="Use the tools to find an answer",
tools=[search_web, visit_webpage],
callbacks=[]
),
)
```
## Providing your own Callbacks
Callbacks are provided to the agent using the [`AgentConfig.callbacks`][any_agent.AgentConfig.callbacks] property.
=== "Alongside the default callbacks"
You can use [`get_default_callbacks`][any_agent.callbacks.get_default_callbacks]:
```py
from any_agent import AgentConfig, AnyAgent
from any_agent.callbacks import get_default_callbacks
from any_agent.tools import search_web, visit_webpage
agent = AnyAgent.create(
"tinyagent",
AgentConfig(
model_id="gpt-4.1-nano",
instructions="Use the tools to find an answer",
tools=[search_web, visit_webpage],
callbacks=[
CountSearchWeb(),
LimitSearchWeb(max_calls=3),
*get_default_callbacks()
]
),
)
```
=== "Override the default callbacks"
```py
from any_agent import AgentConfig, AnyAgent
from any_agent.tools import search_web, visit_webpage
agent = AnyAgent.create(
"tinyagent",
AgentConfig(
model_id="gpt-4.1-nano",
instructions="Use the tools to find an answer",
tools=[search_web, visit_webpage],
callbacks=[
CountSearchWeb(),
LimitSearchWeb(max_calls=3)
]
),
)
```
!!! warning
Callbacks will be called in the order that they are added, so it is important to pay attention to the order
in which you set the callback configuration.
In the above example, passing:
```py
callbacks=[
LimitSearchWeb(max_calls=3)
CountSearchWeb()
]
```
Would fail because `context.shared["search_web_count"]`
was not set yet.
## Example: Offloading sensitive information
Some inputs and/or outputs in your traces might contain sensitive information that you don't want
to be exposed in the [traces](#../tracing.md).
You can use callbacks to offload the sensitive information to an external location and replace the span
attributes with a reference to that location:
```python
import json
from pathlib import Path
from any_agent.callbacks.base import Callback
from any_agent.callbacks.context import Context
from any_agent.tracing.attributes import GenAI
class SensitiveDataOffloader(Callback):
def __init__(self, output_dir: str) -> None:
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True, parents=True)
def before_llm_call(self, context: Context, *args, **kwargs) -> Context:
span = context.current_span
if input_messages := span.attributes.get(GenAI.INPUT_MESSAGES):
output_file = self.output_dir / f"{span.get_span_context().trace_id}.txt"
output_file.write_text(str(input_messages))
span.set_attribute(
GenAI.INPUT_MESSAGES,
json.dumps(
{"ref": str(output_file)}
)
)
return context
```
You can find a working example in the [Callbacks Cookbook](../cookbook/callbacks.ipynb).
---
## agents/frameworks/agno.md
# Agno
[https://github.com/agno-agi/agno](https://github.com/agno-agi/agno)
## Default Agent Type
We use [`agno.agent.Agent`](https://docs.agno.com/reference/agents/agent) as default.
Check the reference to find additional supported `agent_args`.
## Default Model Type
We use [`agno.models.litellm.LiteLLM`](https://docs.agno.com/models/litellm) as default.
Check the reference to find additional supported `model_args`.
## Examples
### Limiting the number of steps
```python
from any_agent import AnyAgent, AgentConfig
from any_agent.tools import search_web, visit_webpage
agent = AnyAgent.create(
"agno",
AgentConfig(
model_id="mistral/mistral-small-latest",
instructions="You must use the available tools to find an answer",
tools=[search_web, visit_webpage]
),
agent_args={
"tool_call_limit": 3
}
)
agent.run("Which Agent Framework is the best??")
```
---
## agents/frameworks/google_adk.md
# Google Agent Development Kit (ADK)
[https://github.com/google/adk-python](https://github.com/google/adk-python)
## Default Agent Type
We use [`google.adk.agents.llm_agent.LlmAgent`](https://google.github.io/adk-docs/agents/llm-agents/) as default.
Check the reference to find additional supported `agent_args`.
## Default Model Type
We use [`google.adk.models.lite_llm.LiteLLM`](https://google.github.io/adk-docs/agents/models/#using-cloud-proprietary-models-via-litellm) as default.
Check the reference to find additional supported `model_args`.
## Run args
Check [`RunConfig`](https://google.github.io/adk-docs/runtime/runconfig/) to find additional supported `AnyAgent.run` args.
## Examples
### Limiting the number of steps
```python
from any_agent import AnyAgent, AgentConfig
from any_agent.tools import search_web, visit_webpage
from google.adk.agents.run_config import RunConfig
agent = AnyAgent.create(
"google",
AgentConfig(
model_id="mistral/mistral-small-latest",
instructions="You must use the available tools to find an answer",
tools=[search_web, visit_webpage]
)
)
agent.run(
"Which Agent Framework is the best??",
run_config=RunConfig(
max_llm_calls=3
)
)
```
---
## agents/frameworks/langchain.md
# LangChain
[https://github.com/langchain-ai/langchain](https://github.com/langchain-ai/langchain)
[https://github.com/langchain-ai/langgraph](https://github.com/langchain-ai/langgraph)
## Default Agent Type
We use [`langgraph.prebuilt.create_react_agent`](https://langchain-ai.github.io/langgraph/reference/agents/?h=create_rea#langgraph.prebuilt.chat_agent_executor.create_react_agent) as default.
Check the reference to find additional supported `agent_args`.
## Default Model Type
We use [`langchain_litellm.ChatLiteLLM`](https://python.langchain.com/docs/integrations/chat/litellm/#chatlitellm)
Check the reference to find additional supported `model_args`.
## Run args
Check [`RunnableConfig`](https://python.langchain.com/api_reference/core/runnables/langchain_core.runnables.config.RunnableConfig.html) to find additional supported `AnyAgent.run` args.
## Examples
### Limiting the number of steps
```python
from any_agent import AnyAgent, AgentConfig
from any_agent.tools import search_web, visit_webpage
from langchain_core.runnables import RunnableConfig
agent = AnyAgent.create(
"langchain",
AgentConfig(
model_id="mistral/mistral-small-latest",
instructions="You must use the available tools to find an answer",
tools=[search_web, visit_webpage]
)
)
agent.run(
"Which Agent Framework is the best??",
config=RunnableConfig(
recursion_limit=3
)
)
```
---
## agents/frameworks/llama_index.md
# LlamaIndex
[https://github.com/run-llama/llama_index](https://github.com/run-llama/llama_index)
## Default Agent Type
We use [`llama_index.core.agent.workflow.react_agent.FunctionAgent`](https://docs.llamaindex.ai/en/stable/api_reference/agent/#llama_index.core.agent.workflow.FunctionAgent) as default.
However, this agent requires that tools are used. If no tools are used, any-agent will default to [`llama_index.core.agent.workflow.react_agent.ReActAgent`](https://docs.llamaindex.ai/en/stable/api_reference/agent/#llama_index.core.agent.workflow.ReActAgent).
Check the reference to find additional supported `agent_args`.
## Default Model Type
We use [`llama_index.llms.litellm.LiteLLM`](https://docs.llamaindex.ai/en/stable/examples/llm/litellm/) as default.
Check the reference to find additional supported `model_args`.
## Examples
### Limiting the number of steps
Pending on https://github.com/run-llama/llama_index/issues/18535
---
## agents/frameworks/openai.md
# OpenAI Agents SDK
[https://github.com/openai/openai-agents-python](https://github.com/openai/openai-agents-python)
## Default Agent Type
We use [`agents.Agent`](ttps://openai.github.io/openai-agents-python/ref/agent/#agents.agent.Agent) as default.
Check the reference to find additional supported `agent_args`.
## Default Model Type
We use [`agents.extensions.models.litellm_model.LitellmModel`](https://openai.github.io/openai-agents-python/ref/extensions/litellm/) as default.
Check the reference to find additional supported `model_args`.
## Run args
Check [`agents.run.Runner.run`](https://openai.github.io/openai-agents-python/ref/run/#agents.run.Runner.run) to find additional supported `AnyAgent.run` args.
## Examples
### Limiting the number of steps
```python
from any_agent import AnyAgent, AgentConfig
from any_agent.tools import search_web, visit_webpage
agent = AnyAgent.create(
"openai",
AgentConfig(
model_id="mistral/mistral-small-latest",
instructions="You must use the available tools to find an answer",
tools=[search_web, visit_webpage]
)
)
agent.run(
"Which Agent Framework is the best??",
max_turns=3
)
```
---
## agents/frameworks/smolagents.md
# smolagents
[https://github.com/huggingface/smolagents](https://github.com/huggingface/smolagents)
## Default Agent Type
We use [`smolagents.ToolCallingAgent`](https://huggingface.co/docs/smolagents/reference/agents#smolagents.ToolCallingAgent) as default.
Check the reference to find additional supported `agent_args`.
## Default Model Type
We use [`smolagents.LiteLLMModel`](https://huggingface.co/docs/smolagents/reference/models#smolagents.LiteLLMModel) as default.
Check the reference to find additional supported `model_args`.
## Run args
Check [`smolagents.MultiStepAgent.run`](https://huggingface.co/docs/smolagents/main/en/reference/agents#smolagents.MultiStepAgent.run) to find additional supported `AnyAgent.run` args.
## Examples
### Limiting the number of steps
```python
from any_agent import AnyAgent, AgentConfig
from any_agent.tools import search_web, visit_webpage
agent = AnyAgent.create(
"smolagents",
AgentConfig(
model_id="mistral/mistral-small-latest",
instructions="You must use the available tools to find an answer",
tools=[search_web, visit_webpage]
)
)
agent.run(
"Which Agent Framework is the best??",
max_steps=3
)
```
---
## agents/frameworks/tinyagent.md
# TinyAgent
As part of the bare bones library, we provide our own Python implementation based on [HuggingFace Tiny Agents](https://huggingface.co/blog/tiny-agents).
You can find it in [`any_agent.frameworks.tinyagent`](https://github.com/mozilla-ai/any-agent/blob/main/src/any_agent/frameworks/tinyagent.py).
## Examples
### Use MCP Tools
```python
from any_agent import AnyAgent, AgentConfig
from any_agent.config import MCPStdio
agent = AnyAgent.create(
"tinyagent",
AgentConfig(
model_id="mistral/mistral-small-latest",
instructions="You must use the available tools to find an answer",
tools=[
MCPStdio(
command="uvx",
args=["duckduckgo-mcp-server"]
)
]
)
)
result = agent.run(
"Which Agent Framework is the best??"
)
print(result.final_output)
```
---
## agents/tools.md
# Agent Tools
`any-agent` provides 2 options to specify what `tools` are available to your agent: `Callables` and `MCP` ([Model Context Protocol](https://modelcontextprotocol.io/introduction)). In order to support multi-agent systems, any agents served via A2A can also be integrated by wrapping the A2A connection in a callable function tool as described [below](#a2a-tools).
You can use any combination of options within the same agent.
## Callables
Any Python callable can be directly passed as tools.
You can define them in the same script, import it from an external package, etc.
Under the hood, `any-agent` takes care of wrapping the
tool so it becomes usable by the selected framework.
!!! tip
Check all the [built-in callable tools](#../api/tools.md) that any-agent provides.
```python
from any_agent import AgentConfig
from any_agent.tools import search_web
main_agent = AgentConfig(
model_id="mistral/mistral-small-latest",
tools=[search_web]
)
```
## MCP
MCP can either be run locally ([MCPStdio][any_agent.config.MCPStdio]) or you can connect to an MCP that is running elsewhere ([MCPSse][any_agent.config.MCPSse]).
!!! tip
There are tools like [SuperGateway](https://github.com/supercorp-ai/supergateway) providing an easy way to turn a Stdio server into an SSE server.
=== "MCP (Stdio)"
See the [MCPStdio][any_agent.config.MCPStdio] API Reference.
```python
from any_agent import AgentConfig
from any_agent.config import MCPStdio
main_agent = AgentConfig(
model_id="mistral/mistral-small-latest",
tools=[
MCPStdio(
command="docker",
args=["run", "-i", "--rm", "mcp/fetch"],
tools=["fetch"]
),
]
)
```
=== "MCP (SSE)"
See the [MCPSse][any_agent.config.MCPSse] API Reference.
```python
from any_agent import AgentConfig
from any_agent.config import MCPSse
main_agent = AgentConfig(
model_id="mistral/mistral-small-latest",
tools=[
MCPSse(
url="http://localhost:8000/sse"
),
]
)
```
## Using Agents-As-Tools
To directly use one agent as a tool for another agent, `any-agent` provides 3 different approaches:
The agent to be used as a tool can be defined as usual:
```py
from any_agent import AgentConfig, AnyAgent
from any_agent.tools import search_web
google_agent = await AnyAgent.create_async(
"google",
AgentConfig(
name="google_expert",
model_id="mistral/mistral-small-latest",
instructions="Use the available tools to answer questions about the Google ADK",
description="An agent that can answer questions about the Google Agents Development Kit (ADK).",
tools=[search_web]
)
)
```
You can then choose to wrap the agent using different approaches:
=== "Agent as Callable"
```py
async def google_agent_as_tool(query: str) -> str:
agent_trace = await google_agent.run_async(prompt=query)
return str(agent_trace.final_output)
google_agent_as_tool.__doc__ = google_agent.config.description
```
=== "Agent as MCP"
```py
from any_agent.config import MCPSse
from any_agent.serving import MCPServingConfig
mcp_handle = await google_agent.serve_async(
MCPServingConfig(port=5001, endpoint="/google-agent"))
google_agent_as_tool = MCPSse(
url="http://localhost:5001/google-agent/sse")
```
=== "Agent as A2A"
```py
from any_agent.serving import A2AServingConfig
from any_agent.tools import a2a_tool_async
a2a_handle = await google_agent.serve_async(
A2AServingConfig(port=5001, endpoint="/google-agent"))
google_agent_as_tool = await a2a_tool_async(
url="http://localhost:5001/google-agent")
```
Finally, regardless of the option chosen above, you can pass the agent as a tool to another agent:
```py
main_agent = await AnyAgent.create_async(
"tinyagent",
AgentConfig(
name="main_agent",
model_id="mistral-small-latest",
instructions="Use the available tools to obtain additional information to answer the query.",
tools=[google_agent_as_tool],
)
)
```
---
## tracing.md
# Agent Tracing
An [`AgentTrace`][any_agent.tracing.agent_trace.AgentTrace] is returned when calling [`agent.run`][any_agent.AnyAgent.run] or [`agent.run_async`][any_agent.AnyAgent.run_async].
`any-agent` generates standardized (regarding the structure) [OpenTelemetry](https://opentelemetry.io/) traces regardless of the framework used, based on the [Semantic conventions for generative AI systems](https://opentelemetry.io/docs/specs/semconv/gen-ai/). This means that any OpenTelemetry exporter compatible with the Python SDK can be added. More information can be found [below](#adding-an-opentelemetry-exporter).
!!! info
Check the exposed attributes in [`GenAI`][any_agent.tracing.attributes.GenAI]
You can try to 🔍 find 🔍 the subtle differences (regarding the content) across frameworks in the examples below.
!!! tip
The following are real traces generated by executing [one of our integration tests](https://github.com/mozilla-ai/any-agent/blob/main/tests/integration/test_load_and_run_agent.py).
## Console
Here is what the console output looks like:
=== "AGNO"
{% include "../tests/assets/AGNO_trace.html" %}
=== "GOOGLE"
{% include "../tests/assets/GOOGLE_trace.html" %}
=== "LANGCHAIN"
{% include "../tests/assets/LANGCHAIN_trace.html" %}
=== "LLAMA_INDEX"
{% include "../tests/assets/LLAMA_INDEX_trace.html" %}
=== "OPENAI"
{% include "../tests/assets/OPENAI_trace.html" %}
=== "SMOLAGENTS"
{% include "../tests/assets/SMOLAGENTS_trace.html" %}
=== "TINYAGENT"
{% include "../tests/assets/TINYAGENT_trace.html" %}
!!! tip
The spans are printed to the console by default, using the `callback` mechanism.
See [Default Callbacks](./agents/callbacks.md#default-callbacks) for more information
and how to disable this behavior.
## Spans
Here's what the returned [`agent_trace.spans`][any_agent.AgentTrace.spans] look like when [dumped to JSON format](#dumping-to-file):
=== "AGNO"
~~~json
{% include "../tests/assets/AGNO_trace.json" %}
~~~
=== "GOOGLE"
~~~json
{% include "../tests/assets/GOOGLE_trace.json" %}
~~~
=== "LANGCHAIN"
~~~json
{% include "../tests/assets/LANGCHAIN_trace.json" %}
~~~
=== "LLAMA_INDEX"
~~~json
{% include "../tests/assets/LLAMA_INDEX_trace.json" %}
~~~
=== "OPENAI"
~~~json
{% include "../tests/assets/OPENAI_trace.json" %}
~~~
=== "SMOLAGENTS"
~~~json
{% include "../tests/assets/SMOLAGENTS_trace.json" %}
~~~
=== "TINYAGENT"
~~~json
{% include "../tests/assets/TINYAGENT_trace.json" %}
~~~
## Dumping to File
The AgentTrace object is a pydantic model and can be saved to disk via standard pydantic practices:
```python
from any_agent import AgentConfig, AnyAgent
from any_agent.tools import search_web
agent = AnyAgent.create(
"openai",
agent_config=AgentConfig(
model_id="mistral/mistral-small-latest",
tools=[search_web],
)
)
agent_trace = agent.run("Which agent framework is the best?")
with open("agent_trace.json", "w", encoding="utf-8") as f:
f.write(agent_trace.model_dump_json(indent=2))
```
## Adding an OpenTelemetry exporter
Before starting to use the library, you can add new OpenTelemetry exporters and processors as needed. Note that this does not affect the existing processor that returns the traces to the user, and optionally prints them on `stdout`.
The following code will use the OpenTelemetry Python SDK to send the agent traces to an additional endpoint using OTLP over HTTP in the indicated URL:
```python
from opentelemetry.trace import get_tracer_provider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
tp = get_tracer_provider()
http_exporter = OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")
tp.add_span_processor(SimpleSpanProcessor(http_exporter))
```
---
## evaluation.md
# Agent Evaluation
The any-agent evaluation module encourages three approaches for evaluating agent traces:
1. **Custom Code Evaluation**: Direct programmatic inspection of traces for deterministic checks
2. **[`LlmJudge`][any_agent.evaluation.LlmJudge]**: LLM-as-a-judge for evaluations that can be answered with a direct LLM call alongside a custom context
3. **[`AgentJudge`][any_agent.evaluation.AgentJudge]**: Complex LLM-based evaluations that utilize built-in and customizable tools to inspect specific parts of the trace or other custom information provided to the agent as a tool
## Choosing the Right Evaluation Method
| Method | Best For | Pros | Cons |
|--------|----------|------|------|
| **Custom Code** | Deterministic checks, performance metrics, specific criteria | Fast, reliable, cost-effective, precise control | Requires manual coding, limited to predefined checks |
| **LlmJudge** | Simple qualitative assessments, text-based evaluations | Easy to set up, flexible questions, good for subjective evaluation | Can be inconsistent, costs tokens, slower than code |
| **AgentJudge** | Complex multi-step evaluations, tool usage analysis | Most flexible, can support tool access to custom additional information sources | Highest cost, slowest, most complex setup |
Both judges work with any-agent's unified tracing format and return structured evaluation results.
## Custom Code Evaluation
Before automatically using an LLM based approach, it is worthwhile to consider whether it is necessary. For deterministic evaluations where you know exactly what to check, you may not want an LLM-based judge at all. Writing a custom evaluation function that directly examines the trace can be more efficient, reliable, and cost-effective. the any-agent [`AgentTrace`][any_agent.tracing.agent_trace.AgentTrace] provides a few helpful methods that can be used to extract common information.
### Example: Custom Evaluation Function
```python
from any_agent.tracing.agent_trace import AgentTrace
def evaluate_efficiency(trace: AgentTrace) -> dict:
"""Custom evaluation function for efficiency criteria."""
# Direct access to trace properties
token_count = trace.tokens.total_tokens
step_count = len(trace.spans)
final_output = trace.final_output
# Apply your specific criteria
results = {
"token_efficient": token_count < 1000,
"step_efficient": step_count <= 5,
"has_output": final_output is not None,
"token_count": token_count,
"step_count": step_count
}
# Calculate overall pass/fail
results["passed"] = all([
results["token_efficient"],
results["step_efficient"],
results["has_output"]
])
return results
# Usage
from any_agent import AgentConfig, AnyAgent
from any_agent.evaluation import LlmJudge
from any_agent.tools import search_web
# First, run an agent to get a trace
agent = AnyAgent.create(
"tinyagent",
AgentConfig(
model_id="mistral/mistral-small-latest",
tools=[search_web]
),
)
trace = agent.run("What is the capital of France?")
evaluation = evaluate_efficiency(trace)
print(f"Evaluation results: {evaluation}")
```
### Working with Trace Messages
You can also examine the conversation flow directly:
```python
def check_tool_usage(trace: AgentTrace, required_tool: str) -> bool:
"""Check if a specific tool was used in the trace."""
messages = trace.spans_to_messages()
for message in messages:
if message.role == "tool" and required_tool in message.content:
return True
return False
# Usage
used_search = check_tool_usage(trace, "search_web")
print(f"Used web search: {used_search}")
```
## LlmJudge
The `LlmJudge` is ideal for straightforward evaluation questions that can be answered by examining the complete trace text. It's efficient and works well for:
- Basic pass/fail assessments
- Simple criteria checking
- Text-based evaluations
### Example: Evaluating Response Quality and Helpfulness
```python
from any_agent import AnyAgent, AgentConfig
from any_agent.tools import search_web
from any_agent.evaluation import LlmJudge
# Run an agent on a customer support task
agent = AnyAgent.create(
"tinyagent",
AgentConfig(
model_id="mistral/mistral-small-latest",
tools=[search_web]
),
)
trace = agent.run(
"A customer is asking about setting up a new email account on the latest version of iOS. "
"They mention they're not very tech-savvy and seem frustrated. "
"Help them with clear, step-by-step instructions."
)
# Evaluate the quality of the agent's response multiple times
judge = LlmJudge(model_id="mistral/mistral-small-latest")
evaluation_questions = [
"Did it provide clear, step-by-step instructions?",
"Was the tone empathetic and appropriate for a frustrated, non-technical customer?",
"Did it avoid using technical jargon without explanation?",
"Was the response complete and actionable?",
"Does the description specify which version of iOS this works with?"
]
# Run evaluation 4 times to check consistency
results = []
for evaluation_question in evaluation_questions:
question = f"Evaluate whether the agent's response demonstrates good customer service by considering: {evaluation_question}."
result = judge.run(context=str(trace.spans_to_messages()), question=evaluation_question)
results.append(result)
# Print all results
for i, result in enumerate(results, 1):
print(f"Run {i} - Passed: {result.passed}")
print(f"Run {i} - Reasoning: {result.reasoning}")
print("-" * 50)
```
!!! tip "Async Usage"
For async applications, use `judge.run_async()` instead of `judge.run()`.
## AgentJudge
The `AgentJudge` is designed for complex evaluations that require inspecting specific aspects of the trace. It comes equipped with evaluation tools and can accept additional custom tools for specialized assessments.
### Built-in Evaluation Tools
The `AgentJudge` automatically has access to these evaluation tools:
- `get_final_output()`: Get the agent's final output
- `get_tokens_used()`: Get total token usage
- `get_steps_taken()`: Get number of steps taken
- `get_messages_from_trace()`: Get formatted trace messages
- `get_duration()`: Get the duration in seconds of the trace
### Example: Agent Judge with Tool Access
```python
from any_agent.evaluation import AgentJudge
# Create an agent judge
judge = AgentJudge(model_id="mistral/mistral-small-latest")
# Evaluate with access to trace inspection tools
eval_trace = judge.run(
trace=trace,
question="Does the final answer provided by the trace mention and correctly specify the most recent major version of iOS? You may need to do a web search to determine the most recent version of iOS. If the final answer does not mention the version at all, this criteria should fail",
additional_tools=[search_web]
)
result = eval_trace.final_output
print(f"Passed: {result.passed}")
print(f"Reasoning: {result.reasoning}")
```
!!! tip "Async Usage"
For async applications, use `judge.run_async()` instead of `judge.run()`.
### Adding Custom Tools
You can extend the `AgentJudge` with additional tools for specialized evaluations:
```python
def current_ios_version() -> str:
"""Custom tool to retrieve the most recent version of iOS
Returns:
The version of iOS
"""
return "iOS 18.5"
judge = AgentJudge(model_id="mistral/mistral-small-latest")
eval_trace = judge.run(
trace=trace,
question="Does the final answer provided by the trace mention and correctly specify the most recent major version of iOS? If the final answer does not mention the version at all, this criteria should fail",
additional_tools=[current_ios_version]
)
```
## Custom Output Types
Both judges support custom output schemas using Pydantic models:
```python
from pydantic import BaseModel
class DetailedEvaluation(BaseModel):
passed: bool
reasoning: str
confidence_score: float
suggestions: list[str]
judge = LlmJudge(
model_id="mistral/mistral-small-latest",
output_type=DetailedEvaluation
)
result = judge.run(trace=trace, question="Evaluate the agent's performance")
print(f"Confidence: {result.confidence_score}")
print(f"Suggestions: {result.suggestions}")
```
---
## serving.md
# Serving
`any-agent` provides a simple way of serving agents from any of the supported frameworks using the
[Agent2Agent Protocol (A2A)](https://google.github.io/A2A/), via the [A2A Python SDK](https://github.com/google-a2a/a2a-python), or using the [Model Context Protocol (MCP)](https://modelcontextprotocol.io/specification/2025-03-26), via the [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk). You can refer to the links for more information on
these protocols, as explaining them is out of the scope of this page.
!!! warning
The A2A protocol is in early stages of development and so is the functionality provided by `any-agent` here.
In order to use A2A serving, you must first install the 'a2a' extra: `pip install 'any-agent[a2a]'`
You can configure and serve an agent using the [`A2AServingConfig`][any_agent.serving.A2AServingConfig] and the [`AnyAgent.serve_async`][any_agent.AnyAgent.serve_async] method.
## Running Async Servers in Sync Environments
Since `any-agent` uses async/await patterns for better performance and resource management, the serving functions are async by default. However, you can easily run async servers in sync environments using Python's `asyncio` utilities:
### Using `asyncio.run()`
The simplest approach is to wrap your async code in `asyncio.run()`:
```python
import asyncio
from any_agent import AgentConfig, AnyAgent
from any_agent.serving import A2AServingConfig
async def main():
agent = await AnyAgent.create_async(
"tinyagent",
AgentConfig(
name="my_agent",
model_id="mistral/mistral-small-latest",
description="A helpful agent"
)
)
server_handle = await agent.serve_async(A2AServingConfig(port=8080))
try:
# Keep the server running
await server_handle.task
except KeyboardInterrupt:
await server_handle.shutdown()
asyncio.run(main())
```
## Serving via A2A
### Example
For illustrative purposes, we are going to define 2 separate scripts, each defining an agent to answer questions about a specific agent framework (either OpenAI Agents SDK or Google ADK):
=== "Google Expert"
```python
# google_expert.py
import asyncio
from any_agent import AgentConfig, AnyAgent
from any_agent.serving import A2AServingConfig
from any_agent.tools import search_web
async def main():
agent = await AnyAgent.create_async(
"google",
AgentConfig(
name="google_expert",
model_id="mistral/mistral-small-latest",
description="An agent that can answer questions specifically and only about the Google Agents Development Kit (ADK). Reject questions about anything else.",
tools=[search_web]
)
)
server_handle = await agent.serve_async(A2AServingConfig(port=5001))
await server_handle.task
asyncio.run(main())
```
=== "OpenAI Expert"
```python
# openai_expert.py
import asyncio
from any_agent import AgentConfig, AnyAgent
from any_agent.serving import A2AServingConfig
from any_agent.tools import search_web
async def main():
agent = await AnyAgent.create_async(
"openai",
AgentConfig(
name="openai_expert",
model_id="mistral/mistral-small-latest",
instructions="You can answer questions about the OpenAI Agents SDK but nothing else.",
description="An agent that can answer questions specifically about the OpenAI Agents SDK.",
tools=[search_web]
)
)
server_handle = await agent.serve_async(A2AServingConfig(port=5002))
await server_handle.task
asyncio.run(main())
```
We can then run each of the scripts in a separate terminal and leave them running in the background.
Now, using a simple python script that implements the A2A client, we can communicate with these agents! For this example,
we use the [A2A Python SDK](https://github.com/google-a2a/a2a-python)
```python
from uuid import uuid4
import asyncio
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import AgentCard, MessageSendParams, SendMessageRequest
async def main():
async with httpx.AsyncClient() as httpx_client:
agent_card: AgentCard = await A2ACardResolver(
httpx_client,
base_url="http://localhost:5001",
).get_agent_card(http_kwargs=None)
client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)
send_message_payload = {
"message": {
"role": "user",
"parts": [{"kind": "text", "text": "What do you know about the Google ADK?"}],
"messageId": uuid4().hex,
},
}
request = SendMessageRequest(params=MessageSendParams(**send_message_payload))
response = await client.send_message(request, http_kwargs={"timeout": 60})
print(f" Response from first agent: {response.model_dump_json(indent=2)}")
agent_card: AgentCard = await A2ACardResolver(
httpx_client,
base_url="http://localhost:5002",
).get_agent_card(http_kwargs=None)
client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)
send_message_payload = {
"message": {
"role": "user",
"parts": [{"kind": "text", "text": "What do you know about the Google ADK?"}],
"messageId": uuid4().hex,
},
}
request = SendMessageRequest(params=MessageSendParams(**send_message_payload))
response = await client.send_message(request, http_kwargs={"timeout": 60})
print(f" Response from second agent: {response.model_dump_json(indent=2)}")
if __name__ == "__main__":
asyncio.run(main())
```
You will see that the first agent answered the question, but the second agent did not answer the question.
This is because the question was about Google ADK,
but the agent was told it could only answer questions about the OpenAI Agents SDK.
### Advanced Configuration
#### Custom Skills
By default, an agent's skills are automatically inferred from its tools. However, you can explicitly define skills for more control over the agent card:
```python
from a2a.types import AgentSkill
from any_agent.serving import A2AServingConfig
# Define custom skills
custom_skills = [
AgentSkill(
id="web-search",
name="search_web",
description="Search the web for current information",
tags=["search", "web", "information"]
),
AgentSkill(
id="data-analysis",
name="analyze_data",
description="Analyze datasets and provide insights",
tags=["analysis", "data", "insights"]
)
]
config = A2AServingConfig(
port=8080,
skills=custom_skills
)
```
### More Examples
Check out our cookbook example for building and serving an agent via A2A:
👉 [Serve an Agent with A2A (Jupyter Notebook)](./cookbook/serve_a2a.ipynb)
### Accessing an A2A agent using tools
As described in the [tools section](./agents/tools.md#a2a-tools), an agent can request actions from other agents by using the `a2a_tool` or `a2a_tool_async` function. It retrieves the agent card, and builds another function that relays the request via the A2A protocol and unpacks the result.
## Serving via MCP
### Example
In a similar way to [the A2A example](#example), we are going to define two agents served over MCP:
=== "Google Expert"
```python
# google_expert.py
import asyncio
from any_agent import AgentConfig, AnyAgent
from any_agent.serving import MCPServingConfig
from any_agent.tools import search_web
async def main():
agent = await AnyAgent.create_async(
"google",
AgentConfig(
name="google_expert",
model_id="mistral/mistral-small-latest",
description="An agent that can answer questions specifically and only about the Google Agents Development Kit (ADK). Reject questions about anything else.",
tools=[search_web]
)
)
server_handle = await agent.serve_async(MCPServingConfig(port=5001, endpoint="/google"))
await server_handle.task
asyncio.run(main())
```
=== "OpenAI Expert"
```python
# openai_expert.py
import asyncio
from any_agent import AgentConfig, AnyAgent
from any_agent.serving import MCPServingConfig
from any_agent.tools import search_web
async def main():
agent = await AnyAgent.create_async(
"openai",
AgentConfig(
name="openai_expert",
model_id="mistral/mistral-small-latest",
instructions="You can provide information about the OpenAI Agents SDK but nothing else (specially, nothing about the Google SDK).",
description="An agent that can answer questions specifically about the OpenAI Agents SDK.",
tools=[search_web]
)
)
server_handle = await agent.serve_async(MCPServingConfig(port=5002, endpoint="/openai"))
await server_handle.task
asyncio.run(main())
```
We can then run each of the scripts in a separate terminal and leave them running in the background.
Then, we run another python script containing the main agent that will contact one of the other two via MCP:
```python
# Main agent
from uuid import uuid4
import asyncio
from any_agent.config import MCPSse
from any_agent import AgentConfig, AgentFramework, AnyAgent
async def main():
prompt = "What do you know about the Google ADK?"
google_server_url = f"http://localhost:5001/google/sse"
openai_server_url = f"http://localhost:5002/openai/sse"
main_agent_cfg = AgentConfig(
instructions="Use the available tools to obtain additional information to answer the query.",
description="The orchestrator that can use other agents via tools using the MCP protocol.",
tools=[
MCPSse(url=google_server_url, client_session_timeout_seconds=300),
MCPSse(url=openai_server_url, client_session_timeout_seconds=300),
],
model_id="mistral/mistral-small-latest",
)
main_agent = await AnyAgent.create_async(
agent_framework=AgentFramework.OPENAI,
agent_config=main_agent_cfg,
)
agent_trace = await main_agent.run_async(prompt)
print(agent_trace)
if __name__ == "__main__":
asyncio.run(main())
```
---
## api/agent.md
## Agent
::: any_agent.AnyAgent
::: any_agent.AgentRunError
---
## api/callbacks.md
# Callbacks
::: any_agent.callbacks.base.Callback
::: any_agent.callbacks.context.Context
::: any_agent.callbacks.span_cost.AddCostInfo
::: any_agent.callbacks.span_print.ConsolePrintSpan
::: any_agent.callbacks.get_default_callbacks
---
## api/config.md
# Config
::: any_agent.config.AgentConfig
::: any_agent.config.MCPStdio
::: any_agent.config.MCPSse
::: any_agent.serving.A2AServingConfig
::: any_agent.serving.MCPServingConfig
::: any_agent.config.AgentFramework
---
## api/evaluation.md
# Evaluation
::: any_agent.evaluation.LlmJudge
::: any_agent.evaluation.AgentJudge
---
## api/logging.md
# Logging with `any-agent`
`any-agent` comes with a logger powered by [Rich](https://github.com/Textualize/rich)
## Quick Start
By default, logging is set up for you. But if you want to customize it, you can call:
```python
from any_agent.logging import setup_logger
setup_logger()
```
## Customizing the Logger
View the docstring in [`setup_logger`][any_agent.logging.setup_logger] for a description of the arguments available .
### Example: Set Log Level to DEBUG
```python
from any_agent.logging import setup_logger
import logging
setup_logger(level=logging.DEBUG)
```
### Example: Custom Log Format
```python
setup_logger(log_format="%(asctime)s - %(levelname)s - %(message)s")
```
### Example: Propagate Logs
```python
setup_logger(propagate=True)
```
::: any_agent.logging.setup_logger
---
## api/serving.md
# Serving
::: any_agent.serving.ServerHandle
---
## api/tools.md
# Tools
::: any_agent.tools
---
## api/tracing.md
# Tracing
::: any_agent.tracing.agent_trace.AgentTrace
::: any_agent.tracing.agent_trace.AgentSpan
::: any_agent.tracing.agent_trace.CostInfo
::: any_agent.tracing.agent_trace.TokenInfo
::: any_agent.tracing.attributes.GenAI
---