fix(server): treat request bodies that are not valid UTF-8 as parse errors

src/a2a/compat/v0_3/rest_handler.py

@@ -3,7 +3,7 @@
 from collections.abc import AsyncIterator
 from typing import TYPE_CHECKING, Any
 
-from google.protobuf.json_format import MessageToDict, Parse
+from google.protobuf.json_format import MessageToDict
 
 
 if TYPE_CHECKING:
@@ -29,6 +29,7 @@
 from a2a.server.context import ServerCallContext
 from a2a.utils import constants
 from a2a.utils.errors import InvalidParamsError
+from a2a.utils.proto_utils import parse_lenient
 from a2a.utils.telemetry import SpanKind, trace_class
 from a2a.utils.version_validator import validate_version
 
@@ -68,7 +69,7 @@ async def on_message_send(
         """
         body = await request.body()
         v03_pb_msg = pb2_v03.SendMessageRequest()
-        Parse(body, v03_pb_msg, ignore_unknown_fields=True)
+        parse_lenient(body, v03_pb_msg)
         v03_params_msg = proto_utils.FromProto.message_send_params(v03_pb_msg)
         rpc_req = types_v03.SendMessageRequest(id='', params=v03_params_msg)
 
@@ -94,7 +95,7 @@ async def on_message_send_stream(
         """
         body = await request.body()
         v03_pb_msg = pb2_v03.SendMessageRequest()
-        Parse(body, v03_pb_msg, ignore_unknown_fields=True)
+        parse_lenient(body, v03_pb_msg)
         v03_params_msg = proto_utils.FromProto.message_send_params(v03_pb_msg)
         rpc_req = types_v03.SendMessageRequest(id='', params=v03_params_msg)
 
@@ -212,7 +213,7 @@ async def set_push_notification(
         body = await request.body()
 
         v03_pb_push = pb2_v03.CreateTaskPushNotificationConfigRequest()
-        Parse(body, v03_pb_push, ignore_unknown_fields=True)
+        parse_lenient(body, v03_pb_push)
 
         v03_params_push = (
             proto_utils.FromProto.task_push_notification_config_request(

src/a2a/server/routes/jsonrpc_dispatcher.py

@@ -340,7 +340,7 @@ async def handle_requests(self, request: Request) -> Response:  # noqa: PLR0911,
                 except A2AError as e:
                     handler_result = build_error_response(request_id, e)
             return self._create_response(call_context, handler_result)
-        except json.decoder.JSONDecodeError as e:
+        except (json.decoder.JSONDecodeError, UnicodeDecodeError) as e:
             return self._generate_error_response(
                 None, JSONParseError(message=str(e))
             )

src/a2a/server/routes/rest_dispatcher.py

@@ -3,7 +3,7 @@
 from collections.abc import AsyncIterator, Awaitable, Callable
 from typing import TYPE_CHECKING, Any, TypeVar
 
-from google.protobuf.json_format import MessageToDict, Parse
+from google.protobuf.json_format import MessageToDict
 
 from a2a.server.context import ServerCallContext
 from a2a.server.request_handlers.request_handler import RequestHandler
@@ -180,7 +180,7 @@ async def _handler(
         ) -> a2a_pb2.SendMessageResponse:
             body = await request.body()
             params = a2a_pb2.SendMessageRequest()
-            Parse(body, params, ignore_unknown_fields=True)
+            proto_utils.parse_lenient(body, params)
             task_or_message = await self.request_handler.on_message_send(
                 params, context
             )
@@ -203,7 +203,7 @@ async def _handler(
         ) -> AsyncIterator[dict[str, Any]]:
             body = await request.body()
             params = a2a_pb2.SendMessageRequest()
-            Parse(body, params, ignore_unknown_fields=True)
+            proto_utils.parse_lenient(body, params)
             async for event in self.request_handler.on_message_send_stream(
                 params, context
             ):
@@ -315,7 +315,7 @@ async def _handler(
         ) -> a2a_pb2.TaskPushNotificationConfig:
             body = await request.body()
             params = a2a_pb2.TaskPushNotificationConfig()
-            Parse(body, params, ignore_unknown_fields=True)
+            proto_utils.parse_lenient(body, params)
             params.task_id = request.path_params['id']
             return await self.request_handler.on_create_task_push_notification_config(
                 params, context

src/a2a/utils/proto_utils.py

@@ -67,8 +67,17 @@ def parse_lenient(
 
     Returns:
         The parsed message.
+
+    Raises:
+        ParseError: If the data is not valid ProtoJSON, including bytes that
+            are not valid UTF-8.
     """
-    if isinstance(data, str | bytes):
+    if isinstance(data, bytes):
+        try:
+            data = data.decode('utf-8')
+        except UnicodeDecodeError as e:
+            raise ParseError(f'Failed to load JSON: {e}.') from e
+    if isinstance(data, str):
         return Parse(data, message, ignore_unknown_fields=True)
     return ParseDict(data, message, ignore_unknown_fields=True)
 

tests/compat/v0_3/test_rest_routes_compat.py

@@ -210,6 +210,22 @@ async def test_get_task_v03_non_integer_history_length(
     request_handler.on_get_task.assert_not_called()
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    'path',
+    ['/v1/message:send', '/v1/tasks/test_task_id/pushNotificationConfigs'],
+)
+async def test_invalid_utf8_body_v03_is_bad_request(
+    client: AsyncClient, path: str
+) -> None:
+    response = await client.post(
+        path, content=b'{"message": {"messageId": "\xff"}}'
+    )
+
+    assert response.status_code == 400
+    assert response.json()['error']['status'] == 'INVALID_ARGUMENT'
+
+
 @pytest.mark.anyio
 async def test_list_tasks_not_in_v03_adapter_routes(
     request_handler: RequestHandler,

tests/server/routes/test_jsonrpc_dispatcher.py

@@ -194,6 +194,16 @@ def test_parse_diagnostic_survives_in_metadata(self, parse_error):
 
 
 class TestMalformedJsonBody:
+    def test_invalid_utf8_is_a_parse_error(self, client):
+        response = client.post(
+            '/',
+            content=b'{"jsonrpc": "2.0", "id": 1, "method": "GetTask", '
+            b'"params": {"id": "\xff"}}',
+            headers={'Content-Type': 'application/json'},
+        )
+
+        assert response.json()['error']['code'] == -32700
+
     def test_reported_through_logging_not_stderr(self, client, capsys, caplog):
         response = client.post(
             '/',

tests/server/routes/test_rest_routes.py

@@ -59,6 +59,25 @@ def test_rest_endpoints_routing(agent_card, mock_handler):
     assert mock_handler.on_message_send.called
 
 
+@pytest.mark.parametrize(
+    'path',
+    ['/message:send', '/message:stream', '/tasks/t1/pushNotificationConfigs'],
+)
+def test_rest_invalid_utf8_body_is_bad_request(mock_handler, path):
+    """A body that is not UTF-8 is malformed JSON, so 400 rather than 500."""
+    app = Starlette(routes=create_rest_routes(request_handler=mock_handler))
+    client = TestClient(app)
+
+    response = client.post(
+        path,
+        content=b'{"message": {"messageId": "\xff"}}',
+        headers={'A2A-Version': '1.0'},
+    )
+
+    assert response.status_code == 400
+    assert response.json()['error']['status'] == 'INVALID_ARGUMENT'
+
+
 def test_rest_endpoints_routing_tenant(agent_card, mock_handler):
     """Tests that mounted routes with {tenant} route to the handler endpoints."""
     mock_handler.on_message_send.return_value = Task(id='123')

tests/utils/test_proto_utils.py

@@ -477,3 +477,8 @@ def test_malformed_known_field_still_raises(self) -> None:
         """Leniency covers unknown names, not a known field of the wrong type."""
         with pytest.raises(ParseError):
             proto_utils.parse_lenient({'messageId': 5}, Message())
+
+    def test_invalid_utf8_bytes_raise_parse_error(self) -> None:
+        """Bytes that are not UTF-8 are malformed JSON, not an internal error."""
+        with pytest.raises(ParseError):
+            proto_utils.parse_lenient(b'{"messageId": "\xff"}', Message())