Introduce parse_response_to_parts helper

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,