feat: Make the client transport agnostic

client/pom.xml → client/base/pom.xml

@@ -8,6 +8,7 @@
         <groupId>io.github.a2asdk</groupId>
         <artifactId>a2a-java-sdk-parent</artifactId>
         <version>0.3.0.Beta1-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
     </parent>
     <artifactId>a2a-java-sdk-client</artifactId>
 
@@ -19,20 +20,24 @@
     <dependencies>
         <dependency>
             <groupId>${project.groupId}</groupId>
-            <artifactId>a2a-java-sdk-common</artifactId>
-            <version>${project.version}</version>
+            <artifactId>a2a-java-sdk-client-config</artifactId>
         </dependency>
         <dependency>
             <groupId>${project.groupId}</groupId>
-            <artifactId>a2a-java-sdk-spec</artifactId>
-            <version>${project.version}</version>
+            <artifactId>a2a-java-sdk-http-client</artifactId>
         </dependency>
         <dependency>
             <groupId>${project.groupId}</groupId>
-            <artifactId>a2a-java-sdk-spec-grpc</artifactId>
-            <version>${project.version}</version>
+            <artifactId>a2a-java-sdk-client-transport-spi</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>a2a-java-sdk-common</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>a2a-java-sdk-spec</artifactId>
         </dependency>
-
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-api</artifactId>

client/src/main/java/io/a2a/A2A.java → client/base/src/main/java/io/a2a/A2A.java

@@ -4,8 +4,8 @@
 import java.util.Map;
 
 import io.a2a.client.A2ACardResolver;
-import io.a2a.http.A2AHttpClient;
-import io.a2a.http.JdkA2AHttpClient;
+import io.a2a.client.http.A2AHttpClient;
+import io.a2a.client.http.JdkA2AHttpClient;
 import io.a2a.spec.A2AClientError;
 import io.a2a.spec.A2AClientJSONError;
 import io.a2a.spec.AgentCard;

client/src/main/java/io/a2a/client/A2ACardResolver.java → client/base/src/main/java/io/a2a/client/A2ACardResolver.java

@@ -9,8 +9,9 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.type.TypeReference;
-import io.a2a.http.A2AHttpClient;
-import io.a2a.http.A2AHttpResponse;
+import io.a2a.client.http.A2AHttpClient;
+import io.a2a.client.http.A2AHttpResponse;
+import io.a2a.client.http.JdkA2AHttpClient;
 import io.a2a.spec.A2AClientError;
 import io.a2a.spec.A2AClientJSONError;
 import io.a2a.spec.AgentCard;
