Client transport config

client/base/pom.xml

@@ -0,0 +1,63 @@
+<?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.3.0.Beta1-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+    <artifactId>a2a-java-sdk-client</artifactId>
+
+    <packaging>jar</packaging>
+
+    <name>Java SDK A2A Client</name>
+    <description>Java SDK for the Agent2Agent Protocol (A2A) - Client</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>a2a-java-sdk-http-client</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>a2a-java-sdk-client-transport-spi</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>a2a-java-sdk-client-transport-jsonrpc</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>a2a-java-sdk-client-transport-grpc</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </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>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mock-server</groupId>
+            <artifactId>mockserver-netty</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

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

@@ -3,9 +3,9 @@
 import java.util.Collections;
 import java.util.Map;
 
-import io.a2a.client.A2ACardResolver;
-import io.a2a.http.A2AHttpClient;
-import io.a2a.http.JdkA2AHttpClient;
+import io.a2a.client.http.A2ACardResolver;
+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/base/src/main/java/io/a2a/client/Client.java

@@ -0,0 +1,240 @@
+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.transport.spi.interceptors.ClientCallContext;
+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;
+
+import static io.a2a.util.Assert.checkNotNullParam;
+
+public class Client extends AbstractClient {
+
+    private final ClientConfig clientConfig;
+    private final ClientTransport clientTransport;
+    private AgentCard agentCard;
+
+    Client(AgentCard agentCard, ClientConfig clientConfig, ClientTransport clientTransport,
+                  List<BiConsumer<ClientEvent, AgentCard>> consumers, Consumer<Throwable> streamingErrorHandler) {
+        super(consumers, streamingErrorHandler);
+        checkNotNullParam("agentCard", agentCard);
+
+        this.agentCard = agentCard;
+        this.clientConfig = clientConfig;
+        this.clientTransport = clientTransport;
+    }
+
+    public static ClientBuilder from(AgentCard agentCard) {
+        return new ClientBuilder(agentCard);
+    }
+
+    @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();
+    }
+}