fix(a2a): map Event.output into to_a2a artifacts

src/google/adk/a2a/converters/event_converter.py

@@ -34,6 +34,7 @@
 from ...events.event import Event
 from ...flows.llm_flows.functions import REQUEST_EUC_FUNCTION_CALL_NAME
 from ..experimental import a2a_experimental
+from .from_adk_event import parts_from_event_output
 from .part_converter import A2A_DATA_PART_METADATA_IS_LONG_RUNNING_KEY
 from .part_converter import A2A_DATA_PART_METADATA_TYPE_FUNCTION_CALL
 from .part_converter import A2A_DATA_PART_METADATA_TYPE_KEY
@@ -406,20 +407,31 @@ def convert_event_to_a2a_message(
   if not event:
     raise ValueError("Event cannot be None")
 
-  if not event.content or not event.content.parts:
-    return None
-
   try:
-    output_parts = []
-    for part in event.content.parts:
-      a2a_parts = part_converter(part)
-      if not isinstance(a2a_parts, list):
-        a2a_parts = [a2a_parts] if a2a_parts else []
-      for a2a_part in a2a_parts:
-        output_parts.append(a2a_part)
-        _process_long_running_tool(a2a_part, event)
+    # Prefer Event.output when there is no content, so Workflow output_schema
+    # finals become the A2A message / artifact. When content is present (e.g.
+    # finish_task function-call parts), keep converting content.
+    output_parts: list[A2APart] = []
+    if event.output is not None and (
+        not event.content or not event.content.parts
+    ):
+      output_parts = parts_from_event_output(
+          event.output, part_converter=part_converter
+      )
+    if not output_parts:
+      if not event.content or not event.content.parts:
+        return None
+      for part in event.content.parts:
+        a2a_parts = part_converter(part)
+        if not isinstance(a2a_parts, list):
+          a2a_parts = [a2a_parts] if a2a_parts else []
+        for a2a_part in a2a_parts:
+          if a2a_part is not None:
+            output_parts.append(a2a_part)
 
     if output_parts:
+      for a2a_part in output_parts:
+        _process_long_running_tool(a2a_part, event)
       return Message(
           message_id=platform_uuid.new_uuid(), role=role, parts=output_parts
       )

src/google/adk/a2a/converters/from_adk_event.py

@@ -74,12 +74,86 @@
 """
 
 
+def _serialize_output_value(value: Any) -> Optional[Any]:
+  """JSON-safe serialization for ``Event.output`` values sent over A2A.
+
+  Uses Pydantic JSON mode so types like ``datetime`` / ``Decimal`` become
+  JSON-native values instead of Python objects that break wire encoding.
+  """
+  if value is None:
+    return None
+
+  if hasattr(value, "model_dump"):
+    try:
+      dumped = value.model_dump(mode="json", exclude_none=True, by_alias=True)
+      return dumped if dumped else None
+    except Exception as e:
+      logger.warning("Failed to serialize event.output model: %s", e)
+      return str(value)
+
+  if isinstance(value, dict):
+    return {
+        (k if isinstance(k, str) else str(k)): _serialize_output_value(v)
+        for k, v in value.items()
+    }
+  if isinstance(value, list):
+    return [_serialize_output_value(item) for item in value]
+  if isinstance(value, (int, float, bool, str)):
+    return value
+  return str(value)
+
+
+def parts_from_event_output(
+    output: Any,
+    part_converter: GenAIPartToA2APartConverter = convert_genai_part_to_a2a_part,
+) -> List[A2APart]:
+  """Converts an ADK ``Event.output`` value into A2A parts.
+
+  Structured outputs (Pydantic models / dicts) become DataParts. GenAI
+  ``Content`` outputs are converted part-by-part. Scalars become TextParts.
+  Returns an empty list when ``output`` is None or serializes to nothing.
+  """
+  if output is None:
+    return []
+
+  # GenAI Content (e.g. finish_task unwrapping that sets output=content).
+  parts = getattr(output, "parts", None)
+  if parts is not None and hasattr(output, "role"):
+    output_parts: list[A2APart] = []
+    for part in parts:
+      a2a_parts = part_converter(part)
+      if not isinstance(a2a_parts, list):
+        a2a_parts = [a2a_parts] if a2a_parts else []
+      output_parts.extend(p for p in a2a_parts if p is not None)
+    return output_parts
+
+  if isinstance(output, str):
+    return [_compat.make_text_part(output)]
+
+  serialized = _serialize_output_value(output)
+  if serialized is None:
+    return []
+  if isinstance(serialized, str):
+    return [_compat.make_text_part(serialized)]
+  if isinstance(serialized, dict):
+    return [_compat.make_data_part(data=serialized)]
+  if isinstance(serialized, list):
+    return [_compat.make_data_part(data={"items": serialized})]
+  if isinstance(serialized, (int, float, bool)):
+    return [_compat.make_text_part(str(serialized))]
+  return [_compat.make_text_part(str(serialized))]
+
+
 def _convert_adk_parts_to_a2a_parts(
     event: Event,
     part_converter: GenAIPartToA2APartConverter = convert_genai_part_to_a2a_part,
 ) -> Optional[List[A2APart]]:
   """Converts an ADK event to an A2A parts list.
 
+  Prefers ``event.output`` when set so Workflow ``output_schema`` results
+  (and other structured node outputs) become the A2A artifact instead of
+  being dropped in favor of the last agent node's text ``content``.
+
   Args:
     event: The ADK event to convert.
     part_converter: The function to convert GenAI part to A2A part.
@@ -93,19 +167,33 @@ def _convert_adk_parts_to_a2a_parts(
   if not event:
     raise ValueError("Event cannot be None")
 
-  if not event.content or not event.content.parts:
-    return []
-
   try:
-    output_parts = []
+    # Prefer structured ``event.output`` when the event has no content parts.
+    # Workflow ``output_schema`` finals are emitted as Event(output=...) with
+    # empty content; without this they are dropped and to_a2a() keeps only the
+    # last agent node's text. When content is present (e.g. finish_task FC
+    # parts), keep converting content so tool-call metadata is preserved.
+    if event.output is not None and (
+        not event.content or not event.content.parts
+    ):
+      output_parts = parts_from_event_output(
+          event.output, part_converter=part_converter
+      )
+      if output_parts:
+        return output_parts
+
+    if not event.content or not event.content.parts:
+      return []
+
+    content_parts = []
     for part in event.content.parts:
       a2a_parts = part_converter(part)
       if not isinstance(a2a_parts, list):
         a2a_parts = [a2a_parts] if a2a_parts else []
       for a2a_part in a2a_parts:
-        output_parts.append(a2a_part)
+        content_parts.append(a2a_part)
 
-    return output_parts
+    return content_parts
 
   except Exception as e:
     logger.error("Failed to convert event to status message: %s", e)

tests/unittests/a2a/converters/test_event_converter.py

@@ -66,6 +66,7 @@ def setup_method(self):
     self.mock_event.error_code = None
     self.mock_event.error_message = None
     self.mock_event.content = None
+    self.mock_event.output = None
     self.mock_event.long_running_tool_ids = None
     self.mock_event.actions = None
 
@@ -688,6 +689,86 @@ def test_convert_event_to_a2a_message_with_multiple_parts_returned(self):
     assert _compat.part_text(result.parts[1]) == "part 2"
     mock_convert_part.assert_called_once_with(mock_genai_part)
 
+  def test_convert_event_to_a2a_message_prefers_event_output(self):
+    """Structured Event.output becomes the A2A message over content text."""
+    from google.adk.a2a.converters.event_converter import convert_event_to_a2a_message
+    from google.adk.a2a.executor.task_result_aggregator import TaskResultAggregator
+    from pydantic import BaseModel
+
+    class Report(BaseModel):
+      title: str
+      summary: str
+      note: str
+
+    # Simulate the #6762 stream: agent texts, then workflow structured output.
+    stream = [
+        Event(
+            author="agent_a",
+            content=genai_types.Content(
+                role="model", parts=[genai_types.Part(text='{"title":"T"}')]
+            ),
+            partial=False,
+        ),
+        Event(
+            author="agent_c",
+            content=genai_types.Content(
+                role="model", parts=[genai_types.Part(text="note-text")]
+            ),
+            partial=False,
+        ),
+        Event(
+            author="wf",
+            output=Report(title="T", summary="S", note="note-text"),
+            partial=False,
+        ),
+    ]
+
+    aggregator = TaskResultAggregator()
+    for event in stream:
+      for a2a_event in convert_event_to_a2a_events(
+          event,
+          self.mock_invocation_context,
+          task_id="task-1",
+          context_id="ctx-1",
+      ):
+        aggregator.process_event(a2a_event)
+
+    parts = aggregator.task_status_message.parts
+    assert parts and _compat.is_data_part(parts[0])
+    assert _compat.data_part_dict(parts[0]) == {
+        "title": "T",
+        "summary": "S",
+        "note": "note-text",
+    }
+
+    # Direct message conversion uses output when content is absent.
+    output_only = Event(
+        author="wf",
+        output=Report(title="T", summary="S", note="note-text"),
+        partial=False,
+    )
+    message = convert_event_to_a2a_message(
+        output_only, self.mock_invocation_context
+    )
+    assert message is not None
+    assert _compat.is_data_part(message.parts[0])
+    assert _compat.data_part_dict(message.parts[0])["title"] == "T"
+
+    # Content still wins when both are present (e.g. finish_task FC parts).
+    mixed = Event(
+        author="wf",
+        content=genai_types.Content(
+            role="model", parts=[genai_types.Part(text="kept-content")]
+        ),
+        output=Report(title="T", summary="S", note="note-text"),
+        partial=False,
+    )
+    mixed_message = convert_event_to_a2a_message(
+        mixed, self.mock_invocation_context
+    )
+    assert mixed_message is not None
+    assert _compat.part_text(mixed_message.parts[0]) == "kept-content"
+
 
 class TestA2AToEventConverters:
   """Test suite for A2A to Event conversion functions."""

tests/unittests/a2a/converters/test_from_adk.py

@@ -37,6 +37,7 @@ def setup_method(self):
     self.mock_event.author = "test-author"
     self.mock_event.branch = None
     self.mock_event.content = None
+    self.mock_event.output = None
     self.mock_event.error_code = None
     self.mock_event.error_message = None
     self.mock_event.grounding_metadata = None
@@ -126,6 +127,52 @@ def test_convert_event_to_a2a_events_with_actions(self):
     assert "adk_actions" in metadata
     assert metadata["adk_actions"]["artifactDelta"] == {"image": 0}
 
+  def test_convert_event_prefers_structured_output_over_content(self):
+    """Workflow output_schema results must become the A2A artifact (#6762).
+
+    When a later Event carries structured ``output`` (and no content), that
+    value must be published as a DataPart artifact — not dropped in favor of
+    earlier agent text left in the task history.
+    """
+    from pydantic import BaseModel
+
+    class Report(BaseModel):
+      title: str
+      note: str
+
+    agents_artifacts: dict[str, str] = {}
+    text_event = Event(
+        author="last_agent",
+        content=genai_types.Content(
+            role="model", parts=[genai_types.Part(text="note-only")]
+        ),
+        partial=False,
+    )
+    workflow_event = Event(
+        author="wf",
+        output=Report(title="T", note="note-only"),
+        partial=False,
+    )
+
+    text_a2a = convert_event_to_a2a_events(
+        text_event, agents_artifacts, task_id="t", context_id="c"
+    )
+    assert len(text_a2a) == 1
+    assert isinstance(text_a2a[0], TaskArtifactUpdateEvent)
+
+    output_a2a = convert_event_to_a2a_events(
+        workflow_event, agents_artifacts, task_id="t", context_id="c"
+    )
+    assert len(output_a2a) == 1
+    assert isinstance(output_a2a[0], TaskArtifactUpdateEvent)
+    parts = output_a2a[0].artifact.parts
+    assert len(parts) == 1
+    assert _compat.is_data_part(parts[0])
+    assert _compat.data_part_dict(parts[0]) == {
+        "title": "T",
+        "note": "note-only",
+    }
+
 
 class TestSerializeValue:
   """Tests for _serialize_value preserving JSON-native types."""