@@ -25,6 +26,19 @@ public class A2ACardResolver {
     private static final TypeReference<AgentCard> AGENT_CARD_TYPE_REFERENCE = new TypeReference<>() {};
 
     /**
+     * Get the agent card for an A2A agent.
+     * The {@code JdkA2AHttpClient} will be used to fetch the agent card.
+     *
+     * @param baseUrl the base URL for the agent whose agent card we want to retrieve
+     * @throws A2AClientError if the URL for the agent is invalid
+     */
+    public A2ACardResolver(String baseUrl) throws A2AClientError {
+        this(new JdkA2AHttpClient(), baseUrl, null, null);
+    }
+
+    /**
+     /**Get the agent card for an A2A agent.
+     *
      * @param httpClient the http client to use
      * @param baseUrl the base URL for the agent whose agent card we want to retrieve
      * @throws A2AClientError if the URL for the agent is invalid

client/base/src/main/java/io/a2a/client/Client.java

@@ -0,0 +1,234 @@
+package io.a2a.client;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+import io.a2a.client.config.ClientCallContext;
+import io.a2a.client.config.ClientConfig;
+import io.a2a.client.transport.spi.ClientTransport;
+import io.a2a.spec.A2AClientError;
+import io.a2a.spec.A2AClientException;
+import io.a2a.spec.A2AClientInvalidStateError;
+import io.a2a.spec.AgentCard;
+import io.a2a.spec.DeleteTaskPushNotificationConfigParams;
+import io.a2a.spec.EventKind;
+import io.a2a.spec.GetTaskPushNotificationConfigParams;
+import io.a2a.spec.ListTaskPushNotificationConfigParams;
+import io.a2a.spec.Message;
+import io.a2a.spec.MessageSendConfiguration;
+import io.a2a.spec.MessageSendParams;
+import io.a2a.spec.PushNotificationConfig;
+import io.a2a.spec.StreamingEventKind;
+import io.a2a.spec.Task;
+import io.a2a.spec.TaskArtifactUpdateEvent;
+import io.a2a.spec.TaskIdParams;
+import io.a2a.spec.TaskPushNotificationConfig;
+import io.a2a.spec.TaskQueryParams;
+import io.a2a.spec.TaskStatusUpdateEvent;
+
+public class Client extends AbstractClient {
+
+    private final ClientConfig clientConfig;
+    private final ClientTransport clientTransport;
+    private AgentCard agentCard;
+
+    public Client(AgentCard agentCard, ClientConfig clientConfig, ClientTransport clientTransport,
+                  List<BiConsumer<ClientEvent, AgentCard>> consumers, Consumer<Throwable> streamingErrorHandler) {
+        super(consumers, streamingErrorHandler);
+        this.agentCard = agentCard;
+        this.clientConfig = clientConfig;
+        this.clientTransport = clientTransport;
+    }
+
+
+    @Override
+    public void sendMessage(Message request, ClientCallContext context) throws A2AClientException {
+        MessageSendParams messageSendParams = getMessageSendParams(request, clientConfig);
+        sendMessage(messageSendParams, null, null, context);
+    }
+
+    @Override
+    public void sendMessage(Message request, List<BiConsumer<ClientEvent, AgentCard>> consumers,
+                            Consumer<Throwable> streamingErrorHandler, ClientCallContext context) throws A2AClientException {
+        MessageSendParams messageSendParams = getMessageSendParams(request, clientConfig);
+        sendMessage(messageSendParams, consumers, streamingErrorHandler, context);
+    }
+
+    @Override
+    public void sendMessage(Message request, PushNotificationConfig pushNotificationConfiguration,
+                            Map<String, Object> metatadata, ClientCallContext context) throws A2AClientException {
+        MessageSendConfiguration messageSendConfiguration = new MessageSendConfiguration.Builder()
+                .acceptedOutputModes(clientConfig.getAcceptedOutputModes())
+                .blocking(clientConfig.isPolling())
+                .historyLength(clientConfig.getHistoryLength())
+                .pushNotification(pushNotificationConfiguration)
+                .build();
+
+        MessageSendParams messageSendParams = new MessageSendParams.Builder()
+                .message(request)
+                .configuration(messageSendConfiguration)
+                .metadata(metatadata)
+                .build();
+
+        sendMessage(messageSendParams, null, null, context);
+    }
+
+    @Override
+    public Task getTask(TaskQueryParams request, ClientCallContext context) throws A2AClientException {
+        return clientTransport.getTask(request, context);
+    }
+
+    @Override
+    public Task cancelTask(TaskIdParams request, ClientCallContext context) throws A2AClientException {
+        return clientTransport.cancelTask(request, context);
+    }
+
+    @Override
+    public TaskPushNotificationConfig setTaskPushNotificationConfiguration(
+            TaskPushNotificationConfig request, ClientCallContext context) throws A2AClientException {
+        return clientTransport.setTaskPushNotificationConfiguration(request, context);
+    }
+
+    @Override
+    public TaskPushNotificationConfig getTaskPushNotificationConfiguration(
+            GetTaskPushNotificationConfigParams request, ClientCallContext context) throws A2AClientException {
+        return clientTransport.getTaskPushNotificationConfiguration(request, context);
+    }
+
+    @Override
+    public List<TaskPushNotificationConfig> listTaskPushNotificationConfigurations(
+            ListTaskPushNotificationConfigParams request, ClientCallContext context) throws A2AClientException {
+        return clientTransport.listTaskPushNotificationConfigurations(request, context);
+    }
+
+    @Override
+    public void deleteTaskPushNotificationConfigurations(
+            DeleteTaskPushNotificationConfigParams request, ClientCallContext context) throws A2AClientException {
+        clientTransport.deleteTaskPushNotificationConfigurations(request, context);
+    }
+
+    @Override
+    public void resubscribe(TaskIdParams request, ClientCallContext context) throws A2AClientException {
+       resubscribeToTask(request, null, null, context);
+    }
+
+    @Override
+    public void resubscribe(TaskIdParams request, List<BiConsumer<ClientEvent, AgentCard>> consumers,
+                            Consumer<Throwable> streamingErrorHandler, ClientCallContext context) throws A2AClientException {
+        resubscribeToTask(request, consumers, streamingErrorHandler, context);
+    }
+
+    @Override
+    public AgentCard getAgentCard(ClientCallContext context) throws A2AClientException {
+        agentCard = clientTransport.getAgentCard(context);
+        return agentCard;
+    }
+
+    @Override
+    public void close() {
+        clientTransport.close();
+    }
+
+    private ClientEvent getClientEvent(StreamingEventKind event, ClientTaskManager taskManager) throws A2AClientError {
+        if (event instanceof Message message) {
+            return new MessageEvent(message);
+        } else if (event instanceof Task task) {
+            taskManager.saveTaskEvent(task);
+            return new TaskEvent(taskManager.getCurrentTask());
+        } else if (event instanceof TaskStatusUpdateEvent updateEvent) {
+            taskManager.saveTaskEvent(updateEvent);
+            return new TaskUpdateEvent(taskManager.getCurrentTask(), updateEvent);
+        } else if (event instanceof TaskArtifactUpdateEvent updateEvent) {
+            taskManager.saveTaskEvent(updateEvent);
+            return new TaskUpdateEvent(taskManager.getCurrentTask(), updateEvent);
+        } else {
+            throw new A2AClientInvalidStateError("Invalid client event");
+        }
+    }
+
+    private void sendMessage(MessageSendParams messageSendParams, List<BiConsumer<ClientEvent, AgentCard>> consumers,
+                             Consumer<Throwable> errorHandler, ClientCallContext context) throws A2AClientException {
+        if (! clientConfig.isStreaming() || ! agentCard.capabilities().streaming()) {
+            EventKind eventKind = clientTransport.sendMessage(messageSendParams, context);
+            ClientEvent clientEvent;
+            if (eventKind instanceof Task task) {
+                clientEvent = new TaskEvent(task);
+            } else {
+                // must be a message
+                clientEvent = new MessageEvent((Message) eventKind);
+            }
+            consume(clientEvent, agentCard, consumers);
+        } else {
+            ClientTaskManager tracker = new ClientTaskManager();
+            Consumer<Throwable> overriddenErrorHandler = getOverriddenErrorHandler(errorHandler);
+            Consumer<StreamingEventKind> eventHandler = event -> {
+                try {
+                    ClientEvent clientEvent = getClientEvent(event, tracker);
+                    consume(clientEvent, agentCard, consumers);
+                } catch (A2AClientError e) {
+                    overriddenErrorHandler.accept(e);
+                }
+            };
+            clientTransport.sendMessageStreaming(messageSendParams, eventHandler, overriddenErrorHandler, context);
+        }
+    }
+
+    private void resubscribeToTask(TaskIdParams request, List<BiConsumer<ClientEvent, AgentCard>> consumers,
+                                   Consumer<Throwable> errorHandler, ClientCallContext context) throws A2AClientException {
+        if (! clientConfig.isStreaming() || ! agentCard.capabilities().streaming()) {
+            throw new A2AClientException("Client and/or server does not support resubscription");
+        }
+        ClientTaskManager tracker = new ClientTaskManager();
+        Consumer<Throwable> overriddenErrorHandler = getOverriddenErrorHandler(errorHandler);
+        Consumer<StreamingEventKind> eventHandler = event -> {
+            try {
+                ClientEvent clientEvent = getClientEvent(event, tracker);
+                consume(clientEvent, agentCard, consumers);
+            } catch (A2AClientError e) {
+                overriddenErrorHandler.accept(e);
+            }
+        };
+        clientTransport.resubscribe(request, eventHandler, overriddenErrorHandler, context);
+    }
+
+    private Consumer<Throwable> getOverriddenErrorHandler(Consumer<Throwable> errorHandler) {
+        return e -> {
+            if (errorHandler != null) {
+                errorHandler.accept(e);
+            } else {
+                if (getStreamingErrorHandler() != null) {
+                    getStreamingErrorHandler().accept(e);
+                }
+            }
+        };
+    }
+
+    private void consume(ClientEvent clientEvent, AgentCard agentCard, List<BiConsumer<ClientEvent, AgentCard>> consumers) {
+        if (consumers != null) {
+            // use specified consumers
+            for (BiConsumer<ClientEvent, AgentCard> consumer : consumers) {
+                consumer.accept(clientEvent, agentCard);
+            }
+        } else {
+            // use configured consumers
+            consume(clientEvent, agentCard);
+        }
+    }
+
+    private MessageSendParams getMessageSendParams(Message request, ClientConfig clientConfig) {
+        MessageSendConfiguration messageSendConfiguration = new MessageSendConfiguration.Builder()
+                .acceptedOutputModes(clientConfig.getAcceptedOutputModes())
+                .blocking(clientConfig.isPolling())
+                .historyLength(clientConfig.getHistoryLength())
+                .pushNotification(clientConfig.getPushNotificationConfig())
+                .build();
+
+        return new MessageSendParams.Builder()
+                .message(request)
+                .configuration(messageSendConfiguration)
+                .metadata(clientConfig.getMetadata())
+                .build();
+    }
+}

client/base/src/main/java/io/a2a/client/ClientEvent.java

@@ -0,0 +1,4 @@
+package io.a2a.client;
+
+public sealed interface ClientEvent permits MessageEvent, TaskEvent, TaskUpdateEvent {
+}