[Feat]: Non blocking with intermediary responses (via push notification)

[fernando-torres-blip-ai]

Is your feature request related to a problem? Please describe.

I want to avoid long-lived persistent connections (SSE) but also be able to receive multiple Agent Responses.

Imagine I sended a request to an Agent to perform a heavy task with multiple steps. I want to receive intermediary messages where the agent can notify steps beeing finished (between some tool calls for example).

Today, the push-notification only works for "final" states like task completed or input required. And streaming responses require persistent connections via SSE.

Describe the solution you'd like

Use the already defined blocking attribute of MessageSendConfiguration to indicate whether the request processing should last for the entire event stream or stop immediately at the task submitted status.

Then, push notifications should also be able to send artifacts generated during processing.

Describe alternatives you've considered

No response

Additional context

Some points to help us decide the best solution.

The protocol specification metion "partial responses" as a good use case for streaming spec reference

When to Use Streaming:

• Interactive, conversational exchanges where immediate feedback or partial responses are beneficial.

Also in protocol specification is mentioned that push notification payloads are not strictly task objects, so I belive that TaskStatusUpdateEvent and TaskArtifactUpdateEvent could also be sent in streaming request push notification ? spec reference

Notification Payload: The A2A protocol itself does not strictly define the HTTP body payload of the push notification sent by the server to the client's webhook. However, the notification SHOULD contain sufficient information for the client to identify the Task ID and understand the general nature of the update (e.g., the new TaskState). Servers might send a minimal payload (just Task ID and new state) or a more comprehensive one (e.g., a summary or even the full Task object).

Code of Conduct

• I agree to follow this project's Code of Conduct

[njbrake]

Ok, if I understand you correctly: you are saying that when a client invokes the 'execute' function of the executor, if the client request is send_message and not send_message_stream, all of the partial notifications don't go out to anyone who may be subscribed to push notifications about the task?

So if we look at my hacky example script https://github.com/mozilla-ai/any-agent/blob/414-a2a-push-not/tests/integration/a2a/test_push_notification.py#L54

In that test,

• I send a non-streaming message and a task is created.
• I make another call to subscribe to push notifications about the task.
• I make another non-streaming message with the same task id, and verify that I get at least one push notification

What you are saying is that in this scenario, even if there are multiple task updates caused by me making that second non-streaming message, I would only get one push notification about the final result of the call?

I haven't checked this behavior so I don't know if this is true or false. Thanks!

[fernando-torres-blip-ai]

Exactly !

Using the actual sdk, I am unable to define an implementation capable of carrying out the flow below

[Request - Response]
Client --Message--> Remote Agent --Task Submitted--> Client

[Push Notifications]
Remote Agent --Artifact--> Client (This beeing a complete artifact response, not a chunk)
Remote Agent --Artifact--> Client (This beeing a complete artifact response, not a chunk)
Remote Agent --Artifact--> Client (This beeing a complete artifact response, not a chunk)
Remote Agent --Task Completed--> Client

[fernando-torres-blip-ai]

I’ve been debugging the SDK and exploring possible implementations, but I’ve run into a couple of roadblocks.

Option 1: Using message/send

I don’t think this will work because of the consume_and_break_on_interrupt method in (see source).

await result_aggregator.consume_and_break_on_interrupt(consumer)

That method loops over the generator returned by consumer.consume_all() (see source), and consume_all() only stops when it encounters an interrupt or terminal state:

is_final_event = (
    (isinstance(event, TaskStatusUpdateEvent) and event.final)
    or isinstance(event, Message)
    or (
        isinstance(event, Task)
        and event.status.state
        in (
            TaskState.completed,
            TaskState.canceled,
            TaskState.failed,
            TaskState.rejected,
            TaskState.unknown,
            TaskState.input_required,
        )
    )
)

Because of this, you can’t send intermediate events—only “final” ones.

Option 2: Using message/stream (with SSE)

This approach relies on a persistent SSE connection, which I’m trying to avoid. It seems there’s no non-blocking request option.

Also, result_aggregator.current_result always returns a Task, so only Task objects ever get pushed:

latest_task = await result_aggregator.current_result
if isinstance(latest_task, Task):
    await self._push_notifier.send_notification(latest_task)

That means the client will receive a Task both when you call updater.add_artifact() and again when you call updater.complete(), making it hard to tell when actual “artifact” events should be forwarded to the end user.

Am I missing something here?

[lukehinds]

I if understand correctly, I did come across this issue too. DefaultRequestHandler has a gap in push notification support for non-streaming message/send requests. When a new task is created (as opposed to updating an existing task), the push notification configuration is not stored, preventing webhook delivery for task completion events.

This was my own hacky approach, but I am unsure if I am foot gunning myself here as I had to drop the work with a plan to return later and really dig in. I apologies if you're having a different issue and I am clouding things up:

import logging
from typing import Union

from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.context import ServerCallContext
from a2a.types import (
    Message,
    Task,
    MessageSendParams,
)

logger = logging.getLogger(__name__)


class CustomRequestHandler(DefaultRequestHandler):
    """
    Enhanced request handler that fixes push notification storage for non-streaming endpoints.

    The default a2a-sdk DefaultRequestHandler only stores push notification configs
    for existing tasks in non-streaming mode, not for new tasks. This fixes that, but
    we should eventually upstream this change to the a2a-sdk, or find out why it
    has not been done already.
    """

    async def on_message_send(
        self,
        params: MessageSendParams,
        context: ServerCallContext | None = None,
    ) -> Union[Message, Task]:
        """
        Override to ensure push notification configs are saved for new tasks.
        """
        # Call parent implementation
        result = await super().on_message_send(params, context)

        # If we got a task back and have push notification config, save it
        if (
            isinstance(result, Task)
            and self._push_notifier
            and params.configuration
            and params.configuration.pushNotificationConfig
        ):
            logger.info(f"Saving push notification config for new task {result.id}")
            await self._push_notifier.set_info(
                result.id,
                params.configuration.pushNotificationConfig,
            )

            # Also send the initial notification until upsteam fixes this
            await self._push_notifier.send_notification(result)

        return result

[lukehinds]

Ahh, @njbrake has already got this fixed! noice!

[lambertjf]

Exactly !

Using the actual sdk, I am unable to define an implementation capable of carrying out the flow below

[Request - Response]
Client --Message--> Remote Agent --Task Submitted--> Client

[Push Notifications]
Remote Agent --Artifact--> Client (This beeing a complete artifact response, not a chunk)
Remote Agent --Artifact--> Client (This beeing a complete artifact response, not a chunk)
Remote Agent --Artifact--> Client (This beeing a complete artifact response, not a chunk)
Remote Agent --Task Completed--> Client

Been trying to implement something like the above. I can see the other day they fixed the issue of push notifications simply not being sent with message/send but it still seems like this flow is not possible. I can't seem to find any way to send any messages before agent execution is finished (agentexecutor execute function) when I send non-streaming requests.

Is this functionality possible? Or is it even intended? I'm imagining an agent that takes multiple hours to run and we want it to just respond with an acknowledgement when the client sends a message, then can send push notifications the rest of the way - that seems like what the push notifications should have been for, but it seems nothing gets sent to the push notification server until the execute function is finished.

My hacky solution is to just manually send the push notifications inside the execution function - otherwise they only get sent at the end.. basically just run this function every time I update the task status: await self.push_sender._dispatch_notification(task, push_config)