Add ArtifactStreamer helper for streaming text artifact updates

new_text_artifact generates a fresh UUID on every call, making it
unsuitable for streaming where append semantics require a stable
artifact ID across chunks. ArtifactStreamer holds a single artifact
ID and provides append() and finalize() methods that emit correctly
shaped TaskArtifactUpdateEvents with append=True.

Author: kevinlu310

src/a2a/utils/__init__.py

@@ -1,6 +1,7 @@
 """Utility functions for the A2A Python SDK."""
 
 from a2a.utils.artifact import (
+    ArtifactStreamer,
     get_artifact_text,
     new_artifact,
     new_data_artifact,
@@ -23,22 +24,16 @@
     new_agent_parts_message,
     new_agent_text_message,
 )
-from a2a.utils.parts import (
-    get_data_parts,
-    get_file_parts,
-    get_text_parts,
-)
-from a2a.utils.task import (
-    completed_task,
-    new_task,
-)
+from a2a.utils.parts import get_data_parts, get_file_parts, get_text_parts
+from a2a.utils.task import completed_task, new_task
 
 
 __all__ = [
     'AGENT_CARD_WELL_KNOWN_PATH',
     'DEFAULT_RPC_URL',
     'EXTENDED_AGENT_CARD_PATH',
     'PREV_AGENT_CARD_WELL_KNOWN_PATH',
+    'ArtifactStreamer',
     'append_artifact_to_task',
     'are_modalities_compatible',
     'build_text_artifact',

src/a2a/utils/artifact.py

@@ -4,7 +4,13 @@
 
 from typing import Any
 
-from a2a.types import Artifact, DataPart, Part, TextPart
+from a2a.types import (
+    Artifact,
+    DataPart,
+    Part,
+    TaskArtifactUpdateEvent,
+    TextPart,
+)
 from a2a.utils.parts import get_text_parts
 
 
@@ -86,3 +92,79 @@ def get_artifact_text(artifact: Artifact, delimiter: str = '\n') -> str:
         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))
+
+
+class ArtifactStreamer:
+    """Helper for streaming text into a single artifact across multiple events.
+
+    Creates a stable artifact ID on construction so all chunks reference
+    the same artifact, enabling proper append semantics per the A2A spec.
+
+    Example::
+
+        streamer = ArtifactStreamer(context_id, task_id, name='response')
+
+        async for chunk in llm.stream(prompt):
+            await event_queue.enqueue_event(streamer.append(chunk))
+
+        await event_queue.enqueue_event(streamer.finalize())
+
+    Args:
+        context_id: The context ID associated with the task.
+        task_id: The task ID associated with the streaming session.
+        name: A human-readable name for the artifact.
+        artifact_id: An explicit artifact ID. If omitted a UUID is generated.
+    """
+
+    def __init__(
+        self,
+        context_id: str,
+        task_id: str,
+        name: str = 'response',
+        artifact_id: str | None = None,
+    ) -> None:
+        self.context_id = context_id
+        self.task_id = task_id
+        self.name = name
+        self.artifact_id = artifact_id or str(uuid.uuid4())
+
+    def append(self, text: str) -> TaskArtifactUpdateEvent:
+        """Emit a chunk to be appended to the streaming artifact.
+
+        Args:
+            text: The incremental text content for this chunk.
+
+        Returns:
+            A ``TaskArtifactUpdateEvent`` with ``append=True`` and
+            ``last_chunk=False``.
+        """
+        return TaskArtifactUpdateEvent(
+            context_id=self.context_id,
+            task_id=self.task_id,
+            append=True,
+            last_chunk=False,
+            artifact=Artifact(
+                artifact_id=self.artifact_id,
+                name=self.name,
+                parts=[Part(root=TextPart(text=text))],
+            ),
+        )
+
+    def finalize(self) -> TaskArtifactUpdateEvent:
+        """Signal that the artifact stream is complete.
+
+        Returns:
+            A ``TaskArtifactUpdateEvent`` with ``append=True`` and
+            ``last_chunk=True``.
+        """
+        return TaskArtifactUpdateEvent(
+            context_id=self.context_id,
+            task_id=self.task_id,
+            append=True,
+            last_chunk=True,
+            artifact=Artifact(
+                artifact_id=self.artifact_id,
+                name=self.name,
+                parts=[],
+            ),
+        )