Python: allow empty A2A attachment payloads

python/packages/a2a/agent_framework_a2a/_utils.py

@@ -2,7 +2,7 @@
 
 import re
 
-URI_PATTERN = re.compile(r"^data:(?P<media_type>[^;,]+(?:;[^;,=]+=[^;,]+)*);base64,(?P<base64_data>[A-Za-z0-9+/=]+)\Z")
+URI_PATTERN = re.compile(r"^data:(?P<media_type>[^;,]+(?:;[^;,=]+=[^;,]+)*);base64,(?P<base64_data>[A-Za-z0-9+/=]*)\Z")
 
 
 def get_uri_data(uri: str) -> str:

python/packages/a2a/tests/test_a2a_agent.py

@@ -287,6 +287,28 @@ async def test_run_with_message_response(a2a_agent: A2AAgent, mock_a2a_client: M
     assert mock_a2a_client.call_count == 1
 
 
+@mark.parametrize("stream", [False, True])
+async def test_run_with_empty_data_content(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient, stream: bool) -> None:
+    """An empty attachment reaches the remote agent as a raw binary part."""
+    mock_a2a_client.add_message_response("msg-empty", "Received empty file")
+    message = Message(role="user", contents=[Content.from_data(b"", "application/octet-stream")])
+
+    if stream:
+        updates = [update async for update in a2a_agent.run(message, stream=True)]
+        assert updates[0].text == "Received empty file"
+    else:
+        response = await a2a_agent.run(message)
+        assert response.text == "Received empty file"
+
+    assert mock_a2a_client.call_count == 1
+    assert len(mock_a2a_client.last_message.parts) == 1
+    part = mock_a2a_client.last_message.parts[0]
+    assert part.WhichOneof("content") == "raw"
+    wire_part = MessageToDict(part)
+    assert wire_part["raw"] == ""
+    assert wire_part["mediaType"] == "application/octet-stream"
+
+
 async def test_run_with_task_response_single_artifact(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
     """Test run() method with Task response containing single artifact."""
     artifacts = [{"id": "art-1", "content": "Generated report content"}]

python/packages/a2a/tests/test_a2a_executor.py

@@ -13,7 +13,7 @@
 )
 from agent_framework._types import AgentResponse
 from agent_framework.a2a import A2AExecutor
-from pytest import fixture, raises
+from pytest import fixture, mark, raises
 
 
 @fixture
@@ -648,10 +648,10 @@ async def test_handle_multiple_text_contents(self, executor: A2AExecutor, mock_u
         mock_updater.update_status.assert_called_once()
         assert mock_updater.new_agent_message.called
 
-    async def test_handle_data_content(self, executor: A2AExecutor, mock_updater: MagicMock) -> None:
-        """Test handling messages with data content."""
+    @mark.parametrize("data", [b"test file data", b""])
+    async def test_handle_data_content(self, executor: A2AExecutor, mock_updater: MagicMock, data: bytes) -> None:
+        """Test handling messages with data content, including empty attachments."""
         # Arrange
-        data = b"test file data"
         message = Message(
             contents=[Content.from_data(data=data, media_type="application/octet-stream")],
             role="assistant",
@@ -664,6 +664,11 @@ async def test_handle_data_content(self, executor: A2AExecutor, mock_updater: Ma
         mock_updater.update_status.assert_called_once()
         call_args = mock_updater.update_status.call_args
         assert call_args.kwargs["state"] == TaskState.TASK_STATE_WORKING
+        parts = mock_updater.new_agent_message.call_args.kwargs["parts"]
+        assert len(parts) == 1
+        assert parts[0].WhichOneof("content") == "raw"
+        assert parts[0].raw == data
+        assert parts[0].media_type == "application/octet-stream"
 
     async def test_handle_uri_content(self, executor: A2AExecutor, mock_updater: MagicMock) -> None:
         """Test handling messages with URI content."""

python/packages/a2a/tests/test_utils.py

@@ -28,6 +28,15 @@ def test_get_uri_data_valid() -> None:
     assert get_uri_data(uri) == "SGVsbG8sIFdvcmxkIQ=="
 
 
+@pytest.mark.parametrize(
+    "uri",
+    ["data:application/octet-stream;base64,", "data:text/plain;charset=utf-8;base64,"],
+)
+def test_get_uri_data_empty_payload(uri: str) -> None:
+    """An empty base64 payload represents a valid zero-byte attachment."""
+    assert get_uri_data(uri) == ""
+
+
 def test_get_uri_data_invalid_format() -> None:
     """Test get_uri_data with invalid URI formats."""
     invalid_uris = [