Tracing Agno Agents: Observability for Python Multi-Agent Pipelines

What Agno is

Agno (formerly phidata) is a Python-native framework for building multi-agent systems. Its core abstractions are Agent (a single LLM-powered agent with tools and instructions) and Team (a coordinator that routes tasks across multiple member agents). Agno agents are lightweight by design: you give an agent a model, a list of tools, and instructions — then call agent.run().

That simplicity introduces a few observability blind spots:

• Tool call failures are silent: When a tool raises an exception inside an agent run, Agno may surface it as a plain text response rather than an error. Without spans, you can’t tell whether the agent used the tool, failed to call it, or got a bad result back.
• Team routing is a black box: A Team’s coordinator decides which member agent handles a task. If the wrong agent is selected, or if a member agent produces a poor output that the coordinator passes through, there’s no built-in record of the routing decision.
• Multi-step runs accumulate latency: A single agent.run() call may trigger multiple LLM calls and tool invocations. Total latency hides which step is the bottleneck.

Tracing a basic Agno agent run

Install the SDK and wrap your agent calls with Nexus traces:

pip install agno nexus-sdk

import os
import time
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
from nexus_sdk import NexusClient

nexus = NexusClient(api_key=os.environ["NEXUS_API_KEY"])

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[DuckDuckGoTools()],
    instructions="You are a concise research assistant. Summarize findings in 2-3 sentences.",
    markdown=False,
)

def run_with_tracing(query: str, user_id: str) -> str:
    trace = nexus.start_trace({
        "agent_id": "agno-research-agent",
        "name": f"agno: {query[:60]}",
        "status": "running",
        "started_at": nexus.now(),
        "metadata": {
            "user_id": user_id,
            "query": query[:200],
            "environment": os.environ.get("APP_ENV", "dev"),
        },
    })
    trace_id = trace["trace_id"]

    t0 = time.time()
    try:
        response = agent.run(query, stream=False)
        elapsed_ms = int((time.time() - t0) * 1000)

        nexus.end_trace(trace_id, {
            "status": "success",
            "latency_ms": elapsed_ms,
            "metadata": {
                "output_length": len(response.content) if response.content else 0,
            },
        })
        return response.content or ""

    except Exception as e:
        nexus.end_trace(trace_id, {
            "status": "error",
            "latency_ms": int((time.time() - t0) * 1000),
            "error": str(e),
        })
        raise

Each call to run_with_tracing() creates one trace in Nexus. The trace captures the query, user ID, total latency, and success or error status.

Tracing tool calls

Agno tools are Python callables passed to the tools list. The cleanest way to trace them is to wrap each tool function with a span before passing it to the agent:

import os
import time
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from nexus_sdk import NexusClient

nexus = NexusClient(api_key=os.environ["NEXUS_API_KEY"])

def make_traced_search(trace_id: str):
    """Returns a search function that emits a Nexus span on each call."""
    def search_web(query: str) -> str:
        """Search the web for current information about a topic."""
        t0 = time.time()
        try:
            from duckduckgo_search import DDGS
            with DDGS() as ddgs:
                results = list(ddgs.text(query, max_results=5))
            output = "\n".join(r["body"] for r in results)
            nexus.add_span(trace_id, {
                "name": "tool:search_web",
                "status": "success",
                "latency_ms": int((time.time() - t0) * 1000),
                "metadata": {
                    "query": query[:120],
                    "result_count": len(results),
                },
            })
            return output
        except Exception as e:
            nexus.add_span(trace_id, {
                "name": "tool:search_web",
                "status": "error",
                "latency_ms": int((time.time() - t0) * 1000),
                "error": str(e),
            })
            raise
    return search_web

def run_agent_with_tool_tracing(query: str, user_id: str) -> str:
    trace = nexus.start_trace({
        "agent_id": "agno-research-agent",
        "name": f"agno: {query[:60]}",
        "status": "running",
        "started_at": nexus.now(),
        "metadata": {"user_id": user_id, "query": query[:200]},
    })
    trace_id = trace["trace_id"]

    # Build agent with traced tool for this request
    agent = Agent(
        model=OpenAIChat(id="gpt-4o"),
        tools=[make_traced_search(trace_id)],
        instructions="You are a concise research assistant.",
        markdown=False,
    )

    t0 = time.time()
    try:
        response = agent.run(query, stream=False)
        elapsed_ms = int((time.time() - t0) * 1000)
        nexus.end_trace(trace_id, {
            "status": "success",
            "latency_ms": elapsed_ms,
        })
        return response.content or ""
    except Exception as e:
        nexus.end_trace(trace_id, {
            "status": "error",
            "latency_ms": int((time.time() - t0) * 1000),
            "error": str(e),
        })
        raise

Building the agent per request (with the traced tool injected) is the reliable pattern — it ties each tool call span back to its parent trace without needing global state.

Tracing Agno Team hierarchies

Agno’s Team class lets you compose multiple agents under a coordinator. The coordinator decides which member agent handles each subtask. To trace this hierarchy, emit one span per member agent invocation alongside the parent trace:

import os
import time
from agno.agent import Agent
from agno.team.team import Team
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.yfinance import YFinanceTools
from nexus_sdk import NexusClient

nexus = NexusClient(api_key=os.environ["NEXUS_API_KEY"])

