fix: a2a use artifact update event (strands-agents#1401)

fix: update tests

fix: simplify code by storing in class

fix: remove uneeded code change

fix: hide a2a artifact streaming under feature flag

fix: use walrus operator

fix: use star to signify end of unnamed

fix: add check for walrus legacy

fix: clarify enable_a2a_compliant_streaming parameter in StrandsA2AExecutor initialization

fix: update tests

refactor: streamline artifact addition logic in StrandsA2AExecutor

Author: brycewcole

src/strands/multiagent/a2a/executor.py

@@ -12,6 +12,8 @@
 import json
 import logging
 import mimetypes
+import uuid
+import warnings
 from typing import Any, Literal
 
 from a2a.server.agent_execution import AgentExecutor, RequestContext
@@ -49,13 +51,21 @@ class StrandsA2AExecutor(AgentExecutor):
     # Handle special cases where format differs from extension
     FORMAT_MAPPINGS = {"jpg": "jpeg", "htm": "html", "3gp": "three_gp", "3gpp": "three_gp", "3g2": "three_gp"}
 
-    def __init__(self, agent: SAAgent):
+    # A2A-compliant streaming mode
+    _current_artifact_id: str | None
+    _is_first_chunk: bool
+
+    def __init__(self, agent: SAAgent, *, enable_a2a_compliant_streaming: bool = False):
         """Initialize a StrandsA2AExecutor.
 
         Args:
             agent: The Strands Agent instance to adapt to the A2A protocol.
+            enable_a2a_compliant_streaming: If True, uses A2A-compliant streaming with
+                artifact updates. If False, uses legacy status updates streaming behavior
+                for backwards compatibility. Defaults to False.
         """
         self.agent = agent
+        self.enable_a2a_compliant_streaming = enable_a2a_compliant_streaming
 
     async def execute(
         self,
@@ -104,12 +114,30 @@ async def _execute_streaming(self, context: RequestContext, updater: TaskUpdater
         else:
             raise ValueError("No content blocks available")
 
+        if not self.enable_a2a_compliant_streaming:
+            warnings.warn(
+                "The default A2A response stream implemented in the strands sdk does not conform to "
+                "what is expected in the A2A spec. Please set the `enable_a2a_compliant_streaming` "
+                "boolean to `True` on your `A2AServer` class to properly conform to the spec. "
+                "In the next major version release, this will be the default behavior.",
+                UserWarning,
+                stacklevel=3,
+            )
+
+        if self.enable_a2a_compliant_streaming:
+            self._current_artifact_id = str(uuid.uuid4())
+            self._is_first_chunk = True
+
         try:
             async for event in self.agent.stream_async(content_blocks):
                 await self._handle_streaming_event(event, updater)
         except Exception:
             logger.exception("Error in streaming execution")
             raise
+        finally:
+            if self.enable_a2a_compliant_streaming:
+                self._current_artifact_id = None
+                self._is_first_chunk = True
 
     async def _handle_streaming_event(self, event: dict[str, Any], updater: TaskUpdater) -> None:
         """Handle a single streaming event from the Strands Agent.
@@ -125,28 +153,60 @@ async def _handle_streaming_event(self, event: dict[str, Any], updater: TaskUpda
         logger.debug("Streaming event: %s", event)
         if "data" in event:
             if text_content := event["data"]:
-                await updater.update_status(
-                    TaskState.working,
-                    new_agent_text_message(
-                        text_content,
-                        updater.context_id,
-                        updater.task_id,
-                    ),
-                )
+                if self.enable_a2a_compliant_streaming:
+                    await updater.add_artifact(
+                        [Part(root=TextPart(text=text_content))],
+                        artifact_id=self._current_artifact_id,
+                        name="agent_response",
+                        append=not self._is_first_chunk,
+                    )
+                    self._is_first_chunk = False
+                else:
+                    # Legacy use update_status with agent message
+                    await updater.update_status(
+                        TaskState.working,
+                        new_agent_text_message(
+                            text_content,
+                            updater.context_id,
+                            updater.task_id,
+                        ),
+                    )
         elif "result" in event:
             await self._handle_agent_result(event["result"], updater)
 
     async def _handle_agent_result(self, result: SAAgentResult | None, updater: TaskUpdater) -> None:
         """Handle the final result from the Strands Agent.
 
-        Processes the agent's final result, extracts text content from the response,
-        and adds it as an artifact to the task before marking the task as complete.
+        For A2A-compliant streaming: sends the final artifact chunk marker and marks
+        the task as complete. If no data chunks were previously sent, includes the
+        result content.
+
+        For legacy streaming: adds the final result as a simple artifact without
+        artifact_id tracking.
 
         Args:
             result: The agent result object containing the final response, or None if no result.
             updater: The task updater for managing task state and adding the final artifact.
         """
-        if final_content := str(result):
+        if self.enable_a2a_compliant_streaming:
+            if self._is_first_chunk:
+                final_content = str(result) if result else ""
+                parts = [Part(root=TextPart(text=final_content))] if final_content else []
+                await updater.add_artifact(
+                    parts,
+                    artifact_id=self._current_artifact_id,
+                    name="agent_response",
+                    last_chunk=True,
+                )
+            else:
+                await updater.add_artifact(
+                    [],
+                    artifact_id=self._current_artifact_id,
+                    name="agent_response",
+                    append=True,
+                    last_chunk=True,
+                )
+        elif final_content := str(result):
             await updater.add_artifact(
                 [Part(root=TextPart(text=final_content))],
                 name="agent_response",

src/strands/multiagent/a2a/server.py

@@ -42,6 +42,7 @@ def __init__(
         queue_manager: QueueManager | None = None,
         push_config_store: PushNotificationConfigStore | None = None,
         push_sender: PushNotificationSender | None = None,
+        enable_a2a_compliant_streaming: bool = False,
     ):
         """Initialize an A2A-compatible server from a Strands agent.
 
@@ -66,6 +67,9 @@ def __init__(
                 no push notification configuration is used.
             push_sender: Custom push notification sender implementation. If None,
                 no push notifications are sent.
+            enable_a2a_compliant_streaming: If True, uses A2A-compliant streaming with
+                artifact updates. If False, uses legacy status updates streaming behavior
+                for backwards compatibility. Defaults to False.
         """
         self.host = host
         self.port = port
@@ -90,7 +94,9 @@ def __init__(
         self.description = self.strands_agent.description
         self.capabilities = AgentCapabilities(streaming=True)
         self.request_handler = DefaultRequestHandler(
-            agent_executor=StrandsA2AExecutor(self.strands_agent),
+            agent_executor=StrandsA2AExecutor(
+                self.strands_agent, enable_a2a_compliant_streaming=enable_a2a_compliant_streaming
+            ),
             task_store=task_store or InMemoryTaskStore(),
             queue_manager=queue_manager,
             push_config_store=push_config_store,

tests/strands/multiagent/a2a/test_executor.py

@@ -1020,3 +1020,96 @@ def test_default_formats_modularization():
     assert executor._get_file_format_from_mime_type("", "document") == "txt"
     assert executor._get_file_format_from_mime_type("", "image") == "png"
     assert executor._get_file_format_from_mime_type("", "video") == "mp4"
+
+
+# Tests for enable_a2a_compliant_streaming parameter
+
+
+@pytest.mark.asyncio
+async def test_legacy_mode_emits_deprecation_warning(mock_strands_agent, mock_request_context, mock_event_queue):
+    """Test that legacy streaming (default) emits deprecation warning."""
+    from a2a.types import TextPart
+
+    executor = StrandsA2AExecutor(mock_strands_agent)  # Default is False
+
+    # Mock stream_async
+    async def mock_stream(content_blocks):
+        yield {"result": None}
+
+    mock_strands_agent.stream_async = MagicMock(return_value=mock_stream([]))
+
+    # Mock task
+    mock_task = MagicMock()
+    mock_task.id = "test-task-id"
+    mock_task.context_id = "test-context-id"
+    mock_request_context.current_task = mock_task
+
+    # Mock message
+    mock_text_part = MagicMock(spec=TextPart)
+    mock_text_part.text = "test"
+    mock_part = MagicMock()
+    mock_part.root = mock_text_part
+    mock_message = MagicMock()
+    mock_message.parts = [mock_part]
+    mock_request_context.message = mock_message
+
+    with pytest.warns(UserWarning, match="does not conform to what is expected in the A2A spec"):
+        await executor.execute(mock_request_context, mock_event_queue)
+
+
+@pytest.mark.asyncio
+async def test_a2a_compliant_mode_no_warning(mock_strands_agent, mock_request_context, mock_event_queue):
+    """Test that A2A-compliant mode does not emit warning."""
+    import warnings
+
+    from a2a.types import TextPart
+
+    executor = StrandsA2AExecutor(mock_strands_agent, enable_a2a_compliant_streaming=True)
+
+    # Mock stream_async
+    async def mock_stream(content_blocks):
+        yield {"result": None}
+
+    mock_strands_agent.stream_async = MagicMock(return_value=mock_stream([]))
+
+    # Mock task
+    mock_task = MagicMock()
+    mock_task.id = "test-task-id"
+    mock_task.context_id = "test-context-id"
+    mock_request_context.current_task = mock_task
+
+    # Mock message
+    mock_text_part = MagicMock(spec=TextPart)
+    mock_text_part.text = "test"
+    mock_part = MagicMock()
+    mock_part.root = mock_text_part
+    mock_message = MagicMock()
+    mock_message.parts = [mock_part]
+    mock_request_context.message = mock_message
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        try:
+            await executor.execute(mock_request_context, mock_event_queue)
+        except UserWarning:
+            pytest.fail("Should not emit warning")
+
+
+@pytest.mark.asyncio
+async def test_a2a_compliant_mode_uses_add_artifact(mock_strands_agent):
+    """Test that A2A-compliant mode uses add_artifact with artifact_id."""
+    executor = StrandsA2AExecutor(mock_strands_agent, enable_a2a_compliant_streaming=True)
+    executor._current_artifact_id = "artifact-123"
+    executor._is_first_chunk = True
+
+    mock_updater = MagicMock()
+    mock_updater.add_artifact = AsyncMock()
+    mock_updater.update_status = AsyncMock()
+
+    event = {"data": "content"}
+    await executor._handle_streaming_event(event, mock_updater)
+
+    mock_updater.add_artifact.assert_called_once()
+    assert mock_updater.add_artifact.call_args[1]["artifact_id"] == "artifact-123"
+    assert mock_updater.add_artifact.call_args[1]["append"] is False
+    mock_updater.update_status.assert_not_called()