refactored artifact and message; added parts utils

Author: martimfasantos

src/a2a/utils/__init__.py

@@ -4,6 +4,17 @@
     new_artifact,
     new_data_artifact,
     new_text_artifact,
+    get_artifact_text,
+)
+from a2a.utils.message import (
+    new_agent_text_message,
+    new_agent_parts_message,
+    get_message_text,
+)
+from a2a.utils.parts import (
+    get_data_parts,
+    get_file_parts,
+    get_text_parts,
 )
 from a2a.utils.constants import (
     AGENT_CARD_WELL_KNOWN_PATH,
@@ -17,14 +28,6 @@
     build_text_artifact,
     create_task_obj,
 )
-from a2a.utils.message import (
-    get_data_parts,
-    get_file_parts,
-    get_message_text,
-    get_text_parts,
-    new_agent_parts_message,
-    new_agent_text_message,
-)
 from a2a.utils.task import (
     completed_task,
     new_task,
@@ -41,6 +44,7 @@
     'build_text_artifact',
     'completed_task',
     'create_task_obj',
+    'get_artifact_text',
     'get_data_parts',
     'get_file_parts',
     'get_message_text',

src/a2a/utils/artifact.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 from a2a.types import Artifact, DataPart, Part, TextPart
+from a2a.utils import get_text_parts
 
 
 def new_artifact(
@@ -70,3 +71,16 @@ def new_data_artifact(
         name,
         description,
     )
+
+
+def get_artifact_text(artifact: Artifact, delimiter: str = '\n') -> str:
+    """Extracts and joins all text content from an Artifact's parts.
+
+    Args:
+        artifact: The `Artifact` object.
+        delimiter: The string to use when joining text from multiple TextParts.
+
+    Returns:
+        A single string containing all text content, or an empty string if no text parts are found.
+    """
+    return delimiter.join(get_text_parts(artifact.parts))

src/a2a/utils/message.py

@@ -2,18 +2,13 @@
 
 import uuid
 
-from typing import Any
-
 from a2a.types import (
-    DataPart,
-    FilePart,
-    FileWithBytes,
-    FileWithUri,
     Message,
     Part,
     Role,
     TextPart,
 )
+from a2a.utils import get_text_parts
 
 
 def new_agent_text_message(
@@ -64,42 +59,6 @@ def new_agent_parts_message(
     )
 
 
-def get_text_parts(parts: list[Part]) -> list[str]:
-    """Extracts text content from all TextPart objects in a list of Parts.
-
-    Args:
-        parts: A list of `Part` objects.
-
-    Returns:
-        A list of strings containing the text content from any `TextPart` objects found.
-    """
-    return [part.root.text for part in parts if isinstance(part.root, TextPart)]
-
-
-def get_data_parts(parts: list[Part]) -> list[dict[str, Any]]:
-    """Extracts dictionary data from all DataPart objects in a list of Parts.
-
-    Args:
-        parts: A list of `Part` objects.
-
-    Returns:
-        A list of dictionaries containing the data from any `DataPart` objects found.
-    """
-    return [part.root.data for part in parts if isinstance(part.root, DataPart)]
-
-
-def get_file_parts(parts: list[Part]) -> list[FileWithBytes | FileWithUri]:
-    """Extracts file data from all FilePart objects in a list of Parts.
-
-    Args:
-        parts: A list of `Part` objects.
-
-    Returns:
-        A list of `FileWithBytes` or `FileWithUri` objects containing the file data from any `FilePart` objects found.
-    """
-    return [part.root.file for part in parts if isinstance(part.root, FilePart)]
-
-
 def get_message_text(message: Message, delimiter: str = '\n') -> str:
     """Extracts and joins all text content from a Message's parts.
 

src/a2a/utils/parts.py

@@ -0,0 +1,50 @@
+"""Utility functions for creating and handling A2A Parts objects."""
+
+import uuid
+
+from typing import Any
+
+from a2a.types import (
+    DataPart,
+    FilePart,
+    FileWithBytes,
+    FileWithUri,
+    Part,
+    TextPart,
+)
+
+
+def get_text_parts(parts: list[Part]) -> list[str]:
+    """Extracts text content from all TextPart objects in a list of Parts.
+
+    Args:
+        parts: A list of `Part` objects.
+
+    Returns:
+        A list of strings containing the text content from any `TextPart` objects found.
+    """
+    return [part.root.text for part in parts if isinstance(part.root, TextPart)]
+
+
+def get_data_parts(parts: list[Part]) -> list[dict[str, Any]]:
+    """Extracts dictionary data from all DataPart objects in a list of Parts.
+
+    Args:
+        parts: A list of `Part` objects.
+
+    Returns:
+        A list of dictionaries containing the data from any `DataPart` objects found.
+    """
+    return [part.root.data for part in parts if isinstance(part.root, DataPart)]
+
+
+def get_file_parts(parts: list[Part]) -> list[FileWithBytes | FileWithUri]:
+    """Extracts file data from all FilePart objects in a list of Parts.
+
+    Args:
+        parts: A list of `Part` objects.
+
+    Returns:
+        A list of `FileWithBytes` or `FileWithUri` objects containing the file data from any `FilePart` objects found.
+    """
+    return [part.root.file for part in parts if isinstance(part.root, FilePart)]