fix(server): preserve non-ASCII characters in streaming JSON responses (#1080)

# Description
Pass ensure_ascii=False to json.dumps() at the 5 streaming call sites. 

## Reasoning
- Consistency. Matches Starlette's JSONResponse (which uses
ensure_ascii=False), so the SDK now emits the same wire format for unary
and streaming responses.
- Smaller payloads. CJK/Arabic/Cyrillic/Devanagari/emoji content drops
to ~half the byte size.
- Better debuggability. Logs, curl, devtools, and tcpdump show actual
text instead of escape soup.
- Inclusivity. Non-Latin scripts become first-class on the wire instead
of being silently doubled in size.
-
[Spec-compliant](https://a2a-protocol.org/latest/specification/#1411-applicationa2ajson):
"UTF-8 encoding MUST be used for JSON text"

Fixes #1078 🦕

Author: sokoliva

src/a2a/compat/v0_3/rest_adapter.py

@@ -1,5 +1,4 @@
 import functools
-import json
 import logging
 
 from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable
@@ -37,6 +36,7 @@
     DefaultServerCallContextBuilder,
     ServerCallContextBuilder,
 )
+from a2a.utils import json_utils
 from a2a.utils.error_handlers import (
     rest_error_handler,
     rest_stream_error_handler,
@@ -94,7 +94,7 @@ async def event_generator(
             stream: AsyncIterable[Any],
         ) -> AsyncIterator[str]:
             async for item in stream:
-                yield json.dumps(item)
+                yield json_utils.dumps(item)
 
         return EventSourceResponse(
             event_generator(method(request, call_context))

src/a2a/server/routes/jsonrpc_dispatcher.py

@@ -41,7 +41,7 @@
     Task,
     TaskPushNotificationConfig,
 )
-from a2a.utils import constants, proto_utils
+from a2a.utils import constants, json_utils, proto_utils
 from a2a.utils.errors import (
     A2AError,
     TaskNotFoundError,
@@ -573,7 +573,7 @@ async def event_generator(
                 try:
                     async for item in stream:
                         event: dict[str, str] = {
-                            'data': json.dumps(item),
+                            'data': json_utils.dumps(item),
                         }
                         if 'error' in item:
                             event['event'] = 'error'
@@ -592,7 +592,7 @@ async def event_generator(
                     )
                     yield {
                         'event': 'error',
-                        'data': json.dumps(error_response),
+                        'data': json_utils.dumps(error_response),
                     }
 
             return EventSourceResponse(event_generator(handler_result))  # ty:ignore[invalid-argument-type]

src/a2a/server/routes/rest_dispatcher.py

@@ -1,4 +1,3 @@
-import json
 import logging
 
 from collections.abc import AsyncIterator, Awaitable, Callable
@@ -18,7 +17,7 @@
     GetTaskPushNotificationConfigRequest,
     SubscribeToTaskRequest,
 )
-from a2a.utils import constants, proto_utils
+from a2a.utils import constants, json_utils, proto_utils
 from a2a.utils.error_handlers import (
     build_rest_error_payload,
     rest_error_handler,
@@ -140,14 +139,14 @@ async def _handle_streaming(
             return EventSourceResponse(iter([]))
 
         async def event_generator() -> AsyncIterator[ServerSentEvent]:
-            yield ServerSentEvent(data=json.dumps(first_item))
+            yield ServerSentEvent(data=json_utils.dumps(first_item))
             try:
                 async for item in stream:
-                    yield ServerSentEvent(data=json.dumps(item))
+                    yield ServerSentEvent(data=json_utils.dumps(item))
             except Exception as e:
                 logger.exception('Error during REST SSE stream')
                 yield ServerSentEvent(
-                    data=json.dumps(build_rest_error_payload(e)),
+                    data=json_utils.dumps(build_rest_error_payload(e)),
                     event='error',
                 )
 

src/a2a/utils/json_utils.py

@@ -0,0 +1,19 @@
+"""JSON serialization helpers for the A2A Python SDK."""
+
+import json
+
+from typing import Any
+
+
+def dumps(obj: Any) -> str:
+    r"""Serialize ``obj`` to a JSON-formatted ``str`` with UTF-8 defaults.
+
+    Use this in SSE/streaming code paths where payloads are serialized
+    manually before being written to the wire. Unary HTTP responses do
+    not need it because Starlette's ``JSONResponse.render`` already calls
+    ``json.dumps(content, ensure_ascii=False, ...)`` internally; this
+    helper makes the streaming paths behave identically so non-ASCII
+    characters (CJK, emoji, etc.) reach clients as raw UTF-8 rather than
+    escape sequences.
+    """
+    return json.dumps(obj, ensure_ascii=False)

tests/compat/v0_3/test_rest_routes_compat.py

@@ -1,6 +1,7 @@
 import logging
 
-from unittest.mock import MagicMock
+from collections.abc import AsyncIterator
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
@@ -22,6 +23,8 @@
 from google.protobuf import json_format
 from httpx import ASGITransport, AsyncClient
 from starlette.applications import Starlette
+from starlette.datastructures import Headers
+from starlette.requests import Request
 
 
 logger = logging.getLogger(__name__)
@@ -200,3 +203,42 @@ async def test_cancel_task_v03(
     actual_response = a2a_v0_3_pb2.Task()
     json_format.Parse(response.text, actual_response)
     assert expected_response == actual_response
+
+
+@pytest.mark.anyio
+async def test_v03_streaming_does_not_ascii_escape_non_ascii(
+    request_handler: RequestHandler,
+) -> None:
+    """v0.3 REST streaming must emit raw UTF-8 for non-ASCII characters.
+
+    Regression test for https://github.com/a2aproject/a2a-python/issues/1078.
+    """
+    adapter = REST03Adapter(http_handler=request_handler)
+    non_ascii_text = '你好'
+
+    async def stream_with_non_ascii(
+        request: Request, context: object
+    ) -> AsyncIterator[dict]:
+        yield {'msg': {'text': non_ascii_text}}
+
+    mock_req = MagicMock(spec=Request)
+    mock_req.body = AsyncMock(return_value=b'{}')
+    mock_req.headers = Headers({'a2a-version': '0.3'})
+    mock_req.user = MagicMock(is_authenticated=False)
+    mock_req.auth = None
+    mock_req.scope = {}
+
+    response = await adapter._handle_streaming_request(
+        stream_with_non_ascii, mock_req
+    )
+    chunks = []
+    async for chunk in response.body_iterator:
+        chunks.append(chunk)
+
+    assert len(chunks) == 1
+    chunk = chunks[0]
+    payload = getattr(chunk, 'data', chunk)
+    if isinstance(payload, bytes):
+        payload = payload.decode('utf-8')
+    assert non_ascii_text in payload
+    assert '\\u4f60\\u597d' not in payload

tests/server/routes/test_jsonrpc_dispatcher.py

@@ -591,6 +591,65 @@ async def stream_generator():
         call_context = handler.on_subscribe_to_task.call_args[0][1]
         assert call_context.state['method'] == 'SubscribeToTask'
 
+    @pytest.mark.asyncio
+    async def test_streaming_response_does_not_ascii_escape_non_ascii(
+        self, handler, agent_card
+    ):
+        """Non-ASCII characters (e.g. CJK) must be emitted as raw UTF-8."""
+        non_ascii_text = '你好'
+
+        async def stream_generator():
+            yield TaskArtifactUpdateEvent(
+                artifact=Artifact(
+                    artifact_id='a1',
+                    name='result',
+                    parts=[Part(text=non_ascii_text)],
+                ),
+                task_id='task1',
+                context_id='ctx1',
+                append=False,
+                last_chunk=True,
+            )
+
+        handler.on_message_send_stream = MagicMock(
+            return_value=stream_generator()
+        )
+
+        jsonrpc_routes = create_jsonrpc_routes(
+            request_handler=handler,
+            rpc_url='/',
+        )
+        from starlette.applications import Starlette
+
+        app = Starlette(routes=jsonrpc_routes)
+        client = TestClient(app, headers={'A2A-Version': '1.0'})
+
+        try:
+            with client.stream(
+                'POST',
+                '/',
+                json=_make_jsonrpc_request(
+                    'SendStreamingMessage',
+                    {
+                        'message': {
+                            'messageId': '1',
+                            'role': 'ROLE_USER',
+                            'parts': [{'text': non_ascii_text}],
+                        }
+                    },
+                ),
+            ) as response:
+                assert response.status_code == 200
+                content = b''
+                for chunk in response.iter_bytes():
+                    content += chunk
+        finally:
+            client.close()
+            await asyncio.sleep(0.1)
+
+        assert non_ascii_text.encode('utf-8') in content
+        assert b'\\u4f60\\u597d' not in content
+
 
 if __name__ == '__main__':
     pytest.main([__file__])

tests/server/routes/test_rest_dispatcher.py

@@ -288,3 +288,38 @@ async def test_on_message_send_stream_handler_error(self, mock_handler):
 
         response = await dispatcher.on_message_send_stream(req)
         assert response.status_code == 400
+
+    async def test_streaming_does_not_ascii_escape_non_ascii(
+        self, rest_dispatcher_instance
+    ):
+        """Non-ASCII characters in SSE payloads must pass through as raw UTF-8.
+
+        Regression test for https://github.com/a2aproject/a2a-python/issues/1078.
+        """
+        non_ascii_text = '你好'
+
+        async def stream_with_non_ascii(
+            context: ServerCallContext,
+        ) -> AsyncIterator[dict]:
+            yield {'msg': {'text': non_ascii_text}}
+            yield {'msg': {'text': f'echo: {non_ascii_text}'}}
+
+        # Drive _handle_streaming directly to assert on the serialized bytes
+        # produced by json.dumps, independent of protobuf encoding details.
+        req = make_mock_request(method='POST')
+        response = await rest_dispatcher_instance._handle_streaming(
+            req, stream_with_non_ascii
+        )
+        assert response.status_code == 200
+
+        chunks = []
+        async for chunk in response.body_iterator:
+            chunks.append(chunk)
+
+        assert len(chunks) == 2
+        for chunk in chunks:
+            payload = getattr(chunk, 'data', chunk)
+            if isinstance(payload, bytes):
+                payload = payload.decode('utf-8')
+            assert non_ascii_text in payload
+            assert '\\u4f60\\u597d' not in payload

tests/utils/test_json_utils.py

@@ -0,0 +1,24 @@
+"""Tests for a2a.utils.json_utils module."""
+
+import json
+
+from a2a.utils import json_utils
+
+
+def test_dumps_emits_raw_utf8_for_non_ascii() -> None:
+    """Non-ASCII characters must serialize as raw UTF-8, not \\uXXXX escapes."""
+    out = json_utils.dumps({'text': '你好'})
+    assert '你好' in out
+    assert '\\u4f60\\u597d' not in out
+
+
+def test_dumps_emits_emoji_as_raw_utf8() -> None:
+    """Emoji (outside the BMP) must also pass through unescaped."""
+    out = json_utils.dumps({'emoji': '🎉'})
+    assert '🎉' in out
+
+
+def test_dumps_round_trips_through_json_loads() -> None:
+    """Wrapper output must remain a valid JSON document."""
+    payload = {'msg': '你好', 'list': ['a', 'é', '日本語'], 'n': 1}
+    assert json.loads(json_utils.dumps(payload)) == payload