[Feat]: Callback for completed sendMessage request

[DataWorm]

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

When I implement the a2a client and send a message/prompt to the remote agent (LLM), then I want to know when my request has been handled completely. With the hello world example server, I get one message event and that is all. In case of streaming I have no reliable way to determine if I should expect further events or if the connection has already been closed.

Describe the solution you'd like

I just need a reliable event/callback/notification that is triggered when the processing of my request has finalized (completed, failed, connection interrupted...). It could be another consumer that I can add as a third consumer parameter, it could be a new kind of ClientEvent, the sendMessage method could return a CompletableFuture object, there could be a reactive interface using Mono/Flux from Project Reactor...

Describe alternatives you've considered

No response

Additional context

No response

Code of Conduct

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

[fjuma]

There's a more involved A2A Java Client example here:

https://github.com/a2aproject/a2a-samples/blob/main/samples/java/agents/dice_agent_multi_transport/client/src/main/java/com/samples/a2a/client/TestClient.java#L156-L196

When creating your event consumers that handle responses received from the server, it's possible to detect when a task has reached a final state. In particular, when handling a TaskUpdateEvent that is a TaskStatusUpdateEvent, we can check if this is a final update. If true, this is the final event in the stream for this interaction. There's an example of this here:

https://github.com/a2aproject/a2a-samples/blob/main/samples/java/agents/dice_agent_multi_transport/client/src/main/java/com/samples/a2a/client/TestClient.java#L173

[DataWorm]

Hi fjuma,

thank you for your feedback. I already found those Task Status Updates but apparently messages are not neccessarily tasks. And at least in non-streaming mode they don't seem to produce any task updates. Can I expect that in case of streaming I get a Message object that contains a task id and then of course also a task status update event that tells me when streaming has completed? In that case everything should be fine but considering the current documentation, that does not seem so obvious to me.

[fjuma]

A Message represents a single turn of communication. An agent returns a Message as opposed to a Task when it can respond immediately. It is used for immediate, self-contained interactions that conclude without requiring further state management.

For the streaming case, a Message can only be returned in a stream in response to Message that does not reference a Task or does not cause a Task to be created.

https://a2a-protocol.org/latest/topics/key-concepts/#messages-and-parts
https://a2a-protocol.org/latest/topics/life-of-a-task/

[scarvel8]

Hello, so we have a sort of similar problem ourselves where we use this library with Spring WebFlux and in streaming contexts where we want to add certain headers for tracing/authentication we don't properly close out our reactive streams because our Flux's never complete, in our tracing use case we need context to restore when the stream completes normally versus an error
We would like to distinguish between normal completions, errors, or connection issues like early termination/timeouts/etc.

The HTTP layer has the completion callback in A2AHttpClient.java, but its discarded at the transport layer in JSONRPCTransport.java and not exposed at ClientTransport.java, similar issue with the resubscribe method

We'd like to add an optional completion handler method to streaming methods to handle reactive streams properly as DataWorm was suggesting above

If the SSE connection does before the a final TaskStatusUpdateEvent sends we have no way of detecting it, adding a timeout for Message responses seems fragile and the fix would be pretty simple, just need to add lines to thread through the completion signal from the http layer, and it would be backwards compatible and not a breaking change

I can put a PR in for this if it helps