any-agent

When implementing custom callbacks by subclassing Callback, you must adhere to the following contract:

1. Receive Context: Callbacks receive a context object of type Context.
2. Shared State: You can store and persist data across different callback executions using context.shared (a dictionary-like object).
3. Return Context: Every callback method must return the context object to allow the execution chain to continue.
4. Modify Spans: You can access and modify the current trace span via context.current_span to inspect or scrub attributes (e.g., using span.set_attribute).

from any_agent.callbacks import Callback, Context

class MyCustomCallback(Callback):
    def before_tool_execution(self, context: Context, *args, **kwargs) -> Context:
        # 1. Use context.shared for state
        context.shared["my_key"] = 123
        
        # 2. MUST return context
        return context

An AgentTrace object contains the execution history of an agent. You can access metadata like duration, cost, and tokens, as well as iterate through spans to inspect specific steps.

To verify if a specific tool was used, iterate through the spans and check if span.is_tool_execution() is true, then inspect the span.attributes using the GenAI.TOOL_NAME key.

Key properties available on AgentTrace:

• duration: Total execution time.
• cost: Total monetary cost of the run.
• tokens: Total token count (trace.tokens.total_tokens).
• final_output: The agent's final response.
• spans: A list of execution steps/spans.

from any_agent.tracing.agent_trace import AgentTrace
from any_agent.tracing.attributes import GenAI

# Example: Checking if a tool was used
def check_tool_usage(trace: AgentTrace, required_tool: str) -> bool:
    return any(
        span.attributes[GenAI.TOOL_NAME] == required_tool
        for span in trace.spans
        if span.is_tool_execution()
    )

To define an agent system, use the AnyAgent.create method with an AgentConfig object. This allows you to specify the framework (e.g., tinyagent), the model, instructions, and a list of tools.

Before running, ensure you have set the necessary environment variables for your model provider (e.g., MISTRAL_API_KEY or OPENAI_API_KEY).

Supported frameworks include: Agno, Google ADK, LangChain, LlamaIndex, OpenAI Agents SDK, smolagents, and TinyAgent.

from any_agent import AgentConfig, AnyAgent
from any_agent.tools import search_web, visit_webpage

# Ensure environment variables like MISTRAL_API_KEY are set

agent = AnyAgent.create(
    "tinyagent",  # Framework name
    AgentConfig(
        model_id="mistral:mistral-small-latest",
        instructions="Use the tools to find an answer",
        tools=[search_web, visit_webpage]
    )
)

agent_trace = agent.run("Which Agent Framework is the best??")
print(agent_trace)

You can expose an agent to other applications using the A2A protocol by calling await agent.serve_async(). This method returns a handle that provides access to the server's port and a shutdown() method.

To configure the server, pass an A2AServingConfig object to serve_async. Setting port=0 allows the system to assign an available ephemeral port.

Example setup with an MCP tool:

from any_agent import AgentConfig, AnyAgent
from any_agent.config import MCPStdio
from any_agent.serving import A2AServingConfig

# Define a tool using MCP (Model Context Protocol)
time_tool = MCPStdio(
    command="python",
    args=["-u", "-m", "mcp_server_time", "--local-timezone", "America/New_York"],
    tools=["get_current_time"],
    client_session_timeout_seconds=30,
)

# Create the agent
agent = await AnyAgent.create_async(
    "my-agent",
    AgentConfig(
        model_id="mistral:mistral-small-latest",
        description="An agent that tells time",
        tools=[time_tool],
    ),
)

# Start the A2A server
time_handle = await agent.serve_async(A2AServingConfig(port=0))
server_port = time_handle.port

time_handle = await time.serve_async(A2AServingConfig(port=0))
server_port = time_handle.port