Add workflow type in genai utils (#4310)

* Add workflow type in genai utils

* fixed errors

* fixed errors

* fixed test

* fixed test

* fixed precommit

* moved to unreleased

* reordered

Author: wrisa

util/opentelemetry-util-genai/CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Populate schema_url on metrics
   ([#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))
 
 ## Version 0.3b0 (2026-02-20)
 

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

@@ -203,6 +203,23 @@ class GenAIInvocation:
     attributes: dict[str, Any] = field(default_factory=_new_str_any_dict)
 
 
+@dataclass
+class WorkflowInvocation(GenAIInvocation):
+    """
+    Represents predetermined static sequence of operations eg: Agent, LLM, tool, and retrieval invocations.
+    A workflow groups multiple operations together, accepting input(s) and producing final output(s).
+    """
+
+    name: str = ""
+    operation_name: str = "invoke_workflow"
+    input_messages: list[InputMessage] = field(
+        default_factory=_new_input_messages
+    )
+    output_messages: list[OutputMessage] = field(
+        default_factory=_new_output_messages
+    )
+
+
 @dataclass
 class LLMInvocation(GenAIInvocation):
     """

util/opentelemetry-util-genai/tests/test_workflow_invocation.py

@@ -0,0 +1,75 @@
+from opentelemetry.util.genai.types import (
+    InputMessage,
+    OutputMessage,
+    Text,
+    WorkflowInvocation,
+)
+
+
+class TestWorkflowInvocation:  # pylint: disable=no-self-use
+    def test_default_values(self):
+        invocation = WorkflowInvocation()
+        assert invocation.name == ""
+        assert invocation.operation_name == "invoke_workflow"
+        assert not invocation.input_messages
+        assert not invocation.output_messages
+        assert invocation.span is None
+        assert invocation.context_token is None
+        assert not invocation.attributes
+
+    def test_custom_name(self):
+        invocation = WorkflowInvocation(name="customer_support_pipeline")
+        assert invocation.name == "customer_support_pipeline"
+
+    def test_custom_operation_name(self):
+        invocation = WorkflowInvocation(operation_name="run_pipeline")
+        assert invocation.operation_name == "run_pipeline"
+
+    def test_with_input_messages(self):
+        msg = InputMessage(role="user", parts=[Text(content="hello")])
+        invocation = WorkflowInvocation(input_messages=[msg])
+        assert len(invocation.input_messages) == 1
+        assert invocation.input_messages[0].role == "user"
+
+    def test_with_output_messages(self):
+        msg = OutputMessage(
+            role="assistant", parts=[Text(content="hi")], finish_reason="stop"
+        )
+        invocation = WorkflowInvocation(output_messages=[msg])
+        assert len(invocation.output_messages) == 1
+        assert invocation.output_messages[0].finish_reason == "stop"
+
+    def test_inherits_genai_invocation(self):
+        invocation = WorkflowInvocation(attributes={"key": "value"})
+        assert invocation.attributes == {"key": "value"}
+
+    def test_default_lists_are_independent(self):
+        """Ensure default factory creates separate list instances."""
+        inv1 = WorkflowInvocation()
+        inv2 = WorkflowInvocation()
+        inv1.input_messages.append(InputMessage(role="user", parts=[]))
+        assert len(inv2.input_messages) == 0
+
+    def test_default_attributes_are_independent(self):
+        inv1 = WorkflowInvocation()
+        inv2 = WorkflowInvocation()
+        inv1.attributes["foo"] = "bar"
+        assert "foo" not in inv2.attributes
+
+    def test_full_construction(self):
+        inp = InputMessage(role="user", parts=[Text(content="query")])
+        out = OutputMessage(
+            role="assistant",
+            parts=[Text(content="answer")],
+            finish_reason="stop",
+        )
+        invocation = WorkflowInvocation(
+            name="my_workflow",
+            operation_name="invoke_workflow",
+            input_messages=[inp],
+            output_messages=[out],
+        )
+        assert invocation.name == "my_workflow"
+        assert len(invocation.input_messages) == 1
+        assert len(invocation.output_messages) == 1
+        assert invocation.output_messages[0].parts[0].content == "answer"