feat: gRPC transport implementation

client/pom.xml

@@ -22,7 +22,11 @@
             <artifactId>a2a-java-sdk-common</artifactId>
             <version>${project.version}</version>
         </dependency>
-
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>a2a-java-sdk-grpc</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>a2a-java-sdk-spec</artifactId>

client/src/main/java/io/a2a/client/A2AGrpcClient.java

@@ -0,0 +1,200 @@
+package io.a2a.client;
+
+import static io.a2a.grpc.A2AServiceGrpc.A2AServiceBlockingV2Stub;
+import static io.a2a.grpc.A2AServiceGrpc.A2AServiceStub;
+import static io.a2a.grpc.utils.ProtoUtils.FromProto;
+import static io.a2a.grpc.utils.ProtoUtils.ToProto;
+import static io.a2a.util.Assert.checkNotNullParam;
+
+import java.util.function.Consumer;
+
+import io.a2a.client.sse.SSEStreamObserver;
+import io.a2a.grpc.A2AServiceGrpc;
+import io.a2a.grpc.CancelTaskRequest;
+import io.a2a.grpc.CreateTaskPushNotificationConfigRequest;
+import io.a2a.grpc.GetTaskPushNotificationConfigRequest;
+import io.a2a.grpc.GetTaskRequest;
+import io.a2a.grpc.SendMessageRequest;
+import io.a2a.grpc.SendMessageResponse;
+import io.a2a.grpc.StreamResponse;
+import io.a2a.grpc.utils.ProtoUtils;
+import io.a2a.spec.A2AServerException;
+import io.a2a.spec.AgentCard;
+import io.a2a.spec.EventKind;
+import io.a2a.spec.GetTaskPushNotificationConfigParams;
+import io.a2a.spec.MessageSendParams;
+import io.a2a.spec.StreamingEventKind;
+import io.a2a.spec.Task;
+import io.a2a.spec.TaskIdParams;
+import io.a2a.spec.TaskPushNotificationConfig;
+import io.a2a.spec.TaskQueryParams;
+import io.grpc.Channel;
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
+
+/**
+ * A2A Client for interacting with an A2A agent via gRPC.
+ */
+public class A2AGrpcClient {
+
+    private A2AServiceBlockingV2Stub blockingStub;
+    private A2AServiceStub asyncStub;
+    private AgentCard agentCard;
+
+    /**
+     * Create an A2A client for interacting with an A2A agent via gRPC.
+     *
+     * @param channel the gRPC channel
+     * @param agentCard the agent card for the A2A server this client will be communicating with
+     */
+    public A2AGrpcClient(Channel channel, AgentCard agentCard) {
+        checkNotNullParam("channel", channel);
+        checkNotNullParam("agentCard", agentCard);
+        this.asyncStub = A2AServiceGrpc.newStub(channel);
+        this.blockingStub = A2AServiceGrpc.newBlockingV2Stub(channel);
+        this.agentCard = agentCard;
+    }
+
+    /**
+     * Send a message to the remote agent.
+     *
+     * @param messageSendParams the parameters for the message to be sent
+     * @return the response, may be a message or a task
+     * @throws A2AServerException if sending the message fails for any reason
+     */
+    public EventKind sendMessage(MessageSendParams messageSendParams) throws A2AServerException {
+        SendMessageRequest request = createGrpcSendMessageRequestFromMessageSendParams(messageSendParams);
+        try {
+            SendMessageResponse response = blockingStub.sendMessage(request);
+            if (response.hasMsg()) {
+                return FromProto.message(response.getMsg());
+            } else if (response.hasTask()) {
+                return FromProto.task(response.getTask());
+            } else {
+                throw new A2AServerException("Server response did not contain a message or task");
+            }
+        } catch (StatusRuntimeException e) {
+            throw new A2AServerException("Failed to send message: " + e, e);
+        }
+    }
+
+    /**
+     * Retrieves the current state and history of a specific task.
+     *
+     * @param taskQueryParams the params for the task to be queried
+     * @return the task
+     * @throws A2AServerException if retrieving the task fails for any reason
+     */
+    public Task getTask(TaskQueryParams taskQueryParams) throws A2AServerException {
+        GetTaskRequest.Builder requestBuilder = GetTaskRequest.newBuilder();
+        requestBuilder.setName("tasks/" + taskQueryParams.id());
+        if (taskQueryParams.historyLength() != null) {
+            requestBuilder.setHistoryLength(taskQueryParams.historyLength());
+        }
+        GetTaskRequest getTaskRequest = requestBuilder.build();
+        try {
+            return FromProto.task(blockingStub.getTask(getTaskRequest));
+        } catch (StatusRuntimeException e) {
+            throw new A2AServerException("Failed to get task: " + e, e);
+        }
+    }
+
+    /**
+     * Cancel a task that was previously submitted to the A2A server.
+     *
+     * @param taskIdParams the params for the task to be cancelled
+     * @return the updated task
+     * @throws A2AServerException if cancelling the task fails for any reason
+     */
+    public Task cancelTask(TaskIdParams taskIdParams) throws A2AServerException {
+        CancelTaskRequest cancelTaskRequest = CancelTaskRequest.newBuilder()
+                .setName("tasks/" + taskIdParams.id())
+                .build();
+        try {
+            return FromProto.task(blockingStub.cancelTask(cancelTaskRequest));
+        } catch (StatusRuntimeException e) {
+            throw new A2AServerException("Failed to cancel task: " + e, e);
+        }
+    }
+
+    /**
+     * Set push notification configuration for a task.
+     *
+     * @param taskPushNotificationConfig the task push notification configuration
+     * @return the task push notification config
+     * @throws A2AServerException if setting the push notification configuration fails for any reason
+     */
+    public TaskPushNotificationConfig setTaskPushNotificationConfig(TaskPushNotificationConfig taskPushNotificationConfig) throws A2AServerException {
+        String configId = taskPushNotificationConfig.pushNotificationConfig().id();
+        CreateTaskPushNotificationConfigRequest request = CreateTaskPushNotificationConfigRequest.newBuilder()
+                .setParent("tasks/" + taskPushNotificationConfig.taskId())
+                .setConfig(ToProto.taskPushNotificationConfig(taskPushNotificationConfig))
+                .setConfigId(configId == null ? "" : configId)
+                .build();
+        try {
+            return FromProto.taskPushNotificationConfig(blockingStub.createTaskPushNotificationConfig(request));
+        } catch (StatusRuntimeException e) {
+            throw new A2AServerException("Failed to set the task push notification config: " + e, e);
+        }
+    }
+
+    /**
+     * Get the push notification configuration for a task.
+     *
+     * @param getTaskPushNotificationConfigParams the params for the task
+     * @return the push notification configuration
+     * @throws A2AServerException if getting the push notification configuration fails for any reason
+     */
+    public TaskPushNotificationConfig getTaskPushNotificationConfig(GetTaskPushNotificationConfigParams getTaskPushNotificationConfigParams) throws A2AServerException {
+        GetTaskPushNotificationConfigRequest getTaskPushNotificationConfigRequest = GetTaskPushNotificationConfigRequest.newBuilder()
+                .setName(getTaskPushNotificationConfigName(getTaskPushNotificationConfigParams))
+                .build();
+        try {
+            return FromProto.taskPushNotificationConfig(blockingStub.getTaskPushNotificationConfig(getTaskPushNotificationConfigRequest));
+        } catch (StatusRuntimeException e) {
+            throw new A2AServerException("Failed to get the task push notification config: " + e, e);
+        }
+    }
+
+    /**
+     * Send a streaming message request 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 an error occurs
+     * @throws A2AServerException if sending the streaming message fails for any reason
+     */
+    public void sendMessageStreaming(MessageSendParams messageSendParams, Consumer<StreamingEventKind> eventHandler,
+                                     Consumer<Throwable> errorHandler) throws A2AServerException {
+        SendMessageRequest request = createGrpcSendMessageRequestFromMessageSendParams(messageSendParams);
+        StreamObserver<StreamResponse> streamObserver = new SSEStreamObserver(eventHandler, errorHandler);
+        try {
+            asyncStub.sendStreamingMessage(request, streamObserver);
+        } catch (StatusRuntimeException e) {
+            throw new A2AServerException("Failed to send streaming message: " + e, e);
+        }
+    }
+
+    private SendMessageRequest createGrpcSendMessageRequestFromMessageSendParams(MessageSendParams messageSendParams) {
+        SendMessageRequest.Builder builder = SendMessageRequest.newBuilder();
+        builder.setRequest(ToProto.message(messageSendParams.message()));
+        if (messageSendParams.configuration() != null) {
+            builder.setConfiguration(ToProto.messageSendConfiguration(messageSendParams.configuration()));
+        }
+        if (messageSendParams.metadata() != null) {
+            builder.setMetadata(ToProto.struct(messageSendParams.metadata()));
+        }
+        return builder.build();
+    }
+
+    private String getTaskPushNotificationConfigName(GetTaskPushNotificationConfigParams getTaskPushNotificationConfigParams) {
+        StringBuilder name = new StringBuilder();
+        name.append("tasks/");
+        name.append(getTaskPushNotificationConfigParams.id());
+        if (getTaskPushNotificationConfigParams.pushNotificationConfigId() != null) {
+            name.append("/pushNotificationConfigs/");
+            name.append(getTaskPushNotificationConfigParams.pushNotificationConfigId());
+        }
+        return name.toString();
+    }
+}

