[Fix #1087] A2A implementation

impl/a2a/pom.xml

@@ -0,0 +1,30 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>io.serverlessworkflow</groupId>
+        <artifactId>serverlessworkflow-impl</artifactId>
+        <version>8.0.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>serverlessworkflow-impl-a2a</artifactId>
+    <name>Serverless Workflow :: Impl :: A2A</name>
+    <dependencies>
+        <dependency>
+            <groupId>io.serverlessworkflow</groupId>
+            <artifactId>serverlessworkflow-impl-core</artifactId>
+        </dependency>
+        <dependency>
+          <groupId>org.a2aproject.sdk</groupId>
+          <artifactId>a2a-java-sdk-client</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

impl/a2a/src/main/java/io/serverlessworkflow/impl/executors/a2a/A2AExceptionHandler.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.a2a;
+
+import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowPosition;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+class A2AExceptionHandler implements Consumer<Throwable> {
+
+  private final CompletableFuture<WorkflowModel> future;
+  private final WorkflowPosition position;
+
+  A2AExceptionHandler(CompletableFuture<WorkflowModel> future, WorkflowPosition position) {
+    this.future = future;
+    this.position = position;
+  }
+
+  @Override
+  public void accept(Throwable ex) {
+    future.completeExceptionally(A2AUtils.workflowException(position, ex));
+  }
+}

impl/a2a/src/main/java/io/serverlessworkflow/impl/executors/a2a/A2AExecutor.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.a2a;
+
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowContext;
+import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowValueResolver;
+import io.serverlessworkflow.impl.executors.CallableTask;
+import java.net.URI;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import org.a2aproject.sdk.client.Client;
+import org.a2aproject.sdk.client.config.ClientConfig;
+import org.a2aproject.sdk.client.http.A2ACardResolver;
+import org.a2aproject.sdk.client.transport.jsonrpc.JSONRPCTransport;
+import org.a2aproject.sdk.client.transport.jsonrpc.JSONRPCTransportConfig;
+import org.a2aproject.sdk.spec.A2AClientException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class A2AExecutor implements CallableTask {
+
+  private final WorkflowValueResolver<URI> uriSupplier;
+  private final A2ARequestDispatcher dispatcher;
+  private final Optional<WorkflowValueResolver<Map<String, Object>>> mapResolver;
+
+  private static final Logger logger = LoggerFactory.getLogger(A2AExecutor.class);
+
+  public A2AExecutor(
+      WorkflowValueResolver<URI> uriSupplier,
+      A2ARequestDispatcher dispatcher,
+      Optional<WorkflowValueResolver<Map<String, Object>>> mapResolver) {
+    this.uriSupplier = uriSupplier;
+    this.dispatcher = dispatcher;
+    this.mapResolver = mapResolver;
+  }
+
+  @Override
+  public CompletableFuture<WorkflowModel> apply(
+      WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
+    URI uri = uriSupplier.apply(workflowContext, taskContext, input);
+
+    return CompletableFuture.supplyAsync(
+            () -> {
+              try {
+                return A2ACardResolver.builder()
+                    .baseUrl(uri.resolve("/").toString())
+                    .agentCardPath(uri.getPath())
+                    .build()
+                    .getAgentCard();
+              } catch (A2AClientException ex) {
+                throw A2AUtils.workflowException(taskContext.position(), ex);
+              }
+            },
+            workflowContext.definition().application().executorService())
+        .thenCompose(
+            agentCard -> {
+              logger.debug("Agent card is {}", agentCard);
+              try {
+                return dispatcher.apply(
+                    agentCard,
+                    Client.builder(agentCard)
+                        .clientConfig(new ClientConfig.Builder().build())
+                        .withTransport(JSONRPCTransport.class, new JSONRPCTransportConfig())
+                        .build(),
+                    mapResolver
+                        .map(m -> m.apply(workflowContext, taskContext, input))
+                        .orElse(Map.of()),
+                    workflowContext,
+                    taskContext);
+              } catch (A2AClientException ex) {
+                throw A2AUtils.workflowException(taskContext.position(), ex);
+              }
+            });
+  }
+}

impl/a2a/src/main/java/io/serverlessworkflow/impl/executors/a2a/A2AExecutorBuilder.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.a2a;
+
+import io.serverlessworkflow.api.types.A2AArguments;
+import io.serverlessworkflow.api.types.CallA2A;
+import io.serverlessworkflow.api.types.Parameters;
+import io.serverlessworkflow.api.types.TaskBase;
+import io.serverlessworkflow.api.types.WithA2AParameters;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowMutablePosition;
+import io.serverlessworkflow.impl.WorkflowUtils;
+import io.serverlessworkflow.impl.WorkflowValueResolver;
+import io.serverlessworkflow.impl.executors.CallableTaskBuilder;
+import io.serverlessworkflow.impl.executors.CallableTaskFactory;
+import java.net.URI;
+import java.util.Map;
+import java.util.Optional;
+
+public class A2AExecutorBuilder implements CallableTaskBuilder<CallA2A> {
+
+  @Override
+  public boolean accept(Class<? extends TaskBase> clazz) {
+    return CallA2A.class.equals(clazz);
+  }
+
+  @Override
+  public CallableTaskFactory init(
+      CallA2A task, WorkflowDefinition definition, WorkflowMutablePosition position) {
+    A2AArguments args = task.getWith();
+
+    WorkflowValueResolver<URI> uriSupplier;
+    if (args.getServer() != null) {
+      uriSupplier = definition.resourceLoader().uriSupplier(args.getServer());
+    } else if (args.getAgentCard() != null) {
+      uriSupplier = definition.resourceLoader().uriSupplier(args.getAgentCard().getEndpoint());
+    } else {
+      throw new IllegalArgumentException("Neither server nor agent card is set for task: " + task);
+    }
+
+    A2ARequestDispatcher dispatcher =
+        switch (args.getMethod()) {
+          case MESSAGE_SEND ->
+              new MessageDispatcher(
+                  (workflowContext, taskContext, completableFuture) ->
+                      new MessageSendConsumer(workflowContext.definition(), completableFuture));
+          case MESSAGE_STREAM ->
+              new MessageDispatcher(
+                  (workflowContext, taskContext, completableFuture) ->
+                      new MessageStreamConsumer(
+                          workflowContext.definition(), completableFuture, taskContext.position()));
+          case TASKS_LIST -> new ListTaskDispatcher();
+          case TASKS_GET -> new GetTaskDispatcher();
+          case TASKS_CANCEL -> new CancelTaskDispatcher();
+          // TODO handle missing cases
+          case AGENT_GET_AUTHENTICATED_EXTENDED_CARD,
+              TASKS_PUSH_NOTIFICATION_CONFIG_DELETE,
+              TASKS_PUSH_NOTIFICATION_CONFIG_GET,
+              TASKS_PUSH_NOTIFICATION_CONFIG_LIST,
+              TASKS_PUSH_NOTIFICATION_CONFIG_SET,
+              TASKS_RESUBSCRIBE ->
+              throw new UnsupportedOperationException("Unimplemented case: " + args.getMethod());
+        };
+
+    Parameters parameters = args.getParameters();
+    Optional<WorkflowValueResolver<Map<String, Object>>> mapResolver;
+    if (parameters == null) {
+      mapResolver = Optional.empty();
+    } else {
+      WithA2AParameters a2aParameters = parameters.getWithA2AParameters();
+      mapResolver =
+          Optional.of(
+              WorkflowUtils.buildMapResolver(
+                  definition.application(),
+                  parameters.getString(),
+                  a2aParameters != null ? a2aParameters.getAdditionalProperties() : null));
+    }
+    return () -> new A2AExecutor(uriSupplier, dispatcher, mapResolver);
+  }
+}

impl/a2a/src/main/java/io/serverlessworkflow/impl/executors/a2a/A2ARequestDispatcher.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.a2a;
+
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowContext;
+import io.serverlessworkflow.impl.WorkflowModel;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.a2aproject.sdk.client.Client;
+import org.a2aproject.sdk.spec.AgentCard;
+
+@FunctionalInterface
+interface A2ARequestDispatcher {
+  CompletableFuture<WorkflowModel> apply(
+      AgentCard agentCard,
+      Client client,
+      Map<String, Object> parameters,
+      WorkflowContext workflowContext,
+      TaskContext taskContext);
+}