Agno Integration
Agno (formerly phidata) is a Python framework for building agentic systems — single agents, multi-agent teams, and autonomous workflows. This guide shows how to add full distributed tracing with Nexus: per-run traces, tool call spans, and Team routing visibility.
Installation
pip install keylightdigital-nexus agno openaiGet your API key from Dashboard → API Keys and set it as an environment variable:
export NEXUS_API_KEY="nxs_your_key_here"Basic agent trace
Wrap each agent.run() call in a Nexus trace. Every run appears in the dashboard with its status, duration, and metadata:
import os
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(
name="research-agent",
model=OpenAIChat(id="gpt-4o"),
instructions=["You are a concise research assistant."],
)
def run_agent(task: str) -> str:
trace = nexus.start_trace({
"agent_id": "agno-research-agent",
"name": f"agent: {task[:60]}",
"status": "running",
"started_at": nexus.now(),
"metadata": {
"task": task[:200],
"model": "gpt-4o",
},
})
try:
result = agent.run(task)
nexus.end_trace(trace["trace_id"], {"status": "success"})
return result.content
except Exception as e:
nexus.end_trace(trace["trace_id"], {
"status": "error",
"metadata": {"error": str(e)},
})
raiseTool call spans
Agno tools are plain Python functions passed to the Agent. Wrap them in a Nexus span to capture inputs, outputs, and errors per tool invocation:
import os
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_tool(tool_fn, tool_name: str, trace_id: str):
"""Returns a version of tool_fn that records a Nexus span per call."""
def wrapper(*args, **kwargs):
span = nexus.start_span(trace_id, {
"name": f"tool:{tool_name}",
"type": "tool",
"metadata": {
"tool": tool_name,
"args": str(args)[:200],
"kwargs": str(kwargs)[:200],
},
})
try:
result = tool_fn(*args, **kwargs)
nexus.end_span(span["id"], {"output": str(result)[:500]})
return result
except Exception as e:
nexus.end_span(span["id"], {"error": str(e)})
raise
wrapper.__name__ = tool_fn.__name__
wrapper.__doc__ = tool_fn.__doc__
return wrapper
# Define your tools as plain functions
def web_search(query: str) -> str:
"""Search the web for up-to-date information."""
# ... your search implementation ...
return f"Results for: {query}"
def get_weather(location: str) -> str:
"""Get current weather for a location."""
# ... your weather API call ...
return f"Sunny, 22C in {location}"
def run_agent_with_tool_spans(task: str) -> str:
trace = nexus.start_trace({
"agent_id": "agno-tool-agent",
"name": f"agent: {task[:60]}",
"status": "running",
"started_at": nexus.now(),
"metadata": {"task": task[:200]},
})
try:
# Wrap tools with the current trace_id before each run
traced_tools = [
make_traced_tool(web_search, "web_search", trace["trace_id"]),
make_traced_tool(get_weather, "get_weather", trace["trace_id"]),
]
agent = Agent(
name="tool-agent",
model=OpenAIChat(id="gpt-4o"),
tools=traced_tools,
)
result = agent.run(task)
nexus.end_trace(trace["trace_id"], {"status": "success"})
return result.content
except Exception as e:
nexus.end_trace(trace["trace_id"], {
"status": "error",
"metadata": {"error": str(e)},
})
raiseTeam routing
An Agno Team routes tasks to member agents. Add a span per member run to capture which agent handled the task and what it returned:
import os
from agno.agent import Agent
from agno.team import Team
from agno.models.openai import OpenAIChat
from nexus_sdk import NexusClient
nexus = NexusClient(api_key=os.environ["NEXUS_API_KEY"])
# Define member agents
researcher = Agent(
name="researcher",
role="Research specialist — find facts and cite sources.",
model=OpenAIChat(id="gpt-4o"),
)
writer = Agent(
name="writer",
role="Content writer — turn facts into clear prose.",
model=OpenAIChat(id="gpt-4o"),
)
team = Team(
members=[researcher, writer],
model=OpenAIChat(id="gpt-4o"),
instructions=["Route tasks to the most appropriate member agent."],
)
def run_team_traced(task: str) -> str:
trace = nexus.start_trace({
"agent_id": "agno-team",
"name": f"team: {task[:60]}",
"status": "running",
"started_at": nexus.now(),
"metadata": {
"task": task[:200],
"members": ["researcher", "writer"],
},
})
try:
# Add a routing span to capture which member was selected
routing_span = nexus.start_span(trace["trace_id"], {
"name": "team:route",
"type": "tool",
"metadata": {"task": task[:200]},
})
result = team.run(task)
# Capture the responding member from the run result
responding_member = getattr(result, "member_name", "unknown")
nexus.end_span(routing_span["id"], {
"output": f"routed to: {responding_member}",
})
nexus.end_trace(trace["trace_id"], {
"status": "success",
"metadata": {"responding_member": responding_member},
})
return result.content
except Exception as e:
nexus.end_trace(trace["trace_id"], {
"status": "error",
"metadata": {"error": str(e)},
})
raiseThe team:route span in the Nexus dashboard waterfall shows exactly which member agent handled each task and the full output — critical for debugging misconfigured routing prompts.
Full example
A complete script combining agent tracing, tool spans, and Team routing in a single runnable file:
import os
from agno.agent import Agent
from agno.team import Team
from agno.models.openai import OpenAIChat
from nexus_sdk import NexusClient
nexus = NexusClient(api_key=os.environ["NEXUS_API_KEY"])
# ----- Tools -----
def make_traced_tool(tool_fn, tool_name: str, trace_id: str):
def wrapper(*args, **kwargs):
span = nexus.start_span(trace_id, {
"name": f"tool:{tool_name}",
"type": "tool",
"metadata": {"tool": tool_name, "args": str(args)[:200]},
})
try:
result = tool_fn(*args, **kwargs)
nexus.end_span(span["id"], {"output": str(result)[:500]})
return result
except Exception as e:
nexus.end_span(span["id"], {"error": str(e)})
raise
wrapper.__name__ = tool_fn.__name__
wrapper.__doc__ = tool_fn.__doc__
return wrapper
def web_search(query: str) -> str:
"""Search the web for information."""
return f"[search results for '{query}']"
def calculator(expression: str) -> str:
"""Evaluate a mathematical expression."""
try:
return str(eval(expression)) # noqa: S307
except Exception as e:
return f"error: {e}"
# ----- Agents -----
def build_agents(trace_id: str):
traced_search = make_traced_tool(web_search, "web_search", trace_id)
traced_calc = make_traced_tool(calculator, "calculator", trace_id)
researcher = Agent(
name="researcher",
role="Research specialist. Use web_search to find facts.",
model=OpenAIChat(id="gpt-4o"),
tools=[traced_search],
)
analyst = Agent(
name="analyst",
role="Data analyst. Use calculator for numeric tasks.",
model=OpenAIChat(id="gpt-4o"),
tools=[traced_calc],
)
return researcher, analyst
# ----- Runner -----
def run(task: str) -> str:
trace = nexus.start_trace({
"agent_id": "agno-demo-team",
"name": f"team: {task[:60]}",
"status": "running",
"started_at": nexus.now(),
"metadata": {"task": task[:200], "model": "gpt-4o"},
})
try:
researcher, analyst = build_agents(trace["trace_id"])
team = Team(
members=[researcher, analyst],
model=OpenAIChat(id="gpt-4o"),
instructions=["Route research tasks to researcher, math tasks to analyst."],
)
routing_span = nexus.start_span(trace["trace_id"], {
"name": "team:route",
"type": "tool",
"metadata": {"task": task[:200]},
})
result = team.run(task)
member = getattr(result, "member_name", "unknown")
nexus.end_span(routing_span["id"], {"output": f"routed to: {member}"})
nexus.end_trace(trace["trace_id"], {
"status": "success",
"metadata": {"responding_member": member},
})
return result.content
except Exception as e:
nexus.end_trace(trace["trace_id"], {
"status": "error",
"metadata": {"error": str(e)},
})
raise
if __name__ == "__main__":
print(run("What is the GDP of Germany?"))
print(run("What is 17 * 89 + 42?"))Debugging patterns
Team always routes to the same member
Check the team:route span in the trace waterfall. If the same agent handles every task, the team’s routing instructions are too vague. Add explicit role boundaries in each agent’s role field and the team’s instructions.
Tool calls not appearing in spans
The make_traced_tool wrapper must be applied before the Agent is constructed, with the correct trace_id in the closure. If you cache agents across requests, rebuild them per trace so each tool wrapper captures the right trace_id.
Traces stuck in “running” state
A trace stays running if nexus.end_trace() is never called. Always wrap agent.run() and team.run() in a try/except so the trace closes even when an exception propagates.
Ready to instrument your Agno agents?
Start for free — no credit card required. See traces in under 5 minutes.
Start free →