Skip to content

feat(agno): trace external agents configured via Agno [WIP] - #4018

Open
kausmeows wants to merge 3 commits into
Arize-ai:mainfrom
kausmeows:feat/agno-external-agents-tracing
Open

kausmeows wants to merge 3 commits into
Arize-ai:mainfrom
kausmeows:feat/agno-external-agents-tracing

Conversation

@kausmeows

@kausmeows kausmeows commented Oct 10, 2026 •

Copy link
Copy Markdown
Contributor

Summary

Extend AgnoInstrumentor to trace adapters built on BaseExternalAgent, including Claude Agent SDK, Codex, LangGraph, DSPy, and Antigravity. Previously, these runs were not covered by Agno instrumentation.

  • Capture agent spans with input/output, run/session/user identity, status, and available token/cost metrics.
  • Support sync, async, streaming, and background execution, preserving parent context across sync worker threads.
  • Create child tool spans from streaming lifecycle events and preserve nested tool relationships when Agno supplies parent_tool_call_id.
  • Retain non-streamed tool calls/results as output messages without inventing execution durations.
  • Respect tracing suppression, context attributes, and privacy masking.
  • Restore wrapped methods on uninstrumentation and remain compatible with Agno versions without external adapters.
  • Add documentation, a runnable example, and regression tests.

These spans work with Agno’s existing DatabaseSpanExporter; no separate collector or tracing database migration is required.

Example in AgentOS tracing platform with a run using ClaudeAgent from Agno-

image
from agno.agents.claude import ClaudeAgent
from agno.os import AgentOS
from agno.db.postgres import PostgresDb

db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
db = PostgresDb(db_url=db_url)

# ---------------------------------------------------------------------------
# Create the Claude Agent SDK agent
# ---------------------------------------------------------------------------
claude_agent = ClaudeAgent(
    name="Claude Assistant",
    description="A Claude-powered assistant served through AgentOS",
    model="claude-sonnet-4-6",
    allowed_tools=["Read", "Bash"],
    permission_mode="acceptEdits",
    max_turns=10,
    db=db,
)

# ---------------------------------------------------------------------------
# Setup AgentOS
# ---------------------------------------------------------------------------
agent_os = AgentOS(
    name="Claude Agent SDK Example",
    description="AgentOS serving a Claude Agent SDK agent",
    agents=[claude_agent],
    tracing=True,
    db=db
)
app = agent_os.get_app()

# ---------------------------------------------------------------------------
# Run
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    agent_os.serve(app="test:app", reload=True)

Validation

  • 230 instrumenter tests passed against local editable Agno.
  • Live Claude and Codex runs verified agent/tool spans, identity, outputs, and token metrics persisted in Agno’s SQLite tracing database.
  • A live Claude subagent run verified nested Bash/Read spans alongside a top-level Bash span.
  • Ruff and focused mypy checks passed.

Compatibility and scope

Nested tool spans require an Agno version that preserves ToolExecution.parent_tool_call_id. Without that field, external tool spans remain directly under the run.

This instruments the Agno adapter boundary and exposed tool events. It does not reconstruct individual model requests hidden inside harness subprocesses or rewrite existing traces.

@kausmeows
kausmeows requested a review from a team as a code owner October 10, 2026 19:36
@kausmeows kausmeows changed the title feat(agno): trace external agents configured via Agno feat(agno): trace external agents configured via Agno [WIP] Oct 10, 2026
ysolanky pushed a commit to agno-agi/agno that referenced this pull request Oct 11, 2026
## Summary

Claude SDK messages include `parent_tool_use_id`, but the adapter
previously discarded it when creating Agno tool events. This prevented
tracing integrations from distinguishing subagent tools from top-level
tools.

- Add optional `parent_tool_call_id` to `ToolExecution`, including
serialization and deserialization.
- Preserve Claude’s parent IDs in streaming tool events and non-streamed
tool records.
- Carry lineage from tool-start events into completion events when
completion messages omit it.
- Preserve the SDK’s tool-error flag in streaming completion events.
- Add regression tests and a cookbook that verifies nested tool spans
persisted in Agno’s tracing database.

