You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@eventmesh.apache.org by "qqeasonchen (via GitHub)" <gi...@apache.org> on 2026/06/28 02:00:13 UTC
[PR] [ISSUE #5259] Add A2A Gateway: REST API, SSE streaming, Task lifecy… (eventmesh)
qqeasonchen opened a new pull request, #5260:
URL: https://github.com/apache/eventmesh/pull/5260
…cle, Java SDK and tests
## What's New
### A2A Gateway Runtime (eventmesh-runtime)
- A2AGatewayServer: standalone HTTP server entry point
- A2AGatewayService: core gateway logic with task timeout, agent validation, cancel notification, and concurrent-safe status subscribers
- A2AGatewayHttpHandler: Netty-based REST handler with CORS, SSE streaming, health check, task list pagination, and precise error codes
- TaskRegistry: in-memory task lifecycle management with TTL cleanup, CopyOnWriteArrayList for parent-child concurrency safety
- InMemoryA2AMessageTransport: default transport implementation
- A2ACardHttpHandler: enhanced with CORS headers
### A2A Protocol Plugin (eventmesh-protocol-a2a)
- A2AClient: typed Java SDK with sync/async task send, SSE streaming (streamTaskStatus), agent discovery, heartbeat, and SSE-aware socket timeout
- A2AMessageTransport: transport abstraction interface
- A2ATopicFactory: A2A topic convention standardization
- A2ATopicFactoryTest: unit tests
### Examples
- A2AGatewayDemo: end-to-end HTTP client demo
### Tests (all passing)
- A2AGatewayServiceTest: 15+ tests covering timeout, agent validation, cancel, SSE, pagination
- TaskRegistryTest: TTL, concurrent child creation, lifecycle transitions
- A2AClientServerIntegrationTest: full client-server integration with SSE streaming, CORS, health, heartbeat
- A2AGatewayEndToEndTest: end-to-end pipeline verification
- InMemoryA2AMessageTransportTest: transport unit tests
### Documentation (7 files updated)
- eventmesh-a2a-design.md, ARCHITECTURE.md, IMPLEMENTATION_SUMMARY.md, IMPLEMENTATION_SUMMARY_EN.md, TEST_RESULTS.md, README.md, README_EN.md
### Key Engineering Decisions
- Task timeout: auto-fail non-terminal tasks after configurable threshold
- SSE: heartbeat mechanism + socketTimeout(0) for long-lived connections
- Concurrency: CopyOnWriteArrayList + ConcurrentHashMap throughout
- CORS: Access-Control-Allow-Origin on all A2A endpoints
- Agent validation: reject tasks for unregistered agents
<!--
### Contribution Checklist
- Name the pull request in the form "[ISSUE #XXXX] Title of the pull request",
where *XXXX* should be replaced by the actual issue number.
Skip *[ISSUE #XXXX]* if there is no associated github issue for this pull request.
- Fill out the template below to describe the changes contributed by the pull request.
That will give reviewers the context they need to do the review.
- Each pull request should address only one issue.
Please do not mix up code from multiple issues.
- Each commit in the pull request should have a meaningful commit message.
- Once all items of the checklist are addressed, remove the above text and this checklist,
leaving only the filled out template below.
(The sections below can be removed for hotfixes of typos)
-->
<!--
(If this PR fixes a GitHub issue, please add `Fixes #<XXX>` or `Closes #<XXX>`.)
-->
Fixes #issue_id
### Motivation
*Explain the content here.*
*Explain why you want to make the changes and what problem you're trying to solve.*
### Modifications
*Describe the modifications you've done.*
### Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
- If a feature is not applicable for documentation, explain why?
- If a feature is not documented yet in this PR, please create a followup issue for adding the documentation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org
Re: [PR] [ISSUE #5259] Add A2A Gateway: REST API, SSE streaming, Task lifecycle, Java SDK and tests (eventmesh)
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #5260:
URL: https://github.com/apache/eventmesh/pull/5260#discussion_r3488699720
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/TaskRegistry.java:
##########
@@ -0,0 +1,348 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * In-memory registry for A2A task lifecycle management.
+ *
+ * <p>Maintains task state transitions and parent-child task relationships.
+ *
+ * <p>State machine:
+ * <pre>
+ * SUBMITTED -> WORKING -> COMPLETED
+ * \-> FAILED
+ * \-> CANCELLED
+ * </pre>
+ */
+@Slf4j
+public class TaskRegistry {
+
+ public enum TaskState {
+ SUBMITTED,
+ WORKING,
+ COMPLETED,
+ FAILED,
+ CANCELLED
+ }
+
+ public static class TaskEntry {
+
+ private final String taskId;
+ private final String parentTaskId;
+ private final String targetAgent;
+ private final String gatewayId;
+ private final long createdAt;
+ private volatile long updatedAt;
+ private volatile TaskState state;
+ private volatile String result;
+ private volatile String errorMessage;
+
+ public TaskEntry(String taskId, String parentTaskId, String targetAgent, String gatewayId) {
+ this.taskId = taskId;
+ this.parentTaskId = parentTaskId;
+ this.targetAgent = targetAgent;
+ this.gatewayId = gatewayId;
+ this.createdAt = System.currentTimeMillis();
+ this.updatedAt = this.createdAt;
+ this.state = TaskState.SUBMITTED;
+ }
+
+ public String getTaskId() {
+ return taskId;
+ }
+
+ public String getParentTaskId() {
+ return parentTaskId;
+ }
+
+ public String getTargetAgent() {
+ return targetAgent;
+ }
+
+ public String getGatewayId() {
+ return gatewayId;
+ }
+
+ public long getCreatedAt() {
+ return createdAt;
+ }
+
+ public long getUpdatedAt() {
+ return updatedAt;
+ }
+
+ public TaskState getState() {
+ return state;
+ }
+
+ public String getResult() {
+ return result;
+ }
+
+ public String getErrorMessage() {
+ return errorMessage;
+ }
+ }
+
+ private final ConcurrentHashMap<String, TaskEntry> tasks = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, List<String>> parentToChildren = new ConcurrentHashMap<>();
+
+ // TTL cleanup configuration
+ private static final long DEFAULT_TASK_TTL_MS = 300_000L; // 5 minutes
+ private static final long DEFAULT_TASK_CLEANUP_INTERVAL_MS = 60_000L; // 60 seconds
+ private final long taskTtlMs;
+ private final long cleanupIntervalMs;
+ private ScheduledExecutorService cleanupExecutor;
+ private volatile boolean started = false;
+
+ public TaskRegistry() {
+ this(DEFAULT_TASK_TTL_MS, DEFAULT_TASK_CLEANUP_INTERVAL_MS);
+ }
+
+ public TaskRegistry(long taskTtlMs, long cleanupIntervalMs) {
+ this.taskTtlMs = taskTtlMs;
+ this.cleanupIntervalMs = cleanupIntervalMs;
+ }
+
+ /**
+ * Starts the TTL cleanup scheduler. Terminal-state tasks older than TTL are automatically removed.
+ */
+ public void start() {
+ if (started) {
+ return;
+ }
+ started = true;
+ cleanupExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "a2a-task-ttl-cleanup");
+ t.setDaemon(true);
+ return t;
+ });
+ cleanupExecutor.scheduleAtFixedRate(this::cleanupExpiredTasks,
+ cleanupIntervalMs, cleanupIntervalMs, TimeUnit.MILLISECONDS);
+ log.info("TaskRegistry started (TTL={}ms, cleanup={}ms)", taskTtlMs, cleanupIntervalMs);
+ }
+
+ /**
+ * Shuts down the TTL cleanup scheduler.
+ */
+ public void shutdown() {
+ if (!started) {
+ return;
+ }
+ started = false;
+ if (cleanupExecutor != null) {
+ cleanupExecutor.shutdownNow();
+ cleanupExecutor = null;
+ }
+ log.info("TaskRegistry shut down.");
+ }
+
+ /**
+ * Removes terminal-state tasks whose updatedAt timestamp is older than TTL.
+ */
+ void cleanupExpiredTasks() {
+ if (!started) {
+ return;
+ }
+ long now = System.currentTimeMillis();
+ int removed = 0;
+ Iterator<Map.Entry<String, TaskEntry>> it = tasks.entrySet().iterator();
+ while (it.hasNext()) {
+ Map.Entry<String, TaskEntry> entry = it.next();
+ TaskEntry te = entry.getValue();
+ if (isTerminalState(te.getState()) && (now - te.getUpdatedAt()) > taskTtlMs) {
+ it.remove();
Review Comment:
`ConcurrentHashMap` iterators do not support `Iterator.remove()`. Calling `it.remove()` here will throw `UnsupportedOperationException`, breaking the TTL cleanup scheduler and potentially leaking tasks indefinitely.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2APublishSubscribeService.java:
##########
@@ -57,27 +61,102 @@ public class A2APublishSubscribeService {
private AgentCardValidator cardValidator;
+ // TTL heartbeat configuration
+ private static final long DEFAULT_CARD_TTL_MS = 60_000L; // 60 seconds
+ private static final long DEFAULT_CLEANUP_INTERVAL_MS = 15_000L; // 15 seconds
+ private final long cardTtlMs;
+ private final long cleanupIntervalMs;
+ private ScheduledExecutorService cleanupExecutor;
+
public A2APublishSubscribeService(EventMeshServer eventMeshServer) {
+ this(eventMeshServer, DEFAULT_CARD_TTL_MS, DEFAULT_CLEANUP_INTERVAL_MS);
+ }
+
+ public A2APublishSubscribeService(EventMeshServer eventMeshServer, long cardTtlMs, long cleanupIntervalMs) {
this.eventMeshServer = eventMeshServer;
+ this.cardTtlMs = cardTtlMs;
+ this.cleanupIntervalMs = cleanupIntervalMs;
}
public void init() throws Exception {
this.cardValidator = new AgentCardValidator(true);
- log.info("A2APublishSubscribeService initialized with Agent Card Registry.");
+ log.info("A2APublishSubscribeService initialized with Agent Card Registry (TTL={}ms, cleanup={}ms).",
+ cardTtlMs, cleanupIntervalMs);
}
public void start() throws Exception {
isStarted = true;
+ startCleanupScheduler();
log.info("A2APublishSubscribeService started.");
}
public void shutdown() throws Exception {
isStarted = false;
+ if (cleanupExecutor != null) {
+ cleanupExecutor.shutdownNow();
+ cleanupExecutor = null;
+ }
cardRegistry.clear();
agentStatusMap.clear();
log.info("A2APublishSubscribeService shutdown.");
}
+ // -------------------------------------------------------------------------
+ // TTL Heartbeat Cleanup
+ // -------------------------------------------------------------------------
+
+ private void startCleanupScheduler() {
+ cleanupExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "a2a-card-ttl-cleanup");
+ t.setDaemon(true);
+ return t;
+ });
+ cleanupExecutor.scheduleAtFixedRate(this::cleanupExpiredCards,
+ cleanupIntervalMs, cleanupIntervalMs, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Removes AgentCards that have not been refreshed within the TTL window.
+ */
+ void cleanupExpiredCards() {
+ if (!isStarted) {
+ return;
+ }
+ long now = System.currentTimeMillis();
+ int removed = 0;
+ Iterator<Map.Entry<AgentIdentity, RegisteredCard>> it = cardRegistry.entrySet().iterator();
+ while (it.hasNext()) {
+ Map.Entry<AgentIdentity, RegisteredCard> entry = it.next();
+ RegisteredCard rc = entry.getValue();
+ if (now - rc.lastHeartbeat > cardTtlMs) {
+ it.remove();
Review Comment:
`ConcurrentHashMap` iterators do not support `Iterator.remove()`. This `it.remove()` will throw `UnsupportedOperationException`, so expired AgentCards will never be cleaned up and the cleanup task may fail repeatedly.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayHttpHandler.java:
##########
@@ -0,0 +1,648 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a;
+
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpContent;
+import io.netty.handler.codec.http.DefaultHttpResponse;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.handler.codec.http.QueryStringDecoder;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * HTTP handler for A2A Gateway API.
+ *
+ * <p>Endpoints:
+ * <ul>
+ * <li>POST /a2a/tasks - submit a new task (sync or async)</li>
+ * <li>GET /a2a/tasks - list tasks (optional ?state=COMPLETED&limit=100&offset=0)</li>
+ * <li>GET /a2a/tasks/{taskId} - get task status</li>
+ * <li>DELETE /a2a/tasks/{taskId} - cancel a task</li>
+ * <li>GET /a2a/tasks/{taskId}/wait - wait for task result (long-poll)</li>
+ * <li>GET /a2a/tasks/{taskId}/stream - SSE stream of task status updates</li>
+ * <li>GET /a2a/agents - list registered agents</li>
+ * <li>POST /a2a/heartbeat - agent heartbeat</li>
+ * <li>GET /a2a/health - health check (for liveness/readiness probes)</li>
+ * <li>OPTIONS * - CORS preflight</li>
+ * </ul>
+ */
+@Slf4j
+public class A2AGatewayHttpHandler {
+
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ private static final String PATH_TASKS = "/a2a/tasks";
+ private static final String PATH_AGENTS = "/a2a/agents";
+ private static final String PATH_HEARTBEAT = "/a2a/heartbeat";
+ private static final String PATH_HEALTH = "/a2a/health";
+
+ private static final long DEFAULT_WAIT_TIMEOUT_MS = 30_000L;
+
+ private final A2AGatewayService gatewayService;
+ private final A2APublishSubscribeService a2aService;
+
+ public A2AGatewayHttpHandler(A2AGatewayService gatewayService,
+ A2APublishSubscribeService a2aService) {
+ this.gatewayService = gatewayService;
+ this.a2aService = a2aService;
+ }
+
+ /**
+ * Handles an HTTP request. Returns a FullHttpResponse, or null if the response was written
+ * directly to the channel (e.g. SSE streaming).
+ */
+ public FullHttpResponse handle(HttpRequest httpRequest, ChannelHandlerContext ctx) throws Exception {
+ String uri = httpRequest.uri();
+ QueryStringDecoder decoder = new QueryStringDecoder(uri);
+ String path = decoder.path();
+ HttpMethod method = httpRequest.method();
+
+ try {
+ // Handle CORS preflight
+ if (method == HttpMethod.OPTIONS) {
+ return corsResponse(HttpResponseStatus.OK);
+ }
+
+ // POST /a2a/tasks
+ if (PATH_TASKS.equals(path) && method == HttpMethod.POST) {
+ return handleSubmitTask(httpRequest, decoder);
+ }
+
+ // GET /a2a/tasks (list tasks with optional pagination)
+ if (PATH_TASKS.equals(path) && method == HttpMethod.GET) {
+ return handleListTasks(decoder);
+ }
+
+ // /a2a/tasks/{taskId} or /a2a/tasks/{taskId}/wait
+ if (path.startsWith(PATH_TASKS + "/")) {
+ String suffix = path.substring(PATH_TASKS.length() + 1);
+ String[] parts = suffix.split("/", 2);
+ String taskId = parts[0];
+
+ if (taskId.isEmpty()) {
+ return jsonResponse(HttpResponseStatus.BAD_REQUEST, errorBody("Missing taskId"));
+ }
+
+ // GET /a2a/tasks/{taskId}
+ if (method == HttpMethod.GET && parts.length == 1) {
+ return handleGetTask(taskId);
+ }
+
+ // DELETE /a2a/tasks/{taskId}
+ if (method == HttpMethod.DELETE && parts.length == 1) {
+ return handleCancelTask(taskId);
+ }
+
+ // GET /a2a/tasks/{taskId}/wait
+ if (method == HttpMethod.GET && parts.length == 2 && "wait".equals(parts[1])) {
+ return handleWaitTask(taskId, decoder);
+ }
+
+ // GET /a2a/tasks/{taskId}/stream (SSE)
+ if (method == HttpMethod.GET && parts.length == 2 && "stream".equals(parts[1])) {
+ return handleStreamTask(taskId, ctx);
+ }
+ }
+
+ // GET /a2a/health
+ if (PATH_HEALTH.equals(path) && method == HttpMethod.GET) {
+ return handleHealth();
+ }
+
+ // GET /a2a/agents
+ if (PATH_AGENTS.equals(path) && method == HttpMethod.GET) {
+ return handleListAgents();
+ }
+
+ // POST /a2a/heartbeat
+ if (PATH_HEARTBEAT.equals(path) && method == HttpMethod.POST) {
+ return handleHeartbeat(httpRequest);
+ }
+
+ return jsonResponse(HttpResponseStatus.NOT_FOUND, errorBody("Not found: " + path));
+ } catch (Exception e) {
+ log.error("A2A gateway handler error: {}", e.getMessage(), e);
+ return jsonResponse(HttpResponseStatus.INTERNAL_SERVER_ERROR, errorBody(e.getMessage()));
+ }
+ }
+
+ // =========================================================================
+ // POST /a2a/tasks
+ // =========================================================================
+
+ private FullHttpResponse handleSubmitTask(HttpRequest request, QueryStringDecoder decoder) throws Exception {
+ String body = readBody(request);
+ TaskRequest taskRequest;
+ try {
+ taskRequest = objectMapper.readValue(body, TaskRequest.class);
+ } catch (Exception e) {
+ return jsonResponse(HttpResponseStatus.BAD_REQUEST, errorBody("Invalid JSON: " + e.getMessage()));
+ }
+
+ if (taskRequest.getTargetAgent() == null || taskRequest.getTargetAgent().isEmpty()) {
+ return jsonResponse(HttpResponseStatus.BAD_REQUEST, errorBody("targetAgent is required"));
+ }
+ if (taskRequest.getMessage() == null) {
+ taskRequest.setMessage("");
+ }
+
+ String mode = getQueryParam(decoder, "mode");
+ String parentTaskId = taskRequest.getParentTaskId();
+
+ // Generate taskId first so we can include it in the response
+ String taskId = "task-" + java.util.UUID.randomUUID().toString().substring(0, 8);
+
+ // Submit the task with the explicit taskId
+ java.util.concurrent.CompletableFuture<A2AGatewayService.TaskResult> future =
+ gatewayService.submitTask(taskId, taskRequest.getTargetAgent(), taskRequest.getMessage(), parentTaskId);
+
+ // For sync mode, wait for result
+ if ("sync".equalsIgnoreCase(mode) || mode == null) {
+ try {
+ A2AGatewayService.TaskResult result = future.get(DEFAULT_WAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ Map<String, Object> response = new LinkedHashMap<>();
+ response.put("taskId", taskId);
+ response.put("state", result.getState().name());
+ response.put("data", result.getData());
+ if (result.getErrorMessage() != null) {
+ response.put("error", result.getErrorMessage());
+ }
+ return jsonResponse(HttpResponseStatus.OK, objectMapper.writeValueAsString(response));
+ } catch (java.util.concurrent.TimeoutException e) {
+ Map<String, Object> response = new LinkedHashMap<>();
+ response.put("taskId", taskId);
+ response.put("state", "TIMEOUT");
+ response.put("error", "Task accepted but timed out waiting for response");
+ return jsonResponse(HttpResponseStatus.ACCEPTED, objectMapper.writeValueAsString(response));
+ } catch (java.util.concurrent.ExecutionException e) {
+ // Unwrap cause to determine correct status code
+ Throwable cause = e.getCause() != null ? e.getCause() : e;
+ if (cause instanceof IllegalArgumentException || cause instanceof IllegalStateException) {
+ return jsonResponse(HttpResponseStatus.BAD_REQUEST,
+ errorBody(cause.getMessage()));
+ }
+ return jsonResponse(HttpResponseStatus.INTERNAL_SERVER_ERROR,
+ errorBody("Task failed: " + cause.getMessage()));
+ } catch (Exception e) {
+ return jsonResponse(HttpResponseStatus.INTERNAL_SERVER_ERROR,
+ errorBody("Task failed: " + e.getMessage()));
+ }
+ } else {
+ // Async mode: return task ID immediately
+ Map<String, Object> response = new LinkedHashMap<>();
+ response.put("taskId", taskId);
+ response.put("status", "accepted");
+ response.put("message", "Task submitted. Use GET /a2a/tasks/" + taskId + " to check status.");
+ return jsonResponse(HttpResponseStatus.ACCEPTED, objectMapper.writeValueAsString(response));
+ }
Review Comment:
In async mode, the handler always returns HTTP 202 even when `submitTask(...)` has already failed (e.g., gateway not started or target agent not registered). In those cases no task is created, and the client receives a taskId that will 404 forever.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServer.java:
##########
@@ -0,0 +1,338 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a;
+
+import org.apache.eventmesh.protocol.a2a.A2AProtocolConstants;
+import org.apache.eventmesh.protocol.a2a.A2ATopicFactory;
+import org.apache.eventmesh.protocol.a2a.AgentIdentity;
+import org.apache.eventmesh.protocol.a2a.model.AgentCard;
+import org.apache.eventmesh.protocol.a2a.model.AgentSkill;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.UUID;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpRequestDecoder;
+import io.netty.handler.codec.http.HttpResponseEncoder;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Standalone A2A Gateway Server.
+ *
+ * <p>Starts an embedded Netty HTTP server that exposes the A2A Gateway REST API.
+ * Pre-registers a mock "weather-agent" that auto-responds to task requests.
+ *
+ * <p>Usage:
+ * <pre>
+ * java org.apache.eventmesh.runtime.a2a.A2AGatewayServer [port]
+ * </pre>
+ *
+ * <p>Then run the client demo (A2AGatewayDemo) to submit tasks via HTTP.
+ */
+@Slf4j
+public class A2AGatewayServer {
+
+ private static final String NAMESPACE = "global";
+ private static final String GATEWAY_ID = "demo-gateway";
+ private static final int DEFAULT_PORT = 10105;
+
+ private final int port;
+ private EventLoopGroup bossGroup;
+ private EventLoopGroup workerGroup;
+ private ChannelFuture serverChannel;
+
+ private InMemoryA2AMessageTransport transport;
+ private TaskRegistry taskRegistry;
+ private A2APublishSubscribeService a2aService;
+ private A2AGatewayService gatewayService;
+ private A2AGatewayHttpHandler gatewayHandler;
+ private A2ACardHttpHandler cardHandler;
+
+ public A2AGatewayServer(int port) {
+ this.port = port;
+ }
+
+ public void start() throws Exception {
+ // 1. Initialize components
+ transport = new InMemoryA2AMessageTransport();
+ taskRegistry = new TaskRegistry();
+ a2aService = new A2APublishSubscribeService(null);
+ a2aService.init();
+ a2aService.start();
+
+ gatewayService = new A2AGatewayService(NAMESPACE, GATEWAY_ID, transport, taskRegistry, a2aService);
+ gatewayService.start();
+
+ gatewayHandler = new A2AGatewayHttpHandler(gatewayService, a2aService);
+ cardHandler = new A2ACardHttpHandler(a2aService);
+
+ // 2. Pre-register a mock weather agent
+ registerMockWeatherAgent();
+
+ // 3. Start Netty HTTP server
+ bossGroup = new NioEventLoopGroup(1);
+ workerGroup = new NioEventLoopGroup();
+
+ ServerBootstrap bootstrap = new ServerBootstrap();
+ bootstrap.group(bossGroup, workerGroup)
+ .channel(NioServerSocketChannel.class)
+ .childHandler(new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) {
+ ch.pipeline()
+ .addLast(new HttpRequestDecoder())
+ .addLast(new HttpObjectAggregator(65536))
+ .addLast(new HttpResponseEncoder())
+ .addLast(new A2AGatewayRequestHandler(gatewayHandler, cardHandler));
+ }
+ })
+ .option(ChannelOption.SO_BACKLOG, 128)
+ .childOption(ChannelOption.SO_KEEPALIVE, true);
+
+ serverChannel = bootstrap.bind(port).sync();
+ log.info("=== A2A Gateway Server started on port {} ===", port);
+ log.info("Gateway ID: {}, Namespace: {}", GATEWAY_ID, NAMESPACE);
+ log.info("Endpoints:");
+ log.info(" POST /a2a/tasks - submit task (sync/async)");
+ log.info(" GET /a2a/tasks - list tasks (?state=COMPLETED&limit=100&offset=0)");
+ log.info(" GET /a2a/tasks/{{taskId}} - get task status");
+ log.info(" DELETE /a2a/tasks/{{taskId}} - cancel task");
+ log.info(" GET /a2a/tasks/{{taskId}}/wait - wait for result (long-poll)");
+ log.info(" GET /a2a/tasks/{{taskId}}/stream - SSE stream of task status updates");
+ log.info(" GET /a2a/agents - list registered agents");
+ log.info(" POST /a2a/heartbeat - agent heartbeat");
+ log.info(" GET /a2a/health - health check");
+ log.info(" GET /a2a/cards/list - list agent cards");
+ log.info(" POST /a2a/cards/card/{{org}}/{{unit}}/{{agent}} - register card");
+ }
+
+ /**
+ * Registers a mock weather agent that auto-responds to task requests.
+ */
+ private void registerMockWeatherAgent() throws Exception {
+ AgentCard weatherCard = AgentCard.builder()
+ .name("weather-agent")
+ .description("A weather agent that can tell you the weather")
+ .version("1.0.0")
+ .supportedInterfaces(Arrays.asList(
+ org.apache.eventmesh.protocol.a2a.model.AgentInterface.builder()
+ .url("http://localhost:" + port + "/a2a")
+ .protocolBinding("JSONRPC")
+ .protocolVersion("0.3")
+ .build()
+ ))
+ .capabilities(org.apache.eventmesh.protocol.a2a.model.AgentCapabilities.builder()
+ .streaming(false)
+ .pushNotifications(false)
+ .build())
+ .skills(Arrays.asList(
+ AgentSkill.builder()
+ .id("get-weather")
+ .name("Get Weather")
+ .description("Get the current weather for a city")
+ .tags(Arrays.asList("weather", "test"))
+ .build()
+ ))
+ .defaultInputModes(Arrays.asList("text/plain"))
+ .defaultOutputModes(Arrays.asList("text/plain"))
+ .build();
+
+ AgentIdentity identity = AgentIdentity.builder()
+ .orgId("default").unitId("default").agentId("weather-agent").build();
+ A2APublishSubscribeService.RegistrationResult regResult =
+ a2aService.registerCard(identity, weatherCard);
+ if (!regResult.isSuccess()) {
+ log.warn("Failed to register weather-agent card: {}", regResult.getErrorMessage());
+ }
+
+ // Subscribe to agent request topic and auto-respond
+ String requestTopic = A2ATopicFactory.agentRequestTopic(NAMESPACE, "weather-agent");
+ transport.subscribe(requestTopic, (topic, event) -> {
+ String taskId = event.getId();
+ String message = event.getData() != null
+ ? new String(event.getData().toBytes(), StandardCharsets.UTF_8) : "";
+
+ log.info("[Weather Agent] Received request: taskId={}, message={}", taskId, message);
+
+ // Simulate processing delay
+ try {
+ Thread.sleep(300);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ // Publish response
+ String gwId = extractGatewayId(event);
+ String responseTopic = A2ATopicFactory.gatewayResponseTopic(NAMESPACE, gwId, taskId);
+ String result = "The weather in " + message + " is sunny, 25°C";
+ CloudEvent responseEvent = CloudEventBuilder.v1()
+ .withId(UUID.randomUUID().toString())
+ .withType(A2AProtocolConstants.CE_TYPE_PREFIX + "task.response")
+ .withSource(java.net.URI.create("agent/weather-agent"))
+ .withDataContentType("application/json")
+ .withData(result.getBytes(StandardCharsets.UTF_8))
+ .withExtension(A2AProtocolConstants.CE_EXTENSION_A2A_METHOD, A2AProtocolConstants.OP_SEND_MESSAGE)
+ .build();
+
+ try {
+ transport.publish(responseTopic, responseEvent);
+ log.info("[Weather Agent] Response published: taskId={}", taskId);
+ } catch (Exception e) {
+ log.error("[Weather Agent] Failed to publish response", e);
+ }
+ });
+
+ log.info("Mock weather-agent registered and listening for requests.");
+ }
+
+ private String extractGatewayId(CloudEvent event) {
+ if (event.getSource() == null) {
+ return "default-gateway";
+ }
+ String source = event.getSource().toString();
+ if (source.startsWith("gateway/")) {
+ return source.substring("gateway/".length());
+ }
+ return "default-gateway";
+ }
+
+ /**
+ * Returns the actual bound port. Useful when port 0 was specified to let the OS pick a port.
+ */
+ public int getBoundPort() {
+ if (serverChannel != null) {
+ return ((io.netty.channel.socket.ServerSocketChannel) serverChannel.channel()).localAddress().getPort();
+ }
+ return port;
+ }
+
+ public void shutdown() throws Exception {
+ if (serverChannel != null) {
+ serverChannel.channel().close().sync();
+ }
+ if (workerGroup != null) {
+ workerGroup.shutdownGracefully();
+ }
+ if (bossGroup != null) {
+ bossGroup.shutdownGracefully();
+ }
+ if (gatewayService != null) {
+ gatewayService.shutdown();
+ }
+ if (a2aService != null) {
+ a2aService.shutdown();
+ }
+ log.info("A2A Gateway Server shut down.");
+ }
+
+ public static void main(String[] args) throws Exception {
+ int port = args.length > 0 ? Integer.parseInt(args[0]) : DEFAULT_PORT;
+
+ A2AGatewayServer server = new A2AGatewayServer(port);
+
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ try {
+ server.shutdown();
+ } catch (Exception e) {
+ log.error("Error during shutdown", e);
+ }
+ }));
+
+ server.start();
+
+ // Keep running until interrupted
+ Thread.currentThread().join();
+ }
+
+ // =========================================================================
+ // Netty Channel Handler
+ // =========================================================================
+
+ private static class A2AGatewayRequestHandler extends io.netty.channel.SimpleChannelInboundHandler<FullHttpRequest> {
+
+ private final A2AGatewayHttpHandler gatewayHandler;
+ private final A2ACardHttpHandler cardHandler;
+
+ A2AGatewayRequestHandler(A2AGatewayHttpHandler gatewayHandler, A2ACardHttpHandler cardHandler) {
+ this.gatewayHandler = gatewayHandler;
+ this.cardHandler = cardHandler;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
+ String uri = request.uri();
+ try {
+ FullHttpResponse response;
+ if (uri.startsWith("/a2a/cards")) {
+ response = cardHandler.handle(request, ctx);
+ } else if (uri.startsWith("/a2a/")) {
+ response = gatewayHandler.handle(request, ctx);
+ } else {
+ response = jsonError(HttpResponseStatus.NOT_FOUND, "Not found: " + uri);
+ }
+
+ // null means the handler wrote the response directly (e.g. SSE streaming)
+ if (response != null) {
+ ctx.writeAndFlush(response);
+ }
+ } catch (Exception e) {
+ log.error("Error handling request: {} - {}", uri, e.getMessage(), e);
+ FullHttpResponse errorResponse = jsonError(HttpResponseStatus.INTERNAL_SERVER_ERROR, e.getMessage());
+ ctx.writeAndFlush(errorResponse);
+ }
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ log.error("Channel exception: {}", cause.getMessage(), cause);
+ ctx.close();
+ }
+
+ private FullHttpResponse jsonError(HttpResponseStatus status, String message) {
+ String body = "{\"error\":\"" + message + "\"}";
Review Comment:
The error JSON is assembled via string concatenation without escaping `message`. If the message contains quotes/backslashes/newlines (common in exception messages), this produces invalid JSON and can enable response-splitting style issues for naive clients.
##########
eventmesh-protocol-plugin/eventmesh-protocol-a2a/src/main/java/org/apache/eventmesh/protocol/a2a/A2AClient.java:
##########
@@ -0,0 +1,585 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.eventmesh.protocol.a2a;
+
+import org.apache.eventmesh.protocol.a2a.model.AgentCard;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.util.EntityUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * A2A Java SDK client for Agent developers.
+ *
+ * <p>Provides a simple API to:
+ * <ul>
+ * <li>Register and publish AgentCard</li>
+ * <li>Send A2A tasks to other agents via Gateway REST API</li>
+ * <li>Subscribe to and handle incoming A2A requests via transport</li>
+ * <li>Send heartbeat to keep agent card alive</li>
+ * </ul>
+ */
+public class A2AClient implements AutoCloseable {
+
+ private static final Logger log = LoggerFactory.getLogger(A2AClient.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ private final String gatewayUrl;
+ private final String namespace;
+ private final String agentName;
+ private final AgentCard agentCard;
+ private final long heartbeatIntervalMs;
+ private final CloseableHttpClient httpClient;
+
+ private volatile boolean started = false;
+ private ScheduledExecutorService heartbeatExecutor;
+ private RequestHandler requestHandler;
+
+ private A2AMessageTransport transport;
+ private String requestSubscriptionId;
+
+ A2AClient(Builder builder) {
+ this.gatewayUrl = builder.gatewayUrl;
+ this.namespace = builder.namespace;
+ this.agentName = builder.agentName;
+ this.agentCard = builder.agentCard;
+ this.heartbeatIntervalMs = builder.heartbeatIntervalMs;
+ RequestConfig config = RequestConfig.custom()
+ .setConnectTimeout(10_000)
+ .setSocketTimeout(builder.socketTimeoutMs)
+ .build();
+ this.httpClient = HttpClients.custom().setDefaultRequestConfig(config).build();
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ // =========================================================================
+ // Lifecycle
+ // =========================================================================
+
+ public void start() throws Exception {
+ if (started) {
+ return;
+ }
+ registerAgentCard();
+ startHeartbeat();
+ started = true;
+ log.info("A2AClient started: agent={}, namespace={}", agentName, namespace);
+ }
+
+ public void shutdown() throws Exception {
+ if (heartbeatExecutor != null) {
+ heartbeatExecutor.shutdownNow();
+ }
+ if (transport != null && requestSubscriptionId != null) {
+ transport.unsubscribe(requestSubscriptionId);
+ requestSubscriptionId = null;
+ }
+ httpClient.close();
+ started = false;
+ log.info("A2AClient shutdown: agent={}", agentName);
+ }
+
+ @Override
+ public void close() throws Exception {
+ shutdown();
+ }
+
+ // =========================================================================
+ // AgentCard Registration
+ // =========================================================================
+
+ public void registerAgentCard() throws Exception {
+ String[] parts = agentName.split("/");
+ String orgId = parts.length >= 1 ? parts[0] : "default";
+ String unitId = parts.length >= 2 ? parts[1] : "default";
+ String agentId = parts.length >= 3 ? parts[2] : agentName;
+
+ String url = String.format("%s/a2a/cards/card/%s/%s/%s", gatewayUrl, orgId, unitId, agentId);
+ String body = objectMapper.writeValueAsString(agentCard);
+
+ HttpPost post = new HttpPost(url);
+ post.setHeader("Content-Type", "application/json");
+ post.setEntity(new StringEntity(body, StandardCharsets.UTF_8));
+
+ HttpResponse response = httpClient.execute(post);
+ int statusCode = response.getStatusLine().getStatusCode();
+ if (statusCode >= 400) {
+ String respBody = response.getEntity() != null
+ ? EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8) : "";
+ throw new RuntimeException("Failed to register agent card: " + statusCode + " " + respBody);
+ }
+ log.info("AgentCard registered: {} -> {}", agentName, statusCode);
+ }
+
+ // =========================================================================
+ // Heartbeat
+ // =========================================================================
+
+ private void startHeartbeat() {
+ heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "a2a-client-heartbeat");
+ t.setDaemon(true);
+ return t;
+ });
+ heartbeatExecutor.scheduleAtFixedRate(this::sendHeartbeat,
+ heartbeatIntervalMs, heartbeatIntervalMs, TimeUnit.MILLISECONDS);
+ }
+
+ private void sendHeartbeat() {
+ try {
+ String[] parts = agentName.split("/");
+ String orgId = parts.length >= 1 ? parts[0] : "default";
+ String unitId = parts.length >= 2 ? parts[1] : "default";
+ String agentId = parts.length >= 3 ? parts[2] : agentName;
+
+ Map<String, String> body = new HashMap<>();
+ body.put("orgId", orgId);
+ body.put("unitId", unitId);
+ body.put("agentId", agentId);
+
+ HttpPost post = new HttpPost(gatewayUrl + "/a2a/heartbeat");
+ post.setHeader("Content-Type", "application/json");
+ post.setEntity(new StringEntity(objectMapper.writeValueAsString(body), StandardCharsets.UTF_8));
+
+ httpClient.execute(post);
+ log.debug("Heartbeat sent for agent: {}", agentName);
+ } catch (Exception e) {
+ log.warn("Heartbeat failed: {}", e.getMessage());
+ }
+ }
+
+ // =========================================================================
+ // Task Operations (via Gateway REST)
+ // =========================================================================
+
+ public CompletableFuture<TaskResult> sendTask(String targetAgent, String message) {
+ return CompletableFuture.supplyAsync(() -> {
+ try {
+ return sendTaskSync(targetAgent, message, null);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ /**
+ * Submits a task asynchronously. Returns the taskId immediately without waiting for agent response.
+ * Use {@link #getTaskStatus(String)} to poll for result.
+ */
+ public String sendTaskAsync(String targetAgent, String message, String parentTaskId) throws Exception {
+ Map<String, Object> bodyMap = new HashMap<>();
+ bodyMap.put("targetAgent", targetAgent);
+ bodyMap.put("message", message);
+ if (parentTaskId != null) {
+ bodyMap.put("parentTaskId", parentTaskId);
+ }
+ String body = objectMapper.writeValueAsString(bodyMap);
+
+ HttpPost post = new HttpPost(gatewayUrl + "/a2a/tasks?mode=async");
+ post.setHeader("Content-Type", "application/json");
+ post.setEntity(new StringEntity(body, StandardCharsets.UTF_8));
+
+ HttpResponse response = httpClient.execute(post);
+ int statusCode = response.getStatusLine().getStatusCode();
+ String respBody = response.getEntity() != null
+ ? EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8) : "";
+
+ if (statusCode >= 400) {
+ throw new RuntimeException("Task submission failed: " + statusCode + " " + respBody);
+ }
+
+ TaskResult result = objectMapper.readValue(respBody, TaskResult.class);
+ return result.getTaskId();
+ }
+
+ public TaskResult sendTaskSync(String targetAgent, String message, String parentTaskId) throws Exception {
+ Map<String, Object> bodyMap = new HashMap<>();
+ bodyMap.put("targetAgent", targetAgent);
+ bodyMap.put("message", message);
+ if (parentTaskId != null) {
+ bodyMap.put("parentTaskId", parentTaskId);
+ }
+ String body = objectMapper.writeValueAsString(bodyMap);
+
+ HttpPost post = new HttpPost(gatewayUrl + "/a2a/tasks?mode=sync");
+ post.setHeader("Content-Type", "application/json");
+ post.setEntity(new StringEntity(body, StandardCharsets.UTF_8));
+
+ HttpResponse response = httpClient.execute(post);
+ int statusCode = response.getStatusLine().getStatusCode();
+ String respBody = response.getEntity() != null
+ ? EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8) : "";
+
+ if (statusCode >= 400) {
+ throw new RuntimeException("Task submission failed: " + statusCode + " " + respBody);
+ }
+
+ return objectMapper.readValue(respBody, TaskResult.class);
+ }
+
+ public TaskResult getTaskStatus(String taskId) throws Exception {
+ HttpGet get = new HttpGet(gatewayUrl + "/a2a/tasks/" + taskId);
+ HttpResponse response = httpClient.execute(get);
+ String respBody = response.getEntity() != null
+ ? EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8) : "";
+ return objectMapper.readValue(respBody, TaskResult.class);
+ }
+
+ public boolean cancelTask(String taskId) throws Exception {
+ HttpDelete delete = new HttpDelete(gatewayUrl + "/a2a/tasks/" + taskId);
+ HttpResponse response = httpClient.execute(delete);
+ return response.getStatusLine().getStatusCode() == 200;
+ }
+
+ /**
+ * Subscribes to SSE status updates for a task via the /stream endpoint.
+ *
+ * <p>This method blocks until the stream ends (task reaches terminal state) or the
+ * consumer returns {@code false} to stop early.
+ *
+ * @param taskId the task ID to stream updates for
+ * @param consumer callback invoked for each SSE event; receives (taskId, state, data).
+ * Return {@code false} from the consumer to stop streaming.
+ * @throws Exception if the HTTP connection fails
+ */
+ public void streamTaskStatus(String taskId, SseEventConsumer consumer) throws Exception {
+ HttpGet get = new HttpGet(gatewayUrl + "/a2a/tasks/" + taskId + "/stream");
+ get.setHeader("Accept", "text/event-stream");
+
+ HttpResponse response = httpClient.execute(get);
+ int statusCode = response.getStatusLine().getStatusCode();
+ if (statusCode != 200) {
+ String body = response.getEntity() != null
+ ? EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8) : "";
+ throw new RuntimeException("SSE stream failed: " + statusCode + " " + body);
+ }
+
+ try (java.io.InputStream is = response.getEntity().getContent();
+ java.io.BufferedReader reader = new java.io.BufferedReader(
+ new java.io.InputStreamReader(is, StandardCharsets.UTF_8))) {
+ String line;
+ StringBuilder eventBuffer = new StringBuilder();
+ while ((line = reader.readLine()) != null) {
+ if (line.isEmpty()) {
+ // Empty line = event boundary
+ if (eventBuffer.length() > 0) {
+ String json = eventBuffer.toString();
+ if (json.startsWith("data: ")) {
+ json = json.substring(6);
+ }
+ try {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> event = objectMapper.readValue(json, Map.class);
+ String eventTaskId = event.get("taskId") != null ? event.get("taskId").toString() : taskId;
+ String state = event.get("state") != null ? event.get("state").toString() : "";
+ String data = event.get("data") != null ? event.get("data").toString() : null;
+ if (!consumer.onEvent(eventTaskId, state, data)) {
+ return; // Consumer requested stop
+ }
+ } catch (Exception e) {
+ log.warn("Failed to parse SSE event: {}", json, e);
+ }
+ eventBuffer.setLength(0);
+ }
+ } else if (line.startsWith(":")) {
+ // SSE comment / heartbeat line, ignore.
+ } else {
+ if (eventBuffer.length() > 0) {
+ eventBuffer.append("\n");
+ }
+ eventBuffer.append(line);
+ }
+ }
+ }
+ }
+
+ public List<String> listAgents() throws Exception {
+ HttpGet get = new HttpGet(gatewayUrl + "/a2a/agents");
+ HttpResponse response = httpClient.execute(get);
+ String respBody = response.getEntity() != null
+ ? EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8) : "[]";
+ List<Map<String, Object>> cards = objectMapper.readValue(respBody,
+ new com.fasterxml.jackson.core.type.TypeReference<List<Map<String, Object>>>() {});
+ List<String> names = new ArrayList<>();
+ for (Map<String, Object> card : cards) {
+ Object name = card.get("name");
+ if (name == null) {
+ name = card.get("id");
+ }
+ if (name != null) {
+ names.add(name.toString());
+ }
+ }
+ return names;
+ }
+
+ // =========================================================================
+ // Request Handler (for agent-to-agent via transport)
+ // =========================================================================
+
+ public void setRequestHandler(RequestHandler handler) {
+ this.requestHandler = handler;
+ }
+
+ public void setTransport(A2AMessageTransport transport) throws Exception {
+ this.transport = transport;
+ String requestTopic = A2ATopicFactory.agentRequestTopic(namespace, agentName);
+ requestSubscriptionId = transport.subscribe(requestTopic, this::handleRequestMessage);
+ log.info("Subscribed to agent request topic: {}", requestTopic);
Review Comment:
`setTransport()` subscribes using `agentName` directly as the topic segment. If `agentName` is the hierarchical form used elsewhere (e.g. `default/default/test-client-agent`), it contains `/` and will fail `A2ATopicFactory.validateSegment(...)`. Even without slashes, subscribing by full identity vs agentId is ambiguous.
##########
eventmesh-protocol-plugin/eventmesh-protocol-a2a/src/main/java/org/apache/eventmesh/protocol/a2a/A2AClient.java:
##########
@@ -0,0 +1,585 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.eventmesh.protocol.a2a;
+
+import org.apache.eventmesh.protocol.a2a.model.AgentCard;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.util.EntityUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * A2A Java SDK client for Agent developers.
+ *
+ * <p>Provides a simple API to:
+ * <ul>
+ * <li>Register and publish AgentCard</li>
+ * <li>Send A2A tasks to other agents via Gateway REST API</li>
+ * <li>Subscribe to and handle incoming A2A requests via transport</li>
+ * <li>Send heartbeat to keep agent card alive</li>
+ * </ul>
+ */
+public class A2AClient implements AutoCloseable {
+
+ private static final Logger log = LoggerFactory.getLogger(A2AClient.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ private final String gatewayUrl;
+ private final String namespace;
+ private final String agentName;
+ private final AgentCard agentCard;
+ private final long heartbeatIntervalMs;
+ private final CloseableHttpClient httpClient;
+
+ private volatile boolean started = false;
+ private ScheduledExecutorService heartbeatExecutor;
+ private RequestHandler requestHandler;
+
+ private A2AMessageTransport transport;
+ private String requestSubscriptionId;
+
+ A2AClient(Builder builder) {
+ this.gatewayUrl = builder.gatewayUrl;
+ this.namespace = builder.namespace;
+ this.agentName = builder.agentName;
+ this.agentCard = builder.agentCard;
+ this.heartbeatIntervalMs = builder.heartbeatIntervalMs;
+ RequestConfig config = RequestConfig.custom()
+ .setConnectTimeout(10_000)
+ .setSocketTimeout(builder.socketTimeoutMs)
+ .build();
+ this.httpClient = HttpClients.custom().setDefaultRequestConfig(config).build();
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ // =========================================================================
+ // Lifecycle
+ // =========================================================================
+
+ public void start() throws Exception {
+ if (started) {
+ return;
+ }
+ registerAgentCard();
+ startHeartbeat();
+ started = true;
+ log.info("A2AClient started: agent={}, namespace={}", agentName, namespace);
+ }
+
+ public void shutdown() throws Exception {
+ if (heartbeatExecutor != null) {
+ heartbeatExecutor.shutdownNow();
+ }
+ if (transport != null && requestSubscriptionId != null) {
+ transport.unsubscribe(requestSubscriptionId);
+ requestSubscriptionId = null;
+ }
+ httpClient.close();
+ started = false;
+ log.info("A2AClient shutdown: agent={}", agentName);
+ }
+
+ @Override
+ public void close() throws Exception {
+ shutdown();
+ }
+
+ // =========================================================================
+ // AgentCard Registration
+ // =========================================================================
+
+ public void registerAgentCard() throws Exception {
+ String[] parts = agentName.split("/");
+ String orgId = parts.length >= 1 ? parts[0] : "default";
+ String unitId = parts.length >= 2 ? parts[1] : "default";
+ String agentId = parts.length >= 3 ? parts[2] : agentName;
+
+ String url = String.format("%s/a2a/cards/card/%s/%s/%s", gatewayUrl, orgId, unitId, agentId);
+ String body = objectMapper.writeValueAsString(agentCard);
+
+ HttpPost post = new HttpPost(url);
+ post.setHeader("Content-Type", "application/json");
+ post.setEntity(new StringEntity(body, StandardCharsets.UTF_8));
+
+ HttpResponse response = httpClient.execute(post);
+ int statusCode = response.getStatusLine().getStatusCode();
+ if (statusCode >= 400) {
+ String respBody = response.getEntity() != null
+ ? EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8) : "";
+ throw new RuntimeException("Failed to register agent card: " + statusCode + " " + respBody);
+ }
+ log.info("AgentCard registered: {} -> {}", agentName, statusCode);
+ }
+
+ // =========================================================================
+ // Heartbeat
+ // =========================================================================
+
+ private void startHeartbeat() {
+ heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "a2a-client-heartbeat");
+ t.setDaemon(true);
+ return t;
+ });
+ heartbeatExecutor.scheduleAtFixedRate(this::sendHeartbeat,
+ heartbeatIntervalMs, heartbeatIntervalMs, TimeUnit.MILLISECONDS);
+ }
+
+ private void sendHeartbeat() {
+ try {
+ String[] parts = agentName.split("/");
+ String orgId = parts.length >= 1 ? parts[0] : "default";
+ String unitId = parts.length >= 2 ? parts[1] : "default";
+ String agentId = parts.length >= 3 ? parts[2] : agentName;
Review Comment:
Heartbeat uses the same `orgId/unitId/agentId` parsing as registration, so the same issue applies: `agentName` without slashes becomes `orgId=<agentName>`. This also duplicates parsing logic, making it easy for registration/heartbeat to drift.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayHttpHandler.java:
##########
@@ -0,0 +1,648 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a;
+
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpContent;
+import io.netty.handler.codec.http.DefaultHttpResponse;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.handler.codec.http.QueryStringDecoder;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * HTTP handler for A2A Gateway API.
+ *
+ * <p>Endpoints:
+ * <ul>
+ * <li>POST /a2a/tasks - submit a new task (sync or async)</li>
+ * <li>GET /a2a/tasks - list tasks (optional ?state=COMPLETED&limit=100&offset=0)</li>
+ * <li>GET /a2a/tasks/{taskId} - get task status</li>
+ * <li>DELETE /a2a/tasks/{taskId} - cancel a task</li>
+ * <li>GET /a2a/tasks/{taskId}/wait - wait for task result (long-poll)</li>
+ * <li>GET /a2a/tasks/{taskId}/stream - SSE stream of task status updates</li>
+ * <li>GET /a2a/agents - list registered agents</li>
+ * <li>POST /a2a/heartbeat - agent heartbeat</li>
+ * <li>GET /a2a/health - health check (for liveness/readiness probes)</li>
+ * <li>OPTIONS * - CORS preflight</li>
+ * </ul>
+ */
+@Slf4j
+public class A2AGatewayHttpHandler {
+
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ private static final String PATH_TASKS = "/a2a/tasks";
+ private static final String PATH_AGENTS = "/a2a/agents";
+ private static final String PATH_HEARTBEAT = "/a2a/heartbeat";
+ private static final String PATH_HEALTH = "/a2a/health";
+
+ private static final long DEFAULT_WAIT_TIMEOUT_MS = 30_000L;
+
+ private final A2AGatewayService gatewayService;
+ private final A2APublishSubscribeService a2aService;
+
+ public A2AGatewayHttpHandler(A2AGatewayService gatewayService,
+ A2APublishSubscribeService a2aService) {
+ this.gatewayService = gatewayService;
+ this.a2aService = a2aService;
+ }
+
+ /**
+ * Handles an HTTP request. Returns a FullHttpResponse, or null if the response was written
+ * directly to the channel (e.g. SSE streaming).
+ */
+ public FullHttpResponse handle(HttpRequest httpRequest, ChannelHandlerContext ctx) throws Exception {
+ String uri = httpRequest.uri();
+ QueryStringDecoder decoder = new QueryStringDecoder(uri);
+ String path = decoder.path();
+ HttpMethod method = httpRequest.method();
+
+ try {
+ // Handle CORS preflight
+ if (method == HttpMethod.OPTIONS) {
+ return corsResponse(HttpResponseStatus.OK);
+ }
+
+ // POST /a2a/tasks
+ if (PATH_TASKS.equals(path) && method == HttpMethod.POST) {
+ return handleSubmitTask(httpRequest, decoder);
+ }
+
+ // GET /a2a/tasks (list tasks with optional pagination)
+ if (PATH_TASKS.equals(path) && method == HttpMethod.GET) {
+ return handleListTasks(decoder);
+ }
+
+ // /a2a/tasks/{taskId} or /a2a/tasks/{taskId}/wait
+ if (path.startsWith(PATH_TASKS + "/")) {
+ String suffix = path.substring(PATH_TASKS.length() + 1);
+ String[] parts = suffix.split("/", 2);
+ String taskId = parts[0];
+
+ if (taskId.isEmpty()) {
+ return jsonResponse(HttpResponseStatus.BAD_REQUEST, errorBody("Missing taskId"));
+ }
+
+ // GET /a2a/tasks/{taskId}
+ if (method == HttpMethod.GET && parts.length == 1) {
+ return handleGetTask(taskId);
+ }
+
+ // DELETE /a2a/tasks/{taskId}
+ if (method == HttpMethod.DELETE && parts.length == 1) {
+ return handleCancelTask(taskId);
+ }
+
+ // GET /a2a/tasks/{taskId}/wait
+ if (method == HttpMethod.GET && parts.length == 2 && "wait".equals(parts[1])) {
+ return handleWaitTask(taskId, decoder);
+ }
+
+ // GET /a2a/tasks/{taskId}/stream (SSE)
+ if (method == HttpMethod.GET && parts.length == 2 && "stream".equals(parts[1])) {
+ return handleStreamTask(taskId, ctx);
+ }
+ }
+
+ // GET /a2a/health
+ if (PATH_HEALTH.equals(path) && method == HttpMethod.GET) {
+ return handleHealth();
+ }
+
+ // GET /a2a/agents
+ if (PATH_AGENTS.equals(path) && method == HttpMethod.GET) {
+ return handleListAgents();
+ }
+
+ // POST /a2a/heartbeat
+ if (PATH_HEARTBEAT.equals(path) && method == HttpMethod.POST) {
+ return handleHeartbeat(httpRequest);
+ }
+
+ return jsonResponse(HttpResponseStatus.NOT_FOUND, errorBody("Not found: " + path));
+ } catch (Exception e) {
+ log.error("A2A gateway handler error: {}", e.getMessage(), e);
+ return jsonResponse(HttpResponseStatus.INTERNAL_SERVER_ERROR, errorBody(e.getMessage()));
+ }
+ }
+
+ // =========================================================================
+ // POST /a2a/tasks
+ // =========================================================================
+
+ private FullHttpResponse handleSubmitTask(HttpRequest request, QueryStringDecoder decoder) throws Exception {
+ String body = readBody(request);
+ TaskRequest taskRequest;
+ try {
+ taskRequest = objectMapper.readValue(body, TaskRequest.class);
+ } catch (Exception e) {
+ return jsonResponse(HttpResponseStatus.BAD_REQUEST, errorBody("Invalid JSON: " + e.getMessage()));
+ }
+
+ if (taskRequest.getTargetAgent() == null || taskRequest.getTargetAgent().isEmpty()) {
+ return jsonResponse(HttpResponseStatus.BAD_REQUEST, errorBody("targetAgent is required"));
+ }
+ if (taskRequest.getMessage() == null) {
+ taskRequest.setMessage("");
+ }
+
+ String mode = getQueryParam(decoder, "mode");
+ String parentTaskId = taskRequest.getParentTaskId();
+
+ // Generate taskId first so we can include it in the response
+ String taskId = "task-" + java.util.UUID.randomUUID().toString().substring(0, 8);
+
+ // Submit the task with the explicit taskId
+ java.util.concurrent.CompletableFuture<A2AGatewayService.TaskResult> future =
+ gatewayService.submitTask(taskId, taskRequest.getTargetAgent(), taskRequest.getMessage(), parentTaskId);
+
+ // For sync mode, wait for result
+ if ("sync".equalsIgnoreCase(mode) || mode == null) {
+ try {
+ A2AGatewayService.TaskResult result = future.get(DEFAULT_WAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ Map<String, Object> response = new LinkedHashMap<>();
+ response.put("taskId", taskId);
+ response.put("state", result.getState().name());
+ response.put("data", result.getData());
+ if (result.getErrorMessage() != null) {
+ response.put("error", result.getErrorMessage());
+ }
+ return jsonResponse(HttpResponseStatus.OK, objectMapper.writeValueAsString(response));
+ } catch (java.util.concurrent.TimeoutException e) {
+ Map<String, Object> response = new LinkedHashMap<>();
+ response.put("taskId", taskId);
+ response.put("state", "TIMEOUT");
+ response.put("error", "Task accepted but timed out waiting for response");
+ return jsonResponse(HttpResponseStatus.ACCEPTED, objectMapper.writeValueAsString(response));
+ } catch (java.util.concurrent.ExecutionException e) {
+ // Unwrap cause to determine correct status code
+ Throwable cause = e.getCause() != null ? e.getCause() : e;
+ if (cause instanceof IllegalArgumentException || cause instanceof IllegalStateException) {
+ return jsonResponse(HttpResponseStatus.BAD_REQUEST,
+ errorBody(cause.getMessage()));
+ }
+ return jsonResponse(HttpResponseStatus.INTERNAL_SERVER_ERROR,
+ errorBody("Task failed: " + cause.getMessage()));
+ } catch (Exception e) {
+ return jsonResponse(HttpResponseStatus.INTERNAL_SERVER_ERROR,
+ errorBody("Task failed: " + e.getMessage()));
+ }
+ } else {
+ // Async mode: return task ID immediately
+ Map<String, Object> response = new LinkedHashMap<>();
+ response.put("taskId", taskId);
+ response.put("status", "accepted");
+ response.put("message", "Task submitted. Use GET /a2a/tasks/" + taskId + " to check status.");
+ return jsonResponse(HttpResponseStatus.ACCEPTED, objectMapper.writeValueAsString(response));
+ }
+ }
+
+ // =========================================================================
+ // GET /a2a/tasks (list with optional pagination)
+ // =========================================================================
+
+ private FullHttpResponse handleListTasks(QueryStringDecoder decoder) throws Exception {
+ String stateFilter = getQueryParam(decoder, "state");
+ int limit = parseIntQueryParam(decoder, "limit", 100, 1, 1_000);
+ int offset = parseIntQueryParam(decoder, "offset", 0, 0, Integer.MAX_VALUE);
+
+ List<TaskRegistry.TaskEntry> allTasks = gatewayService.getTaskRegistry().listTasks();
+ List<Map<String, Object>> filteredTasks = new java.util.ArrayList<>();
+ for (TaskRegistry.TaskEntry entry : allTasks) {
+ if (stateFilter != null && !stateFilter.equalsIgnoreCase(entry.getState().name())) {
+ continue;
+ }
+ Map<String, Object> task = new LinkedHashMap<>();
+ task.put("taskId", entry.getTaskId());
+ task.put("state", entry.getState().name());
+ task.put("targetAgent", entry.getTargetAgent());
+ task.put("parentTaskId", entry.getParentTaskId());
+ task.put("createdAt", entry.getCreatedAt());
+ task.put("updatedAt", entry.getUpdatedAt());
+ filteredTasks.add(task);
+ }
+
+ int total = filteredTasks.size();
Review Comment:
Task listing is built from `ConcurrentHashMap.values()` (via `TaskRegistry.listTasks()`), which has nondeterministic ordering. Using `limit/offset` over an unstable order makes pagination unreliable (items can jump between pages across calls).
##########
eventmesh-protocol-plugin/eventmesh-protocol-a2a/src/main/java/org/apache/eventmesh/protocol/a2a/A2AClient.java:
##########
@@ -0,0 +1,585 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.eventmesh.protocol.a2a;
+
+import org.apache.eventmesh.protocol.a2a.model.AgentCard;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.util.EntityUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * A2A Java SDK client for Agent developers.
+ *
+ * <p>Provides a simple API to:
+ * <ul>
+ * <li>Register and publish AgentCard</li>
+ * <li>Send A2A tasks to other agents via Gateway REST API</li>
+ * <li>Subscribe to and handle incoming A2A requests via transport</li>
+ * <li>Send heartbeat to keep agent card alive</li>
+ * </ul>
+ */
+public class A2AClient implements AutoCloseable {
+
+ private static final Logger log = LoggerFactory.getLogger(A2AClient.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ private final String gatewayUrl;
+ private final String namespace;
+ private final String agentName;
+ private final AgentCard agentCard;
+ private final long heartbeatIntervalMs;
+ private final CloseableHttpClient httpClient;
+
+ private volatile boolean started = false;
+ private ScheduledExecutorService heartbeatExecutor;
+ private RequestHandler requestHandler;
+
+ private A2AMessageTransport transport;
+ private String requestSubscriptionId;
+
+ A2AClient(Builder builder) {
+ this.gatewayUrl = builder.gatewayUrl;
+ this.namespace = builder.namespace;
+ this.agentName = builder.agentName;
+ this.agentCard = builder.agentCard;
+ this.heartbeatIntervalMs = builder.heartbeatIntervalMs;
+ RequestConfig config = RequestConfig.custom()
+ .setConnectTimeout(10_000)
+ .setSocketTimeout(builder.socketTimeoutMs)
+ .build();
+ this.httpClient = HttpClients.custom().setDefaultRequestConfig(config).build();
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ // =========================================================================
+ // Lifecycle
+ // =========================================================================
+
+ public void start() throws Exception {
+ if (started) {
+ return;
+ }
+ registerAgentCard();
+ startHeartbeat();
+ started = true;
+ log.info("A2AClient started: agent={}, namespace={}", agentName, namespace);
+ }
+
+ public void shutdown() throws Exception {
+ if (heartbeatExecutor != null) {
+ heartbeatExecutor.shutdownNow();
+ }
+ if (transport != null && requestSubscriptionId != null) {
+ transport.unsubscribe(requestSubscriptionId);
+ requestSubscriptionId = null;
+ }
+ httpClient.close();
+ started = false;
+ log.info("A2AClient shutdown: agent={}", agentName);
+ }
+
+ @Override
+ public void close() throws Exception {
+ shutdown();
+ }
+
+ // =========================================================================
+ // AgentCard Registration
+ // =========================================================================
+
+ public void registerAgentCard() throws Exception {
+ String[] parts = agentName.split("/");
+ String orgId = parts.length >= 1 ? parts[0] : "default";
+ String unitId = parts.length >= 2 ? parts[1] : "default";
+ String agentId = parts.length >= 3 ? parts[2] : agentName;
Review Comment:
For a simple agent name like `my-agent` (as shown in docs/examples), this logic sets `orgId` to `my-agent` instead of defaulting to `default/default`. That makes the identity semantics surprising and inconsistent with the URL template `/a2a/cards/card/{org}/{unit}/{agent}`.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayHttpHandler.java:
##########
@@ -0,0 +1,648 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a;
+
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpContent;
+import io.netty.handler.codec.http.DefaultHttpResponse;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.handler.codec.http.QueryStringDecoder;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * HTTP handler for A2A Gateway API.
+ *
+ * <p>Endpoints:
+ * <ul>
+ * <li>POST /a2a/tasks - submit a new task (sync or async)</li>
+ * <li>GET /a2a/tasks - list tasks (optional ?state=COMPLETED&limit=100&offset=0)</li>
+ * <li>GET /a2a/tasks/{taskId} - get task status</li>
+ * <li>DELETE /a2a/tasks/{taskId} - cancel a task</li>
+ * <li>GET /a2a/tasks/{taskId}/wait - wait for task result (long-poll)</li>
+ * <li>GET /a2a/tasks/{taskId}/stream - SSE stream of task status updates</li>
+ * <li>GET /a2a/agents - list registered agents</li>
+ * <li>POST /a2a/heartbeat - agent heartbeat</li>
+ * <li>GET /a2a/health - health check (for liveness/readiness probes)</li>
+ * <li>OPTIONS * - CORS preflight</li>
+ * </ul>
+ */
+@Slf4j
+public class A2AGatewayHttpHandler {
+
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ private static final String PATH_TASKS = "/a2a/tasks";
+ private static final String PATH_AGENTS = "/a2a/agents";
+ private static final String PATH_HEARTBEAT = "/a2a/heartbeat";
+ private static final String PATH_HEALTH = "/a2a/health";
+
+ private static final long DEFAULT_WAIT_TIMEOUT_MS = 30_000L;
+
+ private final A2AGatewayService gatewayService;
+ private final A2APublishSubscribeService a2aService;
+
+ public A2AGatewayHttpHandler(A2AGatewayService gatewayService,
+ A2APublishSubscribeService a2aService) {
+ this.gatewayService = gatewayService;
+ this.a2aService = a2aService;
+ }
+
+ /**
+ * Handles an HTTP request. Returns a FullHttpResponse, or null if the response was written
+ * directly to the channel (e.g. SSE streaming).
+ */
+ public FullHttpResponse handle(HttpRequest httpRequest, ChannelHandlerContext ctx) throws Exception {
+ String uri = httpRequest.uri();
+ QueryStringDecoder decoder = new QueryStringDecoder(uri);
+ String path = decoder.path();
+ HttpMethod method = httpRequest.method();
+
+ try {
+ // Handle CORS preflight
+ if (method == HttpMethod.OPTIONS) {
+ return corsResponse(HttpResponseStatus.OK);
+ }
+
+ // POST /a2a/tasks
+ if (PATH_TASKS.equals(path) && method == HttpMethod.POST) {
+ return handleSubmitTask(httpRequest, decoder);
+ }
+
+ // GET /a2a/tasks (list tasks with optional pagination)
+ if (PATH_TASKS.equals(path) && method == HttpMethod.GET) {
+ return handleListTasks(decoder);
+ }
+
+ // /a2a/tasks/{taskId} or /a2a/tasks/{taskId}/wait
+ if (path.startsWith(PATH_TASKS + "/")) {
+ String suffix = path.substring(PATH_TASKS.length() + 1);
+ String[] parts = suffix.split("/", 2);
+ String taskId = parts[0];
+
+ if (taskId.isEmpty()) {
+ return jsonResponse(HttpResponseStatus.BAD_REQUEST, errorBody("Missing taskId"));
+ }
+
+ // GET /a2a/tasks/{taskId}
+ if (method == HttpMethod.GET && parts.length == 1) {
+ return handleGetTask(taskId);
+ }
+
+ // DELETE /a2a/tasks/{taskId}
+ if (method == HttpMethod.DELETE && parts.length == 1) {
+ return handleCancelTask(taskId);
+ }
+
+ // GET /a2a/tasks/{taskId}/wait
+ if (method == HttpMethod.GET && parts.length == 2 && "wait".equals(parts[1])) {
+ return handleWaitTask(taskId, decoder);
+ }
+
+ // GET /a2a/tasks/{taskId}/stream (SSE)
+ if (method == HttpMethod.GET && parts.length == 2 && "stream".equals(parts[1])) {
+ return handleStreamTask(taskId, ctx);
+ }
+ }
+
+ // GET /a2a/health
+ if (PATH_HEALTH.equals(path) && method == HttpMethod.GET) {
+ return handleHealth();
+ }
+
+ // GET /a2a/agents
+ if (PATH_AGENTS.equals(path) && method == HttpMethod.GET) {
+ return handleListAgents();
+ }
+
+ // POST /a2a/heartbeat
+ if (PATH_HEARTBEAT.equals(path) && method == HttpMethod.POST) {
+ return handleHeartbeat(httpRequest);
+ }
+
+ return jsonResponse(HttpResponseStatus.NOT_FOUND, errorBody("Not found: " + path));
+ } catch (Exception e) {
+ log.error("A2A gateway handler error: {}", e.getMessage(), e);
+ return jsonResponse(HttpResponseStatus.INTERNAL_SERVER_ERROR, errorBody(e.getMessage()));
+ }
+ }
+
+ // =========================================================================
+ // POST /a2a/tasks
+ // =========================================================================
+
+ private FullHttpResponse handleSubmitTask(HttpRequest request, QueryStringDecoder decoder) throws Exception {
+ String body = readBody(request);
+ TaskRequest taskRequest;
+ try {
+ taskRequest = objectMapper.readValue(body, TaskRequest.class);
+ } catch (Exception e) {
+ return jsonResponse(HttpResponseStatus.BAD_REQUEST, errorBody("Invalid JSON: " + e.getMessage()));
+ }
+
+ if (taskRequest.getTargetAgent() == null || taskRequest.getTargetAgent().isEmpty()) {
+ return jsonResponse(HttpResponseStatus.BAD_REQUEST, errorBody("targetAgent is required"));
+ }
+ if (taskRequest.getMessage() == null) {
+ taskRequest.setMessage("");
+ }
+
+ String mode = getQueryParam(decoder, "mode");
+ String parentTaskId = taskRequest.getParentTaskId();
+
+ // Generate taskId first so we can include it in the response
+ String taskId = "task-" + java.util.UUID.randomUUID().toString().substring(0, 8);
+
+ // Submit the task with the explicit taskId
+ java.util.concurrent.CompletableFuture<A2AGatewayService.TaskResult> future =
+ gatewayService.submitTask(taskId, taskRequest.getTargetAgent(), taskRequest.getMessage(), parentTaskId);
+
+ // For sync mode, wait for result
+ if ("sync".equalsIgnoreCase(mode) || mode == null) {
+ try {
+ A2AGatewayService.TaskResult result = future.get(DEFAULT_WAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ Map<String, Object> response = new LinkedHashMap<>();
+ response.put("taskId", taskId);
+ response.put("state", result.getState().name());
+ response.put("data", result.getData());
+ if (result.getErrorMessage() != null) {
+ response.put("error", result.getErrorMessage());
+ }
+ return jsonResponse(HttpResponseStatus.OK, objectMapper.writeValueAsString(response));
+ } catch (java.util.concurrent.TimeoutException e) {
+ Map<String, Object> response = new LinkedHashMap<>();
+ response.put("taskId", taskId);
+ response.put("state", "TIMEOUT");
+ response.put("error", "Task accepted but timed out waiting for response");
+ return jsonResponse(HttpResponseStatus.ACCEPTED, objectMapper.writeValueAsString(response));
+ } catch (java.util.concurrent.ExecutionException e) {
+ // Unwrap cause to determine correct status code
+ Throwable cause = e.getCause() != null ? e.getCause() : e;
+ if (cause instanceof IllegalArgumentException || cause instanceof IllegalStateException) {
+ return jsonResponse(HttpResponseStatus.BAD_REQUEST,
+ errorBody(cause.getMessage()));
+ }
+ return jsonResponse(HttpResponseStatus.INTERNAL_SERVER_ERROR,
+ errorBody("Task failed: " + cause.getMessage()));
+ } catch (Exception e) {
+ return jsonResponse(HttpResponseStatus.INTERNAL_SERVER_ERROR,
+ errorBody("Task failed: " + e.getMessage()));
+ }
+ } else {
+ // Async mode: return task ID immediately
+ Map<String, Object> response = new LinkedHashMap<>();
+ response.put("taskId", taskId);
+ response.put("status", "accepted");
+ response.put("message", "Task submitted. Use GET /a2a/tasks/" + taskId + " to check status.");
+ return jsonResponse(HttpResponseStatus.ACCEPTED, objectMapper.writeValueAsString(response));
+ }
+ }
+
+ // =========================================================================
+ // GET /a2a/tasks (list with optional pagination)
+ // =========================================================================
+
+ private FullHttpResponse handleListTasks(QueryStringDecoder decoder) throws Exception {
+ String stateFilter = getQueryParam(decoder, "state");
+ int limit = parseIntQueryParam(decoder, "limit", 100, 1, 1_000);
+ int offset = parseIntQueryParam(decoder, "offset", 0, 0, Integer.MAX_VALUE);
+
+ List<TaskRegistry.TaskEntry> allTasks = gatewayService.getTaskRegistry().listTasks();
+ List<Map<String, Object>> filteredTasks = new java.util.ArrayList<>();
+ for (TaskRegistry.TaskEntry entry : allTasks) {
+ if (stateFilter != null && !stateFilter.equalsIgnoreCase(entry.getState().name())) {
+ continue;
+ }
+ Map<String, Object> task = new LinkedHashMap<>();
+ task.put("taskId", entry.getTaskId());
+ task.put("state", entry.getState().name());
+ task.put("targetAgent", entry.getTargetAgent());
+ task.put("parentTaskId", entry.getParentTaskId());
+ task.put("createdAt", entry.getCreatedAt());
+ task.put("updatedAt", entry.getUpdatedAt());
+ filteredTasks.add(task);
+ }
+
+ int total = filteredTasks.size();
+ int fromIndex = Math.min(offset, total);
+ int toIndex = Math.min(fromIndex + limit, total);
+ List<Map<String, Object>> page = filteredTasks.subList(fromIndex, toIndex);
+
+ Map<String, Object> response = new LinkedHashMap<>();
+ response.put("total", total);
+ response.put("count", page.size());
+ response.put("limit", limit);
+ response.put("offset", offset);
+ response.put("hasMore", toIndex < total);
+ response.put("tasks", page);
+ return jsonResponse(HttpResponseStatus.OK, objectMapper.writeValueAsString(response));
+ }
+
+ // =========================================================================
+ // GET /a2a/tasks/{taskId}
+ // =========================================================================
+
+ private FullHttpResponse handleGetTask(String taskId) throws Exception {
+ TaskRegistry.TaskEntry entry = gatewayService.getTaskStatus(taskId);
+ if (entry == null) {
+ return jsonResponse(HttpResponseStatus.NOT_FOUND, errorBody("Task not found: " + taskId));
+ }
+ Map<String, Object> response = new LinkedHashMap<>();
+ response.put("taskId", entry.getTaskId());
+ response.put("state", entry.getState().name());
+ response.put("targetAgent", entry.getTargetAgent());
+ response.put("gatewayId", entry.getGatewayId());
+ response.put("parentTaskId", entry.getParentTaskId());
+ response.put("createdAt", entry.getCreatedAt());
+ response.put("updatedAt", entry.getUpdatedAt());
+ if (entry.getResult() != null) {
+ response.put("result", entry.getResult());
+ }
+ if (entry.getErrorMessage() != null) {
+ response.put("error", entry.getErrorMessage());
+ }
+ return jsonResponse(HttpResponseStatus.OK, objectMapper.writeValueAsString(response));
+ }
+
+ // =========================================================================
+ // DELETE /a2a/tasks/{taskId}
+ // =========================================================================
+
+ private FullHttpResponse handleCancelTask(String taskId) throws Exception {
+ boolean cancelled = gatewayService.cancelTask(taskId);
+ if (cancelled) {
+ Map<String, Object> response = new LinkedHashMap<>();
+ response.put("taskId", taskId);
+ response.put("state", "CANCELLED");
+ return jsonResponse(HttpResponseStatus.OK, objectMapper.writeValueAsString(response));
+ }
+ return jsonResponse(HttpResponseStatus.NOT_FOUND, errorBody("Task not found or cannot be cancelled: " + taskId));
+ }
+
+ // =========================================================================
+ // GET /a2a/tasks/{taskId}/wait
+ // =========================================================================
+
+ private FullHttpResponse handleWaitTask(String taskId, QueryStringDecoder decoder) throws Exception {
+ String timeoutStr = getQueryParam(decoder, "timeout");
+ long timeout = timeoutStr != null ? Long.parseLong(timeoutStr) : DEFAULT_WAIT_TIMEOUT_MS;
+
Review Comment:
`timeout` query param is parsed with `Long.parseLong` without validation. A non-numeric value (or negative) will bubble up as an exception and get converted into a 500, even though this is a client error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE #5259] Add A2A Gateway: REST API, SSE streaming, Task lifecycle, Java SDK and tests (eventmesh)
Posted by "xwm1992 (via GitHub)" <gi...@apache.org>.
xwm1992 merged PR #5260:
URL: https://github.com/apache/eventmesh/pull/5260
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org