fix: return task immediately when blocking=False in on_message_send

When `blocking=False` is set in MessageSendConfiguration, the handler
now returns the Task object immediately without waiting for executor
events. Event consumption and push notifications are processed entirely
in the background.

Previously, even with `blocking=False`, the handler waited for the first
event from the EventConsumer polling loop (0.5s timeout per iteration).
When the event loop was busy with background work, this caused 5-7s
delays before the HTTP response was sent, leading to client timeouts.

The new non-blocking fast path:
1. Creates the task and starts the executor (via _setup_message_execution)
2. Returns the task in 'submitted' state immediately
3. Consumes events and sends push notifications in a background task

The blocking path remains unchanged for backward compatibility.

Fixes #951

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: ruimgf

src/a2a/server/request_handlers/default_request_handler.py

@@ -41,6 +41,7 @@
     TaskPushNotificationConfig,
     TaskQueryParams,
     TaskState,
+    TaskStatus,
     UnsupportedOperationError,
 )
 from a2a.utils.errors import ServerError
@@ -295,6 +296,13 @@ async def on_message_send(
 
         Starts the agent execution for the message and waits for the final
         result (Task or Message).
+
+        When ``blocking`` is ``False``, the handler returns the task
+        immediately without waiting for executor events and processes
+        everything in the background. Results are delivered via push
+        notifications. This avoids the latency introduced by the
+        ``EventConsumer`` polling loop which can add seconds of delay
+        when the event loop is busy with other work.
         """
         (
             _task_manager,
@@ -311,6 +319,46 @@ async def on_message_send(
         if params.configuration and params.configuration.blocking is False:
             blocking = False
 
+        # Non-blocking fast path: return the task immediately and process
+        # events entirely in the background via push notifications.
+        if not blocking:
+            task = await _task_manager.get_task()
+            if not task:
+                task = Task(
+                    id=task_id,
+                    context_id=params.message.context_id,
+                    status=TaskStatus(state=TaskState.submitted),
+                    history=[params.message],
+                )
+
+            async def _background_consume() -> None:
+                try:
+                    async for _event in result_aggregator.consume_and_emit(
+                        consumer
+                    ):
+                        await self._send_push_notification_if_needed(
+                            task_id, result_aggregator
+                        )
+                except Exception:
+                    logger.exception(
+                        'Background event consumption failed for task %s',
+                        task_id,
+                    )
+                finally:
+                    await self._cleanup_producer(producer_task, task_id)
+
+            bg_task = asyncio.create_task(_background_consume())
+            bg_task.set_name(f'non_blocking_consume:{task_id}')
+            self._track_background_task(bg_task)
+
+            if params.configuration:
+                task = apply_history_length(
+                    task, params.configuration.history_length
+                )
+
+            return task
+
+        # Blocking path: wait for completion or interruption.
         interrupted_or_non_blocking = False
         try:
             # Create async callback for push notifications
@@ -325,7 +373,7 @@ async def push_notification_callback() -> None:
                 bg_consume_task,
             ) = await result_aggregator.consume_and_break_on_interrupt(
                 consumer,
-                blocking=blocking,
+                blocking=True,
                 event_callback=push_notification_callback,
             )