From a82b5527432dc3cc51f243e5d0246491404bf201 Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Mon, 3 Aug 2026 17:05:25 +0000 Subject: [PATCH 01/14] Add instrumentation --- .../.changelog/340.added | 1 + .../README.rst | 12 + .../instrumentation/genai/agno/patch.py | 312 +++++++++++++++++- .../tests/test_agent.py | 171 ++++++++++ 4 files changed, 490 insertions(+), 6 deletions(-) create mode 100644 instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/340.added diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/340.added b/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/340.added new file mode 100644 index 000000000..74c8e3289 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/340.added @@ -0,0 +1 @@ +Add instrumentation for Agno Team, Workflow, Step, and Model response methods. diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst b/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst index 13c67100b..0b80b73d5 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst @@ -25,6 +25,18 @@ Usage # Instrument Agno AgnoInstrumentor().instrument() +Supported Operations +-------------------- + +The instrumentation automatically traces: + +* ``Agent.run`` and ``Agent.arun`` +* ``Team.run`` and ``Team.arun`` +* ``Workflow.run`` and ``Workflow.arun`` +* ``Step.execute`` and ``Step.aexecute`` +* ``Model.response`` and ``Model.aresponse`` +* ``FunctionCall.execute`` and ``FunctionCall.aexecute`` + Configuration ------------- diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py index 67b7bed5a..9669b961e 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py @@ -18,7 +18,9 @@ from opentelemetry.util.genai.handler import TelemetryHandler from opentelemetry.util.genai.invocation import ( AgentInvocation, + InferenceInvocation, ToolInvocation, + WorkflowInvocation, ) from opentelemetry.util.genai.types import ( InputMessage, @@ -30,12 +32,20 @@ _AGNO_MODULE = "agno.agent" _AGENT_CLASS = "Agent" +_AGNO_TEAM_MODULE = "agno.team" +_TEAM_CLASS = "Team" _AGNO_TOOLS_MODULE = "agno.tools.function" _FUNCTION_CALL_CLASS = "FunctionCall" +_AGNO_WORKFLOW_MODULE = "agno.workflow.workflow" +_WORKFLOW_CLASS = "Workflow" +_AGNO_STEP_MODULE = "agno.workflow.step" +_STEP_CLASS = "Step" +_AGNO_MODELS_MODULE = "agno.models.base" +_MODEL_CLASS = "Model" def patch_agent(handler: TelemetryHandler) -> None: - """Apply patches to agno.agent.Agent and agno.tools.function.FunctionCall class methods.""" + """Apply patches to Agno class methods.""" wrap_function_wrapper( _AGNO_MODULE, f"{_AGENT_CLASS}.run", @@ -46,6 +56,19 @@ def patch_agent(handler: TelemetryHandler) -> None: f"{_AGENT_CLASS}.arun", _agent_arun(handler), ) + try: + wrap_function_wrapper( + _AGNO_TEAM_MODULE, + f"{_TEAM_CLASS}.run", + _agent_run(handler), + ) + wrap_function_wrapper( + _AGNO_TEAM_MODULE, + f"{_TEAM_CLASS}.arun", + _agent_arun(handler), + ) + except (ImportError, AttributeError): + pass try: wrap_function_wrapper( _AGNO_TOOLS_MODULE, @@ -59,10 +82,49 @@ def patch_agent(handler: TelemetryHandler) -> None: ) except (ImportError, AttributeError): pass + try: + wrap_function_wrapper( + _AGNO_WORKFLOW_MODULE, + f"{_WORKFLOW_CLASS}.run", + _workflow_run(handler), + ) + wrap_function_wrapper( + _AGNO_WORKFLOW_MODULE, + f"{_WORKFLOW_CLASS}.arun", + _workflow_arun(handler), + ) + except (ImportError, AttributeError): + pass + try: + wrap_function_wrapper( + _AGNO_STEP_MODULE, + f"{_STEP_CLASS}.execute", + _step_execute(handler), + ) + wrap_function_wrapper( + _AGNO_STEP_MODULE, + f"{_STEP_CLASS}.aexecute", + _step_aexecute(handler), + ) + except (ImportError, AttributeError): + pass + try: + wrap_function_wrapper( + _AGNO_MODELS_MODULE, + f"{_MODEL_CLASS}.response", + _model_response(handler), + ) + wrap_function_wrapper( + _AGNO_MODELS_MODULE, + f"{_MODEL_CLASS}.aresponse", + _model_aresponse(handler), + ) + except (ImportError, AttributeError): + pass def unpatch_agent() -> None: - """Remove patches from agno.agent.Agent class methods.""" + """Remove patches from Agno class methods.""" try: import agno.agent # pylint: disable=import-outside-toplevel # noqa: PLC0415 @@ -70,6 +132,13 @@ def unpatch_agent() -> None: unwrap(agno.agent.Agent, "arun") except (ImportError, AttributeError): pass + try: + import agno.team # pylint: disable=import-outside-toplevel # noqa: PLC0415 + + unwrap(agno.team.Team, "run") + unwrap(agno.team.Team, "arun") + except (ImportError, AttributeError): + pass try: import agno.tools.function # pylint: disable=import-outside-toplevel # noqa: PLC0415 @@ -77,6 +146,28 @@ def unpatch_agent() -> None: unwrap(agno.tools.function.FunctionCall, "aexecute") except (ImportError, AttributeError): pass + # Workflow / step depend on optional packages (like fastapi), may fail to import. + try: + import agno.workflow.workflow # pylint: disable=import-outside-toplevel # noqa: PLC0415 + + unwrap(agno.workflow.workflow.Workflow, "run") + unwrap(agno.workflow.workflow.Workflow, "arun") + except (ImportError, AttributeError): + pass + try: + import agno.workflow.step # pylint: disable=import-outside-toplevel # noqa: PLC0415 + + unwrap(agno.workflow.step.Step, "execute") + unwrap(agno.workflow.step.Step, "aexecute") + except (ImportError, AttributeError): + pass + try: + import agno.models.base # pylint: disable=import-outside-toplevel # noqa: PLC0415 + + unwrap(agno.models.base.Model, "response") + unwrap(agno.models.base.Model, "aresponse") + except (ImportError, AttributeError): + pass def _extract_input_content(input_val: Any) -> str: @@ -135,14 +226,16 @@ def _set_tool_invocation_output( def _set_invocation_input( - invocation: AgentInvocation, + invocation: Any, instance: Any, args: tuple[Any, ...], kwargs: dict[str, Any], capture_content: bool, ) -> None: - agent_id = getattr(instance, "agent_id", None) - if agent_id: + agent_id = getattr(instance, "agent_id", None) or getattr( + instance, "id", None + ) + if agent_id and hasattr(invocation, "agent_id"): invocation.agent_id = str(agent_id) if capture_content and (args or "input" in kwargs): @@ -179,7 +272,8 @@ def _start_agent_invocation( kwargs: dict[str, Any], capture_content: bool, ) -> AgentInvocation: - agent_name = getattr(instance, "name", None) or "Agent" + default_name = "Team" if instance.__class__.__name__ == "Team" else "Agent" + agent_name = getattr(instance, "name", None) or default_name invocation = handler.invoke_local_agent(agent_name=agent_name) _set_invocation_input(invocation, instance, args, kwargs, capture_content) invocation.tool_definitions = prepare_tool_definitions( @@ -290,3 +384,209 @@ async def traced_method( return result return cast(Callable[..., Any], traced_method) + + +def _start_workflow_invocation( + handler: TelemetryHandler, + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + capture_content: bool, + *, + default_name: str = "Workflow", +) -> WorkflowInvocation: + workflow_name = getattr(instance, "name", None) or default_name + invocation = handler.workflow(name=workflow_name) + _set_invocation_input(invocation, instance, args, kwargs, capture_content) + return invocation + + +def _start_model_invocation( + handler: TelemetryHandler, + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + capture_content: bool, +) -> InferenceInvocation: + provider_name = ( + getattr(instance, "provider", None) + or instance.__class__.__name__ + or "agno" + ) + request_model = getattr(instance, "id", None) + invocation = handler.inference( + provider=str(provider_name), + request_model=str(request_model) if request_model else None, + ) + if capture_content and (args or "messages" in kwargs): + messages_val: Any = args[0] if args else kwargs.get("messages") + if messages_val and isinstance(messages_val, (list, tuple)): + input_msgs: list[InputMessage] = [] + for msg in cast("list[Any]", messages_val): + role = str(getattr(msg, "role", "user")) + content: Any = getattr(msg, "content", "") + content_str = str(content) if content is not None else "" + input_msgs.append( + InputMessage(role=role, parts=[Text(content=content_str)]) + ) + if input_msgs: + invocation.input_messages = input_msgs + return invocation + + +def _set_model_invocation_output( + invocation: Any, + result: Any, + capture_content: bool, +) -> None: + if capture_content and result is not None: + content = getattr(result, "content", None) + if content is not None: + output_str = ( + str(content) if not isinstance(content, str) else content + ) + invocation.output_messages = [ + OutputMessage( + role="assistant", + parts=[Text(content=output_str)], + finish_reason=str( + getattr(result, "finish_reason", "stop") + ), + ) + ] + + +def _workflow_run( + handler: TelemetryHandler, +) -> Callable[..., Any]: + capture_content = handler.should_capture_content() + + def traced_method( + wrapped: Callable[..., Any], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + with _start_workflow_invocation( + handler, instance, args, kwargs, capture_content + ) as invocation: + result = wrapped(*args, **kwargs) + _set_invocation_output(invocation, result, capture_content) + return result + + return traced_method + + +def _workflow_arun( + handler: TelemetryHandler, +) -> Callable[..., Any]: + capture_content = handler.should_capture_content() + + async def traced_method( + wrapped: Callable[..., Awaitable[Any]], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + with _start_workflow_invocation( + handler, instance, args, kwargs, capture_content + ) as invocation: + result = await wrapped(*args, **kwargs) + _set_invocation_output(invocation, result, capture_content) + return result + + return cast(Callable[..., Any], traced_method) + + +def _step_execute( + handler: TelemetryHandler, +) -> Callable[..., Any]: + capture_content = handler.should_capture_content() + + def traced_method( + wrapped: Callable[..., Any], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + with _start_workflow_invocation( + handler, + instance, + args, + kwargs, + capture_content, + default_name="Step", + ) as invocation: + result = wrapped(*args, **kwargs) + _set_invocation_output(invocation, result, capture_content) + return result + + return traced_method + + +def _step_aexecute( + handler: TelemetryHandler, +) -> Callable[..., Any]: + capture_content = handler.should_capture_content() + + async def traced_method( + wrapped: Callable[..., Awaitable[Any]], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + with _start_workflow_invocation( + handler, + instance, + args, + kwargs, + capture_content, + default_name="Step", + ) as invocation: + result = await wrapped(*args, **kwargs) + _set_invocation_output(invocation, result, capture_content) + return result + + return cast(Callable[..., Any], traced_method) + + +def _model_response( + handler: TelemetryHandler, +) -> Callable[..., Any]: + capture_content = handler.should_capture_content() + + def traced_method( + wrapped: Callable[..., Any], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + with _start_model_invocation( + handler, instance, args, kwargs, capture_content + ) as invocation: + result = wrapped(*args, **kwargs) + _set_model_invocation_output(invocation, result, capture_content) + return result + + return traced_method + + +def _model_aresponse( + handler: TelemetryHandler, +) -> Callable[..., Any]: + capture_content = handler.should_capture_content() + + async def traced_method( + wrapped: Callable[..., Awaitable[Any]], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + with _start_model_invocation( + handler, instance, args, kwargs, capture_content + ) as invocation: + result = await wrapped(*args, **kwargs) + _set_model_invocation_output(invocation, result, capture_content) + return result + + return cast(Callable[..., Any], traced_method) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py index 7b418cd08..49c9ea540 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py @@ -8,8 +8,11 @@ import asyncio from unittest.mock import patch +import pytest from agno.agent import Agent +from agno.models.message import Message from agno.models.response import ModelResponse +from agno.team import Team from agno.tools.function import Function, FunctionCall from tests.mock_model import MockModel @@ -150,3 +153,171 @@ async def _run_async() -> None: assert ( span.attributes.get(GenAIAttributes.GEN_AI_TOOL_CALL_ID) == "call-456" ) + + +def test_team_run_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Team.run emits an invoke_agent span.""" + member = Agent(name="member-agent", model=MockModel(id="mock-model")) + team = Team( + name="test-sync-team", + members=[member], + model=MockModel(id="mock-model"), + ) + mock_output = ModelResponse(content="Hello back from team!") + + with patch("agno.models.base.Model.response", return_value=mock_output): + res = team.run("hello team world") + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "invoke_agent test-sync-team" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "invoke_agent" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_AGENT_NAME) + == "test-sync-team" + ) + + +def test_team_arun_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Team.arun emits an invoke_agent span.""" + member = Agent(name="member-agent", model=MockModel(id="mock-model")) + team = Team( + name="test-async-team", + members=[member], + model=MockModel(id="mock-model"), + ) + mock_output = ModelResponse(content="Async hello back from team!") + + async def _run_async() -> None: + with patch( + "agno.models.base.Model.aresponse", return_value=mock_output + ): + res = await team.arun("hello async team world") + assert res is not None + + asyncio.run(_run_async()) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "invoke_agent test-async-team" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "invoke_agent" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_AGENT_NAME) + == "test-async-team" + ) + + +def test_model_response_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Model.response emits a chat span.""" + model = MockModel(id="test-mock-model") + mock_output = ModelResponse(content="Hello from model!", role="assistant") + + with patch.object(MockModel, "invoke", return_value=mock_output): + res = model.response( + messages=[Message(role="user", content="hello model")] + ) + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "chat test-mock-model" + assert span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) == "chat" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_REQUEST_MODEL) + == "test-mock-model" + ) + + +def test_model_aresponse_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Model.aresponse emits a chat span.""" + model = MockModel(id="test-mock-model-async") + mock_output = ModelResponse( + content="Async hello from model!", role="assistant" + ) + + async def _run_async() -> None: + with patch.object(MockModel, "ainvoke", return_value=mock_output): + res = await model.aresponse( + messages=[Message(role="user", content="hello async model")] + ) + assert res is not None + + asyncio.run(_run_async()) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "chat test-mock-model-async" + assert span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) == "chat" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_REQUEST_MODEL) + == "test-mock-model-async" + ) + + +def test_workflow_run_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Workflow.run emits an invoke_workflow span.""" + pytest.importorskip("fastapi") + from agno.workflow.workflow import Workflow # noqa: PLC0415 + + workflow = Workflow(name="test-workflow") + with patch.object(Workflow, "run", wraps=workflow.run): + try: + workflow.run("test input") + except Exception: + pass + + spans = span_exporter.get_finished_spans() + assert any( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "invoke_workflow" + for span in spans + ) + + +def test_step_execute_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Step.execute emits an invoke_workflow span.""" + pytest.importorskip("fastapi") + from agno.workflow.step import Step # noqa: PLC0415 + + step = Step(name="test-step") + with patch.object(Step, "execute", wraps=step.execute): + try: + step.execute("test input") + except Exception: + pass + + spans = span_exporter.get_finished_spans() + assert any( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "invoke_workflow" + for span in spans + ) From 4947b768f12329876f393292b622cac4cb7133b9 Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Mon, 3 Aug 2026 17:20:04 +0000 Subject: [PATCH 02/14] Fix PR id --- .../.changelog/{340.added => 350.added} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/{340.added => 350.added} (100%) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/340.added b/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/350.added similarity index 100% rename from instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/340.added rename to instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/350.added From 588e16612b7ba5ad15f0de3a6586fb10c2d55e2e Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Mon, 3 Aug 2026 17:50:42 +0000 Subject: [PATCH 03/14] Respond to comments --- .../instrumentation/genai/agno/patch.py | 8 +-- .../tests/test_agent.py | 54 +++++++++++++++++-- 2 files changed, 55 insertions(+), 7 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py index 9669b961e..11de78190 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py @@ -256,9 +256,9 @@ def _set_invocation_output( output_str = _extract_output_content(result) invocation.output_messages = [ OutputMessage( - role="assistant", + role=str(getattr(result, "role", "assistant")), parts=[Text(content=output_str)], - finish_reason="stop", + finish_reason=str(getattr(result, "finish_reason", "stop")), ) ] if hasattr(result, "session_id") and getattr(result, "session_id"): @@ -422,7 +422,7 @@ def _start_model_invocation( messages_val: Any = args[0] if args else kwargs.get("messages") if messages_val and isinstance(messages_val, (list, tuple)): input_msgs: list[InputMessage] = [] - for msg in cast("list[Any]", messages_val): + for msg in cast(list[Any], messages_val): role = str(getattr(msg, "role", "user")) content: Any = getattr(msg, "content", "") content_str = str(content) if content is not None else "" @@ -447,7 +447,7 @@ def _set_model_invocation_output( ) invocation.output_messages = [ OutputMessage( - role="assistant", + role=str(getattr(result, "role", "assistant")), parts=[Text(content=output_str)], finish_reason=str( getattr(result, "finish_reason", "stop") diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py index 49c9ea540..2df6c0ed4 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py @@ -282,7 +282,7 @@ def test_workflow_run_spans( span_exporter, ) -> None: """Test that Workflow.run emits an invoke_workflow span.""" - pytest.importorskip("fastapi") + pytest.importorskip("agno.workflow.workflow") from agno.workflow.workflow import Workflow # noqa: PLC0415 workflow = Workflow(name="test-workflow") @@ -300,15 +300,39 @@ def test_workflow_run_spans( ) +@pytest.mark.asyncio +async def test_workflow_arun_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Workflow.arun emits an invoke_workflow span.""" + pytest.importorskip("agno.workflow.workflow") + from agno.workflow.workflow import Workflow # noqa: PLC0415 + + workflow = Workflow(name="test-workflow-async") + with patch.object(Workflow, "arun", wraps=workflow.arun): + try: + await workflow.arun("test input") + except Exception: + pass + + spans = span_exporter.get_finished_spans() + assert any( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "invoke_workflow" + for span in spans + ) + + def test_step_execute_spans( instrument_agno, span_exporter, ) -> None: """Test that Step.execute emits an invoke_workflow span.""" - pytest.importorskip("fastapi") + pytest.importorskip("agno.workflow.step") from agno.workflow.step import Step # noqa: PLC0415 - step = Step(name="test-step") + step = Step(name="test-step", executor=lambda step_input: "test output") with patch.object(Step, "execute", wraps=step.execute): try: step.execute("test input") @@ -321,3 +345,27 @@ def test_step_execute_spans( == "invoke_workflow" for span in spans ) + + +@pytest.mark.asyncio +async def test_step_aexecute_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Step.aexecute emits an invoke_workflow span.""" + pytest.importorskip("agno.workflow.step") + from agno.workflow.step import Step # noqa: PLC0415 + + step = Step(name="test-step-async", executor=lambda step_input: "test output") + with patch.object(Step, "aexecute", wraps=step.aexecute): + try: + await step.aexecute("test input") + except Exception: + pass + + spans = span_exporter.get_finished_spans() + assert any( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "invoke_workflow" + for span in spans + ) From 16dff3d92ea1cafb2442cf194eda73ed3e9f9801 Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Mon, 3 Aug 2026 19:15:17 +0000 Subject: [PATCH 04/14] fix lint --- .../tests/test_agent.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py index 2df6c0ed4..1e19d3fde 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py @@ -356,7 +356,9 @@ async def test_step_aexecute_spans( pytest.importorskip("agno.workflow.step") from agno.workflow.step import Step # noqa: PLC0415 - step = Step(name="test-step-async", executor=lambda step_input: "test output") + step = Step( + name="test-step-async", executor=lambda step_input: "test output" + ) with patch.object(Step, "aexecute", wraps=step.aexecute): try: await step.aexecute("test input") From d71c36ed2cbb04c6ad5a82a320144111a15b249c Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Mon, 3 Aug 2026 19:36:06 +0000 Subject: [PATCH 05/14] Address comments --- .../instrumentation/genai/agno/patch.py | 15 +---- .../tests/conformance/workflow.py | 65 +++++++++++++++++++ .../tests/requirements.latest.txt | 1 + .../tests/requirements.oldest.txt | 1 + .../tests/test_agent.py | 4 ++ .../tests/test_conformance.py | 2 + 6 files changed, 75 insertions(+), 13 deletions(-) create mode 100644 instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py index 11de78190..8c05e30c2 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py @@ -232,12 +232,6 @@ def _set_invocation_input( kwargs: dict[str, Any], capture_content: bool, ) -> None: - agent_id = getattr(instance, "agent_id", None) or getattr( - instance, "id", None - ) - if agent_id and hasattr(invocation, "agent_id"): - invocation.agent_id = str(agent_id) - if capture_content and (args or "input" in kwargs): input_val = args[0] if args else kwargs.get("input") if input_val is not None: @@ -272,8 +266,7 @@ def _start_agent_invocation( kwargs: dict[str, Any], capture_content: bool, ) -> AgentInvocation: - default_name = "Team" if instance.__class__.__name__ == "Team" else "Agent" - agent_name = getattr(instance, "name", None) or default_name + agent_name = getattr(instance, "name", None) invocation = handler.invoke_local_agent(agent_name=agent_name) _set_invocation_input(invocation, instance, args, kwargs, capture_content) invocation.tool_definitions = prepare_tool_definitions( @@ -392,10 +385,8 @@ def _start_workflow_invocation( args: tuple[Any, ...], kwargs: dict[str, Any], capture_content: bool, - *, - default_name: str = "Workflow", ) -> WorkflowInvocation: - workflow_name = getattr(instance, "name", None) or default_name + workflow_name = getattr(instance, "name", None) invocation = handler.workflow(name=workflow_name) _set_invocation_input(invocation, instance, args, kwargs, capture_content) return invocation @@ -515,7 +506,6 @@ def traced_method( args, kwargs, capture_content, - default_name="Step", ) as invocation: result = wrapped(*args, **kwargs) _set_invocation_output(invocation, result, capture_content) @@ -541,7 +531,6 @@ async def traced_method( args, kwargs, capture_content, - default_name="Step", ) as invocation: result = await wrapped(*args, **kwargs) _set_invocation_output(invocation, result, capture_content) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py new file mode 100644 index 000000000..9b204dd61 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py @@ -0,0 +1,65 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Conformance scenario: basic workflow run for Agno.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import patch + +import pytest + +pytest.importorskip("agno.workflow.workflow") + +from agno.agent import Agent +from agno.models.response import ModelResponse +from agno.workflow.workflow import Workflow +from tests.mock_model import MockModel + +from opentelemetry.instrumentation.genai.agno import AgnoInstrumentor +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.test_util_genai.conformance import Scenario +from opentelemetry.test_util_genai.instrumentor import instrument + + +class WorkflowScenario(Scenario): + expected_spans = {"invoke_workflow": 1, "invoke_agent": 1} + expected_metrics = ("gen_ai.client.operation.duration",) + + def run( + self, + *, + tracer_provider: TracerProvider, + meter_provider: MeterProvider, + logger_provider: LoggerProvider, + vcr: Any, + ) -> None: + with instrument( + AgnoInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + agent = Agent( + name="workflow-agent", + model=MockModel(id="mock-model"), + session_id="session-workflow", + ) + workflow = Workflow( + name="test-conformance-workflow", + agent=agent, + session_id="session-workflow", + ) + mock_output = ModelResponse(content="Workflow Conformance Hello!") + with ( + patch.object(Workflow, "run", wraps=workflow.run), + patch.object(Agent, "run", wraps=agent.run), + patch( + "agno.models.base.Model.response", return_value=mock_output + ), + ): + workflow.run("hello workflow conformance") diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.latest.txt b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.latest.txt index 4706b0aeb..56455ff9f 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.latest.txt +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.latest.txt @@ -39,6 +39,7 @@ agno wrapt>=2.2.2 +fastapi # test with the latest version of opentelemetry-api, sdk, semantic conventions, and instrumentation -e util/opentelemetry-util-genai diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.oldest.txt b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.oldest.txt index 8eecc5b0c..384411328 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.oldest.txt +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.oldest.txt @@ -23,3 +23,4 @@ # # There is currently nothing to pin: agno has no test-only dependency that isn't already # provided by a declared bound or by the shared test fixtures. +fastapi diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py index 1e19d3fde..a490bec2e 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py @@ -282,6 +282,7 @@ def test_workflow_run_spans( span_exporter, ) -> None: """Test that Workflow.run emits an invoke_workflow span.""" + pytest.importorskip("fastapi") pytest.importorskip("agno.workflow.workflow") from agno.workflow.workflow import Workflow # noqa: PLC0415 @@ -306,6 +307,7 @@ async def test_workflow_arun_spans( span_exporter, ) -> None: """Test that Workflow.arun emits an invoke_workflow span.""" + pytest.importorskip("fastapi") pytest.importorskip("agno.workflow.workflow") from agno.workflow.workflow import Workflow # noqa: PLC0415 @@ -329,6 +331,7 @@ def test_step_execute_spans( span_exporter, ) -> None: """Test that Step.execute emits an invoke_workflow span.""" + pytest.importorskip("fastapi") pytest.importorskip("agno.workflow.step") from agno.workflow.step import Step # noqa: PLC0415 @@ -353,6 +356,7 @@ async def test_step_aexecute_spans( span_exporter, ) -> None: """Test that Step.aexecute emits an invoke_workflow span.""" + pytest.importorskip("fastapi") pytest.importorskip("agno.workflow.step") from agno.workflow.step import Step # noqa: PLC0415 diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_conformance.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_conformance.py index 3e8438b54..34d4c6889 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_conformance.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_conformance.py @@ -21,12 +21,14 @@ ) from .conformance.agent import AgentScenario +from .conformance.workflow import WorkflowScenario @pytest.mark.parametrize( "scenario", [ pytest.param(AgentScenario()), + pytest.param(WorkflowScenario()), ], ids=lambda s: type(s).__name__, ) From 9a754a011f2c756b00236af4c0dd68b393730151 Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Mon, 3 Aug 2026 19:42:02 +0000 Subject: [PATCH 06/14] Remove STEP instrumentation --- .../.changelog/350.added | 2 +- .../README.rst | 1 - .../instrumentation/genai/agno/patch.py | 74 +------------------ .../tests/test_agent.py | 50 ------------- 4 files changed, 2 insertions(+), 125 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/350.added b/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/350.added index 74c8e3289..8ac19baa0 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/350.added +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/350.added @@ -1 +1 @@ -Add instrumentation for Agno Team, Workflow, Step, and Model response methods. +Add instrumentation for Agno Team, Workflow, and Model response methods. diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst b/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst index 0b80b73d5..0201df280 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst @@ -33,7 +33,6 @@ The instrumentation automatically traces: * ``Agent.run`` and ``Agent.arun`` * ``Team.run`` and ``Team.arun`` * ``Workflow.run`` and ``Workflow.arun`` -* ``Step.execute`` and ``Step.aexecute`` * ``Model.response`` and ``Model.aresponse`` * ``FunctionCall.execute`` and ``FunctionCall.aexecute`` diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py index 8c05e30c2..8c81c032a 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py @@ -38,8 +38,6 @@ _FUNCTION_CALL_CLASS = "FunctionCall" _AGNO_WORKFLOW_MODULE = "agno.workflow.workflow" _WORKFLOW_CLASS = "Workflow" -_AGNO_STEP_MODULE = "agno.workflow.step" -_STEP_CLASS = "Step" _AGNO_MODELS_MODULE = "agno.models.base" _MODEL_CLASS = "Model" @@ -95,19 +93,6 @@ def patch_agent(handler: TelemetryHandler) -> None: ) except (ImportError, AttributeError): pass - try: - wrap_function_wrapper( - _AGNO_STEP_MODULE, - f"{_STEP_CLASS}.execute", - _step_execute(handler), - ) - wrap_function_wrapper( - _AGNO_STEP_MODULE, - f"{_STEP_CLASS}.aexecute", - _step_aexecute(handler), - ) - except (ImportError, AttributeError): - pass try: wrap_function_wrapper( _AGNO_MODELS_MODULE, @@ -146,7 +131,7 @@ def unpatch_agent() -> None: unwrap(agno.tools.function.FunctionCall, "aexecute") except (ImportError, AttributeError): pass - # Workflow / step depend on optional packages (like fastapi), may fail to import. + # Workflow depends on optional packages (like fastapi), may fail to import. try: import agno.workflow.workflow # pylint: disable=import-outside-toplevel # noqa: PLC0415 @@ -154,13 +139,6 @@ def unpatch_agent() -> None: unwrap(agno.workflow.workflow.Workflow, "arun") except (ImportError, AttributeError): pass - try: - import agno.workflow.step # pylint: disable=import-outside-toplevel # noqa: PLC0415 - - unwrap(agno.workflow.step.Step, "execute") - unwrap(agno.workflow.step.Step, "aexecute") - except (ImportError, AttributeError): - pass try: import agno.models.base # pylint: disable=import-outside-toplevel # noqa: PLC0415 @@ -489,56 +467,6 @@ async def traced_method( return cast(Callable[..., Any], traced_method) -def _step_execute( - handler: TelemetryHandler, -) -> Callable[..., Any]: - capture_content = handler.should_capture_content() - - def traced_method( - wrapped: Callable[..., Any], - instance: Any, - args: tuple[Any, ...], - kwargs: dict[str, Any], - ) -> Any: - with _start_workflow_invocation( - handler, - instance, - args, - kwargs, - capture_content, - ) as invocation: - result = wrapped(*args, **kwargs) - _set_invocation_output(invocation, result, capture_content) - return result - - return traced_method - - -def _step_aexecute( - handler: TelemetryHandler, -) -> Callable[..., Any]: - capture_content = handler.should_capture_content() - - async def traced_method( - wrapped: Callable[..., Awaitable[Any]], - instance: Any, - args: tuple[Any, ...], - kwargs: dict[str, Any], - ) -> Any: - with _start_workflow_invocation( - handler, - instance, - args, - kwargs, - capture_content, - ) as invocation: - result = await wrapped(*args, **kwargs) - _set_invocation_output(invocation, result, capture_content) - return result - - return cast(Callable[..., Any], traced_method) - - def _model_response( handler: TelemetryHandler, ) -> Callable[..., Any]: diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py index a490bec2e..fb6810d51 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py @@ -325,53 +325,3 @@ async def test_workflow_arun_spans( for span in spans ) - -def test_step_execute_spans( - instrument_agno, - span_exporter, -) -> None: - """Test that Step.execute emits an invoke_workflow span.""" - pytest.importorskip("fastapi") - pytest.importorskip("agno.workflow.step") - from agno.workflow.step import Step # noqa: PLC0415 - - step = Step(name="test-step", executor=lambda step_input: "test output") - with patch.object(Step, "execute", wraps=step.execute): - try: - step.execute("test input") - except Exception: - pass - - spans = span_exporter.get_finished_spans() - assert any( - span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) - == "invoke_workflow" - for span in spans - ) - - -@pytest.mark.asyncio -async def test_step_aexecute_spans( - instrument_agno, - span_exporter, -) -> None: - """Test that Step.aexecute emits an invoke_workflow span.""" - pytest.importorskip("fastapi") - pytest.importorskip("agno.workflow.step") - from agno.workflow.step import Step # noqa: PLC0415 - - step = Step( - name="test-step-async", executor=lambda step_input: "test output" - ) - with patch.object(Step, "aexecute", wraps=step.aexecute): - try: - await step.aexecute("test input") - except Exception: - pass - - spans = span_exporter.get_finished_spans() - assert any( - span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) - == "invoke_workflow" - for span in spans - ) From 4f336a62f94c576e20be9e94333466a3e6d77b1b Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Mon, 3 Aug 2026 19:49:07 +0000 Subject: [PATCH 07/14] Fix tests --- .../tests/conformance/workflow.py | 24 ++++--------------- .../tests/test_agent.py | 15 ++++-------- 2 files changed, 8 insertions(+), 31 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py index 9b204dd61..e8db84d5a 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py @@ -6,16 +6,13 @@ from __future__ import annotations from typing import Any -from unittest.mock import patch import pytest +pytest.importorskip("fastapi") pytest.importorskip("agno.workflow.workflow") -from agno.agent import Agent -from agno.models.response import ModelResponse from agno.workflow.workflow import Workflow -from tests.mock_model import MockModel from opentelemetry.instrumentation.genai.agno import AgnoInstrumentor from opentelemetry.sdk._logs import LoggerProvider @@ -26,7 +23,7 @@ class WorkflowScenario(Scenario): - expected_spans = {"invoke_workflow": 1, "invoke_agent": 1} + expected_spans = {"invoke_workflow": 1} expected_metrics = ("gen_ai.client.operation.duration",) def run( @@ -44,22 +41,9 @@ def run( meter_provider=meter_provider, content_capture="SPAN_ONLY", ): - agent = Agent( - name="workflow-agent", - model=MockModel(id="mock-model"), - session_id="session-workflow", - ) workflow = Workflow( name="test-conformance-workflow", - agent=agent, + steps=[], session_id="session-workflow", ) - mock_output = ModelResponse(content="Workflow Conformance Hello!") - with ( - patch.object(Workflow, "run", wraps=workflow.run), - patch.object(Agent, "run", wraps=agent.run), - patch( - "agno.models.base.Model.response", return_value=mock_output - ), - ): - workflow.run("hello workflow conformance") + workflow.run("hello workflow conformance") diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py index fb6810d51..bd3805804 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py @@ -286,12 +286,9 @@ def test_workflow_run_spans( pytest.importorskip("agno.workflow.workflow") from agno.workflow.workflow import Workflow # noqa: PLC0415 - workflow = Workflow(name="test-workflow") + workflow = Workflow(name="test-workflow", steps=[]) with patch.object(Workflow, "run", wraps=workflow.run): - try: - workflow.run("test input") - except Exception: - pass + workflow.run("test input") spans = span_exporter.get_finished_spans() assert any( @@ -311,12 +308,9 @@ async def test_workflow_arun_spans( pytest.importorskip("agno.workflow.workflow") from agno.workflow.workflow import Workflow # noqa: PLC0415 - workflow = Workflow(name="test-workflow-async") + workflow = Workflow(name="test-workflow-async", steps=[]) with patch.object(Workflow, "arun", wraps=workflow.arun): - try: - await workflow.arun("test input") - except Exception: - pass + await workflow.arun("test input") spans = span_exporter.get_finished_spans() assert any( @@ -324,4 +318,3 @@ async def test_workflow_arun_spans( == "invoke_workflow" for span in spans ) - From eedc27097496ee675ad6c38729d354e5fb094f20 Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Mon, 3 Aug 2026 20:00:40 +0000 Subject: [PATCH 08/14] fix(util-genai): record duration metric on workflow invocations Assisted-by: Antigravity --- .../.changelog/350.fixed | 1 + .../util/genai/_workflow_invocation.py | 11 +++++- .../tests/test_handler_workflow.py | 37 +++++++++++++++++++ 3 files changed, 48 insertions(+), 1 deletion(-) create mode 100644 util/opentelemetry-util-genai/.changelog/350.fixed diff --git a/util/opentelemetry-util-genai/.changelog/350.fixed b/util/opentelemetry-util-genai/.changelog/350.fixed new file mode 100644 index 000000000..7d27c3e99 --- /dev/null +++ b/util/opentelemetry-util-genai/.changelog/350.fixed @@ -0,0 +1 @@ +Record `gen_ai.client.operation.duration` metric on workflow invocations (`WorkflowInvocation`). diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py index f06d72dd3..e8ea48bbe 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py @@ -87,6 +87,15 @@ def _get_messages_for_span(self) -> dict[str, AttributeValue]: key: value for key, value in optional_attrs if value is not None } + def _get_metric_attributes(self) -> dict[str, AttributeValue]: + attrs: dict[str, AttributeValue] = { + GenAI.GEN_AI_OPERATION_NAME: self._operation_name, + } + if self._name is not None: + attrs[GenAI.GEN_AI_WORKFLOW_NAME] = self._name + attrs.update(self.metric_attributes) + return attrs + def _apply_finish(self, error: Error | None = None) -> None: attributes: dict[str, AttributeValue] = self._get_messages_for_span() if error is not None: @@ -97,4 +106,4 @@ def _apply_finish(self, error: Error | None = None) -> None: inputs=self.input_messages, outputs=self.output_messages, ) - # TODO: Add workflow metrics when supported + self._metrics_recorder.record(self) diff --git a/util/opentelemetry-util-genai/tests/test_handler_workflow.py b/util/opentelemetry-util-genai/tests/test_handler_workflow.py index cf45f90cb..2afcacd00 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_workflow.py +++ b/util/opentelemetry-util-genai/tests/test_handler_workflow.py @@ -17,6 +17,7 @@ from opentelemetry.semconv._incubating.attributes import ( gen_ai_attributes as GenAI, ) +from opentelemetry.test.test_base import TestBase from opentelemetry.trace import INVALID_SPAN, SpanKind from opentelemetry.trace.status import StatusCode from opentelemetry.util.genai.handler import TelemetryHandler @@ -287,3 +288,39 @@ def test_workflow_context_manager_with_messages(self) -> None: spans[0].attributes[GenAI.GEN_AI_OPERATION_NAME], "invoke_workflow", ) + + +class TestWorkflowInvocationMetrics(TestBase): + def _harvest_metrics(self): + metrics = self.get_sorted_metrics() + metrics_by_name = {} + for metric in metrics or []: + points = metric.data.data_points or [] + metrics_by_name.setdefault(metric.name, []).extend(points) + return metrics_by_name + + def test_workflow_records_duration(self) -> None: + handler = TelemetryHandler( + tracer_provider=self.tracer_provider, + meter_provider=self.meter_provider, + ) + with patch("timeit.default_timer", return_value=1000.0): + invocation = handler.workflow(name="test-workflow") + + with patch("timeit.default_timer", return_value=1002.0): + invocation.stop() + + metrics = self._harvest_metrics() + self.assertIn("gen_ai.client.operation.duration", metrics) + duration_points = metrics["gen_ai.client.operation.duration"] + self.assertEqual(len(duration_points), 1) + duration_point = duration_points[0] + self.assertEqual( + duration_point.attributes[GenAI.GEN_AI_OPERATION_NAME], + "invoke_workflow", + ) + self.assertEqual( + duration_point.attributes[GenAI.GEN_AI_WORKFLOW_NAME], + "test-workflow", + ) + self.assertAlmostEqual(duration_point.sum, 2.0, places=3) From 544ad9c920d951caef0c13910397ea87e9606d06 Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Mon, 3 Aug 2026 20:21:59 +0000 Subject: [PATCH 09/14] build(agno): require opentelemetry-util-genai >= 1.1b0.dev, <2 Assisted-by: Antigravity --- .../opentelemetry-instrumentation-genai-agno/pyproject.toml | 2 +- .../tests/requirements.oldest.txt | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/pyproject.toml b/instrumentation/opentelemetry-instrumentation-genai-agno/pyproject.toml index a4b53d926..9fed56251 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/pyproject.toml +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/pyproject.toml @@ -28,7 +28,7 @@ dependencies = [ "opentelemetry-api ~= 1.43", "opentelemetry-instrumentation >= 0.64b0, <1", "opentelemetry-semantic-conventions >= 0.64b0, <1", - "opentelemetry-util-genai >= 1.0b0, <2", + "opentelemetry-util-genai >= 1.1b0.dev, <2", ] [project.optional-dependencies] diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.oldest.txt b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.oldest.txt index 384411328..37d900aae 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.oldest.txt +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.oldest.txt @@ -21,6 +21,6 @@ # OpenTelemetry SDK and test utilities come transitively from opentelemetry-test-util-genai, which # every oldest env installs. Pin here only test-only deps that nothing else already provides. # -# There is currently nothing to pin: agno has no test-only dependency that isn't already -# provided by a declared bound or by the shared test fixtures. -fastapi +# Drop this once opentelemetry-util-genai 1.1b0 is published. +-e util/opentelemetry-util-genai +fastapi >= 0.100.0 From e5ae91704b3cede6c43af8d28f41a296b6411f4f Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Wed, 5 Aug 2026 14:06:32 +0000 Subject: [PATCH 10/14] Address comments --- .../instrumentation/genai/agno/patch.py | 4 ++-- .../tests/conformance/workflow.py | 2 +- .../tests/conformance/workflow.py | 1 + .../tests/conformance/orchestration.py | 5 ++++- .../.changelog/350.fixed | 2 +- .../util/genai/_workflow_invocation.py | 2 +- .../src/opentelemetry/util/genai/instruments.py | 14 ++++++++++++++ .../src/opentelemetry/util/genai/metrics.py | 17 +++++++++++++++++ .../tests/test_handler_workflow.py | 4 ++-- 9 files changed, 43 insertions(+), 8 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py index 8c81c032a..3df07f03a 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py @@ -204,7 +204,7 @@ def _set_tool_invocation_output( def _set_invocation_input( - invocation: Any, + invocation: AgentInvocation | WorkflowInvocation, instance: Any, args: tuple[Any, ...], kwargs: dict[str, Any], @@ -220,7 +220,7 @@ def _set_invocation_input( def _set_invocation_output( - invocation: Any, + invocation: AgentInvocation | WorkflowInvocation, result: Any, capture_content: bool, ) -> None: diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py index e8db84d5a..1925e0174 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py @@ -24,7 +24,7 @@ class WorkflowScenario(Scenario): expected_spans = {"invoke_workflow": 1} - expected_metrics = ("gen_ai.client.operation.duration",) + expected_metrics = ("gen_ai.invoke_workflow.duration",) def run( self, diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py index b2bbeb045..848b4e1b3 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py @@ -35,6 +35,7 @@ class WorkflowScenario(Scenario): expected_metrics = ( "gen_ai.client.operation.duration", "gen_ai.client.token.usage", + "gen_ai.invoke_workflow.duration", ) # langchain can't populate server.address on chat spans. expected_violations = ( diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai-agents/tests/conformance/orchestration.py b/instrumentation/opentelemetry-instrumentation-genai-openai-agents/tests/conformance/orchestration.py index 85c074cb4..bc6d94f7d 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-openai-agents/tests/conformance/orchestration.py +++ b/instrumentation/opentelemetry-instrumentation-genai-openai-agents/tests/conformance/orchestration.py @@ -75,7 +75,10 @@ class OrchestrationScenario(Scenario): "invoke_agent": 2, "execute_tool": 1, } - expected_metrics = ("gen_ai.client.operation.duration",) + expected_metrics = ( + "gen_ai.client.operation.duration", + "gen_ai.invoke_workflow.duration", + ) expected_violations = ( # `FunctionSpanData` in the openai-agents library doesn't expose # `tool_call_id`, so our `execute_tool` spans can't set diff --git a/util/opentelemetry-util-genai/.changelog/350.fixed b/util/opentelemetry-util-genai/.changelog/350.fixed index 7d27c3e99..52f754115 100644 --- a/util/opentelemetry-util-genai/.changelog/350.fixed +++ b/util/opentelemetry-util-genai/.changelog/350.fixed @@ -1 +1 @@ -Record `gen_ai.client.operation.duration` metric on workflow invocations (`WorkflowInvocation`). +Record `gen_ai.invoke_workflow.duration` metric on workflow invocations (`WorkflowInvocation`). diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py index e8ea48bbe..042c59c72 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py @@ -106,4 +106,4 @@ def _apply_finish(self, error: Error | None = None) -> None: inputs=self.input_messages, outputs=self.output_messages, ) - self._metrics_recorder.record(self) + self._metrics_recorder.record_workflow(self) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/instruments.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/instruments.py index c697e89d2..445ce86d6 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/instruments.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/instruments.py @@ -48,6 +48,20 @@ def create_duration_histogram(meter: Meter) -> Histogram: ) +def create_workflow_duration_histogram(meter: Meter) -> Histogram: + name = getattr( + gen_ai_metrics, + "GEN_AI_INVOKE_WORKFLOW_DURATION", + "gen_ai.invoke_workflow.duration", + ) + return meter.create_histogram( + name=name, + description="Measures the duration of a workflow execution.", + unit="s", + explicit_bucket_boundaries_advisory=_GEN_AI_CLIENT_OPERATION_DURATION_BUCKETS, + ) + + def create_token_histogram(meter: Meter) -> Histogram: return meter.create_histogram( name=gen_ai_metrics.GEN_AI_CLIENT_TOKEN_USAGE, diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/metrics.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/metrics.py index 57796c77d..be45f2105 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/metrics.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/metrics.py @@ -17,6 +17,7 @@ create_time_per_output_chunk_histogram, create_time_to_first_chunk_histogram, create_token_histogram, + create_workflow_duration_histogram, ) from opentelemetry.util.types import Attributes @@ -28,6 +29,9 @@ class InvocationMetricsRecorder: def __init__(self, meter: Meter): self._duration_histogram: Histogram = create_duration_histogram(meter) + self._workflow_duration_histogram: Histogram = ( + create_workflow_duration_histogram(meter) + ) self._token_histogram: Histogram = create_token_histogram(meter) self._time_to_first_chunk_histogram: Histogram = ( create_time_to_first_chunk_histogram(meter) @@ -58,6 +62,19 @@ def record(self, invocation: GenAIInvocation) -> None: context=invocation._span_context, ) + def record_workflow(self, invocation: GenAIInvocation) -> None: + """Record duration metric for a workflow invocation.""" + attributes = invocation._get_metric_attributes() + duration_seconds = max( + timeit.default_timer() - invocation._monotonic_start_s, + 0.0, + ) + self._workflow_duration_histogram.record( + duration_seconds, + attributes=attributes, + context=invocation._span_context, + ) + def record_time_to_first_chunk( self, ttfc_seconds: float, diff --git a/util/opentelemetry-util-genai/tests/test_handler_workflow.py b/util/opentelemetry-util-genai/tests/test_handler_workflow.py index 2afcacd00..5ad8131c2 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_workflow.py +++ b/util/opentelemetry-util-genai/tests/test_handler_workflow.py @@ -311,8 +311,8 @@ def test_workflow_records_duration(self) -> None: invocation.stop() metrics = self._harvest_metrics() - self.assertIn("gen_ai.client.operation.duration", metrics) - duration_points = metrics["gen_ai.client.operation.duration"] + self.assertIn("gen_ai.invoke_workflow.duration", metrics) + duration_points = metrics["gen_ai.invoke_workflow.duration"] self.assertEqual(len(duration_points), 1) duration_point = duration_points[0] self.assertEqual( From 6d18b26334ff2f5e591e7ee09851511a64c33ae5 Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Wed, 5 Aug 2026 14:10:18 +0000 Subject: [PATCH 11/14] Fix patch --- .../src/opentelemetry/instrumentation/genai/agno/patch.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py index 98332db31..f6eb72b67 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py @@ -134,14 +134,14 @@ def unpatch_agent() -> None: pass # Workflow depends on optional packages (like fastapi), may fail to import. try: - import agno.workflow.workflow # pylint: disable=import-outside-toplevel # noqa: PLC0415 + import agno.workflow.workflow # pylint: disable=import-outside-toplevel unwrap(agno.workflow.workflow.Workflow, "run") unwrap(agno.workflow.workflow.Workflow, "arun") except (ImportError, AttributeError): pass try: - import agno.models.base # pylint: disable=import-outside-toplevel # noqa: PLC0415 + import agno.models.base # pylint: disable=import-outside-toplevel unwrap(agno.models.base.Model, "response") unwrap(agno.models.base.Model, "aresponse") From a98eeeb38a5319f1216038a9840f7f6c5ba59ced Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Wed, 5 Aug 2026 14:20:51 +0000 Subject: [PATCH 12/14] fix(util-genai,agno): return Self from GenAIInvocation.__enter__ and guard conversation_id assignment --- .../src/opentelemetry/instrumentation/genai/agno/patch.py | 6 +++++- .../tests/test_agent.py | 4 ++-- .../src/opentelemetry/util/genai/_invocation.py | 4 +++- 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py index f6eb72b67..7d53d610e 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py @@ -234,7 +234,11 @@ def _set_invocation_output( finish_reason=str(getattr(result, "finish_reason", "stop")), ) ] - if hasattr(result, "session_id") and getattr(result, "session_id"): + if ( + isinstance(invocation, AgentInvocation) + and hasattr(result, "session_id") + and getattr(result, "session_id") + ): invocation.conversation_id = str(getattr(result, "session_id")) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py index bd3805804..c6f516867 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py @@ -284,7 +284,7 @@ def test_workflow_run_spans( """Test that Workflow.run emits an invoke_workflow span.""" pytest.importorskip("fastapi") pytest.importorskip("agno.workflow.workflow") - from agno.workflow.workflow import Workflow # noqa: PLC0415 + from agno.workflow.workflow import Workflow workflow = Workflow(name="test-workflow", steps=[]) with patch.object(Workflow, "run", wraps=workflow.run): @@ -306,7 +306,7 @@ async def test_workflow_arun_spans( """Test that Workflow.arun emits an invoke_workflow span.""" pytest.importorskip("fastapi") pytest.importorskip("agno.workflow.workflow") - from agno.workflow.workflow import Workflow # noqa: PLC0415 + from agno.workflow.workflow import Workflow workflow = Workflow(name="test-workflow-async", steps=[]) with patch.object(Workflow, "arun", wraps=workflow.arun): diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py index 8c48dba51..52e0a65b3 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py @@ -12,6 +12,8 @@ from types import TracebackType from typing import TYPE_CHECKING, Any, TypeAlias +from typing_extensions import Self + from opentelemetry._logs import Logger, LogRecord from opentelemetry.context import Context, attach, detach from opentelemetry.semconv._incubating.attributes import ( @@ -211,7 +213,7 @@ def fail(self, error: Error | BaseException) -> None: error = Error.from_exception(error, self._error_type_resolver) self._finish(error) - def __enter__(self) -> GenAIInvocation: + def __enter__(self) -> Self: return self def __exit__( From e64d4a1f390c2a735367169739051bdfd4d7d782 Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Wed, 5 Aug 2026 14:36:31 +0000 Subject: [PATCH 13/14] build(semconv): bump SEMCONV_GENAI_REF to include gen_ai.invoke_workflow.duration and make upstream semconv filter optional --- .../test_util_genai/_setup_weaver.py | 31 ++++++++----------- .../test_util_genai/conformance.py | 4 +-- versions.env | 2 +- 3 files changed, 15 insertions(+), 22 deletions(-) diff --git a/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/_setup_weaver.py b/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/_setup_weaver.py index d22801d96..4915f5b32 100644 --- a/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/_setup_weaver.py +++ b/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/_setup_weaver.py @@ -207,25 +207,20 @@ def _provision_genai_root() -> Path: ) upstream_pins = _load_version_pins(genai_target / "versions.env") - try: - upstream_version = upstream_pins["SEMCONV_VERSION"] - except KeyError as missing: - raise RuntimeError( - f"genai repo's versions.env is missing {missing!s}" - ) from missing - - upstream_target = cache_root / f"upstream-{upstream_version}" - if not (upstream_target / "model").is_dir(): - upstream_archive_url = ( - "https://github.com/open-telemetry/semantic-conventions/" - f"archive/refs/tags/{upstream_version}.tar.gz" - ) - _download_and_extract( - upstream_archive_url, upstream_target, label="upstream-semconv" - ) + upstream_version = upstream_pins.get("SEMCONV_VERSION") + if upstream_version is not None: + upstream_target = cache_root / f"upstream-{upstream_version}" + if not (upstream_target / "model").is_dir(): + upstream_archive_url = ( + "https://github.com/open-telemetry/semantic-conventions/" + f"archive/refs/tags/{upstream_version}.tar.gz" + ) + _download_and_extract( + upstream_archive_url, upstream_target, label="upstream-semconv" + ) - filtered = _materialize_filtered_upstream(genai_target, upstream_target) - _rewrite_manifest_dependency(genai_target, filtered) + filtered = _materialize_filtered_upstream(genai_target, upstream_target) + _rewrite_manifest_dependency(genai_target, filtered) stamp.touch() return genai_target diff --git a/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/conformance.py b/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/conformance.py index 516d99a72..e30cfd947 100644 --- a/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/conformance.py +++ b/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/conformance.py @@ -72,9 +72,7 @@ class ExpectedViolation: def matches(self, violation: dict[str, Any]) -> bool: return violation.get( "id" - ) == self.advice_id and self.message_substring in str( - violation.get("message", "") - ) + ) == self.advice_id and self.message_substring in str(violation) class Scenario(ABC): diff --git a/versions.env b/versions.env index efe17e853..7114c53a7 100644 --- a/versions.env +++ b/versions.env @@ -6,4 +6,4 @@ WEAVER_VERSION=v0.25.1 # The genai semconv registry has no tagged releases yet, so we pin a SHA on `main`. # renovate: datasource=git-refs depName=open-telemetry/semantic-conventions-genai packageName=https://github.com/open-telemetry/semantic-conventions-genai.git versioning=git -SEMCONV_GENAI_REF=f77b9235f2ad49fe95b61e9809ca82bb08ef9d47 +SEMCONV_GENAI_REF=4a39b6ef1363fab57bc40e18c82abb32cc89ebcd From 9efeaf1175f27a879149f2d3f1c5ce98d89f09ea Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Wed, 5 Aug 2026 14:56:12 +0000 Subject: [PATCH 14/14] style: fix ruff formatting in _setup_weaver.py --- .../src/opentelemetry/test_util_genai/_setup_weaver.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/_setup_weaver.py b/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/_setup_weaver.py index 4915f5b32..efa5dadbf 100644 --- a/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/_setup_weaver.py +++ b/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/_setup_weaver.py @@ -219,7 +219,9 @@ def _provision_genai_root() -> Path: upstream_archive_url, upstream_target, label="upstream-semconv" ) - filtered = _materialize_filtered_upstream(genai_target, upstream_target) + filtered = _materialize_filtered_upstream( + genai_target, upstream_target + ) _rewrite_manifest_dependency(genai_target, filtered) stamp.touch() return genai_target