Introduce parse_response_to_parts helper (#785)

* Address review comments

* feat: Introduce parse_response_to_parts helper

In the samples, agent.py parses the LLM response and extracts the text
part and A2UI JSON part. It also validates the JSON part. Then it sends
the original LLM response back to agent_executor.py, which parses again.

This commit consolidates the logic to only parse and validate once by
leveraging a parse_response_to_parts helper.

- tested: the orchestrator sample and all sub-agents worked from
  end-to-end.

* Address review comments

Author: nan-yu

agent_sdks/python/agent_development.md

@@ -116,7 +116,23 @@ After parsing and validating the A2UI JSON payloads, wrap them in an A2A DataPar
 
 To ensure the A2UI Renderers on the frontend recognize the data, add `{"mimeType": "application/json+a2ui"}` to the DataPart's metadata.
 
-**Recommendation:** Use the [create_a2ui_datapart](src/a2ui/a2a.py#L37-L54) helper method to convert A2UI JSON payloads into an A2A DataPart.
+**Recommendation:** Use the [create_a2ui_part](src/a2ui/a2a.py) helper method to convert A2UI JSON payloads into an A2A DataPart.
+
+#### 4d. Complete Agent Output Structure
+
+The most efficient way to generate structured agent output is to use the `parse_response_to_parts` helper. It handles splitting the text, extracting A2UI JSON, optional validation, and wrapping everything into A2A `Part` objects.
+
+```python
+from a2ui.a2a import parse_response_to_parts
+
+# Inside your agent's stream method:
+final_response_content = f"{text_segment}\n{A2UI_DELIMITER}\n{json_payload}"
+
+yield {
+    "is_task_complete": True,
+    "parts": parse_response_to_parts(final_response_content, fallback_text="OK."),
+}
+```
 
 ## Use Cases
 

agent_sdks/python/src/a2ui/a2a.py

@@ -16,7 +16,7 @@
 from typing import Any, Optional, List
 
 from a2a.server.agent_execution import RequestContext
-from a2a.types import AgentExtension, Part, DataPart
+from a2a.types import AgentExtension, Part, DataPart, TextPart
 
 logger = logging.getLogger(__name__)
 
@@ -106,6 +106,49 @@ def get_a2ui_agent_extension(
   )
 
 
+def parse_response_to_parts(
+    content: str,
+    validator: Optional[Any] = None,
+    fallback_text: Optional[str] = None,
+) -> List[Part]:
+  """Helper to parse LLM response content into A2A Parts, with optional validation.
+
+  Args:
+      content: The LLM response content, potentially containing A2UI delimiters.
+      validator: Optional validator to run against extracted JSON payloads.
+      fallback_text: Optional text to return if no parts are successfully created.
+
+  Returns:
+      A list of A2A Part objects (TextPart and/or DataPart).
+  """
+  from a2ui.core.parser import parse_response
+
+  parts = []
+  try:
+    text_part, json_data = parse_response(content)
+
+    if text_part:
+      parts.append(Part(root=TextPart(text=text_part.strip())))
+
+    if validator and json_data is not None and json_data != []:
+      validator.validate(json_data)
+
+    if json_data:
+      if isinstance(json_data, list):
+        for message in json_data:
+          parts.append(create_a2ui_part(message))
+      else:
+        parts.append(create_a2ui_part(json_data))
+
+  except Exception as e:
+    logger.warning(f"Failed to parse or validate A2UI response: {e}")
+
+  if not parts and fallback_text:
+    parts.append(Part(root=TextPart(text=fallback_text)))
+
+  return parts
+
+
 def try_activate_a2ui_extension(context: RequestContext) -> bool:
   """Activates the A2UI extension if requested.
 

agent_sdks/python/src/a2ui/adk/a2a_extension/send_a2ui_to_client_toolset.py

@@ -104,7 +104,11 @@ async def get_examples(ctx: ReadonlyContext) -> str:
 import jsonschema
 
 from a2a import types as a2a_types
-from ...a2a import create_a2ui_part
+from a2ui.a2a import (
+    A2UI_EXTENSION_URI,
+    create_a2ui_part,
+    parse_response_to_parts,
+)
 from a2ui.core.parser import parse_response, parse_and_fix
 from a2ui.core.schema.catalog import A2uiCatalog
 from a2ui.core.schema.constants import A2UI_DELIMITER
@@ -384,42 +388,12 @@ def convert(self, part: genai_types.Part) -> list[a2a_types.Part]:
     # 3. Handle Text-based A2UI (TextPart)
     if text := part.text:
       if A2UI_DELIMITER in text:
-        return self._convert_text_with_a2ui(text)
+        return parse_response_to_parts(text, validator=self._catalog.validator)
 
     # 4. Default conversion for other parts
     converted_part = part_converter.convert_genai_part_to_a2a_part(part)
     return [converted_part] if converted_part else []
 
-  def _convert_text_with_a2ui(self, text: str) -> list[a2a_types.Part]:
-    """Helper to split text and extract/validate A2UI JSON."""
-    parts = []
-    try:
-      text_part, json_data = parse_response(text)
-      self._catalog.validator.validate(json_data)
-
-      if text_part:
-        parts.append(
-            a2a_types.Part(root=a2a_types.TextPart(kind="text", text=text_part))
-        )
-
-      logger.info(f"Found {len(json_data)} messages. Creating individual DataParts.")
-      for message in json_data:
-        parts.append(create_a2ui_part(message))
-
-    except Exception as e:
-      logger.error(f"Failed to parse or validate text-based A2UI JSON: {e}")
-      # Fallback: at least try to return the leading text part if we can split it
-      if not parts:
-        segments = text.split(A2UI_DELIMITER, 1)
-        if segments[0].strip():
-          parts.append(
-              a2a_types.Part(
-                  root=a2a_types.TextPart(kind="text", text=segments[0].strip())
-              )
-          )
-
-    return parts
-
 
 @experimental
 class A2uiEventConverter:

agent_sdks/python/tests/adk/a2a_extension/test_send_a2ui_to_client_toolset.py

@@ -376,24 +376,6 @@ def test_converter_class_convert_text_with_a2ui():
   catalog_mock.validator.validate.assert_called_once_with(valid_a2ui)
 
 
-def test_converter_class_convert_text_multiple_segments():
-  catalog_mock = MagicMock(spec=A2uiCatalog)
-  converter = A2uiPartConverter(catalog_mock)
-
-  ui1 = [{"type": "Text", "text": "one"}]
-  ui2 = [{"type": "Text", "text": "two"}]
-
-  text = f"Intro{A2UI_DELIMITER}{json.dumps(ui1)}{A2UI_DELIMITER}Middle{A2UI_DELIMITER}{json.dumps(ui2)}"
-
-  part = genai_types.Part(text=text)
-  a2a_parts = converter.convert(part)
-
-  # parse_response with single segment assumption will fail to parse the whole trailing string as JSON.
-  # But our robust _convert_text_with_a2ui will still return the leading text part.
-  assert len(a2a_parts) == 1
-  assert a2a_parts[0].root.text == "Intro"
-
-
 def test_converter_class_convert_text_empty_leading():
   catalog_mock = MagicMock(spec=A2uiCatalog)
   converter = A2uiPartConverter(catalog_mock)
@@ -435,10 +417,7 @@ def test_converter_class_convert_text_with_invalid_a2ui():
   part = genai_types.Part(text=text)
 
   a2a_parts = converter.convert(part)
-
-  # Expect only 1 part: the leading TextPart. The invalid A2UI is skipped.
-  assert len(a2a_parts) == 1
-  assert a2a_parts[0].root.text == "Here is the UI:"
+  assert len(a2a_parts) == 0
 
 
 def test_converter_class_convert_other_part():

samples/agent/adk/component_gallery/agent.py

@@ -1,9 +1,13 @@
 """Agent logic for the Component Gallery."""
 
 import logging
-import json
 from collections.abc import AsyncIterable
 from typing import Any
+import json
+
+from a2a.types import DataPart, Part, TextPart
+from a2ui.core.schema.constants import A2UI_DELIMITER
+from a2ui.a2a import create_a2ui_part, parse_response_to_parts
 
 import asyncio
 import datetime
@@ -29,10 +33,9 @@ async def stream(self, query: str, session_id: str) -> AsyncIterable[dict[str, A
       gallery_json = get_gallery_json()
       yield {
           "is_task_complete": True,
-          "payload": {
-              "text": "Here is the component gallery.",
-              "json_string": gallery_json,
-          },
+          "parts": parse_response_to_parts(
+              f"Here is the component gallery.\n{A2UI_DELIMITER}\n{gallery_json}"
+          ),
       }
       return
 
@@ -46,7 +49,7 @@ async def stream(self, query: str, session_id: str) -> AsyncIterable[dict[str, A
 
       timestamp = datetime.datetime.now().strftime("%H:%M:%S")
 
-      response_update = [{
+      response_update = {
           "surfaceUpdate": {
               "surfaceId": "response-surface",
               "components": [{
@@ -63,16 +66,19 @@ async def stream(self, query: str, session_id: str) -> AsyncIterable[dict[str, A
                   },
               }],
           }
-      }]
+      }
 
       yield {
           "is_task_complete": True,
-          "payload": {"text": "Action processed.", "json_data": response_update},
+          "parts": [
+              Part(root=TextPart(text="Action processed.")),
+              create_a2ui_part(response_update),
+          ],
       }
       return
 
     # Fallback for text
     yield {
         "is_task_complete": True,
-        "payload": {"text": "I am the Component Gallery Agent."},
+        "parts": [Part(root=TextPart(text="I am the Component Gallery Agent."))],
     }

samples/agent/adk/component_gallery/agent_executor.py

@@ -8,8 +8,7 @@
 from a2a.types import (DataPart, Part, TaskState, TextPart)
 from a2a.utils import new_agent_parts_message, new_task
 from agent import ComponentGalleryAgent
-from a2ui.a2a import create_a2ui_part, try_activate_a2ui_extension
-from a2ui.core.parser import parse_response
+from a2ui.a2a import try_activate_a2ui_extension
 
 logger = logging.getLogger(__name__)
 
@@ -50,46 +49,7 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non
     updater = TaskUpdater(event_queue, task.id, task.context_id)
 
     async for item in self.agent.stream(query, task.context_id):
-      final_parts = []
-
-      if "payload" in item:
-        payload = item["payload"]
-        text = payload.get("text")
-        if text:
-          final_parts.append(Part(root=TextPart(text=text)))
-
-        json_data = payload.get("json_data")
-        json_string = payload.get("json_string")
-
-        if json_string:
-          try:
-            json_data = json.loads(json_string)
-          except Exception as e:
-            logger.error(f"Failed to parse JSON string: {e}")
-
-        if json_data:
-          if isinstance(json_data, list):
-            for msg in json_data:
-              final_parts.append(create_a2ui_part(msg))
-          else:
-            final_parts.append(create_a2ui_part(json_data))
-      else:
-        content = item.get("content", "")
-        try:
-          text_part, json_data = parse_response(content)
-          if text_part.strip():
-            final_parts.append(Part(root=TextPart(text=text_part.strip())))
-
-          if json_data:
-            if isinstance(json_data, list):
-              for msg in json_data:
-                final_parts.append(create_a2ui_part(msg))
-            else:
-              final_parts.append(create_a2ui_part(json_data))
-        except (ValueError, json.JSONDecodeError) as e:
-          logger.error(f"Failed to parse response: {content}, {e}")
-          if content:
-            final_parts.append(Part(root=TextPart(text=content)))
+      final_parts = item["parts"]
 
       await updater.update_status(
           TaskState.completed,