Python: Capture workflow telemetry input and output

python/packages/core/agent_framework/_workflows/_executor.py

@@ -11,7 +11,11 @@
 from collections.abc import Awaitable, Callable
 from typing import Any, TypeVar, overload
 
-from ..observability import create_processing_span
+from ..observability import (
+    OtelAttr,
+    _set_sensitive_span_attributes,  # pyright: ignore[reportPrivateUsage]
+    create_processing_span,
+)
 from ._events import (
     WorkflowErrorDetails,
     WorkflowEvent,
@@ -277,7 +281,7 @@ async def execute(
                 type(message).__name__,
                 source_trace_contexts=trace_contexts,
                 source_span_ids=source_span_ids,
-            ):
+            ) as span:
                 # Find the handler and handler spec that matches the message type.
                 handler = self._find_handler(message)
 
@@ -286,6 +290,13 @@ async def execute(
                     # Unwrap raw data for handler call
                     message = message.data
 
+                _set_sensitive_span_attributes(
+                    span,
+                    message,
+                    (OtelAttr.EXECUTOR_INPUT, OtelAttr.INPUT_VALUE),
+                    (OtelAttr.INPUT_MIME_TYPE,),
+                )
+
                 # Create the appropriate WorkflowContext based on handler specs
                 context = self._create_context_for_handler(
                     source_executor_ids=source_executor_ids,
@@ -316,6 +327,13 @@ async def execute(
                     sent_messages = context.get_sent_messages()
                     yielded_outputs = context.get_yielded_outputs()
                     completion_data = sent_messages + yielded_outputs
+                    if completion_data:
+                        _set_sensitive_span_attributes(
+                            span,
+                            completion_data,
+                            (OtelAttr.EXECUTOR_OUTPUT, OtelAttr.OUTPUT_VALUE),
+                            (OtelAttr.OUTPUT_MIME_TYPE,),
+                        )
                     completed_event = WorkflowEvent.executor_completed(
                         self.id, completion_data if completion_data else None
                     )

python/packages/core/agent_framework/_workflows/_workflow.py

@@ -19,7 +19,12 @@
 from .._sessions import ContextProvider
 from .._types import ResponseStream
 from ..exceptions import WorkflowException
-from ..observability import OtelAttr, capture_exception, create_workflow_span
+from ..observability import (
+    OtelAttr,
+    _set_sensitive_span_attributes,  # pyright: ignore[reportPrivateUsage]
+    capture_exception,
+    create_workflow_span,
+)
 from ._checkpoint import CheckpointStorage
 from ._const import DEFAULT_MAX_ITERATIONS, GLOBAL_KWARGS_KEY, INTERNAL_SOURCE_ID, WORKFLOW_RUN_KWARGS_KEY
 from ._edge import (
@@ -478,6 +483,7 @@ def get_executors_list(self) -> list[Executor]:
     async def _run_workflow_with_tracing(
         self,
         initial_executor_fn: Callable[[], Awaitable[None]] | None = None,
+        telemetry_input: Any | None = None,
         is_continuation: bool = False,
         streaming: bool = False,
         function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None,
@@ -490,6 +496,7 @@ async def _run_workflow_with_tracing(
 
         Args:
             initial_executor_fn: Optional function to execute initial executor.
+            telemetry_input: Input payload for this run, captured only when sensitive telemetry is enabled.
             is_continuation: True when this run is a continuation of prior
                 work (a checkpoint restore or a responses-only replay) rather
                 than a fresh new turn delivered via the start executor with
@@ -517,9 +524,20 @@ async def _run_workflow_with_tracing(
             OtelAttr.WORKFLOW_RUN_SPAN,
             attributes,
         ) as span:
+            from ..observability import OBSERVABILITY_SETTINGS
+
+            capture_workflow_io = OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and span.is_recording()
             saw_request = False
             emitted_in_progress_pending = False
+            workflow_outputs: list[Any] | None = [] if capture_workflow_io else None
             try:
+                if capture_workflow_io and telemetry_input is not None:
+                    _set_sensitive_span_attributes(
+                        span,
+                        telemetry_input,
+                        (OtelAttr.INPUT_VALUE,),
+                        (OtelAttr.INPUT_MIME_TYPE,),
+                    )
                 # Add workflow started event (telemetry + surface state to consumers)
                 span.add_event(OtelAttr.WORKFLOW_STARTED)
                 # Emit explicit start/status events to the stream
@@ -577,6 +595,8 @@ async def _run_workflow_with_tracing(
                     # Track request events for final status determination
                     if event.type == "request_info":
                         saw_request = True
+                    elif workflow_outputs is not None and event.type == "output":
+                        workflow_outputs.append(event.data)
                     yield event
 
                     if event.type == "request_info" and not emitted_in_progress_pending:
@@ -597,6 +617,13 @@ async def _run_workflow_with_tracing(
                         terminal_status = WorkflowEvent.status(self._status)
                     yield terminal_status
 
+                if workflow_outputs:
+                    _set_sensitive_span_attributes(
+                        span,
+                        workflow_outputs,
+                        (OtelAttr.OUTPUT_VALUE,),
+                        (OtelAttr.OUTPUT_MIME_TYPE,),
+                    )
                 span.add_event(OtelAttr.WORKFLOW_COMPLETED)
             except Exception as exc:
                 # Drain any pending events (for example, executor_failed) before yielding failed event
@@ -872,9 +899,13 @@ async def _run_core(
                     )
 
             initial_executor_fn = self._resolve_execution_mode(message, responses, checkpoint_id, checkpoint_storage)
+            telemetry_input = message.data if isinstance(message, WorkflowMessage) else message
+            if telemetry_input is None:
+                telemetry_input = responses
 
             async for event in self._run_workflow_with_tracing(
                 initial_executor_fn=initial_executor_fn,
+                telemetry_input=telemetry_input,
                 is_continuation=(message is None),
                 streaming=streaming,
                 function_invocation_kwargs=function_invocation_kwargs,

python/packages/core/agent_framework/_workflows/_workflow_context.py

@@ -13,7 +13,11 @@
 from opentelemetry.trace import SpanKind
 from typing_extensions import Never, TypeVar
 
-from ..observability import OtelAttr, create_workflow_span
+from ..observability import (
+    OtelAttr,
+    _set_sensitive_span_attributes,  # pyright: ignore[reportPrivateUsage]
+    create_workflow_span,
+)
 from ._events import (
     WorkflowEvent,
     WorkflowEventSource,
@@ -327,10 +331,20 @@ async def send_message(self, message: OutT, target_id: str | None = None) -> Non
         from ..observability import OBSERVABILITY_SETTINGS
 
         # Create publishing span (inherits current trace context automatically)
-        attributes: dict[str, str] = {OtelAttr.MESSAGE_TYPE: type(message).__name__}
+        attributes: dict[str, str] = {
+            OtelAttr.MESSAGE_TYPE: type(message).__name__,
+            OtelAttr.MESSAGE_SOURCE_ID: self._executor_id,
+        }
         if target_id:
             attributes[OtelAttr.MESSAGE_DESTINATION_EXECUTOR_ID] = target_id
+            attributes[OtelAttr.MESSAGE_TARGET_ID] = target_id
         with create_workflow_span(OtelAttr.MESSAGE_SEND_SPAN, attributes, kind=SpanKind.PRODUCER) as span:
+            _set_sensitive_span_attributes(
+                span,
+                message,
+                (OtelAttr.MESSAGE_CONTENT, OtelAttr.INPUT_VALUE),
+                (OtelAttr.INPUT_MIME_TYPE,),
+            )
             # Create Message wrapper
             msg = WorkflowMessage(data=message, source_id=self._executor_id, target_id=target_id)
 

python/packages/core/agent_framework/observability.py

@@ -44,6 +44,7 @@
 from . import __version__ as version_info
 from ._serialization import (
     _is_serialization_protocol,  # pyright: ignore[reportPrivateUsage]
+    make_json_safe,
 )
 from ._settings import load_settings
 
@@ -304,6 +305,8 @@ class OtelAttr(str, Enum):
     EXECUTOR_PROCESS_SPAN = "executor.process"
     EXECUTOR_ID = "executor.id"
     EXECUTOR_TYPE = "executor.type"
+    EXECUTOR_INPUT = "executor.input"
+    EXECUTOR_OUTPUT = "executor.output"
     # Edge group attributes
     EDGE_GROUP_PROCESS_SPAN = "edge_group.process"
     EDGE_GROUP_TYPE = "edge_group.type"
@@ -317,6 +320,15 @@ class OtelAttr(str, Enum):
     MESSAGE_TYPE = "message.type"
     MESSAGE_PAYLOAD_TYPE = "message.payload_type"
     MESSAGE_DESTINATION_EXECUTOR_ID = "message.destination_executor_id"
+    MESSAGE_CONTENT = "message.content"
+
+    # OpenInference attributes for vendor-neutral input/output ingestion.
+    # https://arize-ai.github.io/openinference/spec/semantic_conventions.html
+    INPUT_VALUE = "input.value"
+    INPUT_MIME_TYPE = "input.mime_type"
+    OUTPUT_VALUE = "output.value"
+    OUTPUT_MIME_TYPE = "output.mime_type"
+    JSON_MIME_TYPE = "application/json"
 
     # Activity events
     EVENT_NAME = "event.name"
@@ -356,6 +368,32 @@ def __str__(self) -> str:
         return self.value
 
 
+def _serialize_for_telemetry(value: Any) -> str:
+    """Serialize heterogeneous telemetry payloads without affecting application execution."""
+    try:
+        return json.dumps(make_json_safe(value), ensure_ascii=False, allow_nan=False)
+    except Exception:
+        value_type = f"{type(value).__module__}.{type(value).__qualname__}"
+        return json.dumps(f"[Unserializable: {value_type}]", ensure_ascii=False)
+
+
+def _set_sensitive_span_attributes(  # pyright: ignore[reportUnusedFunction]
+    span: trace.Span,
+    value: Any,
+    value_attributes: Sequence[str | OtelAttr],
+    mime_type_attributes: Sequence[str | OtelAttr] = (),
+) -> None:
+    """Set serialized payload attributes only when sensitive telemetry is enabled."""
+    if not OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED or not span.is_recording():
+        return
+
+    serialized_value = _serialize_for_telemetry(value)
+    for attribute in value_attributes:
+        span.set_attribute(attribute, serialized_value)
+    for attribute in mime_type_attributes:
+        span.set_attribute(attribute, OtelAttr.JSON_MIME_TYPE)
+
+
 ROLE_EVENT_MAP = {
     "system": OtelAttr.SYSTEM_MESSAGE,
     "user": OtelAttr.USER_MESSAGE,

python/packages/core/tests/workflow/test_workflow_observability.py

@@ -98,6 +98,17 @@ def processed_messages(self) -> list[Any]:
         return self._processed_messages
 
 
+class OutputExecutor(Executor):
+    """Executor that yields a structured workflow output for telemetry tests."""
+
+    def __init__(self, id: str = "output_executor") -> None:
+        super().__init__(id=id)
+
+    @handler
+    async def handle_message(self, message: dict[str, int], ctx: WorkflowContext[Any, dict[str, int]]) -> None:
+        await ctx.yield_output({"result": message["value"] + 1})
+
+
 async def test_span_creation_and_attributes(span_exporter: InMemorySpanExporter) -> None:
     """Test creation and attributes of all span types (workflow, processing, sending)."""
     # Create a mock workflow object
@@ -229,6 +240,122 @@ async def test_trace_context_handling(span_exporter: InMemorySpanExporter) -> No
     assert processing_span.attributes.get("message.payload_type") == "str"
 
 
+@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True)
+async def test_workflow_payloads_are_captured_when_sensitive_data_enabled(
+    span_exporter: InMemorySpanExporter,
+) -> None:
+    workflow = WorkflowBuilder(start_executor=OutputExecutor()).build()
+    span_exporter.clear()
+
+    workflow_message = WorkflowMessage(data={"value": 1}, source_id="external", target_id=None)
+    async for _ in workflow.run(workflow_message, stream=True):
+        pass
+
+    spans = span_exporter.get_finished_spans()
+    workflow_span = next(span for span in spans if span.name == OtelAttr.WORKFLOW_RUN_SPAN)
+    executor_span = next(span for span in spans if span.name == "executor.process output_executor")
+
+    assert workflow_span.attributes is not None
+    assert workflow_span.attributes[OtelAttr.INPUT_VALUE] == '{"value": 1}'
+    assert workflow_span.attributes[OtelAttr.OUTPUT_VALUE] == '[{"result": 2}]'
+    assert workflow_span.attributes[OtelAttr.INPUT_MIME_TYPE] == OtelAttr.JSON_MIME_TYPE
+    assert workflow_span.attributes[OtelAttr.OUTPUT_MIME_TYPE] == OtelAttr.JSON_MIME_TYPE
+
+    assert executor_span.attributes is not None
+    assert executor_span.attributes[OtelAttr.EXECUTOR_INPUT] == '{"value": 1}'
+    assert executor_span.attributes[OtelAttr.EXECUTOR_OUTPUT] == '[{"result": 2}]'
+    assert executor_span.attributes[OtelAttr.INPUT_VALUE] == '{"value": 1}'
+    assert executor_span.attributes[OtelAttr.OUTPUT_VALUE] == '[{"result": 2}]'
+
+
+@pytest.mark.parametrize("enable_sensitive_data", [False], indirect=True)
+async def test_workflow_payloads_are_omitted_when_sensitive_data_disabled(
+    span_exporter: InMemorySpanExporter,
+) -> None:
+    workflow = WorkflowBuilder(start_executor=OutputExecutor()).build()
+    span_exporter.clear()
+
+    async for _ in workflow.run({"value": 1}, stream=True):
+        pass
+
+    payload_attributes = {
+        OtelAttr.EXECUTOR_INPUT,
+        OtelAttr.EXECUTOR_OUTPUT,
+        OtelAttr.MESSAGE_CONTENT,
+        OtelAttr.INPUT_VALUE,
+        OtelAttr.OUTPUT_VALUE,
+        OtelAttr.INPUT_MIME_TYPE,
+        OtelAttr.OUTPUT_MIME_TYPE,
+    }
+    for span in span_exporter.get_finished_spans():
+        assert span.attributes is not None
+        assert payload_attributes.isdisjoint(span.attributes)
+
+
+@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True)
+async def test_message_send_captures_content_and_routing(span_exporter: InMemorySpanExporter) -> None:
+    workflow_ctx: WorkflowContext[str] = WorkflowContext(
+        MockExecutor("source_executor"),
+        ["source"],
+        State(),
+        InProcRunnerContext(),
+    )
+
+    await workflow_ctx.send_message("hello", target_id="target_executor")
+
+    sending_span = next(span for span in span_exporter.get_finished_spans() if span.name == OtelAttr.MESSAGE_SEND_SPAN)
+    assert sending_span.attributes is not None
+    assert sending_span.attributes[OtelAttr.MESSAGE_SOURCE_ID] == "source_executor"
+    assert sending_span.attributes[OtelAttr.MESSAGE_TARGET_ID] == "target_executor"
+    assert sending_span.attributes[OtelAttr.MESSAGE_CONTENT] == '"hello"'
+    assert sending_span.attributes[OtelAttr.INPUT_VALUE] == '"hello"'
+    assert sending_span.attributes[OtelAttr.INPUT_MIME_TYPE] == OtelAttr.JSON_MIME_TYPE
+
+
+@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True)
+async def test_message_send_telemetry_serialization_cannot_break_workflow(
+    span_exporter: InMemorySpanExporter,
+) -> None:
+    class UnserializablePayload:
+        __slots__ = ()
+
+        def __str__(self) -> str:
+            raise RuntimeError("cannot stringify")
+
+    workflow_ctx: WorkflowContext[Any] = WorkflowContext(
+        MockExecutor("source_executor"),
+        ["source"],
+        State(),
+        InProcRunnerContext(),
+    )
+
+    await workflow_ctx.send_message(UnserializablePayload())
+
+    sending_span = next(span for span in span_exporter.get_finished_spans() if span.name == OtelAttr.MESSAGE_SEND_SPAN)
+    assert sending_span.attributes is not None
+    assert "[Unserializable:" in str(sending_span.attributes[OtelAttr.MESSAGE_CONTENT])
+
+
+@pytest.mark.parametrize("non_finite_value", [float("nan"), float("inf"), float("-inf")])
+@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True)
+async def test_message_send_telemetry_rejects_non_finite_json_values(
+    span_exporter: InMemorySpanExporter,
+    non_finite_value: float,
+) -> None:
+    workflow_ctx: WorkflowContext[float] = WorkflowContext(
+        MockExecutor("source_executor"),
+        ["source"],
+        State(),
+        InProcRunnerContext(),
+    )
+
+    await workflow_ctx.send_message(non_finite_value)
+
+    sending_span = next(span for span in span_exporter.get_finished_spans() if span.name == OtelAttr.MESSAGE_SEND_SPAN)
+    assert sending_span.attributes is not None
+    assert sending_span.attributes[OtelAttr.MESSAGE_CONTENT] == '"[Unserializable: builtins.float]"'
+
+
 @pytest.mark.parametrize("enable_instrumentation", [False], indirect=True)
 async def test_trace_context_disabled_when_tracing_disabled(
     enable_instrumentation: bool, span_exporter: InMemorySpanExporter