def run_team_with_tracing(task: str, user_id: str) -> str:
    trace = nexus.start_trace({
        "agent_id": "agno-research-team",
        "name": f"team: {task[:60]}",
        "status": "running",
        "started_at": nexus.now(),
        "metadata": {
            "user_id": user_id,
            "task": task[:200],
            "team_size": 2,
        },
    })
    trace_id = trace["trace_id"]

    def make_web_agent():
        def search(query: str) -> str:
            """Search the web for recent news and information."""
            t0 = time.time()
            try:
                from duckduckgo_search import DDGS
                with DDGS() as ddgs:
                    results = list(ddgs.text(query, max_results=5))
                output = "\n".join(r["body"] for r in results)
                nexus.add_span(trace_id, {
                    "name": "agent:WebSearchAgent:tool:search",
                    "status": "success",
                    "latency_ms": int((time.time() - t0) * 1000),
                    "metadata": {"query": query[:120], "result_count": len(results)},
                })
                return output
            except Exception as e:
                nexus.add_span(trace_id, {
                    "name": "agent:WebSearchAgent:tool:search",
                    "status": "error",
                    "latency_ms": int((time.time() - t0) * 1000),
                    "error": str(e),
                })
                raise

        return Agent(
            name="WebSearchAgent",
            model=OpenAIChat(id="gpt-4o"),
            tools=[search],
            instructions="Search the web for relevant, up-to-date information.",
            markdown=False,
        )

    def make_analyst_agent():
        return Agent(
            name="AnalystAgent",
            model=OpenAIChat(id="gpt-4o"),
            tools=[YFinanceTools(stock_price=True, company_news=True)],
            instructions="Analyze data and provide structured insights.",
            markdown=False,
        )

    team = Team(
        name="ResearchTeam",
        agents=[make_web_agent(), make_analyst_agent()],
        model=OpenAIChat(id="gpt-4o"),
        instructions="Coordinate agents to research and analyze. Delegate to the right agent for each subtask.",
        markdown=False,
    )

    t0 = time.time()
    t_coord = time.time()
    try:
        response = team.run(task, stream=False)
        elapsed_ms = int((time.time() - t0) * 1000)
        coord_ms = int((time.time() - t_coord) * 1000)

        nexus.add_span(trace_id, {
            "name": "coordinator:route_and_aggregate",
            "status": "success",
            "latency_ms": coord_ms,
            "metadata": {
                "output_length": len(response.content) if response.content else 0,
            },
        })
        nexus.end_trace(trace_id, {
            "status": "success",
            "latency_ms": elapsed_ms,
        })
        return response.content or ""

    except Exception as e:
        nexus.end_trace(trace_id, {
            "status": "error",
            "latency_ms": int((time.time() - t0) * 1000),
            "error": str(e),
        })
        raise

With this pattern, each trace shows the full team run in the top-level span plus individual tool call spans named agent:WebSearchAgent:tool:search. That naming convention makes it easy to filter by agent or tool name in the Nexus UI.

Async agents

Agno supports async execution via agent.arun(). The tracing pattern is the same — just use await:

import asyncio
import os
import time
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from nexus_sdk import NexusClient

nexus = NexusClient(api_key=os.environ["NEXUS_API_KEY"])

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    instructions="You are a concise assistant.",
    markdown=False,
)

async def run_async_with_tracing(query: str, user_id: str) -> str:
    trace = nexus.start_trace({
        "agent_id": "agno-async-agent",
        "name": f"agno-async: {query[:60]}",
        "status": "running",
        "started_at": nexus.now(),
        "metadata": {"user_id": user_id, "query": query[:200]},
    })
    trace_id = trace["trace_id"]

    t0 = time.time()
    try:
        response = await agent.arun(query, stream=False)
        elapsed_ms = int((time.time() - t0) * 1000)
        nexus.end_trace(trace_id, {
            "status": "success",
            "latency_ms": elapsed_ms,
            "metadata": {
                "output_length": len(response.content) if response.content else 0,
            },
        })
        return response.content or ""
    except Exception as e:
        nexus.end_trace(trace_id, {
            "status": "error",
            "latency_ms": int((time.time() - t0) * 1000),
            "error": str(e),
        })
        raise

What to watch for in production

Once traces are flowing from your Agno agents, three failure patterns appear most often:

• Tool call loops: An agent may call the same tool multiple times within a single run if the initial result doesn’t satisfy the LLM. Look for traces with 3+ spans to the same tool name — that’s a signal the agent is stuck in a search loop rather than synthesizing what it found.
• Team routing misses: A coordinator that routes every task to the same member agent is effectively ignoring your team setup. Track which agent name appears in spans per trace and alert when only one agent fires for multi-step tasks.
• Empty responses: Agno agents occasionally return an empty or near-empty content string when the model refuses, hits a content filter, or times out. Tracking output_length == 0 in trace metadata catches silent failures that don’t raise exceptions.

Next steps

Agno is gaining adoption quickly as a lightweight alternative to heavier agent frameworks — its Team primitive in particular makes it easy to compose specialist agents without complex orchestration logic. The tracing approach above works with any Agno model provider (OpenAI, Anthropic, Groq, Gemini) since it wraps the run boundary rather than patching internal SDK calls. Sign up for a free Nexus account to start capturing traces from your Agno agents today.