fix: non-blocking send_message server handler not invoke push notification

- Problem: When client use send_message with MessageSendConfiguration.blocking=False, the result_aggregator will enter the logic of _continue_consuming. But it's not push notification to client. The client can't get notification for long-running task(non-blocking invoke) at this situation.
- Solution: Simply add push notification logic to result_aggregator is okay.

Author: weimch

src/a2a/server/request_handlers/default_request_handler.py

@@ -231,7 +231,10 @@ async def _setup_message_execution(
             )
 
         queue = await self._queue_manager.create_or_tap(task_id)
-        result_aggregator = ResultAggregator(task_manager)
+        result_aggregator = ResultAggregator(
+            task_manager,
+            push_sender=self._push_sender,
+        )
         # TODO: to manage the non-blocking flows.
         producer_task = asyncio.create_task(
             self._run_event_stream(request_context, queue)

src/a2a/server/tasks/result_aggregator.py

@@ -4,6 +4,7 @@
 from collections.abc import AsyncGenerator, AsyncIterator
 
 from a2a.server.events import Event, EventConsumer
+from a2a.server.tasks.push_notification_sender import PushNotificationSender
 from a2a.server.tasks.task_manager import TaskManager
 from a2a.types import Message, Task, TaskState, TaskStatusUpdateEvent
 
@@ -24,14 +25,20 @@ class ResultAggregator:
        Task object and emit that Task object.
     """
 
-    def __init__(self, task_manager: TaskManager):
+    def __init__(
+        self,
+        task_manager: TaskManager,
+        push_sender: PushNotificationSender | None = None,
+    ) -> None:
         """Initializes the ResultAggregator.
 
         Args:
             task_manager: The `TaskManager` instance to use for processing events
                           and managing the task state.
+            push_sender: The `PushNotificationSender` instance to use for sending push notifications.
         """
         self.task_manager = task_manager
+        self.push_sender = push_sender
         self._message: Message | None = None
 
     @property
@@ -168,3 +175,11 @@ async def _continue_consuming(
         """
         async for event in event_stream:
             await self.task_manager.process(event)
+            await self._send_push_notification_if_needed()
+
+    async def _send_push_notification_if_needed(self) -> None:
+        """Sends push notification if configured and task is available."""
+        if self.push_sender:
+            latest_task = await self.current_result
+            if isinstance(latest_task, Task):
+                await self.push_sender.send_notification(latest_task)