fix(json-rpc): add stream_send_timeout for JSON-RPC streaming endpoints

src/a2a/server/apps/jsonrpc/fastapi_app.py

@@ -66,38 +66,41 @@
     (SSE).
     """
 
     def __init__(  # noqa: PLR0913
         self,
         agent_card: AgentCard,
         http_handler: RequestHandler,
         extended_agent_card: AgentCard | None = None,
         context_builder: CallContextBuilder | None = None,
         card_modifier: Callable[[AgentCard], AgentCard] | None = None,
         extended_card_modifier: Callable[
             [AgentCard, ServerCallContext], AgentCard
         ]
         | None = None,
         max_content_length: int | None = 10 * 1024 * 1024,  # 10MB
+        stream_send_timeout: float | None = None,
     ) -> None:
         """Initializes the A2AFastAPIApplication.
 
         Args:
             agent_card: The AgentCard describing the agent's capabilities.
             http_handler: The handler instance responsible for processing A2A
               requests via http.
             extended_agent_card: An optional, distinct AgentCard to be served
               at the authenticated extended card endpoint.
             context_builder: The CallContextBuilder used to construct the
               ServerCallContext passed to the http_handler. If None, no
               ServerCallContext is passed.
             card_modifier: An optional callback to dynamically modify the public
               agent card before it is served.
             extended_card_modifier: An optional callback to dynamically modify
               the extended agent card before it is served. It receives the
               call context.
             max_content_length: The maximum allowed content length for incoming
               requests. Defaults to 10MB. Set to None for unbounded maximum.
+            stream_send_timeout: The timeout in seconds for sending events in default timeout. Set to a larger value or None to disable for
+              long-running agents.
         """
         if not _package_fastapi_installed:
             raise ImportError(
                 'The `fastapi` package is required to use the `A2AFastAPIApplication`.'
@@ -112,6 +115,7 @@
             card_modifier=card_modifier,
             extended_card_modifier=extended_card_modifier,
             max_content_length=max_content_length,
+            stream_send_timeout=stream_send_timeout,
         )
 
     def add_routes_to_app(

src/a2a/server/apps/jsonrpc/jsonrpc_app.py

@@ -172,38 +172,43 @@
         for model in A2ARequestModel.__args__
     }
 
     def __init__(  # noqa: PLR0913
         self,
         agent_card: AgentCard,
         http_handler: RequestHandler,
         extended_agent_card: AgentCard | None = None,
         context_builder: CallContextBuilder | None = None,
         card_modifier: Callable[[AgentCard], AgentCard] | None = None,
         extended_card_modifier: Callable[
             [AgentCard, ServerCallContext], AgentCard
         ]
         | None = None,
         max_content_length: int | None = 10 * 1024 * 1024,  # 10MB
+        stream_send_timeout: float | None = None,
     ) -> None:
         """Initializes the JSONRPCApplication.
 
         Args:
             agent_card: The AgentCard describing the agent's capabilities.
             http_handler: The handler instance responsible for processing A2A
               requests via http.
             extended_agent_card: An optional, distinct AgentCard to be served
               at the authenticated extended card endpoint.
             context_builder: The CallContextBuilder used to construct the
               ServerCallContext passed to the http_handler. If None, no
               ServerCallContext is passed.
             card_modifier: An optional callback to dynamically modify the public
               agent card before it is served.
             extended_card_modifier: An optional callback to dynamically modify
               the extended agent card before it is served. It receives the
               call context.
             max_content_length: The maximum allowed content length for incoming
               requests. Defaults to 10MB. Set to None for unbounded maximum.