client/src/main/java/io/a2a/client/sse/SSEStreamObserver.java

@@ -0,0 +1,57 @@
+package io.a2a.client.sse;
+
+
+import static io.a2a.grpc.utils.ProtoUtils.FromProto;
+
+import java.util.function.Consumer;
+import java.util.logging.Logger;
+
+import io.a2a.grpc.StreamResponse;
+import io.a2a.spec.StreamingEventKind;
+import io.grpc.stub.StreamObserver;
+
+public class SSEStreamObserver implements StreamObserver<StreamResponse> {
+
+    private static final Logger log = Logger.getLogger(SSEStreamObserver.class.getName());
+    private final Consumer<StreamingEventKind> eventHandler;
+    private final Consumer<Throwable> errorHandler;
+
+    public SSEStreamObserver(Consumer<StreamingEventKind> eventHandler, Consumer<Throwable> errorHandler) {
+        this.eventHandler = eventHandler;
+        this.errorHandler = errorHandler;
+    }
+
+    @Override
+    public void onNext(StreamResponse response) {
+        StreamingEventKind event;
+        switch (response.getPayloadCase()) {
+            case MSG:
+                event = FromProto.message(response.getMsg());
+                break;
+            case TASK:
+                event = FromProto.task(response.getTask());
+                break;
+            case STATUS_UPDATE:
+                event = FromProto.taskStatusUpdateEvent(response.getStatusUpdate());
+                break;
+            case ARTIFACT_UPDATE:
+                event = FromProto.taskArtifactUpdateEvent(response.getArtifactUpdate());
+                break;
+            default:
+                log.warning("Invalid stream response " + response.getPayloadCase());
+                errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase()));
+                return;
+        }
+        eventHandler.accept(event);
+    }
+
+    @Override
+    public void onError(Throwable t) {
+        errorHandler.accept(t);
+    }
+
+    @Override
+    public void onCompleted() {
+        // done
+    }
+}

grpc/pom.xml

@@ -0,0 +1,39 @@
+<?xml version="1.0"?>
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.github.a2asdk</groupId>
+        <artifactId>a2a-java-sdk-parent</artifactId>
+        <version>0.2.6.Beta1-SNAPSHOT</version>
+    </parent>
+    <artifactId>a2a-java-sdk-grpc</artifactId>
+
+    <packaging>jar</packaging>
+
+    <name>Java SDK A2A gRPC</name>
+    <description>Java SDK for the Agent2Agent Protocol (A2A) - gRPC</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.github.a2asdk</groupId>
+            <artifactId>a2a-java-sdk-spec</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-protobuf</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-stub</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>