Merge pull request #120 from fjuma/client

Update the client docs and A2AC

Author: fjuma

README.md

@@ -139,18 +139,6 @@ public class WeatherAgentExecutorProducer {
             updater.complete();
         }
 
-        private String extractTextFromMessage(Message message) {
-            StringBuilder textBuilder = new StringBuilder();
-            if (message.getParts() != null) {
-                for (Part part : message.getParts()) {
-                    if (part instanceof TextPart textPart) {
-                        textBuilder.append(textPart.getText());
-                    }
-                }
-            }
-            return textBuilder.toString();
-        }
-
         @Override
         public void cancel(RequestContext context, EventQueue eventQueue) throws JSONRPCError {
             Task task = context.getTask();
@@ -169,6 +157,18 @@ public class WeatherAgentExecutorProducer {
             TaskUpdater updater = new TaskUpdater(context, eventQueue);
             updater.cancel();
         }
+
+        private String extractTextFromMessage(Message message) {
+            StringBuilder textBuilder = new StringBuilder();
+            if (message.getParts() != null) {
+                for (Part part : message.getParts()) {
+                    if (part instanceof TextPart textPart) {
+                        textBuilder.append(textPart.getText());
+                    }
+                }
+            }
+            return textBuilder.toString();
+        }
     }
 }
 ```
@@ -202,26 +202,25 @@ OR
 </dependency>
 ```
 
-## Client
+## A2A Client
 