+            stream_send_timeout: The timeout in seconds for sending events in
+              streaming responses. Defaults to None, which uses Starlette's
+              default timeout. Set to a larger value or None to disable for
+              long-running agents.
         """
         if not _package_starlette_installed:
             raise ImportError(
                 'Packages `starlette` and `sse-starlette` are required to use the'
@@ -222,6 +227,7 @@
         )
         self._context_builder = context_builder or DefaultCallContextBuilder()
         self._max_content_length = max_content_length
+        self.stream_send_timeout = stream_send_timeout
 
     def _generate_error_response(
         self, request_id: str | int | None, error: JSONRPCError | A2AError
@@ -540,8 +546,14 @@
                 async for item in stream:
                     yield {'data': item.root.model_dump_json(exclude_none=True)}
 
+            send_timeout = context.state.get(
+                'stream_send_timeout', self.stream_send_timeout
+            )
+
             return EventSourceResponse(
-                event_generator(handler_result), headers=headers
+                event_generator(handler_result),
+                headers=headers,
+                send_timeout=send_timeout,
             )
         if isinstance(handler_result, JSONRPCErrorResponse):
             return JSONResponse(

src/a2a/server/apps/jsonrpc/starlette_app.py

@@ -48,18 +48,19 @@
     (SSE).
     """
 
     def __init__(  # noqa: PLR0913
         self,
         agent_card: AgentCard,
         http_handler: RequestHandler,
         extended_agent_card: AgentCard | None = None,
         context_builder: CallContextBuilder | None = None,
         card_modifier: Callable[[AgentCard], AgentCard] | None = None,
         extended_card_modifier: Callable[
             [AgentCard, ServerCallContext], AgentCard
         ]
         | None = None,
         max_content_length: int | None = 10 * 1024 * 1024,  # 10MB
+        stream_send_timeout: float | None = None,
     ) -> None:
         """Initializes the A2AStarletteApplication.
 
@@ -79,6 +80,10 @@
               call context.
             max_content_length: The maximum allowed content length for incoming
               requests. Defaults to 10MB. Set to None for unbounded maximum.
+            stream_send_timeout: The timeout in seconds for sending events in
+              streaming responses. Defaults to None, which uses Starlette's
+              default timeout. Set to a larger value or None to disable for
+              long-running agents.
         """
         if not _package_starlette_installed:
             raise ImportError(
@@ -94,6 +99,7 @@
             card_modifier=card_modifier,
             extended_card_modifier=extended_card_modifier,
             max_content_length=max_content_length,
+            stream_send_timeout=stream_send_timeout,
         )
 
     def routes(

tests/server/apps/jsonrpc/test_fastapi_app.py

@@ -75,6 +75,23 @@ def test_create_a2a_fastapi_app_with_missing_deps_raises_importerror(
         ):
             _app = A2AFastAPIApplication(**mock_app_params)
 
+    def test_stream_send_timeout_parameter(self, mock_app_params: dict):
+        try:
+            app_default = A2AFastAPIApplication(**mock_app_params)
+            assert app_default.stream_send_timeout is None
+
+            app_custom = A2AFastAPIApplication(
+                **mock_app_params, stream_send_timeout=30.0
+            )
+            assert app_custom.stream_send_timeout == 30.0
+
+            app_none = A2AFastAPIApplication(
+                **mock_app_params, stream_send_timeout=None
+            )
+            assert app_none.stream_send_timeout is None
+        except ImportError:
+            pytest.skip('FastAPI dependencies not available')
+
 
 if __name__ == '__main__':
     pytest.main([__file__])

tests/server/apps/jsonrpc/test_starlette_app.py

@@ -77,6 +77,23 @@ def test_create_a2a_starlette_app_with_missing_deps_raises_importerror(
         ):
             _app = A2AStarletteApplication(**mock_app_params)
 
+    def test_stream_send_timeout_parameter(self, mock_app_params: dict):
+        try:
+            app_default = A2AStarletteApplication(**mock_app_params)
+            assert app_default.stream_send_timeout is None
+
+            app_custom = A2AStarletteApplication(
+                **mock_app_params, stream_send_timeout=30.0
+            )
+            assert app_custom.stream_send_timeout == 30.0
+
+            app_none = A2AStarletteApplication(
+                **mock_app_params, stream_send_timeout=None
+            )
+            assert app_none.stream_send_timeout is None
+        except ImportError:
+            pytest.skip('Starlette dependencies not available')
+
 
 if __name__ == '__main__':
     pytest.main([__file__])