diff --git a/server-common/src/main/java/io/a2a/server/tasks/BasePushNotificationSender.java b/server-common/src/main/java/io/a2a/server/tasks/BasePushNotificationSender.java index e400b7ba0..8027c89d3 100644 --- a/server-common/src/main/java/io/a2a/server/tasks/BasePushNotificationSender.java +++ b/server-common/src/main/java/io/a2a/server/tasks/BasePushNotificationSender.java @@ -3,15 +3,18 @@ import static io.a2a.client.http.A2AHttpClient.APPLICATION_JSON; import static io.a2a.client.http.A2AHttpClient.CONTENT_TYPE; import static io.a2a.common.A2AHeaders.X_A2A_NOTIFICATION_TOKEN; -import jakarta.enterprise.context.ApplicationScoped; -import jakarta.inject.Inject; import java.io.IOException; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; + import com.fasterxml.jackson.core.JsonProcessingException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import io.a2a.client.http.A2AHttpClient; import io.a2a.client.http.JdkA2AHttpClient; @@ -19,9 +22,6 @@ import io.a2a.spec.Task; import io.a2a.util.Utils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - @ApplicationScoped public class BasePushNotificationSender implements PushNotificationSender { @@ -80,7 +80,7 @@ private boolean dispatchNotification(Task task, PushNotificationConfig pushInfo) String body; try { - body = Utils.OBJECT_MAPPER.writeValueAsString(task); + body = Utils.marshalFrom(task); } catch (JsonProcessingException e) { LOGGER.debug("Error writing value as string: {}", e.getMessage(), e); return false; diff --git a/server-common/src/test/java/io/a2a/server/requesthandlers/AbstractA2ARequestHandlerTest.java b/server-common/src/test/java/io/a2a/server/requesthandlers/AbstractA2ARequestHandlerTest.java index d654a83a6..31133be38 100644 --- a/server-common/src/test/java/io/a2a/server/requesthandlers/AbstractA2ARequestHandlerTest.java +++ b/server-common/src/test/java/io/a2a/server/requesthandlers/AbstractA2ARequestHandlerTest.java @@ -6,6 +6,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Properties; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; @@ -15,6 +16,12 @@ import jakarta.enterprise.context.Dependent; +import com.fasterxml.jackson.databind.JsonNode; +import io.quarkus.arc.profile.IfBuildProfile; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; + import io.a2a.client.http.A2AHttpClient; import io.a2a.client.http.A2AHttpResponse; import io.a2a.server.agentexecution.AgentExecutor; @@ -30,20 +37,14 @@ import io.a2a.server.tasks.TaskStore; import io.a2a.spec.AgentCapabilities; import io.a2a.spec.AgentCard; +import io.a2a.spec.Event; import io.a2a.spec.JSONRPCError; import io.a2a.spec.Message; import io.a2a.spec.Task; import io.a2a.spec.TaskState; import io.a2a.spec.TaskStatus; -import io.a2a.spec.Event; import io.a2a.spec.TextPart; import io.a2a.util.Utils; -import io.quarkus.arc.profile.IfBuildProfile; -import java.util.Map; - -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.BeforeEach; public class AbstractA2ARequestHandlerTest { @@ -199,7 +200,11 @@ public PostBuilder body(String body) { @Override public A2AHttpResponse post() throws IOException, InterruptedException { - tasks.add(Utils.OBJECT_MAPPER.readValue(body, Task.TYPE_REFERENCE)); + JsonNode root = Utils.OBJECT_MAPPER.readTree(body); + // This will need to be updated for #490 to unmarshall based on the kind of payload + JsonNode taskNode = root.elements().next(); + Task task = Utils.OBJECT_MAPPER.treeToValue(taskNode, Task.TYPE_REFERENCE); + tasks.add(task); try { return new A2AHttpResponse() { @Override diff --git a/server-common/src/test/java/io/a2a/server/tasks/PushNotificationSenderTest.java b/server-common/src/test/java/io/a2a/server/tasks/PushNotificationSenderTest.java index c52a4cf20..65ae20703 100644 --- a/server-common/src/test/java/io/a2a/server/tasks/PushNotificationSenderTest.java +++ b/server-common/src/test/java/io/a2a/server/tasks/PushNotificationSenderTest.java @@ -1,6 +1,6 @@ package io.a2a.server.tasks; -import static io.a2a.client.http.A2AHttpClient.APPLICATION_JSON; + import static io.a2a.client.http.A2AHttpClient.APPLICATION_JSON; import static io.a2a.client.http.A2AHttpClient.CONTENT_TYPE; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -16,17 +16,18 @@ import java.util.concurrent.TimeUnit; import java.util.function.Consumer; +import com.fasterxml.jackson.databind.JsonNode; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import io.a2a.client.http.A2AHttpClient; import io.a2a.client.http.A2AHttpResponse; import io.a2a.common.A2AHeaders; -import io.a2a.util.Utils; import io.a2a.spec.PushNotificationConfig; import io.a2a.spec.Task; import io.a2a.spec.TaskState; import io.a2a.spec.TaskStatus; +import io.a2a.util.Utils; public class PushNotificationSenderTest { @@ -77,7 +78,11 @@ public A2AHttpResponse post() throws IOException, InterruptedException { } try { - Task task = Utils.OBJECT_MAPPER.readValue(body, Task.TYPE_REFERENCE); + JsonNode root = Utils.OBJECT_MAPPER.readTree(body); + // This assumes there is always one field in the outer JSON object. + // This will need to be updated for #490 to unmarshall based on the kind of payload + JsonNode taskNode = root.elements().next(); + Task task = Utils.OBJECT_MAPPER.treeToValue(taskNode, Task.TYPE_REFERENCE); tasks.add(task); urls.add(url); headers.add(new java.util.HashMap<>(requestHeaders)); diff --git a/spec/src/main/java/io/a2a/util/Utils.java b/spec/src/main/java/io/a2a/util/Utils.java index 59db4fb67..2901d546c 100644 --- a/spec/src/main/java/io/a2a/util/Utils.java +++ b/spec/src/main/java/io/a2a/util/Utils.java @@ -2,6 +2,7 @@ import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.logging.Logger; import com.fasterxml.jackson.core.JsonProcessingException; @@ -11,6 +12,7 @@ import io.a2a.spec.Artifact; import io.a2a.spec.Part; +import io.a2a.spec.StreamingEventKind; import io.a2a.spec.Task; import io.a2a.spec.TaskArtifactUpdateEvent; @@ -69,6 +71,21 @@ public static T unmarshalFrom(String data, TypeReference typeRef) throws return OBJECT_MAPPER.readValue(data, typeRef); } + /** + * Serializes a StreamingEventKind in a JSON string + *

+ * The StreamingEventKind object is wrapped in a JSON field named from its kind (e.g. "task") before + * it is serialized + * + * @param kind the StreamingEventKind to deserialize + * @return a JSON String + * @throws JsonProcessingException if JSON parsing fails + */ + public static String marshalFrom(StreamingEventKind kind) throws JsonProcessingException { + Map wrapper = Map.of(kind.getKind(), kind); + return OBJECT_MAPPER.writeValueAsString(wrapper); + } + /** * Returns the provided value if non-null, otherwise returns the default value. *