[opentelemetry-instrumentation-genai-agno] Adds streaming instrumentation support for Agno Agent, Team, and Workflow executions

[DylanRussell]

Description

Adds streaming instrumentation support for Agno Agent, Team, and Workflow executions across synchronous and asynchronous invocations

Type of change

• New feature (non-breaking change which adds functionality)

How has this been tested?

Unit tests

Checklist

• Followed the style guidelines of this project
• Changelog updated if the change requires an entry
• Unit tests added
• Documentation updated

[Copilot]

🟡 Changes recommended

format_content()’s JSON default serializer doesn’t handle nested Pydantic v1 models (via .dict()), which can break structured serialization in the oldest dependency environment and fail the newly added nested-structured tests.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Adds streaming support to the opentelemetry-instrumentation-genai-agno package so Agno Agent/Team/Workflow streaming executions are instrumented correctly (spans remain open until the stream is drained/closed), with additional structured output handling.

Changes:

• Introduces stream wrapper implementations for sync/async Agno streaming that finalize telemetry on stream end/error.
• Updates Agno patching logic to return instrumented stream proxies for streaming executions and to support structured (e.g., Pydantic/dataclass) content serialization.
• Adds unit tests covering streaming spans, mid-stream errors, caller-aborted streams, and structured output serialization.

File summaries

File	Description
instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_stream.py	New unit tests for sync/async streaming spans + error/abort paths and structured chunk output.
instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py	Adds focused tests for format_content and structured output handling helpers.
instrumentation/opentelemetry-instrumentation-genai-agno/tests/conftest.py	Adds a fixture enabling SPAN_ONLY content capture for streaming tests.
instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/utils.py	Adds format_content() and JSON serialization helpers for structured values.
instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/stream.py	New sync/async stream wrapper implementations built on util-genai stream helpers.
instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py	Updates patches to wrap streaming returns and to use structured content formatting.
instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/590.added	Changelog fragment for the new streaming support feature.

Review details

• Files reviewed: 7/7 changed files
• Comments generated: 1
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[opentelemetry-pr-dashboard]

Pull request dashboard status

Waiting on the author · refreshed 2026-09-04 22:20 UTC

Resolve merge conflicts.

Respond to 3 review items (e.g. link a commit, explain why not, ask a follow-up):

• Inline threads: 1, 2, 3

Status above doesn't look right?

• Just replied or pushed? Anything around or after the refresh time above may not be picked up yet — give it a few minutes.
• Should this be with reviewers? Comment /dashboard route:reviewers to route it to them.
• Anything wrong — including the routing? Report it with what you expected; it helps us improve the dashboard.

[lmolkova]

Starting the invocation synchronously attaches the span before the coroutine is awaited. When concurrent calls are made (e.g. asyncio.gather(agent.arun(...), agent.arun(...))), the spans nest and detaching raises ValueError: was created in a different Context.

For Awaitable returns, start and stop the invocation inside _await_result instead.

def test_agent_arun_concurrent(instrument_agno, span_exporter):
    agent = Agent(name="test-agent", model=MockModel(id="mock-model"))

    async def _run():
        coro1 = agent.arun("input 1")
        coro2 = agent.arun("input 2")
        await asyncio.gather(coro1, coro2)

    asyncio.run(_run())

    spans = span_exporter.get_finished_spans()
    assert len(spans) == 2
    assert spans[0].parent is None
    assert spans[1].parent is None

[lmolkova]

"completed" in event_name.lower() matches intermediate events like ToolCallCompleted and ReasoningCompleted. Since they carry content, _self_completed_content is set to tool or reasoning results and the final output is lost.

Match terminal event names explicitly (e.g. event_name in ("RunCompleted", "TeamRunCompleted")). The same applies to _WorkflowStreamMixin (WorkflowCompleted).

def test_agent_stream_tool_completed_event_preserves_final_content(
    instrument_agno_content_capture, span_exporter
):
    agent = Agent(name="test-agent", model=MockModel(id="mock-model"))

    def fake_stream(*args, **kwargs):
        yield ToolCallCompletedEvent(content="tool output")
        yield RunContentEvent(content="final answer")
        yield RunCompletedEvent()

    with _patch_agent_stream(fake_stream):
        list(agent.run("hello", stream=True))

    span = span_exporter.get_finished_spans()[0]
    output_messages = span.attributes.get(GenAIAttributes.GEN_AI_OUTPUT_MESSAGES)
    assert "final answer" in output_messages
    assert "tool output" not in output_messages

[lmolkova]

Passing invocation=invocation to super().__init__ causes SyncStreamWrapper / AsyncStreamWrapper to record gen_ai.client.operation.time_to_first_chunk and time_per_output_chunk metrics. These metrics are only defined for client LLM calls (e.g. chat), not invoke_agent or invoke_workflow.

Call super().__init__(stream) without invocation (similar to qwen_agent).

def test_agent_stream_no_llm_client_metrics(
    instrument_agno, metric_reader
):
    agent = Agent(name="test-agent", model=MockModel(id="mock-model"))

    def fake_stream(*args, **kwargs):
        yield ModelResponse(content="chunk 1")
        yield ModelResponse(content="chunk 2")

    with patch("agno.models.base.Model.response_stream", side_effect=fake_stream):
        list(agent.run("hello", stream=True))

    metric_names = [
        m.name
        for rm in metric_reader.get_metrics_data().resource_metrics
        for sm in rm.scope_metrics
        for m in sm.metrics
    ]
    assert "gen_ai.client.operation.time_to_first_chunk" not in metric_names
    assert "gen_ai.client.operation.time_per_output_chunk" not in metric_names