With the corresponding OpenInference update, traces can represent:

```text
ClaudeAgent.run
├── Agent
│   ├── Bash
│   └── Read
└── Bash
```

tracing update on openinference-
Arize-ai/openinference#4018

to test locally use `uv pip install -e
../openinference/python/instrumentation/openinference-instrumentation-agno/`

Quick cookbook (not adding in the PR files since the openinference
update needs to be released first, so the cookbook cant go on main yet.-
```python
mport asyncio
from pathlib import Path
from tempfile import TemporaryDirectory
from uuid import uuid4

from claude_agent_sdk import AgentDefinition
from opentelemetry import trace

from agno.agents.claude import ClaudeAgent
from agno.db.sqlite import SqliteDb
from agno.run.agent import RunOutput
from agno.tracing import setup_tracing


async def main() -> None:
    db = SqliteDb(db_file="tmp/claude-nested-traces.db")
    setup_tracing(db=db)
    run_id = str(uuid4())
    with TemporaryDirectory(prefix="agno-trace-probe-") as directory:
        Path(directory, "probe.txt").write_text("nested_read_probe\n")
        agent = ClaudeAgent(
            name="Claude trace probe",
            model="claude-sonnet-4-6",
            db=db,
            cwd=directory,
            allowed_tools=["Agent", "Bash", "Read"],
            max_turns=8,
            max_budget_usd=1.0,
            options_kwargs={
                "agents": {
                    "trace-probe": AgentDefinition(
                        description="Run the tracing verification tools.",
                        prompt=(
                            "Run Bash with `printf nested_shell_probe` once, then Read probe.txt "
                            "in the working directory. Report both outputs. Do not modify files."
                        ),
                        tools=["Bash", "Read"],
                        model="haiku",
                    )
                }
            },
        )
        output = None
        async for event in agent.arun(
            "Use the trace-probe subagent to perform its verification. After it finishes, "
            "run Bash yourself with `printf main_shell_probe` once. Then report the three outputs.",
            stream=True,
            yield_run_output=True,
            run_id=run_id,
            session_id=run_id,
            user_id="tracing-demo",
        ):
            if isinstance(event, RunOutput):
                output = event

    trace.get_tracer_provider().force_flush()
    assert output is not None and output.status.value.lower() == "completed", (
        "Claude run did not complete"
    )
    record = db.get_trace(run_id=run_id)
    assert record is not None, "No trace persisted; check the instrumenter installation"
    spans = db.get_spans(trace_id=record.trace_id)
    root = next(span for span in spans if span.parent_span_id is None)
    tool_spans = {
        span.attributes["tool_call.id"]: span
        for span in spans
        if "tool_call.id" in span.attributes
    }
    nested = [
        span
        for span in tool_spans.values()
        if span.attributes.get("agno.tool.parent_call_id")
    ]
    assert nested, (
        "No nested tools recorded; check both Agno and the instrumenter have lineage support"
    )
    for span in nested:
        parent = tool_spans[span.attributes["agno.tool.parent_call_id"]]
        assert span.parent_span_id == parent.span_id
    assert any(
        span.name == "Bash" and span.parent_span_id == root.span_id
        for span in tool_spans.values()
    )
    assert {"Bash", "Read"}.issubset({span.name for span in nested})
    assert all(span.status_code == "OK" for span in spans)

    def show_tree(parent_id: str | None, depth: int = 0) -> None:
        for span in sorted(spans, key=lambda span: span.start_time):
            if span.parent_span_id == parent_id:
                print("  " * depth + span.name)
                show_tree(span.span_id, depth + 1)

    show_tree(None)
    print("Trace ID:", record.trace_id)
    print(
        "Verified nested tool spans and top-level sibling in tmp/claude-nested-traces.db"
    )


if __name__ == "__main__":
    asyncio.run(asyncio.wait_for(main(), timeout=120))

```

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

Status: No status

Development

Successfully merging this pull request may close these issues.

1 participant