Add support for workflow invocation in genAI utils

util/opentelemetry-util-genai/CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4320](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4320))
 - Add workflow invocation type to genAI utils
   ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4310](#4310))
+- Add support for workflow in genAI utils handler.
+  ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4334](#4334))
 
 ## Version 0.3b0 (2026-02-20)
 

util/opentelemetry-util-genai/src/opentelemetry/util/genai/handler.py

@@ -60,6 +60,7 @@
 
 from __future__ import annotations
 
+import logging
 import timeit
 from contextlib import contextmanager
 from typing import Iterator
@@ -82,11 +83,33 @@
 from opentelemetry.util.genai.span_utils import (
     _apply_error_attributes,
     _apply_llm_finish_attributes,
+    _apply_workflow_finish_attributes,
     _maybe_emit_llm_event,
 )
-from opentelemetry.util.genai.types import Error, LLMInvocation
+from opentelemetry.util.genai.types import (
+    Error,
+    GenAIInvocation,
+    LLMInvocation,
+    WorkflowInvocation,
+)
 from opentelemetry.util.genai.version import __version__
 
+_logger = logging.getLogger(__name__)
+
+
+def _safe_detach(invocation: GenAIInvocation) -> None:
+    """Detach the context token if still present, as a safety net."""
+    if invocation.context_token is not None:
+        try:
+            otel_context.detach(invocation.context_token)
+        except Exception:  # pylint: disable=broad-except
+            pass
+    if invocation.span is not None:
+        try:
+            invocation.span.end()
+        except Exception:  # pylint: disable=broad-except
+            pass
+
 
 class TelemetryHandler:
     """
@@ -211,6 +234,96 @@ def llm(
             raise
         self.stop_llm(invocation)
 
+    def start_workflow(
+        self, invocation: WorkflowInvocation
+    ) -> WorkflowInvocation:
+        """Start a workflow invocation and create a pending span entry."""
+        # Create a span and attach it as current; keep the token to detach later
+        span = self._tracer.start_span(
+            name=f"{invocation.operation_name} {invocation.name}",
+            kind=SpanKind.INTERNAL,
+        )
+        invocation.monotonic_start_s = timeit.default_timer()
+        invocation.span = span
+        invocation.context_token = otel_context.attach(
+            set_span_in_context(span)
+        )
+        return invocation
+
+    def stop_workflow(  # pylint: disable=no-self-use
+        self, invocation: WorkflowInvocation
+    ) -> WorkflowInvocation:
+        """Finalize a workflow successfully and end its span."""
+        if invocation.context_token is None or invocation.span is None:
+            # TODO: Provide feedback that this invocation was not started
+            return invocation
+
+        try:
+            _apply_workflow_finish_attributes(invocation.span, invocation)
+        finally:
+            # Detach context and end span
+            otel_context.detach(invocation.context_token)
+            invocation.span.end()
+        return invocation
+
+    def fail_workflow(  # pylint: disable=no-self-use
+        self, invocation: WorkflowInvocation, error: Error
+    ) -> WorkflowInvocation:
+        """Fail a Workflow invocation and end its span with error status."""
+        if invocation.context_token is None or invocation.span is None:
+            # TODO: Provide feedback that this invocation was not started
+            return invocation
+
+        try:
+            _apply_workflow_finish_attributes(invocation.span, invocation)
+            _apply_error_attributes(invocation.span, error)
+        finally:
+            otel_context.detach(invocation.context_token)
+            invocation.span.end()
+        return invocation
+
+    @contextmanager
+    def workflow(
+        self, invocation: WorkflowInvocation | None = None
+    ) -> Iterator[WorkflowInvocation]:
+        """Context manager for Workflow invocations.
+
+        Only set data attributes on the invocation object, do not modify the span or context.
+
+        Starts the span on entry. On normal exit, finalizes the invocation and ends the span.
+        If an exception occurs inside the context, marks the span as error, ends it, and
+        re-raises the original exception.
+        """
+        if invocation is None:
+            invocation = WorkflowInvocation()
+
+        try:
+            self.start_workflow(invocation)
+        except Exception:  # pylint: disable=broad-except
+            _logger.warning(
+                "Failed to start workflow telemetry", exc_info=True
+            )
+
+        try:
+            yield invocation
+        except Exception as exc:
+            try:
+                self.fail_workflow(
+                    invocation, Error(message=str(exc), type=type(exc))
+                )
+            except Exception:  # pylint: disable=broad-except
+                _logger.warning(
+                    "Failed to record workflow failure", exc_info=True
+                )
+                _safe_detach(invocation)
+            raise
+
+        try:
+            self.stop_workflow(invocation)
+        except Exception:  # pylint: disable=broad-except
+            _logger.warning("Failed to stop workflow telemetry", exc_info=True)
+            _safe_detach(invocation)
+
 
 def get_telemetry_handler(
     tracer_provider: TracerProvider | None = None,

util/opentelemetry-util-genai/src/opentelemetry/util/genai/span_utils.py

@@ -37,6 +37,7 @@
     LLMInvocation,
     MessagePart,
     OutputMessage,
+    WorkflowInvocation,
 )
 from opentelemetry.util.genai.utils import (
     ContentCapturingMode,
@@ -279,6 +280,71 @@ def _get_llm_response_attributes(
     return {key: value for key, value in optional_attrs if value is not None}
 
 
+def _apply_workflow_finish_attributes(
+    span: Span, invocation: WorkflowInvocation
+) -> None:
+    """Apply attributes/messages common to finish() paths."""
+
+    # Build all attributes by reusing the attribute getter functions
+    attributes: dict[str, Any] = {}
+    attributes.update(_get_workflow_common_attributes(invocation))
+    attributes.update(
+        _get_workflow_messages_attributes_for_span(
+            invocation.input_messages,
+            invocation.output_messages,
+        )
+    )
+    attributes.update(invocation.attributes)
+
+    # Set all attributes on the span
+    if attributes:
+        span.set_attributes(attributes)
+
+
+def _get_workflow_common_attributes(
+    invocation: WorkflowInvocation,
+) -> dict[str, Any]:
+    """Get common Workflow attributes shared by finish() and error() paths.
+
+    Returns a dictionary of attributes.
+    """
+    return {
+        GenAI.GEN_AI_OPERATION_NAME: invocation.operation_name,
+    }
+
+
+def _get_workflow_messages_attributes_for_span(
+    input_messages: list[InputMessage],
+    output_messages: list[OutputMessage],
+) -> dict[str, Any]:
+    """Get message attributes formatted for span (JSON string format).
+
+    Returns empty dict if not in experimental mode or content capturing is disabled.
+    """
+    if not is_experimental_mode() or get_content_capturing_mode() not in (
+        ContentCapturingMode.SPAN_ONLY,
+        ContentCapturingMode.SPAN_AND_EVENT,
+    ):
+        return {}
+
+    optional_attrs = (
+        (
+            GenAI.GEN_AI_INPUT_MESSAGES,
+            gen_ai_json_dumps([asdict(m) for m in input_messages])
+            if input_messages
+            else None,
+        ),
+        (
+            GenAI.GEN_AI_OUTPUT_MESSAGES,
+            gen_ai_json_dumps([asdict(m) for m in output_messages])
+            if output_messages
+            else None,
+        ),
+    )
+
+    return {key: value for key, value in optional_attrs if value is not None}
+
+
 __all__ = [
     "_apply_llm_finish_attributes",
     "_apply_error_attributes",
@@ -287,4 +353,6 @@ def _get_llm_response_attributes(
     "_get_llm_response_attributes",
     "_get_llm_span_name",
     "_maybe_emit_llm_event",
+    "_get_workflow_common_attributes",
+    "_apply_workflow_finish_attributes",
 ]

util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py

@@ -201,6 +201,12 @@ class GenAIInvocation:
     context_token: ContextToken | None = None
     span: Span | None = None
     attributes: dict[str, Any] = field(default_factory=_new_str_any_dict)
+    monotonic_start_s: float | None = None
+    """
+    Monotonic start time in seconds (from timeit.default_timer) used
+    for duration calculations to avoid mixing clock sources. This is
+    populated by the TelemetryHandler when starting an invocation.
+    """
 
 
 @dataclass
@@ -267,10 +273,6 @@ class LLMInvocation(GenAIInvocation):
     seed: int | None = None
     server_address: str | None = None
     server_port: int | None = None
-    # Monotonic start time in seconds (from timeit.default_timer) used
-    # for duration calculations to avoid mixing clock sources. This is
-    # populated by the TelemetryHandler when starting an invocation.
-    monotonic_start_s: float | None = None
 
 
 @dataclass