diff --git a/src/google/adk/a2a/converters/event_converter.py b/src/google/adk/a2a/converters/event_converter.py index 2e6904b16bd..0560354dc0c 100644 --- a/src/google/adk/a2a/converters/event_converter.py +++ b/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 ) diff --git a/src/google/adk/a2a/converters/from_adk_event.py b/src/google/adk/a2a/converters/from_adk_event.py index 2f668baadb7..92440fe029c 100644 --- a/src/google/adk/a2a/converters/from_adk_event.py +++ b/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) diff --git a/tests/unittests/a2a/converters/test_event_converter.py b/tests/unittests/a2a/converters/test_event_converter.py index 3e7a3658d6f..fe6ddca0dde 100644 --- a/tests/unittests/a2a/converters/test_event_converter.py +++ b/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.""" diff --git a/tests/unittests/a2a/converters/test_from_adk.py b/tests/unittests/a2a/converters/test_from_adk.py index e1cda390676..95548b437ab 100644 --- a/tests/unittests/a2a/converters/test_from_adk.py +++ b/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."""