up

Author: lmolkova

instrumentation/opentelemetry-instrumentation-genai-openai-agents/.changelog/86.changed

@@ -0,0 +1 @@
+Rewrite on top of `opentelemetry-util-genai`. The instrumentor now registers a single `TracingProcessor` with the agents library (`agents.tracing.add_trace_processor`) and turns its `Trace` / `AgentSpan` / `FunctionSpan` callbacks into `invoke_workflow` / `invoke_agent` / `execute_tool` spans. The workflow span carries a `gen_ai.workflow.name` attribute sourced from the agents library's `RunConfig.workflow_name`. LLM-level spans are not emitted by this package — install `opentelemetry-instrumentation-genai-openai` alongside to capture them. Removes the `OTEL_INSTRUMENTATION_OPENAI_AGENTS_*` environment variables and the matching instrumentor kwargs. New `disable_openai_trace_export=True` instrumentor kwarg replaces the default OpenAI-hosted trace exporter so traces flow only via OTel. Handoff / guardrail / speech / transcription spans are no longer emitted pending semconv. `execute_tool` spans are missing `gen_ai.tool.call.id` because the agents library's `FunctionSpanData` does not expose it — tracked in [#86](https://github.com/open-telemetry/opentelemetry-python-genai/issues/86).

instrumentation/opentelemetry-instrumentation-genai-openai-agents/src/opentelemetry/instrumentation/genai/openai_agents/processor.py

@@ -0,0 +1,143 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Bridges agents-library tracing callbacks to opentelemetry-util-genai.
+
+The agents library exposes a public extension API
+(:func:`agents.tracing.add_trace_processor`) for plugging custom
+:class:`TracingProcessor` implementations into its own tracing system.
+``Trace.start()`` / ``Span.start()`` invoke the registered processors'
+``on_*_start`` callbacks *synchronously* on whichever asyncio task
+started the agents-library span:
+
+* Workflow (``Trace``) and agent (``AgentSpanData``) spans start in the
+  ``Runner.run`` task itself.
+* Function tool (``FunctionSpanData``) spans start inside the per-tool
+  ``asyncio.Task`` the agents library creates for tool dispatch. That
+  sub-task inherits a snapshot of the run-loop context (so workflow +
+  agent are already active in OTel contextvars).
+
+Because every ``*_end`` callback fires on the same task as its
+matching ``*_start``, util-genai's auto-``attach()`` / ``detach()`` of
+OTel context is balanced and no context tokens leak across tasks.
+OTel's natural parent tracking nests the tree:
+
+    workflow > invoke_agent > [chat from openai instrumentation,
+                               execute_tool]
+
+LLM-level spans (``chat`` / ``responses`` / ``embeddings``) are not
+emitted here — ``opentelemetry-instrumentation-genai-openai`` patches
+the openai SDK directly and produces those.
+"""
+
+from __future__ import annotations
+
+import weakref
+from typing import Any
+
+from agents.tracing import Span, Trace, TracingProcessor
+from agents.tracing.span_data import (
+    AgentSpanData,
+    FunctionSpanData,
+)
+
+from opentelemetry.semconv._incubating.attributes import (
+    gen_ai_attributes as GenAI,
+)
+from opentelemetry.util.genai.handler import TelemetryHandler
+from opentelemetry.util.genai.invocation import (
+    GenAIInvocation,
+    ToolInvocation,
+)
+
+# Non-semconv attribute: surfaces the workflow name on the workflow span
+# so callers can query/filter by it. util-genai's WorkflowInvocation
+# only puts the name in the span name, not as an attribute.
+_WORKFLOW_NAME_ATTR = "gen_ai.workflow.name"
+
+
+class GenAITracingProcessor(TracingProcessor):
+    """Translate agents-library tracing into util-genai invocations.
+
+    Stateful only for span lifetime: each in-flight Trace/Span has one
+    entry in a :class:`weakref.WeakKeyDictionary` keyed by the
+    agents-library object itself. Entries are removed on ``*_end`` or
+    garbage-collected with the agents-library span/trace if the library
+    drops it before ``end`` (which it shouldn't, but the weak reference
+    is belt-and-suspenders against any future leak).
+    """
+
+    def __init__(self, handler: TelemetryHandler, provider: str) -> None:
+        self._handler = handler
+        self._provider = provider
+        self._invocations: weakref.WeakKeyDictionary[
+            Any, GenAIInvocation
+        ] = weakref.WeakKeyDictionary()
+
+    def on_trace_start(self, trace: Trace) -> None:
+        # ``trace.name`` comes from ``RunConfig.workflow_name`` (default
+        # "Agent workflow"). Callers customize it via the agents library's
+        # own ``Runner.run(..., run_config=RunConfig(workflow_name=...))``;
+        # we don't expose a second knob.
+        invocation = self._handler.workflow(name=trace.name)
+        if trace.name:
+            invocation.attributes[_WORKFLOW_NAME_ATTR] = trace.name
+        self._invocations[trace] = invocation
+
+    def on_trace_end(self, trace: Trace) -> None:
+        invocation = self._invocations.pop(trace, None)
+        if invocation is not None:
+            invocation.stop()
+
+    def on_span_start(self, span: Span[Any]) -> None:
+        span_data = span.span_data
+        if isinstance(span_data, AgentSpanData):
+            invocation = self._handler.invoke_local_agent(
+                provider=self._provider,
+                agent_name=span_data.name,
+            )
+            self._invocations[span] = invocation
+            return
+        if isinstance(span_data, FunctionSpanData):
+            invocation = self._handler.tool(
+                name=span_data.name,
+                arguments=span_data.input,
+                tool_type="function",
+            )
+            # ToolInvocation does not include provider in metric attributes
+            # by default; set it so gen_ai.client.operation.duration carries
+            # the required gen_ai.provider.name attribute.
+            invocation.metric_attributes[GenAI.GEN_AI_PROVIDER_NAME] = (
+                self._provider
+            )
+            self._invocations[span] = invocation
+            return
+        # Other span_data types (GenerationSpanData, ResponseSpanData,
+        # HandoffSpanData, GuardrailSpanData, Speech/TranscriptionSpanData)
+        # are intentionally ignored. LLM-level spans come from the openai
+        # instrumentation; the rest have no semconv yet.
+
+    def on_span_end(self, span: Span[Any]) -> None:
+        invocation = self._invocations.pop(span, None)
+        if invocation is None:
+            return
+        if isinstance(invocation, ToolInvocation) and isinstance(
+            span.span_data, FunctionSpanData
+        ):
+            output = span.span_data.output
+            if output is not None:
+                invocation.tool_result = (
+                    output if isinstance(output, str) else str(output)
+                )
+        invocation.stop()
+
+    def shutdown(self) -> None:
+        for invocation in list(self._invocations.values()):
+            try:
+                invocation.stop()
+            except Exception:  # pylint: disable=broad-except
+                pass
+        self._invocations.clear()
+
+    def force_flush(self) -> None:  # pragma: no cover - nothing to flush
+        pass

instrumentation/opentelemetry-instrumentation-genai-openai-agents/tests/test_instrumentor.py

@@ -0,0 +1,85 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import agents.tracing
+
+from opentelemetry.instrumentation.genai.openai_agents import (
+    OpenAIAgentsInstrumentor,
+)
+from opentelemetry.instrumentation.genai.openai_agents.package import (
+    _instruments,
+)
+from opentelemetry.instrumentation.genai.openai_agents.processor import (
+    GenAITracingProcessor,
+)
+
+
+def _registered_processors() -> tuple:
+    provider = agents.tracing.get_trace_provider()
+    multi = getattr(provider, "_multi_processor", None)
+    return tuple(getattr(multi, "_processors", ()))
+
+
+def _our_processors():
+    return [
+        p
+        for p in _registered_processors()
+        if isinstance(p, GenAITracingProcessor)
+    ]
+
+
+def test_instrumentation_dependencies_exposed() -> None:
+    instrumentor = OpenAIAgentsInstrumentor()
+    assert instrumentor.instrumentation_dependencies() == _instruments
+
+
+def test_instrument_adds_processor_alongside_default() -> None:
+    instrumentor = OpenAIAgentsInstrumentor()
+    pre_count = len(_registered_processors())
+    try:
+        instrumentor.instrument()
+        post = _registered_processors()
+        # Default processor stays in place, ours is appended.
+        assert len(post) == pre_count + 1
+        assert len(_our_processors()) == 1
+    finally:
+        instrumentor.uninstrument()
+    assert len(_our_processors()) == 0
+
+
+def test_instrument_with_disable_openai_trace_export_replaces_processors() -> None:
+    # Make sure the default processor is registered before we start,
+    # so the "replace" behavior is observable.
+    agents.tracing.set_trace_processors(
+        [agents.tracing.processors.default_processor()]
+    )
+    instrumentor = OpenAIAgentsInstrumentor()
+    try:
+        instrumentor.instrument(disable_openai_trace_export=True)
+        post = _registered_processors()
+        assert len(post) == 1
+        assert isinstance(post[0], GenAITracingProcessor)
+    finally:
+        instrumentor.uninstrument()
+
+
+def test_double_instrument_is_noop() -> None:
+    instrumentor = OpenAIAgentsInstrumentor()
+    try:
+        instrumentor.instrument()
+        first = _our_processors()
+        instrumentor.instrument()
+        second = _our_processors()
+        assert len(first) == 1 and len(second) == 1
+        assert first[0] is second[0]
+    finally:
+        instrumentor.uninstrument()
+
+
+def test_double_uninstrument_is_noop() -> None:
+    instrumentor = OpenAIAgentsInstrumentor()
+    instrumentor.instrument()
+    instrumentor.uninstrument()
+    instrumentor.uninstrument()  # must not raise