-An *initial* [A2AClient](https://github.com/fjuma/a2a-java-sdk/blob/main/src/main/java/io/a2a/client/A2AClient.java) class has been added. This is very much work in progress, we are working on implementing the methods required by the client side of the protocol.
+The A2A Java SDK provides a Java client implementation of the [Agent2Agent (A2A) Protocol](https://google-a2a.github.io/A2A), allowing communication with A2A servers.
 
 ### Sample Usage
 
-#### Create a client
+#### Create an A2A client
 
 ```java
-// Create an A2AClient (the URL specified is the server agent's URL)
+// Create an A2AClient (the URL specified is the server agent's URL, be sure to replace it with the actual URL of the A2A server you want to connect to)
 A2AClient client = new A2AClient("http://localhost:1234");
 ```
 
-#### Send a message
+#### Send a message to the A2A server agent
 
 ```java
-// Send a text message to the server agent
+// Send a text message to the A2A server agent
 Message message = A2A.toUserMessage("tell me a joke"); // the message ID will be automatically generated for you
 MessageSendParams params = new MessageSendParams.Builder()
-        .id("task-1234") // id is optional
         .message(message)
         .build();
 SendMessageResponse response = client.sendMessage(params);        
@@ -234,7 +233,7 @@ if you don't specify it. You can also explicitly specify a message ID like this:
 Message message = A2A.toUserMessage("tell me a joke", "message-1234"); // messageId is message-1234
 ```
 
-#### Get a task
+#### Get the current state of a task
 
 ```java
 // Retrieve the task with id "task-1234"
@@ -245,7 +244,7 @@ GetTaskResponse response = client.getTask("task-1234");
 GetTaskResponse response = client.getTask(new TaskQueryParams("task-1234", 10));
 ```
 
-#### Cancel a task
+#### Cancel an ongoing task
 
 ```java
 // Cancel the task we previously submitted with id "task-1234"
@@ -259,12 +258,12 @@ CancelTaskResponse response = client.cancelTask(new TaskIdParams("task-1234", me
 #### Get the push notification configuration for a task
 
 ```java
-// Get task push notification
-GetTaskPushNotificationResponse response = client.getTaskPushNotificationConfig("task-1234");
+// Get task push notification configuration
+GetTaskPushNotificationConfigResponse response = client.getTaskPushNotificationConfig("task-1234");
 
 // You can also specify additional properties using a map
 Map<String, Object> metadata = ...
-GetTaskPushNotificationResponse response = client.getTaskPushNotificationConfig(new TaskIdParams("task-1234", metadata));
+GetTaskPushNotificationConfigResponse response = client.getTaskPushNotificationConfig(new TaskIdParams("task-1234", metadata));
 ```
 
 #### Set the push notification configuration for a task
@@ -274,7 +273,7 @@ GetTaskPushNotificationResponse response = client.getTaskPushNotificationConfig(
 PushNotificationConfig pushNotificationConfig = new PushNotificationConfig.Builder()
         .url("https://example.com/callback")
         .authenticationInfo(new AuthenticationInfo(Collections.singletonList("jwt"), null))
-        .build());
+        .build();
 SetTaskPushNotificationResponse response = client.setTaskPushNotificationConfig("task-1234", pushNotificationConfig);
 ```
 
@@ -284,12 +283,11 @@ SetTaskPushNotificationResponse response = client.setTaskPushNotificationConfig(
 // Send a text message to the remote agent
 Message message = A2A.toUserMessage("tell me some jokes"); // the message ID will be automatically generated for you
 MessageSendParams params = new MessageSendParams.Builder()
-        .id("task-1234") // id is optional
         .message(message)
         .build();
 
 // Create a handler that will be invoked for Task, Message, TaskStatusUpdateEvent, and TaskArtifactUpdateEvent
-Consumer<StreamingEventType> eventHandler = event -> {...};
+Consumer<StreamingEventKind> eventHandler = event -> {...};
 
 // Create a handler that will be invoked if an error is received
 Consumer<JSONRPCError> errorHandler = error -> {...};
@@ -301,6 +299,23 @@ Runnable failureHandler = () -> {...};
 client.sendStreamingMessage(params, eventHandler, errorHandler, failureHandler);
 ```
 
+#### Resubscribe to a task
+
+```java
+// Create a handler that will be invoked for Task, Message, TaskStatusUpdateEvent, and TaskArtifactUpdateEvent
+Consumer<StreamingEventKind> eventHandler = event -> {...};
+
+// Create a handler that will be invoked if an error is received
+Consumer<JSONRPCError> errorHandler = error -> {...};
+
+// Create a handler that will be invoked in the event of a failure
+Runnable failureHandler = () -> {...};
+
+// Resubscribe to an ongoing task with id "task-1234"
+TaskIdParams taskIdParams = new TaskIdParams("task-1234");
+client.resubscribeToTask("request-1234", taskIdParams, eventHandler, errorHandler, failureHandler);
+```
+
 #### Retrieve details about the server agent that this client agent is communicating with
 ```java
 AgentCard serverAgentCard = client.getAgentCard();
@@ -312,7 +327,7 @@ An agent card can also be retrieved using the `A2A#getAgentCard` method:
 AgentCard agentCard = A2A.getAgentCard("http://localhost:1234");
 ```
 
-## Examples
+## Additional Examples
 
 ### Hello World Example
 
@@ -324,9 +339,13 @@ A complete example of an A2A client communicating with a Python A2A server is av
 
 The example includes detailed instructions on how to run both the Python server and the Java client using JBang. Check out the [example's README](src/main/java/io/a2a/examples/helloworld/README.md) for more information.
 
-## Server
+## License
+
+This project is licensed under the terms of the [Apache 2.0 License](LICENSE).
+
+## Contributing
 
-TODO
+See [CONTRIBUTING.md](CONTRIBUTING.md) for contribution guidelines.
 
 
 

core/src/main/java/io/a2a/client/A2AClient.java

@@ -6,6 +6,7 @@
 import static io.a2a.spec.A2A.JSONRPC_VERSION;
 import static io.a2a.spec.A2A.SEND_MESSAGE_METHOD;
 import static io.a2a.spec.A2A.SEND_STREAMING_MESSAGE_METHOD;
+import static io.a2a.spec.A2A.SEND_TASK_RESUBSCRIPTION_METHOD;
 import static io.a2a.spec.A2A.SET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD;
 import static io.a2a.util.Assert.checkNotNullParam;
 import static io.a2a.util.Utils.OBJECT_MAPPER;
@@ -45,6 +46,7 @@
 import io.a2a.spec.TaskIdParams;
 import io.a2a.spec.TaskPushNotificationConfig;
 import io.a2a.spec.TaskQueryParams;
+import io.a2a.spec.TaskResubscriptionRequest;
 
 /**
  * An A2A client.
@@ -347,6 +349,20 @@ public SetTaskPushNotificationConfigResponse setTaskPushNotificationConfig(Strin
         }
     }
 
+    /**
+     * Send a streaming message to the remote agent.
+     *
+     * @param messageSendParams the parameters for the message to be sent
+     * @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 sending the streaming message fails for any reason
+     */
+    public void sendStreamingMessage(MessageSendParams messageSendParams, Consumer<StreamingEventKind> eventHandler,
+                                     Consumer<JSONRPCError> errorHandler, Runnable failureHandler) throws A2AServerException {
+        sendStreamingMessage(null, messageSendParams, eventHandler, errorHandler, failureHandler);
+    }
+
     /**
      * Send a streaming message to the remote agent.
      *
@@ -392,6 +408,65 @@ public void sendStreamingMessage(String requestId, MessageSendParams messageSend
         }
     }
 
+    /**
+     * Resubscribe to an ongoing task.
+     *
+     * @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
+     */
+    public void resubscribeToTask(TaskIdParams taskIdParams, Consumer<StreamingEventKind> eventHandler,
+                                  Consumer<JSONRPCError> errorHandler, Runnable failureHandler) throws A2AServerException {
+        resubscribeToTask(null, taskIdParams, eventHandler, errorHandler, failureHandler);
+    }
+
+    /**
+     * 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
+     */
+    public void resubscribeToTask(String requestId, TaskIdParams taskIdParams, Consumer<StreamingEventKind> eventHandler,
+                                  Consumer<JSONRPCError> errorHandler, Runnable failureHandler) throws A2AServerException {
+        checkNotNullParam("taskIdParams", taskIdParams);
+        checkNotNullParam("eventHandler", eventHandler);
+        checkNotNullParam("errorHandler", errorHandler);
+        checkNotNullParam("failureHandler", failureHandler);
+
+        TaskResubscriptionRequest.Builder taskResubscriptionRequestBuilder = new TaskResubscriptionRequest.Builder()
+                .jsonrpc(JSONRPC_VERSION)
+                .method(SEND_TASK_RESUBSCRIPTION_METHOD)
+                .params(taskIdParams);
+
+        if (requestId != null) {
+            taskResubscriptionRequestBuilder.id(requestId);
+        }
+
+        AtomicReference<CompletableFuture<Void>> ref = new AtomicReference<>();
+        SSEEventListener sseEventListener = new SSEEventListener(eventHandler, errorHandler, failureHandler);
+        TaskResubscriptionRequest taskResubscriptionRequest = taskResubscriptionRequestBuilder.build();
+        try {
+            A2AHttpClient.PostBuilder builder = createPostBuilder(taskResubscriptionRequest);
+            ref.set(builder.postAsyncSSE(
+                    msg -> sseEventListener.onMessage(msg, ref.get()),
+                    throwable -> sseEventListener.onError(throwable, ref.get()),
+                    () -> {
+                        // We don't need to do anything special on completion
+                    }));
+
+        } catch (IOException e) {
+            throw new A2AServerException("Failed to send task resubscription request: " + e);
+        } catch (InterruptedException e) {
+            throw new A2AServerException("Task resubscription request timed out: " + e);
+        }
+    }
+
     private String sendPostRequest(Object value) throws IOException, InterruptedException {
         A2AHttpClient.PostBuilder builder = createPostBuilder(value);
         A2AHttpResponse response = builder.post();

core/src/main/java/io/a2a/spec/SendStreamingMessageRequest.java

@@ -4,6 +4,8 @@
 import static io.a2a.spec.A2A.SEND_STREAMING_MESSAGE_METHOD;
 import static io.a2a.util.Utils.defaultIfNull;
 
+import java.util.UUID;
+
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonInclude;
@@ -67,6 +69,9 @@ public Builder params(MessageSendParams params) {
             }
 
             public SendStreamingMessageRequest build() {
+                if (id == null) {
+                    id = UUID.randomUUID().toString();
+                }
                 return new SendStreamingMessageRequest(jsonrpc, id, method, params);
             }
         }

core/src/main/java/io/a2a/spec/TaskResubscriptionRequest.java

@@ -40,4 +40,38 @@ public TaskResubscriptionRequest(@JsonProperty("jsonrpc") String jsonrpc, @JsonP
     public TaskResubscriptionRequest(Object id, TaskIdParams params) {
         this(null, id, SEND_TASK_RESUBSCRIPTION_METHOD, params);
     }
+
+    public static class Builder {
+        private String jsonrpc;
+        private Object id;
+        private String method = SEND_TASK_RESUBSCRIPTION_METHOD;
+        private TaskIdParams params;
+
+        public TaskResubscriptionRequest.Builder jsonrpc(String jsonrpc) {
+            this.jsonrpc = jsonrpc;
+            return this;
+        }
+
+        public TaskResubscriptionRequest.Builder id(Object id) {
+            this.id = id;
+            return this;
+        }
+
+        public TaskResubscriptionRequest.Builder method(String method) {
+            this.method = method;
+            return this;
+        }
+
+        public TaskResubscriptionRequest.Builder params(TaskIdParams params) {
+            this.params = params;
+            return this;
+        }
+
+        public TaskResubscriptionRequest build() {
+            if (id == null) {
+                id = UUID.randomUUID().toString();
+            }
+            return new TaskResubscriptionRequest(jsonrpc, id, method, params);
+        }
+    }
 }