feat: Make A2AClient agnostic to the transport protocol #228
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Summary of Changes
Hello @brasseld, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request introduces a significant architectural refactoring to the A2A SDK client, making it agnostic to the underlying transport protocol. By introducing a Transport Service Provider Interface (SPI) and moving protocol-specific implementations (JSON-RPC and gRPC) into dedicated modules, the core A2AClient is decoupled from communication details. This enhances modularity, simplifies the client's public API by returning direct domain objects, and sets the foundation for easier integration of future transport protocols.
Highlights
- Introduction of Transport Service Provider Interface (SPI): A new
transport/spimodule has been introduced, defining aTransportinterface. This interface serves as a Service Provider Interface (SPI), establishing a common contract for all A2A client operations, thereby abstracting away the underlying communication protocol details. - Modularization of Transport Protocols: Protocol-specific implementations, previously embedded or tightly coupled, have been extracted and refactored into dedicated modules:
transport/jsonrpcandtransport/grpc. This includes moving HTTP client interfaces, JDK HTTP client implementations, and SSE event listeners to thetransport/jsonrpcmodule, and gRPC-related client and server handlers to thetransport/grpcmodule. NewJSONRPCTransportandGrpcTransportclasses have been added, each implementing theTransportinterface. - A2AClient Refactoring for Protocol Agnosticism: The core
A2AClienthas been significantly refactored to depend solely on the newTransportinterface. This change removes all direct HTTP request handling and JSON unmarshalling logic from the client, delegating these responsibilities to the injected transport implementation. This makes theA2AClienttruly agnostic to the communication protocol. - Simplified Client API Return Types: The API of several
A2AClientmethods has been simplified. Instead of returning wrapper response objects (e.g.,SendMessageResponse), methods now directly return the relevant domain objects (e.g.,EventKind,Task,TaskPushNotificationConfig,List<TaskPushNotificationConfig>). This reduces boilerplate and improves the usability of the client API. - Removal of Standalone A2AGrpcClient: The standalone
A2AGrpcClientclass has been removed. Its functionality is now integrated into the new modular transport architecture, specifically within theGrpcTransportimplementation, aligning with the goal of a unified, transport-agnostic client.
Using Gemini Code Assist
The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.
Invoking Gemini
You can request assistance from Gemini at any point in your pull request via creating an issue comment (i.e. comment on the pull request page) using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands.
| Feature | Command | Description |
|---|---|---|
| Code Review | /gemini review |
Performs a code review for the current pull request in its current state. |
| Pull Request Summary | /gemini summary |
Provides a summary of the current pull request in its current state. |
| Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in issue comments and review comments. |
| Help | /gemini help |
Displays a list of available commands. |
Customization
To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.
Limitations & Feedback
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.
You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.
Footnotes
-
Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩
There was a problem hiding this comment.
Code Review
This pull request refactors the A2A client to be transport-agnostic by introducing a Transport SPI. This is a great architectural improvement. However, the implementation has several critical issues and incomplete parts. The Transport interface design is too specific to JSON-RPC, causing compilation errors in the gRPC implementation. The GrpcTransport has multiple unimplemented methods and incorrect method signatures. Furthermore, the A2AClient refactoring is incomplete, leaving it with broken functionality for fetching the agent card and no way to actually use a different transport, defeating the purpose of the abstraction. These issues need to be addressed to make the feature complete and functional.
| /** | ||
| * An A2A client. | ||
| */ | ||
| public class A2AClient { |
There was a problem hiding this comment.
The A2AClient refactoring is incomplete and has introduced some critical issues:
- Broken
getAgentCard()method: ThegetAgentCard()method (not modified in this PR, but affected by it) now causes a compilation error because it tries to use thehttpClientandagentUrlfields, which have been removed from this class. - No transport injection: The client is not truly transport-agnostic from a user's perspective. The constructors hardcode
JSONRPCTransport, and there is no way to provide a differentTransportimplementation (e.g.,GrpcTransport).
To fix this, I suggest the following:
- Add a new constructor to allow injecting a
Transportinstance:
public A2AClient(Transport transport, AgentCard agentCard) {
this.transport = transport;
this.agentCard = agentCard;
}- Refactor the
getAgentCard()method to work with the new design or remove it if it's no longer intended to be used this way.
| } | ||
|
|
||
| @Override | ||
| public TaskPushNotificationConfig setTaskPushNotificationConfig(String requestId, String taskId, TaskPushNotificationConfig taskPushNotificationConfig) throws A2AServerException { |
There was a problem hiding this comment.
The method signature for setTaskPushNotificationConfig does not match the Transport interface it implements. The interface expects (String requestId, String taskId, PushNotificationConfig pushNotificationConfig), but the implementation here is (String requestId, String taskId, TaskPushNotificationConfig taskPushNotificationConfig). This will cause a compilation error.
You should change the signature to match the interface and construct the TaskPushNotificationConfig object inside the method.
public TaskPushNotificationConfig setTaskPushNotificationConfig(String requestId, String taskId, PushNotificationConfig pushNotificationConfig) throws A2AServerException {
TaskPushNotificationConfig taskPushNotificationConfig = new TaskPushNotificationConfig(taskId, pushNotificationConfig);| void sendStreamingMessage(String requestId, MessageSendParams messageSendParams, Consumer<StreamingEventKind> eventHandler, | ||
| Consumer<JSONRPCError> errorHandler, Runnable failureHandler) throws A2AServerException; | ||
|
|
||
| /** | ||
| * Resubscribe to an ongoing task. | ||
| * | ||
| * @param requestId the request ID to use | ||
| * @param taskIdParams the params for the task to resubscribe to | ||
| * @param eventHandler a consumer that will be invoked for each event received from the remote agent | ||
| * @param errorHandler a consumer that will be invoked if the remote agent returns an error | ||
| * @param failureHandler a consumer that will be invoked if a failure occurs when processing events | ||
| * @throws A2AServerException if resubscribing to the task fails for any reason | ||
| */ | ||
| void resubscribeToTask(String requestId, TaskIdParams taskIdParams, Consumer<StreamingEventKind> eventHandler, | ||
| Consumer<JSONRPCError> errorHandler, Runnable failureHandler) throws A2AServerException; |
There was a problem hiding this comment.
The errorHandler parameter in sendStreamingMessage and resubscribeToTask uses Consumer<JSONRPCError>, which is specific to the JSON-RPC transport. This makes the Transport interface not truly transport-agnostic and causes compilation errors in the GrpcTransport implementation, which needs to handle Throwable for gRPC errors.
To make this SPI genuinely transport-agnostic, the error handler should be more generic. I recommend changing Consumer<JSONRPCError> to Consumer<Throwable> in both streaming methods.
| void sendStreamingMessage(String requestId, MessageSendParams messageSendParams, Consumer<StreamingEventKind> eventHandler, | |
| Consumer<JSONRPCError> errorHandler, Runnable failureHandler) throws A2AServerException; | |
| /** | |
| * Resubscribe to an ongoing task. | |
| * | |
| * @param requestId the request ID to use | |
| * @param taskIdParams the params for the task to resubscribe to | |
| * @param eventHandler a consumer that will be invoked for each event received from the remote agent | |
| * @param errorHandler a consumer that will be invoked if the remote agent returns an error | |
| * @param failureHandler a consumer that will be invoked if a failure occurs when processing events | |
| * @throws A2AServerException if resubscribing to the task fails for any reason | |
| */ | |
| void resubscribeToTask(String requestId, TaskIdParams taskIdParams, Consumer<StreamingEventKind> eventHandler, | |
| Consumer<JSONRPCError> errorHandler, Runnable failureHandler) throws A2AServerException; | |
| void sendStreamingMessage(String requestId, MessageSendParams messageSendParams, Consumer<StreamingEventKind> eventHandler, | |
| Consumer<Throwable> errorHandler, Runnable failureHandler) throws A2AServerException; | |
| /** | |
| * Resubscribe to an ongoing task. | |
| * | |
| * @param requestId the request ID to use | |
| * @param taskIdParams the params for the task to resubscribe to | |
| * @param eventHandler a consumer that will be invoked for each event received from the remote agent | |
| * @param errorHandler a consumer that will be invoked if the remote agent returns an error | |
| * @param failureHandler a consumer that will be invoked if a failure occurs when processing events | |
| * @throws A2AServerException if resubscribing to the task fails for any reason | |
| */ | |
| void resubscribeToTask(String requestId, TaskIdParams taskIdParams, Consumer<StreamingEventKind> eventHandler, | |
| Consumer<Throwable> errorHandler, Runnable failureHandler) throws A2AServerException; |
| public List<TaskPushNotificationConfig> listTaskPushNotificationConfig(String requestId, ListTaskPushNotificationConfigParams listTaskPushNotificationConfigParams) throws A2AServerException { | ||
| return List.of(); | ||
| } |
There was a problem hiding this comment.
This method is not implemented. It currently returns an empty list, which is incorrect behavior. Please provide the full implementation for listing task push notification configurations over gRPC.
| public void deleteTaskPushNotificationConfig(String requestId, DeleteTaskPushNotificationConfigParams deleteTaskPushNotificationConfigParams) throws A2AServerException { | ||
|
|
||
| } |
There was a problem hiding this comment.
This method is not implemented. It has an empty body, which is incorrect. Please provide the full implementation for deleting a task push notification configuration over gRPC.
| public void resubscribeToTask(String requestId, TaskIdParams taskIdParams, Consumer<StreamingEventKind> eventHandler, Consumer<JSONRPCError> errorHandler, Runnable failureHandler) throws A2AServerException { | ||
|
|
||
| } |
There was a problem hiding this comment.
This method is not implemented. It has an empty body, which is incorrect. Please provide the full implementation for resubscribing to a task over gRPC.
|
Thanks very much for your PR, @brasseld! As mentioned on chat, it seems we had the same idea and I have a branch in progress as well here :-) https://github.com/fjuma/a2a-java/tree/client_refactor I will take a closer look at your branch. I'm thinking I can likely combine aspects from both our branches since I think we each have some distinct parts. I like the way you've split things out. Will go through this in more detail and combine things. |
|
Just FYI, I've got your branch here locally and am starting to tweak and combine things. For the transport code, I think it will be good to split out the client and server code into different modules. Am also making that change locally. Will let you know as soon as I have something ready to review. |
|
Hi @brasseld, thanks again for your PR! I've combined our approaches together and submitted a draft PR here to start getting feedback: Whenever you get a chance, please take a look and let me know what you think. I split out the transport code further so the client and server transport code are now in separate modules and the code for gRPC is separate from the code for JSON-RPC as you had. Next, I'm going to work through updating the tests to make use of the new ClientFactory to instantiate an appropriate Client for the transport being tested so As mentioned in the PR description, I also need to update This is a pretty significant refactor with breaking changes for the client side so would be great to get feedback from you and others on what you think. Thanks! |
|
Closing this one since it has been incorporated in #231 and further changes are being made there. |
2 participants
Description
Thank you for opening a Pull Request!
Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
CONTRIBUTINGGuide.fix:which represents bug fixes, and correlates to a SemVer patch.feat:represents a new feature, and correlates to a SemVer minor.feat!:, orfix!:,refactor!:, etc., which represent a breaking change (indicated by the!) and will result in a SemVer major.Fixes #<issue_number_goes_here> 🦕