fix: harden SSE transport and enforce request body size limit

src/a2a/compat/v0_3/jsonrpc_adapter.py

@@ -277,4 +277,8 @@ async def event_generator(
                     )
                 }
 
-        return EventSourceResponse(event_generator(stream_gen))
+        return EventSourceResponse(
+            event_generator(stream_gen),
+            ping=constants.SSE_PING_INTERVAL_SECONDS,
+            send_timeout=constants.SSE_SEND_TIMEOUT_SECONDS,
+        )

src/a2a/compat/v0_3/rest_adapter.py

@@ -36,7 +36,7 @@
     DefaultServerCallContextBuilder,
     ServerCallContextBuilder,
 )
-from a2a.utils import json_utils
+from a2a.utils import constants, json_utils
 from a2a.utils.error_handlers import (
     rest_error_handler,
     rest_stream_error_handler,
@@ -97,7 +97,9 @@ async def event_generator(
                 yield json_utils.dumps(item)
 
         return EventSourceResponse(
-            event_generator(method(request, call_context))
+            event_generator(method(request, call_context)),
+            ping=constants.SSE_PING_INTERVAL_SECONDS,
+            send_timeout=constants.SSE_SEND_TIMEOUT_SECONDS,
         )
 
     def routes(self) -> dict[tuple[str, str], Callable[[Request], Any]]:

src/a2a/server/routes/common.py

@@ -13,12 +13,61 @@
         Request = Any
         BaseUser = Any
 
+    try:
+        from starlette.exceptions import HTTPException
+    except ImportError:
+        HTTPException = Any
+
 from a2a.auth.user import UnauthenticatedUser, User
 from a2a.extensions.common import (
     HTTP_EXTENSION_HEADER,
     get_requested_extensions,
 )
 from a2a.server.context import ServerCallContext
+from a2a.utils.constants import MAX_REQUEST_BODY_SIZE
+
+
+try:
+    from starlette.status import HTTP_413_CONTENT_TOO_LARGE
+except ImportError:
+    HTTP_413_CONTENT_TOO_LARGE = 413
+
+
+async def read_request_body_with_limit(request: Request) -> bytes:
+    """Reads a request body, rejecting bodies over ``MAX_REQUEST_BODY_SIZE``.
+
+    The content-length header is checked first (fast reject), and the body
+    is then streamed in chunks with an incremental cap so oversized
+    chunked bodies are rejected while being read instead of buffered
+    unboundedly. The body is cached on the request (``request._body``) so
+    later ``request.body()`` calls return the same bytes.
+
+    Raises:
+        starlette.exceptions.HTTPException: With status 413 (content too
+            large) when the body exceeds the limit.
+    """
+    content_length = request.headers.get('content-length')
+    if content_length:
+        try:
+            if int(content_length) > MAX_REQUEST_BODY_SIZE:
+                raise HTTPException(status_code=HTTP_413_CONTENT_TOO_LARGE)
+        except ValueError:
+            pass
+
+    chunks: list[bytes] = []
+    size = 0
+    async for chunk in request.stream():
+        size += len(chunk)
+        if size > MAX_REQUEST_BODY_SIZE:
+            raise HTTPException(status_code=HTTP_413_CONTENT_TOO_LARGE)
+        chunks.append(chunk)
+
+    body = b''.join(chunks)
+    # Cache the body exactly like Request.body() does (Starlette stores it on
+    # `_body`), so subsequent request.body()/stream() calls return the same
+    # bytes instead of re-reading an already-consumed stream.
+    request._body = body  # noqa: SLF001
+    return body
 
 
 class StarletteUser(User):

src/a2a/server/routes/jsonrpc_dispatcher.py

@@ -26,6 +26,7 @@
 from a2a.server.routes.common import (
     DefaultServerCallContextBuilder,
     ServerCallContextBuilder,
+    read_request_body_with_limit,
 )
 from a2a.types.a2a_pb2 import (
     CancelTaskRequest,
@@ -224,7 +225,7 @@ async def handle_requests(self, request: Request) -> Response:  # noqa: PLR0911,
         body = None
 
         try:
-            body = await request.json()
+            body = json.loads(await read_request_body_with_limit(request))
             if isinstance(body, dict):
                 request_id = body.get('id')
                 # Ensure request_id is valid for JSON-RPC response (str/int/None only)
@@ -595,7 +596,11 @@ async def event_generator(
                         'data': json_utils.dumps(error_response),
                     }
 
-            return EventSourceResponse(event_generator(handler_result))  # ty:ignore[invalid-argument-type]
+            return EventSourceResponse(
+                event_generator(handler_result),  # ty: ignore[invalid-argument-type]
+                ping=constants.SSE_PING_INTERVAL_SECONDS,
+                send_timeout=constants.SSE_SEND_TIMEOUT_SECONDS,
+            )
 
         # handler_result is a dict (JSON-RPC response)
         return JSONResponse(handler_result)

src/a2a/server/routes/rest_dispatcher.py

@@ -10,6 +10,7 @@
 from a2a.server.routes.common import (
     DefaultServerCallContextBuilder,
     ServerCallContextBuilder,
+    read_request_body_with_limit,
 )
 from a2a.types import a2a_pb2
 from a2a.types.a2a_pb2 import (
@@ -34,16 +35,20 @@
 if TYPE_CHECKING:
     from sse_starlette.event import ServerSentEvent
     from sse_starlette.sse import EventSourceResponse
+    from starlette.exceptions import HTTPException
     from starlette.requests import Request
     from starlette.responses import JSONResponse, Response
+    from starlette.status import HTTP_413_CONTENT_TOO_LARGE
 
     _package_starlette_installed = True
 else:
     try:
         from sse_starlette.event import ServerSentEvent
         from sse_starlette.sse import EventSourceResponse
+        from starlette.exceptions import HTTPException
         from starlette.requests import Request
         from starlette.responses import JSONResponse, Response
+        from starlette.status import HTTP_413_CONTENT_TOO_LARGE
 
         _package_starlette_installed = True
     except ImportError:
@@ -52,6 +57,8 @@
         Request = Any
         JSONResponse = Any
         Response = Any
+        HTTPException = Any
+        HTTP_413_CONTENT_TOO_LARGE = Any
 
         _package_starlette_installed = False
 
@@ -98,6 +105,19 @@ def _build_call_context(self, request: Request) -> ServerCallContext:
             call_context.tenant = request.path_params['tenant']
         return call_context
 
+    async def _read_request_body(self, request: Request) -> bytes:
+        """Reads a request body with the size limit enforced.
+
+        Raises:
+            InvalidRequestError: If the body exceeds the configured limit.
+        """
+        try:
+            return await read_request_body_with_limit(request)
+        except HTTPException as e:
+            if e.status_code == HTTP_413_CONTENT_TOO_LARGE:
+                raise InvalidRequestError(message='Payload too large') from e
+            raise
+
     async def _handle_non_streaming(
         self,
         request: Request,
@@ -117,7 +137,7 @@ async def _handle_streaming(
         # This is required because Starlette's request.body() can only be consumed once,
         # and attempting to consume it after EventSourceResponse starts causes deadlock
         try:
-            await request.body()
+            await self._read_request_body(request)
         except (ValueError, RuntimeError, OSError) as e:
             raise InvalidRequestError(
                 message=f'Failed to pre-consume request body: {e}'
@@ -136,7 +156,11 @@ async def _handle_streaming(
         try:
             first_item = await anext(stream)
         except StopAsyncIteration:
-            return EventSourceResponse(iter([]))
+            return EventSourceResponse(
+                iter([]),
+                ping=constants.SSE_PING_INTERVAL_SECONDS,
+                send_timeout=constants.SSE_SEND_TIMEOUT_SECONDS,
+            )
 
         async def event_generator() -> AsyncIterator[ServerSentEvent]:
             yield ServerSentEvent(data=json_utils.dumps(first_item))
@@ -150,7 +174,11 @@ async def event_generator() -> AsyncIterator[ServerSentEvent]:
                     event='error',
                 )
 
-        return EventSourceResponse(event_generator())
+        return EventSourceResponse(
+            event_generator(),
+            ping=constants.SSE_PING_INTERVAL_SECONDS,
+            send_timeout=constants.SSE_SEND_TIMEOUT_SECONDS,
+        )
 
     @rest_error_handler
     async def on_message_send(self, request: Request) -> Response:
@@ -160,7 +188,7 @@ async def on_message_send(self, request: Request) -> Response:
         async def _handler(
             context: ServerCallContext,
         ) -> a2a_pb2.SendMessageResponse:
-            body = await request.body()
+            body = await self._read_request_body(request)
             params = a2a_pb2.SendMessageRequest()
             Parse(body, params)
             task_or_message = await self.request_handler.on_message_send(
@@ -183,7 +211,7 @@ async def on_message_send_stream(
         async def _handler(
             context: ServerCallContext,
         ) -> AsyncIterator[dict[str, Any]]:
-            body = await request.body()
+            body = await self._read_request_body(request)
             params = a2a_pb2.SendMessageRequest()
             Parse(body, params)
             async for event in self.request_handler.on_message_send_stream(
@@ -295,7 +323,7 @@ async def set_push_notification(self, request: Request) -> Response:
         async def _handler(
             context: ServerCallContext,
         ) -> a2a_pb2.TaskPushNotificationConfig:
-            body = await request.body()
+            body = await self._read_request_body(request)
             params = a2a_pb2.TaskPushNotificationConfig()
             Parse(body, params)
             params.task_id = request.path_params['id']

src/a2a/utils/constants.py

@@ -11,6 +11,24 @@
 MAX_LIST_TASKS_PAGE_SIZE = 100
 """Maximum page size for the `tasks/list` method."""
 
+MAX_REQUEST_BODY_SIZE = 10 * 1024 * 1024
+"""Maximum accepted HTTP request body size in bytes (10 MiB).
+
+Requests with a body larger than this are rejected with HTTP 413
+(payload too large) instead of being buffered unboundedly.
+"""
+
+SSE_PING_INTERVAL_SECONDS = 15
+"""Heartbeat interval for SSE streams, in seconds.
+
+A comment-only ``: ping`` frame is sent on every interval to keep the
+connection alive and detect dead clients.
+"""
+
+SSE_SEND_TIMEOUT_SECONDS = 300
+"""Maximum time in seconds without a successful send before an SSE
+stream is torn down, preventing zombie streams from accumulating."""
+
 
 class TransportProtocol(str, Enum):
     """Transport protocol string constants."""

tests/compat/v0_3/test_jsonrpc_app_compat.py

@@ -143,3 +143,41 @@ def test_get_extended_agent_card_v03_compat(
     assert 'result' in data
     # The result should be a v0.3 AgentCard
     assert 'supportsAuthenticatedExtendedCard' in data['result']
+
+
+@pytest.mark.anyio
+async def test_v03_jsonrpc_streaming_response_carries_ping_and_send_timeout():
+    """v0.3 JSON-RPC streaming must configure SSE keep-alive like the v1 route.
+
+    ``ping=None`` (the ``sse_starlette`` default) leaves compat-enabled
+    deployments without heartbeats, so those streams keep the previous
+    dead-connection behaviour the transport hardening removes elsewhere.
+    """
+    from a2a.compat.v0_3 import types as types_v03
+    from a2a.compat.v0_3.jsonrpc_adapter import JSONRPC03Adapter
+    from a2a.server.context import ServerCallContext
+    from a2a.utils import constants
+
+    handler = AsyncMock(spec=RequestHandler)
+
+    async def stream(request_obj, context):
+        yield Message10(message_id='1', role=Role10.ROLE_AGENT, parts=[])
+
+    handler.on_message_send_stream.side_effect = stream
+
+    payload = types_v03.SendStreamingMessageRequest(
+        id='1',
+        params=types_v03.MessageSendParams(
+            message=types_v03.Message(
+                message_id='1', role=types_v03.Role.user, parts=[]
+            )
+        ),
+    )
+
+    adapter = JSONRPC03Adapter(handler)
+    response = await adapter._process_streaming_request(
+        '1', payload, ServerCallContext()
+    )
+
+    assert response.ping_interval == constants.SSE_PING_INTERVAL_SECONDS
+    assert response.send_timeout == constants.SSE_SEND_TIMEOUT_SECONDS

tests/compat/v0_3/test_rest_routes_compat.py

@@ -19,6 +19,7 @@
     TaskState as TaskState10,
     TaskStatus as TaskStatus10,
 )
+from a2a.utils import constants
 from fastapi import FastAPI
 from google.protobuf import json_format
 from httpx import ASGITransport, AsyncClient
@@ -242,3 +243,31 @@ async def stream_with_non_ascii(
         payload = payload.decode('utf-8')
     assert non_ascii_text in payload
     assert '\\u4f60\\u597d' not in payload
+
+
+@pytest.mark.anyio
+async def test_v03_streaming_response_carries_ping_and_send_timeout(
+    request_handler: RequestHandler,
+) -> None:
+    """v0.3 REST streaming must configure SSE keep-alive like the v1 route.
+
+    Without explicit ``ping``/``send_timeout`` the v0.3 streams keep
+    ``sse_starlette`` defaults, so a compat-enabled deployment loses the
+    heartbeat behaviour the transport hardening adds elsewhere.
+    """
+    adapter = REST03Adapter(http_handler=request_handler)
+
+    async def stream(request: Request, context: object) -> AsyncIterator[dict]:
+        yield {'msg': {'text': 'hello'}}
+
+    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, mock_req)
+
+    assert response.ping_interval == constants.SSE_PING_INTERVAL_SECONDS
+    assert response.send_timeout == constants.SSE_SEND_TIMEOUT_SECONDS