[Bug]: Push Notifications Only sent in on_message_send_stream, not sent in on_message_send

[njbrake]

What happened?

I'm not sure if this is a bug or a missing description of the feature (or a miss by me to find this documentation).

I'm working on getting support for Push Notifications in my a2a server mozilla-ai/any-agent#414

The way I understand it, the InMemoryPushNotifier has the send_notification method which sends the notification back to all subscribed clients. However, this method is only called from within the DefaultRequestHandler.on_message_send_stream method, it is not used in the on_message call.

Am I trying to use push notifications the wrong way, or is this a bug? It seems like the on_message call should support sending push notifications, to support a scenario like below:

I send a message, subscribe the task for push notifications, and then send another message.

client = await A2AClient.get_client_from_agent_card_url(
      httpx_client, server_url
  )
  # Configure push notifications in the initial message/send request
  # following the A2A specification example
  first_message_id = str(uuid4())
  context_id = str(uuid4())
  task_id = str(uuid4())
  params = MessageSendParams(
      message=Message(
          role="user",
          parts=[
              Part(
                  kind="text",
                  text="Generate a Q1 sales report. This usually takes a while. Notify me when it's ready.",
              )
          ],
          messageId=first_message_id,
          contextId=context_id,
          taskId=task_id,
      ),
      configuration=MessageSendConfiguration(
          acceptedOutputModes=["text"],
          pushNotificationConfig=PushNotificationConfig(url=webhook_url),
      ),
  )

  request_1 = SendMessageRequest(id=str(uuid4()), params=params)
  response_1 = await client.send_message(request_1)
  assert response_1.root.result.id == task_id

  push_notification_config = PushNotificationConfig(url=webhook_url)

  params_model = TaskPushNotificationConfig(
      taskId=task_id, pushNotificationConfig=push_notification_config
  )

  request_0 = SetTaskPushNotificationConfigRequest(id="", params=params_model)
  response_0 = await client.set_task_callback(request_0)
  assert not hasattr(response_0, "error")

  # now ask the agent a question on the same taskid
  request_1 = SendMessageRequest(id=str(uuid4()), params=params)
  response_1 = await client.send_message(request_1)
  assert response_1.root.result.id == task_id

Thanks!

Relevant log output

Code of Conduct

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

[mikeas1]

Agreed this makes sense to change, but I wanted to point out an issue with your example: clients do not dictate the taskId that gets used for Tasks (they did in our v0.1 of the protocol). Instead, the server generates this when it decides that a task is warranted. This was the major change from v0.1 to v0.2. The purpose of the taskId field in the Message type is to reference an incomplete task that was previously created by the server.

So, because of this, push notifications for message/send requests are in an odd state. The only way you could register for push notifications for a message sent via message/send is to have a taskId for an incomplete task. That could happen via:

1. The client calls message/send with the initial request.
2. The agent decides a task is warranted and so creates one on its side. Since the client is not receiving a stream of events, the client doesn't actually know this happened. It's just blocked waiting for the call to return.
3. The agent hits a state that requires returning a response to the caller, but the agent has not actually completed the work (specifically, either the Task reaches an input-required or an auth-required state).
4. The client's call to the agent returns, and it can see that a Task has been created and is not yet complete.
5. The client registers for push notifications for the task.
6. (for input-required tasks) The client calls message/send again, referencing the incomplete task and providing the requested input.
7. The agent receives the message and continues work. Now, when updates are made to the task they should be sent to the client via push notifications.

I still think receiving push notifications in this situation should be possible, so I agree we need to make the change. But I want to flag that the situation where this occurs is odd. There's an opportunity to improve the protocol here, I think -- both documentation and protocol specifics.

[njbrake]

Thank you for all the details! In my case I was specifically trying to address situation 6.

Should there be an error thrown if the client tries to pass a taskId that doesn't exist? It sounds like I wasn't supposed to be able to create a new taskID from the client, but IIRC I was able to do so without error in the current code.

[mikeas1]

The spec actually doesn't specify what happens in this case. We're working on some additional details here, but mostly focused on what happens when you reference a finalized Task.

I believe in the current implementation of the SDK a taskID for a non-existent task acts the same as not setting the taskID at all. The response you get from the agent will include a new random taskID (if it returns a Task).

[njbrake]

Hmm. I don't think that's the case? I wrote an integration "test" just to exercise the current behavior: https://github.com/mozilla-ai/any-agent/blob/414-a2a-push-not/tests/integration/a2a/test_push_notification.py#L127

In that, I generate a taskID, pass it from the client to the server, and then assert that the task ID I get back from the server is the one I sent it assert response_1.root.result.id == task_id

[mikeas1]

Looks like you're right!

So, in your case you are explicitly using the task ID from the request to create a new task, which certainly explains that behavior: https://github.com/mozilla-ai/any-agent/blob/414-a2a-push-not/src/any_agent/serving/agent_executor.py#L46

Our demo code often uses the context.task_id field to determine the task ID to use. But... it seems like we still fall back to the task ID provided by the client even in that case: https://github.com/a2aproject/a2a-python/blob/main/src/a2a/server/agent_execution/context.py#L145

[fernando-torres-blip-ai]

I'm not sure if this is the most appropriate thread to bring this up, but I’d like to point out another behavior regarding push notifications that isn’t very clear.

According to the spec and the SDK code, push notifications are only sent for "final" states when the request is not in streaming mode.

In the SDK, this is due to the use of result_aggregator.consume_and_break_on_interrupt (code reference), which waits for the end of consumer.consume_all() — a generator that only finishes when an is_final_event is received (code reference).

But what about "intermediary" messages?

For example:

Agent: "Hey user! Please wait a minute while I check a few things."

Agent: "Hey user, I’ve already completed step 1 of your request and everything looks good! Now I’ll move on to step 2 — please hold on."

These aren't input-required or completed status events.
So, would these kinds of messages only be supported in streaming mode?

[njbrake]

@fernando-torres-blip-ai ooo very interesting question! I think this is a slight divergence from the topic I intended to discuss in this issue, but my Github issue title is very generic so it's my fault for not being specific in my description about what issue I was trying to resolve.

If you're ok with it, can we create a new issue to discuss this? I have some thoughts.

[fernando-torres-blip-ai]

Great @njbrake , totally agree ! Thank you for the reply.

I've opened this feature request #239 please feel free to help me adjust the descriptions or format to discuss this other topic.