diff --git a/common/src/main/java/org/a2aproject/sdk/common/A2AHeaders.java b/common/src/main/java/org/a2aproject/sdk/common/A2AHeaders.java index 5d1c05e36..3f7e6d2ed 100644 --- a/common/src/main/java/org/a2aproject/sdk/common/A2AHeaders.java +++ b/common/src/main/java/org/a2aproject/sdk/common/A2AHeaders.java @@ -21,7 +21,13 @@ public final class A2AHeaders { * HTTP header name for a push notification token. */ public static final String X_A2A_NOTIFICATION_TOKEN = "X-A2A-Notification-Token"; - + + /** + * Metadata header identifying the target agent ID for transports without per-path + * routing (e.g. gRPC). + */ + public static final String X_A2A_AGENT_ID = "X-A2A-Agent-Id"; + private A2AHeaders() { // Utility class } diff --git a/reference/grpc/pom.xml b/reference/grpc/pom.xml index 26f21ce74..3298e6160 100644 --- a/reference/grpc/pom.xml +++ b/reference/grpc/pom.xml @@ -94,6 +94,16 @@ rest-assured test + + org.mockito + mockito-core + test + + + org.mockito + mockito-junit-jupiter + test + io.quarkus diff --git a/reference/grpc/src/main/java/org/a2aproject/sdk/server/grpc/quarkus/QuarkusGrpcHandler.java b/reference/grpc/src/main/java/org/a2aproject/sdk/server/grpc/quarkus/QuarkusGrpcHandler.java index 8df6e0699..42b46d6d9 100644 --- a/reference/grpc/src/main/java/org/a2aproject/sdk/server/grpc/quarkus/QuarkusGrpcHandler.java +++ b/reference/grpc/src/main/java/org/a2aproject/sdk/server/grpc/quarkus/QuarkusGrpcHandler.java @@ -1,17 +1,26 @@ package org.a2aproject.sdk.server.grpc.quarkus; +import static java.util.Locale.ROOT; + import java.util.concurrent.Executor; import jakarta.enterprise.inject.Instance; import jakarta.inject.Inject; +import org.a2aproject.sdk.common.A2AHeaders; import org.a2aproject.sdk.server.ExtendedAgentCard; import org.a2aproject.sdk.server.PublicAgentCard; +import org.a2aproject.sdk.server.grpc.quarkus.registry.GrpcAgent; +import org.a2aproject.sdk.server.grpc.quarkus.registry.MultiAgentRegistry; import org.a2aproject.sdk.server.requesthandlers.RequestHandler; import org.a2aproject.sdk.server.util.async.Internal; import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.InvalidRequestError; +import org.a2aproject.sdk.transport.grpc.context.GrpcContextKeys; import org.a2aproject.sdk.transport.grpc.handler.CallContextFactory; import org.a2aproject.sdk.transport.grpc.handler.GrpcHandler; +import io.grpc.Context; +import io.grpc.Metadata; import io.quarkus.grpc.GrpcService; import io.quarkus.grpc.RegisterInterceptor; import io.quarkus.security.Authenticated; @@ -74,10 +83,14 @@ @Blocking public class QuarkusGrpcHandler extends GrpcHandler { - private final AgentCard agentCard; - private final AgentCard extendedAgentCard; - private final RequestHandler requestHandler; + private static final Metadata.Key AGENT_ID_KEY = + Metadata.Key.of(A2AHeaders.X_A2A_AGENT_ID.toLowerCase(ROOT), Metadata.ASCII_STRING_MARSHALLER); + + private final Instance agentCardInstance; + private final Instance extendedAgentCardInstance; + private final Instance requestHandlerInstance; private final Instance callContextFactoryInstance; + private final Instance multiAgentRegistryInstance; private final Executor executor; /** @@ -99,42 +112,70 @@ public class QuarkusGrpcHandler extends GrpcHandler { *
  • {@code callContextFactoryInstance} - Custom context factory (can be unsatisfied)
    • * * - * @param agentCard the public agent card (qualified with {@code @PublicAgentCard}) + * @param agentCard the public agent card instance (qualified with {@code @PublicAgentCard}); may be + * unresolvable when every agent is served through a {@link MultiAgentRegistry} * @param extendedAgentCard the extended agent card instance (qualified with {@code @ExtendedAgentCard}) - * @param requestHandler the request handler for protocol operations + * @param requestHandler the request handler instance for protocol operations; may be unresolvable + * when every agent is served through a {@link MultiAgentRegistry} * @param callContextFactoryInstance the call context factory instance (optional) + * @param multiAgentRegistryInstance the multi-agent registry instance (optional) * @param executor the executor for async operations (qualified with {@code @Internal}) */ @Inject - public QuarkusGrpcHandler(@PublicAgentCard AgentCard agentCard, + public QuarkusGrpcHandler(@PublicAgentCard Instance agentCard, @ExtendedAgentCard Instance extendedAgentCard, - RequestHandler requestHandler, + Instance requestHandler, Instance callContextFactoryInstance, + Instance multiAgentRegistryInstance, @Internal Executor executor) { - this.agentCard = agentCard; - if (extendedAgentCard != null && extendedAgentCard.isResolvable()) { - this.extendedAgentCard = extendedAgentCard.get(); - } else { - this.extendedAgentCard = null; - } - this.requestHandler = requestHandler; + this.agentCardInstance = agentCard; + this.extendedAgentCardInstance = extendedAgentCard; + this.requestHandlerInstance = requestHandler; this.callContextFactoryInstance = callContextFactoryInstance; + this.multiAgentRegistryInstance = multiAgentRegistryInstance; this.executor = executor; } @Override protected RequestHandler getRequestHandler() { - return requestHandler; + return resolveAgent().requestHandler(); } @Override protected AgentCard getAgentCard() { - return agentCard; + return resolveAgent().agentCard(); } @Override protected AgentCard getExtendedAgentCard() { - return extendedAgentCard; + return resolveAgent().extendedAgentCard(); + } + + /** + * Resolves the agent for the current call: by {@code X-A2A-Agent-Id} header via + * {@link MultiAgentRegistry} if present, else the default single-agent beans. + * + * @throws InvalidRequestError if no agent could be resolved for this call + */ + private GrpcAgent resolveAgent() { + if (multiAgentRegistryInstance.isResolvable()) { + String agentId = currentAgentId(); + GrpcAgent agent = agentId != null ? multiAgentRegistryInstance.get().getAgents().get(agentId) : null; + if (agent != null) { + return agent; + } + } + if (agentCardInstance.isResolvable() && requestHandlerInstance.isResolvable()) { + AgentCard extendedAgentCard = extendedAgentCardInstance.isResolvable() ? extendedAgentCardInstance.get() : null; + return new GrpcAgent(agentCardInstance.get(), extendedAgentCard, requestHandlerInstance.get()); + } + throw new InvalidRequestError("No agent configured for this request"); + } + + /** @return the {@code X-A2A-Agent-Id} header from the current call's metadata, or null */ + private @Nullable String currentAgentId() { + Metadata metadata = GrpcContextKeys.METADATA_KEY.get(Context.current()); + return metadata != null ? metadata.get(AGENT_ID_KEY) : null; } @Override diff --git a/reference/grpc/src/main/java/org/a2aproject/sdk/server/grpc/quarkus/registry/GrpcAgent.java b/reference/grpc/src/main/java/org/a2aproject/sdk/server/grpc/quarkus/registry/GrpcAgent.java new file mode 100644 index 000000000..d218ec6d5 --- /dev/null +++ b/reference/grpc/src/main/java/org/a2aproject/sdk/server/grpc/quarkus/registry/GrpcAgent.java @@ -0,0 +1,17 @@ +package org.a2aproject.sdk.server.grpc.quarkus.registry; + +import org.a2aproject.sdk.server.requesthandlers.RequestHandler; +import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.util.Assert; +import org.jspecify.annotations.Nullable; + +/** + * Bundles the pieces needed to serve a single agent over gRPC: its request handler and + * agent card(s). Used by {@link MultiAgentRegistry} to describe each registered agent. + */ +public record GrpcAgent(AgentCard agentCard, @Nullable AgentCard extendedAgentCard, RequestHandler requestHandler) { + public GrpcAgent { + Assert.checkNotNullParam("agentCard", agentCard); + Assert.checkNotNullParam("requestHandler", requestHandler); + } +} diff --git a/reference/grpc/src/main/java/org/a2aproject/sdk/server/grpc/quarkus/registry/MultiAgentRegistry.java b/reference/grpc/src/main/java/org/a2aproject/sdk/server/grpc/quarkus/registry/MultiAgentRegistry.java new file mode 100644 index 000000000..9c0d6ffb1 --- /dev/null +++ b/reference/grpc/src/main/java/org/a2aproject/sdk/server/grpc/quarkus/registry/MultiAgentRegistry.java @@ -0,0 +1,18 @@ +package org.a2aproject.sdk.server.grpc.quarkus.registry; + +import java.util.Map; + +/** + * Registry for supporting multiple agents behind a single Quarkus gRPC service. + * If a CDI bean implements this interface, incoming calls are dispatched to the agent + * identified by the {@code X-A2A-Agent-Id} metadata header. Calls that don't carry the + * header, or name an agent not present in the registry, fall back to the default + * single-agent {@link org.a2aproject.sdk.spec.AgentCard} / {@link org.a2aproject.sdk.server.requesthandlers.RequestHandler} + * beans, if configured. + */ +public interface MultiAgentRegistry { + /** + * @return a map of agent ID (as sent in the {@code X-A2A-Agent-Id} header) to their {@link GrpcAgent} + */ + Map getAgents(); +} diff --git a/reference/grpc/src/test/java/org/a2aproject/sdk/server/grpc/quarkus/QuarkusGrpcHandlerTest.java b/reference/grpc/src/test/java/org/a2aproject/sdk/server/grpc/quarkus/QuarkusGrpcHandlerTest.java new file mode 100644 index 000000000..77b1d5e9a --- /dev/null +++ b/reference/grpc/src/test/java/org/a2aproject/sdk/server/grpc/quarkus/QuarkusGrpcHandlerTest.java @@ -0,0 +1,156 @@ +package org.a2aproject.sdk.server.grpc.quarkus; + +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.Map; +import java.util.concurrent.Executor; + +import jakarta.enterprise.inject.Instance; + +import org.a2aproject.sdk.server.grpc.quarkus.registry.GrpcAgent; +import org.a2aproject.sdk.server.grpc.quarkus.registry.MultiAgentRegistry; +import org.a2aproject.sdk.server.requesthandlers.RequestHandler; +import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.InvalidRequestError; +import org.a2aproject.sdk.transport.grpc.context.GrpcContextKeys; +import io.grpc.Context; +import io.grpc.Metadata; +import org.junit.jupiter.api.Test; + +/** + * Unit test for {@link QuarkusGrpcHandler}'s multi-agent dispatch logic: selecting the + * agent to serve a call based on the {@code X-A2A-Agent-Id} metadata header, with fallback + * to the default single-agent beans. + */ +public class QuarkusGrpcHandlerTest { + + private static final Metadata.Key AGENT_ID_KEY = + Metadata.Key.of("x-a2a-agent-id", Metadata.ASCII_STRING_MARSHALLER); + + @SuppressWarnings("unchecked") + private Instance instanceOf(AgentCard value) { + Instance instance = mock(Instance.class); + when(instance.isResolvable()).thenReturn(value != null); + when(instance.get()).thenReturn(value); + return instance; + } + + @SuppressWarnings("unchecked") + private Instance instanceOf(RequestHandler value) { + Instance instance = mock(Instance.class); + when(instance.isResolvable()).thenReturn(value != null); + when(instance.get()).thenReturn(value); + return instance; + } + + @SuppressWarnings("unchecked") + private Instance instanceOf(MultiAgentRegistry value) { + Instance instance = mock(Instance.class); + when(instance.isResolvable()).thenReturn(value != null); + when(instance.get()).thenReturn(value); + return instance; + } + + @SuppressWarnings("unchecked") + private T runWithAgentIdHeader(String agentId, java.util.function.Supplier action) { + Metadata metadata = new Metadata(); + if (agentId != null) { + metadata.put(AGENT_ID_KEY, agentId); + } + Context context = Context.current().withValue(GrpcContextKeys.METADATA_KEY, metadata); + Context previous = context.attach(); + try { + return action.get(); + } finally { + context.detach(previous); + } + } + + @Test + public void testMultiAgentMode_DispatchesByAgentIdHeader() { + AgentCard agentACard = mock(AgentCard.class); + RequestHandler agentARequestHandler = mock(RequestHandler.class); + GrpcAgent agentA = new GrpcAgent(agentACard, null, agentARequestHandler); + + AgentCard agentBCard = mock(AgentCard.class); + RequestHandler agentBRequestHandler = mock(RequestHandler.class); + GrpcAgent agentB = new GrpcAgent(agentBCard, null, agentBRequestHandler); + + MultiAgentRegistry registry = mock(MultiAgentRegistry.class); + when(registry.getAgents()).thenReturn(Map.of("agentA", agentA, "agentB", agentB)); + + QuarkusGrpcHandler handler = new QuarkusGrpcHandler( + instanceOf((AgentCard) null), + instanceOf((AgentCard) null), + instanceOf((RequestHandler) null), + mock(Instance.class), + instanceOf(registry), + mock(Executor.class)); + + runWithAgentIdHeader("agentB", () -> { + assertSame(agentBRequestHandler, handler.getRequestHandler()); + assertSame(agentBCard, handler.getAgentCard()); + return null; + }); + } + + @Test + public void testMultiAgentMode_UnknownAgentId_FallsBackToDefaultHandler() { + AgentCard defaultCard = mock(AgentCard.class); + RequestHandler defaultRequestHandler = mock(RequestHandler.class); + + MultiAgentRegistry registry = mock(MultiAgentRegistry.class); + when(registry.getAgents()).thenReturn(Map.of()); + + QuarkusGrpcHandler handler = new QuarkusGrpcHandler( + instanceOf(defaultCard), + instanceOf((AgentCard) null), + instanceOf(defaultRequestHandler), + mock(Instance.class), + instanceOf(registry), + mock(Executor.class)); + + runWithAgentIdHeader("unknown-agent", () -> { + assertSame(defaultRequestHandler, handler.getRequestHandler()); + assertSame(defaultCard, handler.getAgentCard()); + return null; + }); + } + + @Test + public void testMultiAgentMode_UnknownAgentId_NoDefaultHandler_Throws() { + MultiAgentRegistry registry = mock(MultiAgentRegistry.class); + when(registry.getAgents()).thenReturn(Map.of()); + + QuarkusGrpcHandler handler = new QuarkusGrpcHandler( + instanceOf((AgentCard) null), + instanceOf((AgentCard) null), + instanceOf((RequestHandler) null), + mock(Instance.class), + instanceOf(registry), + mock(Executor.class)); + + runWithAgentIdHeader("unknown-agent", () -> + assertThrows(InvalidRequestError.class, handler::getRequestHandler)); + } + + @Test + public void testSingleAgentMode_NoRegistry_UsesDefaultHandler() { + AgentCard defaultCard = mock(AgentCard.class); + RequestHandler defaultRequestHandler = mock(RequestHandler.class); + + QuarkusGrpcHandler handler = new QuarkusGrpcHandler( + instanceOf(defaultCard), + instanceOf((AgentCard) null), + instanceOf(defaultRequestHandler), + mock(Instance.class), + instanceOf((MultiAgentRegistry) null), + mock(Executor.class)); + + assertSame(defaultRequestHandler, handler.getRequestHandler()); + assertSame(defaultCard, handler.getAgentCard()); + } +} diff --git a/reference/jsonrpc/src/main/java/org/a2aproject/sdk/server/apps/quarkus/A2AServerRoutes.java b/reference/jsonrpc/src/main/java/org/a2aproject/sdk/server/apps/quarkus/A2AServerRoutes.java index f0349d03f..eb0553acf 100644 --- a/reference/jsonrpc/src/main/java/org/a2aproject/sdk/server/apps/quarkus/A2AServerRoutes.java +++ b/reference/jsonrpc/src/main/java/org/a2aproject/sdk/server/apps/quarkus/A2AServerRoutes.java @@ -64,6 +64,7 @@ import org.a2aproject.sdk.jsonrpc.common.wrappers.SubscribeToTaskRequest; import org.a2aproject.sdk.server.AgentCardCacheMetadata; import org.a2aproject.sdk.server.ServerCallContext; +import org.a2aproject.sdk.server.apps.quarkus.registry.MultiAgentRegistry; import org.a2aproject.sdk.server.auth.AuthenticatedUser; import org.a2aproject.sdk.server.auth.UnauthenticatedUser; import org.a2aproject.sdk.server.auth.User; @@ -174,8 +175,15 @@ public class A2AServerRoutes { private static final Logger LOG = LoggerFactory.getLogger(A2AServerRoutes.class); + // RoutingContext key holding the agent path prefix (e.g. "/agentId") registered for this + // route, so extractTenant() can strip it before computing the tenant from the remaining path. + private static final String AGENT_PATH_PREFIX_CTX_KEY = "a2aAgentPathPrefix"; + + @Inject + Instance jsonRpcHandler; + @Inject - JSONRPCHandler jsonRpcHandler; + Instance multiAgentRegistry; @Inject AgentCardCacheMetadata cacheMetadata; @@ -205,16 +213,32 @@ public class A2AServerRoutes { * @param router the Vert.x Web Router instance to configure */ void setupRoutes(@Observes Router router) { - // Main JSON-RPC endpoint: POST / - // BodyHandler is per-route (not global) to avoid interfering with gRPC routes - // ordered=false: delegation via Vert.x WebClient can share the same event loop context as the outer request; ordered=true would serialize them, causing a 30s deadlock. - router.post("/") + if (multiAgentRegistry.isResolvable()) { + // Multi-agent mode + Map agents = multiAgentRegistry.get().getAgents(); + for (Map.Entry entry : agents.entrySet()) { + String agentId = entry.getKey(); + String pathPrefix = "/" + agentId; + registerAgentRoutes(router, pathPrefix, entry.getValue()); + } + } else if (jsonRpcHandler.isResolvable()) { + // Single-agent mode (default) + registerAgentRoutes(router, "", jsonRpcHandler.get()); + } + } + + private void registerAgentRoutes(Router router, String pathPrefix, JSONRPCHandler handler) { + String rpcPath = pathPrefix.isEmpty() ? "/" : pathPrefix; + String cardPath = pathPrefix + "/.well-known/agent-card.json"; + + router.post(rpcPath) .consumes(APPLICATION_JSON) .handler(BodyHandler.create()) .blockingHandler(ctx -> { try { + ctx.put(AGENT_PATH_PREFIX_CTX_KEY, pathPrefix); vertxSecurityHelper.runInRequestContextDeferred(ctx, () -> { - invokeJSONRPCHandler(ctx.body().asString(), ctx); + invokeJSONRPCHandler(ctx.body().asString(), ctx, handler); }); } catch (UnauthorizedException | ForbiddenException e) { vertxSecurityHelper.handleAuthError(ctx, e); @@ -223,12 +247,11 @@ void setupRoutes(@Observes Router router) { } }, false); - // Agent card endpoint: GET /.well-known/agent-card.json - router.get("/.well-known/agent-card.json") + router.get(cardPath) .produces(APPLICATION_JSON) .handler(ctx -> { try { - String agentCard = getAgentCard(ctx); + String agentCard = getAgentCard(ctx, handler); ctx.response() .setStatusCode(200) .putHeader(CONTENT_TYPE, APPLICATION_JSON) @@ -309,10 +332,11 @@ void setupRoutes(@Observes Router router) { * * @param body the raw JSON-RPC request body * @param rc the Vert.x routing context containing HTTP request/response + * @param handler the handler for the agent this request is addressed to * @throws A2AError if request processing fails */ @Authenticated - public void invokeJSONRPCHandler(String body, RoutingContext rc) { + public void invokeJSONRPCHandler(String body, RoutingContext rc, JSONRPCHandler handler) { boolean streaming = false; ServerCallContext context = createCallContext(rc); A2AResponse nonStreamingResponse = null; @@ -322,10 +346,10 @@ public void invokeJSONRPCHandler(String body, RoutingContext rc) { A2ARequest request = JSONRPCUtils.parseRequestBody(body, extractTenant(rc)); context.getState().put(METHOD_NAME_KEY, request.getMethod()); if (request instanceof NonStreamingJSONRPCRequest nonStreamingRequest) { - nonStreamingResponse = processNonStreamingRequest(nonStreamingRequest, context); + nonStreamingResponse = processNonStreamingRequest(nonStreamingRequest, context, handler); } else { streaming = true; - streamingResponse = processStreamingRequest(request, context); + streamingResponse = processStreamingRequest(request, context, handler); } } catch (A2AError e) { error = new A2AErrorResponse(e); @@ -407,14 +431,15 @@ public void invokeJSONRPCHandler(String body, RoutingContext rc) { * } * * @param rc the Vert.x routing context + * @param handler the handler for the agent this request is addressed to * @return the agent card as a JSON string * @throws JsonProcessingException if serialization fails * @see JSONRPCHandler#getAgentCard() */ - public String getAgentCard(RoutingContext rc) throws JsonProcessingException { + public String getAgentCard(RoutingContext rc, JSONRPCHandler handler) throws JsonProcessingException { // Add caching headers per A2A specification section 8.6 cacheMetadata.getHttpHeadersMap().forEach((k, v) -> rc.response().putHeader(k, v)); - return JsonUtil.toJson(jsonRpcHandler.getAgentCard()); + return JsonUtil.toJson(handler.getAgentCard()); } /** @@ -440,33 +465,33 @@ public String getAgentCard(RoutingContext rc) throws JsonProcessingException { * @param context the server call context * @return the JSON-RPC response */ - private A2AResponse processNonStreamingRequest(NonStreamingJSONRPCRequest request, ServerCallContext context) { + private A2AResponse processNonStreamingRequest(NonStreamingJSONRPCRequest request, ServerCallContext context, JSONRPCHandler handler) { if (request instanceof GetTaskRequest req) { - return jsonRpcHandler.onGetTask(req, context); + return handler.onGetTask(req, context); } if (request instanceof CancelTaskRequest req) { - return jsonRpcHandler.onCancelTask(req, context); + return handler.onCancelTask(req, context); } if (request instanceof ListTasksRequest req) { - return jsonRpcHandler.onListTasks(req, context); + return handler.onListTasks(req, context); } if (request instanceof CreateTaskPushNotificationConfigRequest req) { - return jsonRpcHandler.setPushNotificationConfig(req, context); + return handler.setPushNotificationConfig(req, context); } if (request instanceof GetTaskPushNotificationConfigRequest req) { - return jsonRpcHandler.getPushNotificationConfig(req, context); + return handler.getPushNotificationConfig(req, context); } if (request instanceof SendMessageRequest req) { - return jsonRpcHandler.onMessageSend(req, context); + return handler.onMessageSend(req, context); } if (request instanceof ListTaskPushNotificationConfigsRequest req) { - return jsonRpcHandler.listPushNotificationConfigs(req, context); + return handler.listPushNotificationConfigs(req, context); } if (request instanceof DeleteTaskPushNotificationConfigRequest req) { - return jsonRpcHandler.deletePushNotificationConfig(req, context); + return handler.deletePushNotificationConfig(req, context); } if (request instanceof GetExtendedAgentCardRequest req) { - return jsonRpcHandler.onGetExtendedCardRequest(req, context); + return handler.onGetExtendedCardRequest(req, context); } return generateErrorResponse(request, new UnsupportedOperationError()); } @@ -488,20 +513,20 @@ private A2AResponse processNonStreamingRequest(NonStreamingJSONRPCRequest * @return a Multi stream of JSON-RPC responses */ private Multi> processStreamingRequest( - A2ARequest request, ServerCallContext context) throws A2AError { + A2ARequest request, ServerCallContext context, JSONRPCHandler handler) throws A2AError { if (request instanceof SendStreamingMessageRequest req) { - jsonRpcHandler.authorizeTaskAccess(req.getParams().message().taskId(), context, + handler.authorizeTaskAccess(req.getParams().message().taskId(), context, TaskOperation.MESSAGE_SEND_STREAM); } else if (request instanceof SubscribeToTaskRequest req) { - jsonRpcHandler.authorizeTaskAccess(req.getParams().id(), context, + handler.authorizeTaskAccess(req.getParams().id(), context, TaskOperation.SUBSCRIBE_TO_TASK); } try { Flow.Publisher> publisher; if (request instanceof SendStreamingMessageRequest req) { - publisher = jsonRpcHandler.onMessageSendStream(req, context); + publisher = handler.onMessageSendStream(req, context); } else if (request instanceof SubscribeToTaskRequest req) { - publisher = jsonRpcHandler.onSubscribeToTask(req, context); + publisher = handler.onSubscribeToTask(req, context); } else { return Multi.createFrom().item(generateErrorResponse(request, new UnsupportedOperationError())); } @@ -615,6 +640,10 @@ private String extractTenant(RoutingContext rc) { if (tenantPath == null || tenantPath.isBlank()) { return ""; } + String agentPathPrefix = rc.get(AGENT_PATH_PREFIX_CTX_KEY); + if (agentPathPrefix != null && !agentPathPrefix.isEmpty() && tenantPath.startsWith(agentPathPrefix)) { + tenantPath = tenantPath.substring(agentPathPrefix.length()); + } if (tenantPath.startsWith("/")) { tenantPath = tenantPath.substring(1); } diff --git a/reference/jsonrpc/src/main/java/org/a2aproject/sdk/server/apps/quarkus/registry/MultiAgentRegistry.java b/reference/jsonrpc/src/main/java/org/a2aproject/sdk/server/apps/quarkus/registry/MultiAgentRegistry.java new file mode 100644 index 000000000..71c1ac9c3 --- /dev/null +++ b/reference/jsonrpc/src/main/java/org/a2aproject/sdk/server/apps/quarkus/registry/MultiAgentRegistry.java @@ -0,0 +1,16 @@ +package org.a2aproject.sdk.server.apps.quarkus.registry; + +import java.util.Map; +import org.a2aproject.sdk.transport.jsonrpc.handler.JSONRPCHandler; + +/** + * Registry for supporting multiple agents in a single Quarkus application. + * If a CDI bean implements this interface, the server will register routes for each + * agent in the registry under {@code //} and {@code //.well-known/agent-card.json}. + */ +public interface MultiAgentRegistry { + /** + * @return a map of agent ID (path segment) to their JSONRPCHandler + */ + Map getAgents(); +} diff --git a/reference/jsonrpc/src/test/java/org/a2aproject/sdk/server/apps/quarkus/A2AServerRoutesTest.java b/reference/jsonrpc/src/test/java/org/a2aproject/sdk/server/apps/quarkus/A2AServerRoutesTest.java index f2a746b2c..52d5cffd1 100644 --- a/reference/jsonrpc/src/test/java/org/a2aproject/sdk/server/apps/quarkus/A2AServerRoutesTest.java +++ b/reference/jsonrpc/src/test/java/org/a2aproject/sdk/server/apps/quarkus/A2AServerRoutesTest.java @@ -49,6 +49,7 @@ import org.a2aproject.sdk.jsonrpc.common.wrappers.CreateTaskPushNotificationConfigResponse; import org.a2aproject.sdk.jsonrpc.common.wrappers.SubscribeToTaskRequest; import org.a2aproject.sdk.server.ServerCallContext; +import org.a2aproject.sdk.server.apps.quarkus.registry.MultiAgentRegistry; import org.a2aproject.sdk.spec.AgentCapabilities; import org.a2aproject.sdk.spec.AgentCard; import org.a2aproject.sdk.spec.AgentInterface; @@ -75,7 +76,8 @@ public class A2AServerRoutesTest { private A2AServerRoutes routes; - private JSONRPCHandler mockJsonRpcHandler; + private Instance mockJsonRpcHandler; + private JSONRPCHandler mockHandlerInstance; private Executor mockExecutor; private Instance mockCallContextFactory; private RoutingContext mockRoutingContext; @@ -87,7 +89,14 @@ public class A2AServerRoutesTest { @BeforeEach public void setUp() { routes = new A2AServerRoutes(); - mockJsonRpcHandler = mock(JSONRPCHandler.class); + mockJsonRpcHandler = mock(Instance.class); + mockHandlerInstance = mock(JSONRPCHandler.class); + when(mockJsonRpcHandler.isResolvable()).thenReturn(true); + when(mockJsonRpcHandler.get()).thenReturn(mockHandlerInstance); + + Instance multiAgentRegistry = mock(Instance.class); + when(multiAgentRegistry.isResolvable()).thenReturn(false); + setField(routes, "multiAgentRegistry", multiAgentRegistry); mockExecutor = mock(Executor.class); mockCallContextFactory = mock(Instance.class); mockRoutingContext = mock(RoutingContext.class); @@ -154,16 +163,16 @@ public void testSendMessage_MethodNameSetInContext() { .status(new TaskStatus(TaskState.TASK_STATE_SUBMITTED)) .build(); SendMessageResponse realResponse = new SendMessageResponse("1", responseTask); - when(mockJsonRpcHandler.onMessageSend(any(SendMessageRequest.class), any(ServerCallContext.class))) + when(mockHandlerInstance.onMessageSend(any(SendMessageRequest.class), any(ServerCallContext.class))) .thenReturn(realResponse); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).onMessageSend(any(SendMessageRequest.class), contextCaptor.capture()); + verify(mockHandlerInstance).onMessageSend(any(SendMessageRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); assertEquals(SEND_MESSAGE_METHOD, capturedContext.getState().get(METHOD_NAME_KEY)); @@ -201,16 +210,16 @@ public void testSendStreamingMessage_MethodNameSetInContext() { @SuppressWarnings("unchecked") Flow.Publisher mockPublisher = mock(Flow.Publisher.class); - when(mockJsonRpcHandler.onMessageSendStream(any(SendStreamingMessageRequest.class), + when(mockHandlerInstance.onMessageSendStream(any(SendStreamingMessageRequest.class), any(ServerCallContext.class))).thenReturn(mockPublisher); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).onMessageSendStream(any(SendStreamingMessageRequest.class), + verify(mockHandlerInstance).onMessageSendStream(any(SendStreamingMessageRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); @@ -239,16 +248,16 @@ public void testGetTask_MethodNameSetInContext() { .status(new TaskStatus(TaskState.TASK_STATE_SUBMITTED)) .build(); GetTaskResponse realResponse = new GetTaskResponse("1", responseTask); - when(mockJsonRpcHandler.onGetTask(any(GetTaskRequest.class), any(ServerCallContext.class))) + when(mockHandlerInstance.onGetTask(any(GetTaskRequest.class), any(ServerCallContext.class))) .thenReturn(realResponse); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).onGetTask(any(GetTaskRequest.class), contextCaptor.capture()); + verify(mockHandlerInstance).onGetTask(any(GetTaskRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); assertEquals(GET_TASK_METHOD, capturedContext.getState().get(METHOD_NAME_KEY)); @@ -276,16 +285,16 @@ public void testCancelTask_MethodNameSetInContext() { .status(new TaskStatus(TaskState.TASK_STATE_CANCELED)) .build(); CancelTaskResponse realResponse = new CancelTaskResponse("1", responseTask); - when(mockJsonRpcHandler.onCancelTask(any(CancelTaskRequest.class), any(ServerCallContext.class))) + when(mockHandlerInstance.onCancelTask(any(CancelTaskRequest.class), any(ServerCallContext.class))) .thenReturn(realResponse); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).onCancelTask(any(CancelTaskRequest.class), contextCaptor.capture()); + verify(mockHandlerInstance).onCancelTask(any(CancelTaskRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); assertEquals(CANCEL_TASK_METHOD, capturedContext.getState().get(METHOD_NAME_KEY)); @@ -308,16 +317,16 @@ public void testTaskResubscription_MethodNameSetInContext() { @SuppressWarnings("unchecked") Flow.Publisher mockPublisher = mock(Flow.Publisher.class); - when(mockJsonRpcHandler.onSubscribeToTask(any(SubscribeToTaskRequest.class), + when(mockHandlerInstance.onSubscribeToTask(any(SubscribeToTaskRequest.class), any(ServerCallContext.class))).thenReturn(mockPublisher); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).onSubscribeToTask(any(SubscribeToTaskRequest.class), + verify(mockHandlerInstance).onSubscribeToTask(any(SubscribeToTaskRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); @@ -353,16 +362,16 @@ public void testCreateTaskPushNotificationConfig_MethodNameSetInContext() { .build(); CreateTaskPushNotificationConfigResponse realResponse = new CreateTaskPushNotificationConfigResponse("1", responseConfig); - when(mockJsonRpcHandler.setPushNotificationConfig(any(CreateTaskPushNotificationConfigRequest.class), + when(mockHandlerInstance.setPushNotificationConfig(any(CreateTaskPushNotificationConfigRequest.class), any(ServerCallContext.class))).thenReturn(realResponse); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).setPushNotificationConfig(any(CreateTaskPushNotificationConfigRequest.class), + verify(mockHandlerInstance).setPushNotificationConfig(any(CreateTaskPushNotificationConfigRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); @@ -392,16 +401,16 @@ public void testGetTaskPushNotificationConfig_MethodNameSetInContext() { .url("https://example.com/callback") .build(); GetTaskPushNotificationConfigResponse realResponse = new GetTaskPushNotificationConfigResponse("1", responseConfig); - when(mockJsonRpcHandler.getPushNotificationConfig(any(GetTaskPushNotificationConfigRequest.class), + when(mockHandlerInstance.getPushNotificationConfig(any(GetTaskPushNotificationConfigRequest.class), any(ServerCallContext.class))).thenReturn(realResponse); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).getPushNotificationConfig(any(GetTaskPushNotificationConfigRequest.class), + verify(mockHandlerInstance).getPushNotificationConfig(any(GetTaskPushNotificationConfigRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); @@ -432,16 +441,16 @@ public void testListTaskPushNotificationConfigs_MethodNameSetInContext() { .url("https://example.com/callback") .build(); ListTaskPushNotificationConfigsResponse realResponse = new ListTaskPushNotificationConfigsResponse("1", new ListTaskPushNotificationConfigsResult(singletonList(config))); - when(mockJsonRpcHandler.listPushNotificationConfigs(any(ListTaskPushNotificationConfigsRequest.class), + when(mockHandlerInstance.listPushNotificationConfigs(any(ListTaskPushNotificationConfigsRequest.class), any(ServerCallContext.class))).thenReturn(realResponse); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).listPushNotificationConfigs(any(ListTaskPushNotificationConfigsRequest.class), + verify(mockHandlerInstance).listPushNotificationConfigs(any(ListTaskPushNotificationConfigsRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); @@ -466,16 +475,16 @@ public void testDeleteTaskPushNotificationConfig_MethodNameSetInContext() { // Create a real response with id DeleteTaskPushNotificationConfigResponse realResponse = new DeleteTaskPushNotificationConfigResponse("1"); - when(mockJsonRpcHandler.deletePushNotificationConfig(any(DeleteTaskPushNotificationConfigRequest.class), + when(mockHandlerInstance.deletePushNotificationConfig(any(DeleteTaskPushNotificationConfigRequest.class), any(ServerCallContext.class))).thenReturn(realResponse); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).deletePushNotificationConfig(any(DeleteTaskPushNotificationConfigRequest.class), + verify(mockHandlerInstance).deletePushNotificationConfig(any(DeleteTaskPushNotificationConfigRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); @@ -502,17 +511,17 @@ public void testGetExtendedCard_MethodNameSetInContext() { .supportedInterfaces(Collections.singletonList(new AgentInterface("jsonrpc", "http://localhost:9999"))) .build(); GetExtendedAgentCardResponse realResponse = new GetExtendedAgentCardResponse(1, agentCard); - when(mockJsonRpcHandler.onGetExtendedCardRequest( + when(mockHandlerInstance.onGetExtendedCardRequest( any(GetExtendedAgentCardRequest.class), any(ServerCallContext.class))) .thenReturn(realResponse); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).onGetExtendedCardRequest( + verify(mockHandlerInstance).onGetExtendedCardRequest( any(GetExtendedAgentCardRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); @@ -542,16 +551,16 @@ public void testTenantExtraction_MultiSegmentPath() { .status(new TaskStatus(TaskState.TASK_STATE_SUBMITTED)) .build(); GetTaskResponse realResponse = new GetTaskResponse("1", responseTask); - when(mockJsonRpcHandler.onGetTask(any(GetTaskRequest.class), any(ServerCallContext.class))) + when(mockHandlerInstance.onGetTask(any(GetTaskRequest.class), any(ServerCallContext.class))) .thenReturn(realResponse); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).onGetTask(any(GetTaskRequest.class), contextCaptor.capture()); + verify(mockHandlerInstance).onGetTask(any(GetTaskRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); assertEquals("test/titi", capturedContext.getState().get(TENANT_KEY)); @@ -579,16 +588,16 @@ public void testTenantExtraction_RootPath() { .status(new TaskStatus(TaskState.TASK_STATE_SUBMITTED)) .build(); GetTaskResponse realResponse = new GetTaskResponse("1", responseTask); - when(mockJsonRpcHandler.onGetTask(any(GetTaskRequest.class), any(ServerCallContext.class))) + when(mockHandlerInstance.onGetTask(any(GetTaskRequest.class), any(ServerCallContext.class))) .thenReturn(realResponse); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).onGetTask(any(GetTaskRequest.class), contextCaptor.capture()); + verify(mockHandlerInstance).onGetTask(any(GetTaskRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); assertEquals("", capturedContext.getState().get(TENANT_KEY)); @@ -616,16 +625,16 @@ public void testTenantExtraction_SingleSegmentPath() { .status(new TaskStatus(TaskState.TASK_STATE_SUBMITTED)) .build(); GetTaskResponse realResponse = new GetTaskResponse("1", responseTask); - when(mockJsonRpcHandler.onGetTask(any(GetTaskRequest.class), any(ServerCallContext.class))) + when(mockHandlerInstance.onGetTask(any(GetTaskRequest.class), any(ServerCallContext.class))) .thenReturn(realResponse); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).onGetTask(any(GetTaskRequest.class), contextCaptor.capture()); + verify(mockHandlerInstance).onGetTask(any(GetTaskRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); assertEquals("tenant1", capturedContext.getState().get(TENANT_KEY)); @@ -653,16 +662,16 @@ public void testTenantExtraction_ThreeSegmentPath() { .status(new TaskStatus(TaskState.TASK_STATE_SUBMITTED)) .build(); GetTaskResponse realResponse = new GetTaskResponse("1", responseTask); - when(mockJsonRpcHandler.onGetTask(any(GetTaskRequest.class), any(ServerCallContext.class))) + when(mockHandlerInstance.onGetTask(any(GetTaskRequest.class), any(ServerCallContext.class))) .thenReturn(realResponse); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).onGetTask(any(GetTaskRequest.class), contextCaptor.capture()); + verify(mockHandlerInstance).onGetTask(any(GetTaskRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); assertEquals("tenant1/api/v1", capturedContext.getState().get(TENANT_KEY)); @@ -700,22 +709,61 @@ public void testTenantExtraction_StreamingRequest() { @SuppressWarnings("unchecked") Flow.Publisher mockPublisher = mock(Flow.Publisher.class); - when(mockJsonRpcHandler.onMessageSendStream(any(SendStreamingMessageRequest.class), + when(mockHandlerInstance.onMessageSendStream(any(SendStreamingMessageRequest.class), any(ServerCallContext.class))).thenReturn(mockPublisher); ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert - verify(mockJsonRpcHandler).onMessageSendStream(any(SendStreamingMessageRequest.class), + verify(mockHandlerInstance).onMessageSendStream(any(SendStreamingMessageRequest.class), contextCaptor.capture()); ServerCallContext capturedContext = contextCaptor.getValue(); assertNotNull(capturedContext); assertEquals("myTenant/api", capturedContext.getState().get(TENANT_KEY)); } + @Test + public void testTenantExtraction_AgentPathPrefixStripped() { + // Arrange - simulate a multi-agent route: POST /myagent, where "myagent" is the + // agent ID (from the registered path prefix), not a tenant. + when(mockRoutingContext.normalizedPath()).thenReturn("/myagent"); + when(mockRoutingContext.get("a2aAgentPathPrefix")).thenReturn("/myagent"); + String jsonRpcRequest = """ + { + "jsonrpc": "2.0", + "id": "cd4c76de-d54c-436c-8b9f-4c2703648d64", + "method": "GetTask", + "params": { + "id": "de38c76d-d54c-436c-8b9f-4c2703648d64", + "historyLength": 10 + } + }"""; + when(mockRequestBody.asString()).thenReturn(jsonRpcRequest); + + Task responseTask = Task.builder() + .id("de38c76d-d54c-436c-8b9f-4c2703648d64") + .contextId("context-1234") + .status(new TaskStatus(TaskState.TASK_STATE_SUBMITTED)) + .build(); + GetTaskResponse realResponse = new GetTaskResponse("1", responseTask); + when(mockHandlerInstance.onGetTask(any(GetTaskRequest.class), any(ServerCallContext.class))) + .thenReturn(realResponse); + + ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); + + // Act + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); + + // Assert - the agent path prefix must not leak into the tenant + verify(mockHandlerInstance).onGetTask(any(GetTaskRequest.class), contextCaptor.capture()); + ServerCallContext capturedContext = contextCaptor.getValue(); + assertNotNull(capturedContext); + assertEquals("", capturedContext.getState().get(TENANT_KEY)); + } + @Test public void testJsonParseError_ContentTypeIsApplicationJson() { // Arrange - invalid JSON @@ -723,7 +771,7 @@ public void testJsonParseError_ContentTypeIsApplicationJson() { when(mockRequestBody.asString()).thenReturn(invalidJson); // Act - routes.invokeJSONRPCHandler(invalidJson, mockRoutingContext); + routes.invokeJSONRPCHandler(invalidJson, mockRoutingContext, mockHandlerInstance); // Assert verify(mockHttpResponse).putHeader(CONTENT_TYPE, APPLICATION_JSON); @@ -742,7 +790,7 @@ public void testMethodNotFound_ContentTypeIsApplicationJson() { when(mockRequestBody.asString()).thenReturn(jsonRpcRequest); // Act - routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext); + routes.invokeJSONRPCHandler(jsonRpcRequest, mockRoutingContext, mockHandlerInstance); // Assert verify(mockHttpResponse).putHeader(CONTENT_TYPE, APPLICATION_JSON); diff --git a/reference/multiversion-jsonrpc/src/main/java/org/a2aproject/sdk/server/multiversion/jsonrpc/MultiVersionJSONRPCRoutes.java b/reference/multiversion-jsonrpc/src/main/java/org/a2aproject/sdk/server/multiversion/jsonrpc/MultiVersionJSONRPCRoutes.java index 74ac5acf6..35dfb2b32 100644 --- a/reference/multiversion-jsonrpc/src/main/java/org/a2aproject/sdk/server/multiversion/jsonrpc/MultiVersionJSONRPCRoutes.java +++ b/reference/multiversion-jsonrpc/src/main/java/org/a2aproject/sdk/server/multiversion/jsonrpc/MultiVersionJSONRPCRoutes.java @@ -3,6 +3,7 @@ import static io.vertx.core.http.HttpHeaders.CONTENT_TYPE; import jakarta.enterprise.event.Observes; +import jakarta.enterprise.inject.Instance; import jakarta.inject.Inject; import jakarta.inject.Singleton; @@ -15,6 +16,7 @@ import org.a2aproject.sdk.server.common.quarkus.VertxSecurityHelper; import org.a2aproject.sdk.spec.A2AError; import org.a2aproject.sdk.spec.VersionNotSupportedError; +import org.a2aproject.sdk.transport.jsonrpc.handler.JSONRPCHandler; import io.quarkus.security.ForbiddenException; import io.quarkus.security.UnauthorizedException; @@ -28,6 +30,9 @@ public class MultiVersionJSONRPCRoutes { @Inject A2AServerRoutes_v0_3 v03Routes; + @Inject + Instance jsonRpcHandler; + @Inject VertxSecurityHelper vertxSecurityHelper; @@ -46,7 +51,7 @@ void setupRoutes(@Observes Router router) { String body = ctx.body().asString(); if (VersionRouter.isV10(version)) { - v10Routes.invokeJSONRPCHandler(body, ctx); + v10Routes.invokeJSONRPCHandler(body, ctx, jsonRpcHandler.get()); } else if (VersionRouter.isV03(version)) { v03Routes.invokeJSONRPCHandler(body, ctx); } else { diff --git a/reference/multiversion-rest/src/main/java/org/a2aproject/sdk/server/multiversion/rest/MultiVersionRestRoutes.java b/reference/multiversion-rest/src/main/java/org/a2aproject/sdk/server/multiversion/rest/MultiVersionRestRoutes.java index 816f2639a..f2e89b35c 100644 --- a/reference/multiversion-rest/src/main/java/org/a2aproject/sdk/server/multiversion/rest/MultiVersionRestRoutes.java +++ b/reference/multiversion-rest/src/main/java/org/a2aproject/sdk/server/multiversion/rest/MultiVersionRestRoutes.java @@ -7,6 +7,7 @@ import jakarta.annotation.Priority; import jakarta.enterprise.event.Observes; +import jakarta.enterprise.inject.Instance; import jakarta.inject.Inject; import jakarta.inject.Singleton; @@ -21,6 +22,7 @@ import org.a2aproject.sdk.spec.A2AError; import org.a2aproject.sdk.spec.A2AErrorCodes; import org.a2aproject.sdk.spec.VersionNotSupportedError; +import org.a2aproject.sdk.transport.rest.handler.RestHandler; import io.quarkus.security.ForbiddenException; import io.quarkus.security.UnauthorizedException; @@ -34,6 +36,9 @@ public class MultiVersionRestRoutes { @Inject A2AServerRoutes_v0_3 v03Routes; + @Inject + Instance jsonRestHandler; + @Inject VertxSecurityHelper vertxSecurityHelper; @@ -44,7 +49,7 @@ void setupRoutes(@Observes @Priority(5) Router router) { .handler(BodyHandler.create()) .blockingHandler(versionDispatch(false, MultiVersionRestRoutes::bridgeTenant, - (body, ctx) -> v10Routes.sendMessage(body, ctx), + (body, ctx) -> v10Routes.sendMessage(body, ctx, jsonRestHandler.get()), (body, ctx) -> v03Routes.sendMessage(body, ctx)), false); // POST /v1/message:stream (deferred CDI context destruction) @@ -53,7 +58,7 @@ void setupRoutes(@Observes @Priority(5) Router router) { .handler(BodyHandler.create()) .blockingHandler(versionDispatch(true, MultiVersionRestRoutes::bridgeTenant, - (body, ctx) -> v10Routes.sendMessageStreaming(body, ctx), + (body, ctx) -> v10Routes.sendMessageStreaming(body, ctx, jsonRestHandler.get()), (body, ctx) -> v03Routes.sendMessageStreaming(body, ctx)), false); // GET /v1/tasks/{taskId} @@ -61,7 +66,7 @@ void setupRoutes(@Observes @Priority(5) Router router) { .order(-1) .blockingHandler(versionDispatchNoBody(false, ctx -> { bridgeTenant(ctx); bridgeTaskId(ctx); }, - ctx -> v10Routes.getTask(ctx), + ctx -> v10Routes.getTask(ctx, jsonRestHandler.get()), ctx -> v03Routes.getTask(ctx)), false); // POST /v1/tasks/{taskId}:cancel @@ -70,7 +75,7 @@ void setupRoutes(@Observes @Priority(5) Router router) { .handler(BodyHandler.create()) .blockingHandler(versionDispatch(false, ctx -> { bridgeTenant(ctx); bridgeTaskId(ctx); }, - (body, ctx) -> v10Routes.cancelTask(body, ctx), + (body, ctx) -> v10Routes.cancelTask(body, ctx, jsonRestHandler.get()), (body, ctx) -> v03Routes.cancelTask(ctx)), false); // POST /v1/tasks/{taskId}:subscribe (deferred CDI context destruction) @@ -78,7 +83,7 @@ void setupRoutes(@Observes @Priority(5) Router router) { .order(-1) .blockingHandler(versionDispatchNoBody(true, ctx -> { bridgeTenant(ctx); bridgeTaskId(ctx); }, - ctx -> v10Routes.subscribeToTask(ctx), + ctx -> v10Routes.subscribeToTask(ctx, jsonRestHandler.get()), ctx -> v03Routes.resubscribeTask(ctx)), false); // POST /v1/tasks/{taskId}/pushNotificationConfigs @@ -87,7 +92,7 @@ void setupRoutes(@Observes @Priority(5) Router router) { .handler(BodyHandler.create()) .blockingHandler(versionDispatch(false, ctx -> { bridgeTenant(ctx); bridgeTaskId(ctx); }, - (body, ctx) -> v10Routes.createTaskPushNotificationConfiguration(body, ctx), + (body, ctx) -> v10Routes.createTaskPushNotificationConfiguration(body, ctx, jsonRestHandler.get()), (body, ctx) -> v03Routes.setTaskPushNotificationConfiguration(body, ctx)), false); // GET /v1/tasks/{taskId}/pushNotificationConfigs/{configId} @@ -95,7 +100,7 @@ void setupRoutes(@Observes @Priority(5) Router router) { .order(-1) .blockingHandler(versionDispatchNoBody(false, ctx -> { bridgeTenant(ctx); bridgeTaskId(ctx); }, - ctx -> v10Routes.getTaskPushNotificationConfiguration(ctx), + ctx -> v10Routes.getTaskPushNotificationConfiguration(ctx, jsonRestHandler.get()), ctx -> v03Routes.getTaskPushNotificationConfiguration(ctx)), false); // GET /v1/tasks/{taskId}/pushNotificationConfigs @@ -103,7 +108,7 @@ void setupRoutes(@Observes @Priority(5) Router router) { .order(-1) .blockingHandler(versionDispatchNoBody(false, ctx -> { bridgeTenant(ctx); bridgeTaskId(ctx); }, - ctx -> v10Routes.listTaskPushNotificationConfigurations(ctx), + ctx -> v10Routes.listTaskPushNotificationConfigurations(ctx, jsonRestHandler.get()), ctx -> v03Routes.listTaskPushNotificationConfigurations(ctx)), false); // DELETE /v1/tasks/{taskId}/pushNotificationConfigs/{configId} @@ -111,7 +116,7 @@ void setupRoutes(@Observes @Priority(5) Router router) { .order(-1) .blockingHandler(versionDispatchNoBody(false, ctx -> { bridgeTenant(ctx); bridgeTaskId(ctx); }, - ctx -> v10Routes.deleteTaskPushNotificationConfiguration(ctx), + ctx -> v10Routes.deleteTaskPushNotificationConfiguration(ctx, jsonRestHandler.get()), ctx -> v03Routes.deleteTaskPushNotificationConfiguration(ctx)), false); // GET /v1/card — v0.3 only (v1.0 uses /{tenant}/extendedAgentCard) diff --git a/reference/rest/src/main/java/org/a2aproject/sdk/server/rest/quarkus/A2AServerRoutes.java b/reference/rest/src/main/java/org/a2aproject/sdk/server/rest/quarkus/A2AServerRoutes.java index 8cd9f7af3..62062e91b 100644 --- a/reference/rest/src/main/java/org/a2aproject/sdk/server/rest/quarkus/A2AServerRoutes.java +++ b/reference/rest/src/main/java/org/a2aproject/sdk/server/rest/quarkus/A2AServerRoutes.java @@ -16,6 +16,7 @@ import java.util.concurrent.Flow; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Consumer; +import java.util.regex.Pattern; import org.a2aproject.sdk.server.util.sse.SseFormatter; @@ -31,6 +32,7 @@ import org.a2aproject.sdk.server.auth.UnauthenticatedUser; import org.a2aproject.sdk.server.auth.User; import org.a2aproject.sdk.server.extensions.A2AExtensions; +import org.a2aproject.sdk.server.rest.quarkus.registry.MultiAgentRegistry; import org.a2aproject.sdk.server.util.async.Internal; import org.a2aproject.sdk.spec.A2AError; import org.a2aproject.sdk.spec.ContentTypeNotSupportedError; @@ -135,7 +137,10 @@ public class A2AServerRoutes { private static final String STATUS_TIMESTAMP_AFTER = "statusTimestampAfter"; @Inject - RestHandler jsonRestHandler; + Instance jsonRestHandler; + + @Inject + Instance multiAgentRegistry; // Hook so testing can wait until the SSE subscriber is attached. // Without this we get intermittent failures @@ -161,6 +166,30 @@ public class A2AServerRoutes { * @param router the Vert.x router to configure */ void setupRouter(@Observes @Priority(10) Router router) { + if (multiAgentRegistry.isResolvable()) { + // Multi-agent mode + Map agents = multiAgentRegistry.get().getAgents(); + for (Map.Entry entry : agents.entrySet()) { + String agentId = entry.getKey(); + String pathPrefix = "/" + agentId; + registerAgentRoutes(router, pathPrefix, entry.getValue()); + } + } else if (jsonRestHandler.isResolvable()) { + // Single-agent mode (default) + registerAgentRoutes(router, "", jsonRestHandler.get()); + } + } + + /** + * Builds a route regex anchored at the start of the path, with {@code pathPrefix} + * (e.g. an agent ID segment) matched as a literal prefix before {@code regexTail}. + * An empty {@code pathPrefix} is a no-op, preserving the original single-agent patterns. + */ + private static String prefixedRegex(String pathPrefix, String regexTail) { + return "^" + Pattern.quote(pathPrefix) + regexTail; + } + + private void registerAgentRoutes(Router router, String pathPrefix, RestHandler handler) { // Don't add a global BodyHandler - it interferes with gRPC routes // Instead, BodyHandler is added per-route below @@ -171,86 +200,86 @@ void setupRouter(@Observes @Priority(10) Router router) { // request. ordered=true would serialize them, causing a 30s deadlock. // POST /{tenant}/message:send - Non-streaming message send - router.postWithRegex("^\\/(?[^\\/]*\\/?)message:send$") + router.postWithRegex(prefixedRegex(pathPrefix, "\\/(?[^\\/]*\\/?)message:send$")) .handler(BodyHandler.create()) .blockingHandler(authenticated(ctx -> { String body = extractBody(ctx); - sendMessage(body, ctx); + sendMessage(body, ctx, handler); }), false); // POST /{tenant}/message:stream - Streaming message with SSE - router.postWithRegex("^\\/(?[^\\/]*\\/?)message:stream$") + router.postWithRegex(prefixedRegex(pathPrefix, "\\/(?[^\\/]*\\/?)message:stream$")) .handler(BodyHandler.create()) .blockingHandler(authenticatedStreaming(ctx -> { String body = extractBody(ctx); - sendMessageStreaming(body, ctx); + sendMessageStreaming(body, ctx, handler); }), false); // Task Routes // GET /{tenant}/tasks - List tasks with query params - router.getWithRegex("^\\/(?[^\\/]*\\/?)tasks\\??") + router.getWithRegex(prefixedRegex(pathPrefix, "\\/(?[^\\/]*\\/?)tasks\\??")) .order(0) - .blockingHandler(authenticated(this::listTasks), false); + .blockingHandler(authenticated(ctx -> listTasks(ctx, handler)), false); // GET /{tenant}/tasks/{taskId} - Get specific task - router.getWithRegex("^\\/(?[^\\/]*\\/?)tasks\\/(?[^:^/]+)$") + router.getWithRegex(prefixedRegex(pathPrefix, "\\/(?[^\\/]*\\/?)tasks\\/(?[^:^/]+)$")) .order(1) - .blockingHandler(authenticated(this::getTask), false); + .blockingHandler(authenticated(ctx -> getTask(ctx, handler)), false); // POST /{tenant}/tasks/{taskId}:cancel - Cancel task - router.postWithRegex("^\\/(?[^\\/]*\\/?)tasks\\/(?[^/]+):cancel$") + router.postWithRegex(prefixedRegex(pathPrefix, "\\/(?[^\\/]*\\/?)tasks\\/(?[^/]+):cancel$")) .order(1) .handler(BodyHandler.create()) .blockingHandler(authenticated(ctx -> { String body = extractBody(ctx); - cancelTask(body, ctx); + cancelTask(body, ctx, handler); }), false); // POST /{tenant}/tasks/{taskId}:subscribe - Subscribe to task updates (SSE) - router.postWithRegex("^\\/(?[^\\/]*\\/?)tasks\\/(?[^/]+):subscribe$") + router.postWithRegex(prefixedRegex(pathPrefix, "\\/(?[^\\/]*\\/?)tasks\\/(?[^/]+):subscribe$")) .order(1) - .blockingHandler(authenticatedStreaming(this::subscribeToTask), false); + .blockingHandler(authenticatedStreaming(ctx -> subscribeToTask(ctx, handler)), false); // Push Notification Routes // POST /{tenant}/tasks/{taskId}/pushNotificationConfigs - router.postWithRegex("^\\/(?[^\\/]*\\/?)tasks\\/(?[^/]+)\\/pushNotificationConfigs$") + router.postWithRegex(prefixedRegex(pathPrefix, "\\/(?[^\\/]*\\/?)tasks\\/(?[^/]+)\\/pushNotificationConfigs$")) .order(1) .handler(BodyHandler.create()) .blockingHandler(authenticated(ctx -> { String body = extractBody(ctx); - createTaskPushNotificationConfiguration(body, ctx); + createTaskPushNotificationConfiguration(body, ctx, handler); }), false); // GET /{tenant}/tasks/{taskId}/pushNotificationConfigs/{configId} - router.getWithRegex("^\\/(?[^\\/]*\\/?)tasks\\/(?[^/]+)\\/pushNotificationConfigs\\/(?[^\\/]+)") + router.getWithRegex(prefixedRegex(pathPrefix, "\\/(?[^\\/]*\\/?)tasks\\/(?[^/]+)\\/pushNotificationConfigs\\/(?[^\\/]+)")) .order(2) - .blockingHandler(authenticated(this::getTaskPushNotificationConfiguration), false); + .blockingHandler(authenticated(ctx -> getTaskPushNotificationConfiguration(ctx, handler)), false); // GET /{tenant}/tasks/{taskId}/pushNotificationConfigs - router.getWithRegex("^\\/(?[^\\/]*\\/?)tasks\\/(?[^/]+)\\/pushNotificationConfigs\\/?$") + router.getWithRegex(prefixedRegex(pathPrefix, "\\/(?[^\\/]*\\/?)tasks\\/(?[^/]+)\\/pushNotificationConfigs\\/?$")) .order(3) - .blockingHandler(authenticated(this::listTaskPushNotificationConfigurations), false); + .blockingHandler(authenticated(ctx -> listTaskPushNotificationConfigurations(ctx, handler)), false); // DELETE /{tenant}/tasks/{taskId}/pushNotificationConfigs/{configId} - router.deleteWithRegex("^\\/(?[^\\/]*\\/?)tasks\\/(?[^/]+)\\/pushNotificationConfigs\\/(?[^/]+)") + router.deleteWithRegex(prefixedRegex(pathPrefix, "\\/(?[^\\/]*\\/?)tasks\\/(?[^/]+)\\/pushNotificationConfigs\\/(?[^/]+)")) .order(1) - .blockingHandler(authenticated(this::deleteTaskPushNotificationConfiguration), false); + .blockingHandler(authenticated(ctx -> deleteTaskPushNotificationConfiguration(ctx, handler)), false); // Discovery Routes // GET /.well-known/agent-card.json - Public agent card (no auth required) - router.get("/.well-known/agent-card.json") + router.get(pathPrefix + "/.well-known/agent-card.json") .order(1) .produces(APPLICATION_JSON) - .handler(this::getAgentCard); + .handler(ctx -> getAgentCard(ctx, handler)); // GET /{tenant}/extendedAgentCard - Extended agent card (auth required) - router.getWithRegex("^\\/(?[^\\/]*\\/?)extendedAgentCard$") + router.getWithRegex(prefixedRegex(pathPrefix, "\\/(?[^\\/]*\\/?)extendedAgentCard$")) .order(1) .produces(APPLICATION_JSON) - .blockingHandler(authenticated(this::getExtendedAgentCard), false); + .blockingHandler(authenticated(ctx -> getExtendedAgentCard(ctx, handler)), false); } private Handler authenticated(Consumer action) { @@ -299,19 +328,20 @@ private Handler authenticatedStreaming(Consumer * * @param body the JSON request body * @param rc the Vert.x routing context + * @param handler the handler for the agent this request is addressed to */ @Authenticated - public void sendMessage(String body, RoutingContext rc) { - if(!validateContentType(rc)) { + public void sendMessage(String body, RoutingContext rc, RestHandler handler) { + if(!validateContentType(rc, handler)) { return; } ServerCallContext context = createCallContext(rc, SEND_MESSAGE_METHOD); HTTPRestResponse response = null; try { - response = jsonRestHandler.sendMessage(context, extractTenant(rc), body); + response = handler.sendMessage(context, extractTenant(rc), body); } catch (Throwable t) { LOG.error("Internal error while processing request", t); - response = jsonRestHandler.createErrorResponse(new InternalError("Internal error")); + response = handler.createErrorResponse(new InternalError("Internal error")); } finally { sendResponse(rc, response); } @@ -336,17 +366,18 @@ public void sendMessage(String body, RoutingContext rc) { * * @param body the JSON request body * @param rc the Vert.x routing context + * @param handler the handler for the agent this request is addressed to */ @Authenticated - public void sendMessageStreaming(String body, RoutingContext rc) { - if(!validateContentType(rc)) { + public void sendMessageStreaming(String body, RoutingContext rc, RestHandler handler) { + if(!validateContentType(rc, handler)) { return; } ServerCallContext context = createCallContext(rc, SEND_STREAMING_MESSAGE_METHOD); HTTPRestStreamingResponse streamingResponse = null; HTTPRestResponse error = null; try { - HTTPRestResponse response = jsonRestHandler.sendStreamingMessage(context, extractTenant(rc), body); + HTTPRestResponse response = handler.sendStreamingMessage(context, extractTenant(rc), body); if (response instanceof HTTPRestStreamingResponse hTTPRestStreamingResponse) { streamingResponse = hTTPRestStreamingResponse; } else { @@ -389,9 +420,10 @@ public void sendMessageStreaming(String body, RoutingContext rc) { * * * @param rc the Vert.x routing context + * @param handler the handler for the agent this request is addressed to */ @Authenticated - public void listTasks(RoutingContext rc) { + public void listTasks(RoutingContext rc, RestHandler handler) { ServerCallContext context = createCallContext(rc, LIST_TASK_METHOD); HTTPRestResponse response = null; try { @@ -422,15 +454,15 @@ public void listTasks(RoutingContext rc) { if (includeArtifactsStr != null && !includeArtifactsStr.isEmpty()) { includeArtifacts = Boolean.valueOf(includeArtifactsStr); } - response = jsonRestHandler.listTasks(context, extractTenant(rc), contextId, statusStr, pageSize, pageToken, + response = handler.listTasks(context, extractTenant(rc), contextId, statusStr, pageSize, pageToken, historyLength, statusTimestampAfter, includeArtifacts); } catch (NumberFormatException e) { - response = jsonRestHandler.createErrorResponse(new InvalidParamsError("Invalid number format in parameters")); + response = handler.createErrorResponse(new InvalidParamsError("Invalid number format in parameters")); } catch (IllegalArgumentException e) { - response = jsonRestHandler.createErrorResponse(new InvalidParamsError("Invalid parameter value: " + e.getMessage())); + response = handler.createErrorResponse(new InvalidParamsError("Invalid parameter value: " + e.getMessage())); } catch (Throwable t) { LOG.error("Internal error while processing request", t); - response = jsonRestHandler.createErrorResponse(new InternalError("Internal error")); + response = handler.createErrorResponse(new InternalError("Internal error")); } finally { sendResponse(rc, response); } @@ -445,27 +477,28 @@ public void listTasks(RoutingContext rc) { *

    URL Pattern: {@code /tasks/{taskId}?historyLength=10} * * @param rc the Vert.x routing context (taskId extracted from path) + * @param handler the handler for the agent this request is addressed to */ @Authenticated - public void getTask(RoutingContext rc) { + public void getTask(RoutingContext rc, RestHandler handler) { String taskId = rc.pathParam("taskId"); ServerCallContext context = createCallContext(rc, GET_TASK_METHOD); HTTPRestResponse response = null; try { if (taskId == null || taskId.isEmpty()) { - response = jsonRestHandler.createErrorResponse(new InvalidParamsError("bad task id")); + response = handler.createErrorResponse(new InvalidParamsError("bad task id")); } else { Integer historyLength = null; if (rc.request().params().contains(HISTORY_LENGTH_PARAM)) { historyLength = Integer.valueOf(rc.request().params().get(HISTORY_LENGTH_PARAM)); } - response = jsonRestHandler.getTask(context, extractTenant(rc), taskId, historyLength); + response = handler.getTask(context, extractTenant(rc), taskId, historyLength); } } catch (NumberFormatException e) { - response = jsonRestHandler.createErrorResponse(new InvalidParamsError("bad historyLength")); + response = handler.createErrorResponse(new InvalidParamsError("bad historyLength")); } catch (Throwable t) { LOG.error("Internal error while processing request", t); - response = jsonRestHandler.createErrorResponse(new InternalError("Internal error")); + response = handler.createErrorResponse(new InternalError("Internal error")); } finally { sendResponse(rc, response); } @@ -480,10 +513,11 @@ public void getTask(RoutingContext rc) { *

    URL Pattern: {@code /tasks/{taskId}:cancel} * * @param rc the Vert.x routing context (taskId extracted from path) + * @param handler the handler for the agent this request is addressed to */ @Authenticated - public void cancelTask(String body, RoutingContext rc) { - if (!validateContentTypeForOptionalBody(rc, body)) { + public void cancelTask(String body, RoutingContext rc, RestHandler handler) { + if (!validateContentTypeForOptionalBody(rc, body, handler)) { return; } String taskId = rc.pathParam("taskId"); @@ -491,16 +525,16 @@ public void cancelTask(String body, RoutingContext rc) { HTTPRestResponse response = null; try { if (taskId == null || taskId.isEmpty()) { - response = jsonRestHandler.createErrorResponse(new InvalidParamsError("bad task id")); + response = handler.createErrorResponse(new InvalidParamsError("bad task id")); } else { - response = jsonRestHandler.cancelTask(context, extractTenant(rc), body, taskId); + response = handler.cancelTask(context, extractTenant(rc), body, taskId); } } catch (Throwable t) { if (t instanceof A2AError error) { - response = jsonRestHandler.createErrorResponse(error); + response = handler.createErrorResponse(error); } else { LOG.error("Internal error while processing request", t); - response = jsonRestHandler.createErrorResponse(new InternalError("Internal error")); + response = handler.createErrorResponse(new InternalError("Internal error")); } } finally { sendResponse(rc, response); @@ -547,18 +581,19 @@ private void sendResponse(RoutingContext rc, @Nullable HTTPRestResponse response * * * @param rc the Vert.x routing context (taskId extracted from path) + * @param handler the handler for the agent this request is addressed to */ @Authenticated - public void subscribeToTask(RoutingContext rc) { + public void subscribeToTask(RoutingContext rc, RestHandler handler) { String taskId = rc.pathParam("taskId"); ServerCallContext context = createCallContext(rc, SUBSCRIBE_TO_TASK_METHOD); HTTPRestStreamingResponse streamingResponse = null; HTTPRestResponse error = null; try { if (taskId == null || taskId.isEmpty()) { - error = jsonRestHandler.createErrorResponse(new InvalidParamsError("bad task id")); + error = handler.createErrorResponse(new InvalidParamsError("bad task id")); } else { - HTTPRestResponse response = jsonRestHandler.subscribeToTask(context, extractTenant(rc), taskId); + HTTPRestResponse response = handler.subscribeToTask(context, extractTenant(rc), taskId); if (response instanceof HTTPRestStreamingResponse hTTPRestStreamingResponse) { streamingResponse = hTTPRestStreamingResponse; } else { @@ -594,10 +629,11 @@ public void subscribeToTask(RoutingContext rc) { * * @param body the JSON request body with notification configuration * @param rc the Vert.x routing context (taskId extracted from path) + * @param handler the handler for the agent this request is addressed to */ @Authenticated - public void createTaskPushNotificationConfiguration(String body, RoutingContext rc) { - if(!validateContentType(rc)) { + public void createTaskPushNotificationConfiguration(String body, RoutingContext rc, RestHandler handler) { + if(!validateContentType(rc, handler)) { return; } String taskId = rc.pathParam("taskId"); @@ -605,13 +641,13 @@ public void createTaskPushNotificationConfiguration(String body, RoutingContext HTTPRestResponse response = null; try { if (taskId == null || taskId.isEmpty()) { - response = jsonRestHandler.createErrorResponse(new InvalidParamsError("bad task id")); + response = handler.createErrorResponse(new InvalidParamsError("bad task id")); } else { - response = jsonRestHandler.createTaskPushNotificationConfiguration(context, extractTenant(rc), body, taskId); + response = handler.createTaskPushNotificationConfiguration(context, extractTenant(rc), body, taskId); } } catch (Throwable t) { LOG.error("Internal error while processing request", t); - response = jsonRestHandler.createErrorResponse(new InternalError("Internal error")); + response = handler.createErrorResponse(new InternalError("Internal error")); } finally { sendResponse(rc, response); } @@ -626,24 +662,25 @@ public void createTaskPushNotificationConfiguration(String body, RoutingContext *

    URL Pattern: {@code /tasks/{taskId}/pushNotificationConfigs/{configId}} * * @param rc the Vert.x routing context (taskId and configId extracted from path) + * @param handler the handler for the agent this request is addressed to */ @Authenticated - public void getTaskPushNotificationConfiguration(RoutingContext rc) { + public void getTaskPushNotificationConfiguration(RoutingContext rc, RestHandler handler) { String taskId = rc.pathParam("taskId"); String configId = rc.pathParam("configId"); ServerCallContext context = createCallContext(rc, GET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD); HTTPRestResponse response = null; try { if (taskId == null || taskId.isEmpty()) { - response = jsonRestHandler.createErrorResponse(new InvalidParamsError("bad task id")); - } else if (configId == null || configId.isEmpty()) { - response = jsonRestHandler.createErrorResponse(new InvalidParamsError("bad configuration id")); + response = handler.createErrorResponse(new InvalidParamsError("bad task id")); + } else if (configId == null || configId.isEmpty()) { + response = handler.createErrorResponse(new InvalidParamsError("bad configuration id")); }else { - response = jsonRestHandler.getTaskPushNotificationConfiguration(context, extractTenant(rc), taskId, configId); + response = handler.getTaskPushNotificationConfiguration(context, extractTenant(rc), taskId, configId); } } catch (Throwable t) { LOG.error("Internal error while processing request", t); - response = jsonRestHandler.createErrorResponse(new InternalError("Internal error")); + response = handler.createErrorResponse(new InternalError("Internal error")); } finally { sendResponse(rc, response); } @@ -665,15 +702,16 @@ public void getTaskPushNotificationConfiguration(RoutingContext rc) { * * * @param rc the Vert.x routing context (taskId extracted from path) + * @param handler the handler for the agent this request is addressed to */ @Authenticated - public void listTaskPushNotificationConfigurations(RoutingContext rc) { + public void listTaskPushNotificationConfigurations(RoutingContext rc, RestHandler handler) { String taskId = rc.pathParam("taskId"); ServerCallContext context = createCallContext(rc, LIST_TASK_PUSH_NOTIFICATION_CONFIG_METHOD); HTTPRestResponse response = null; try { if (taskId == null || taskId.isEmpty()) { - response = jsonRestHandler.createErrorResponse(new InvalidParamsError("bad task id")); + response = handler.createErrorResponse(new InvalidParamsError("bad task id")); } else { int pageSize = 0; if (rc.request().params().contains(PAGE_SIZE_PARAM)) { @@ -683,13 +721,13 @@ public void listTaskPushNotificationConfigurations(RoutingContext rc) { if (rc.request().params().contains(PAGE_TOKEN_PARAM)) { pageToken = Utils.defaultIfNull(rc.request().params().get(PAGE_TOKEN_PARAM), ""); } - response = jsonRestHandler.listTaskPushNotificationConfigurations(context, extractTenant(rc), taskId, pageSize, pageToken); + response = handler.listTaskPushNotificationConfigurations(context, extractTenant(rc), taskId, pageSize, pageToken); } } catch (NumberFormatException e) { - response = jsonRestHandler.createErrorResponse(new InvalidParamsError("bad " + PAGE_SIZE_PARAM)); + response = handler.createErrorResponse(new InvalidParamsError("bad " + PAGE_SIZE_PARAM)); } catch (Throwable t) { LOG.error("Internal error while processing request", t); - response = jsonRestHandler.createErrorResponse(new InternalError("Internal error")); + response = handler.createErrorResponse(new InternalError("Internal error")); } finally { sendResponse(rc, response); } @@ -706,24 +744,25 @@ public void listTaskPushNotificationConfigurations(RoutingContext rc) { *

    Response: HTTP 204 No Content on success * * @param rc the Vert.x routing context (taskId and configId extracted from path) + * @param handler the handler for the agent this request is addressed to */ @Authenticated - public void deleteTaskPushNotificationConfiguration(RoutingContext rc) { + public void deleteTaskPushNotificationConfiguration(RoutingContext rc, RestHandler handler) { String taskId = rc.pathParam("taskId"); String configId = rc.pathParam("configId"); ServerCallContext context = createCallContext(rc, DELETE_TASK_PUSH_NOTIFICATION_CONFIG_METHOD); HTTPRestResponse response = null; try { if (taskId == null || taskId.isEmpty()) { - response = jsonRestHandler.createErrorResponse(new InvalidParamsError("bad task id")); + response = handler.createErrorResponse(new InvalidParamsError("bad task id")); } else if (configId == null || configId.isEmpty()) { - response = jsonRestHandler.createErrorResponse(new InvalidParamsError("bad config id")); + response = handler.createErrorResponse(new InvalidParamsError("bad config id")); } else { - response = jsonRestHandler.deleteTaskPushNotificationConfiguration(context, extractTenant(rc), taskId, configId); + response = handler.deleteTaskPushNotificationConfiguration(context, extractTenant(rc), taskId, configId); } } catch (Throwable t) { LOG.error("Internal error while processing request", t); - response = jsonRestHandler.createErrorResponse(new InternalError("Internal error")); + response = handler.createErrorResponse(new InternalError("Internal error")); } finally { sendResponse(rc, response); } @@ -772,10 +811,10 @@ private String extractTenant(RoutingContext rc) { * @param rc the routing context * @return true if the content type is application/json - false otherwise. */ - private boolean validateContentType(RoutingContext rc) { + private boolean validateContentType(RoutingContext rc, RestHandler handler) { String contentType = rc.request().getHeader(CONTENT_TYPE); if (contentType == null || !contentType.trim().startsWith(APPLICATION_JSON)) { - sendResponse(rc, jsonRestHandler.createErrorResponse(new ContentTypeNotSupportedError(null, null, null))); + sendResponse(rc, handler.createErrorResponse(new ContentTypeNotSupportedError(null, null, null))); return false; } return true; @@ -790,14 +829,14 @@ private boolean validateContentType(RoutingContext rc) { * @param body the request body (may be null or empty) * @return true if validation passes, false if Content-Type error should be returned */ - private boolean validateContentTypeForOptionalBody(RoutingContext rc, @Nullable String body) { + private boolean validateContentTypeForOptionalBody(RoutingContext rc, @Nullable String body, RestHandler handler) { // If body is null or empty, Content-Type is not required if (body == null || body.isBlank()) { return true; } // Body has content - validate Content-Type - return validateContentType(rc); + return validateContentType(rc, handler); } /** @@ -820,10 +859,11 @@ private boolean validateContentTypeForOptionalBody(RoutingContext rc, @Nullable * * * @param rc the Vert.x routing context + * @param handler the handler for the agent this request is addressed to */ @PermitAll - public void getAgentCard(RoutingContext rc) { - HTTPRestResponse response = jsonRestHandler.getAgentCard(); + public void getAgentCard(RoutingContext rc, RestHandler handler) { + HTTPRestResponse response = handler.getAgentCard(); sendResponse(rc, response); } @@ -838,10 +878,11 @@ public void getAgentCard(RoutingContext rc) { *

    Authentication: Required (inherits {@code @Authenticated} from class) * * @param rc the Vert.x routing context + * @param handler the handler for the agent this request is addressed to */ @Authenticated - public void getExtendedAgentCard(RoutingContext rc) { - HTTPRestResponse response = jsonRestHandler.getExtendedAgentCard(createCallContext(rc, GET_EXTENDED_AGENT_CARD_METHOD), extractTenant(rc)); + public void getExtendedAgentCard(RoutingContext rc, RestHandler handler) { + HTTPRestResponse response = handler.getExtendedAgentCard(createCallContext(rc, GET_EXTENDED_AGENT_CARD_METHOD), extractTenant(rc)); sendResponse(rc, response); } @@ -855,9 +896,10 @@ public void getExtendedAgentCard(RoutingContext rc) { * instead of generic 404 HTML pages. * * @param rc the Vert.x routing context + * @param handler the handler for the agent this request is addressed to */ - public void methodNotFoundMessage(RoutingContext rc) { - HTTPRestResponse response = jsonRestHandler.createErrorResponse(new MethodNotFoundError()); + public void methodNotFoundMessage(RoutingContext rc, RestHandler handler) { + HTTPRestResponse response = handler.createErrorResponse(new MethodNotFoundError()); sendResponse(rc, response); } diff --git a/reference/rest/src/main/java/org/a2aproject/sdk/server/rest/quarkus/registry/MultiAgentRegistry.java b/reference/rest/src/main/java/org/a2aproject/sdk/server/rest/quarkus/registry/MultiAgentRegistry.java new file mode 100644 index 000000000..78e1962a5 --- /dev/null +++ b/reference/rest/src/main/java/org/a2aproject/sdk/server/rest/quarkus/registry/MultiAgentRegistry.java @@ -0,0 +1,16 @@ +package org.a2aproject.sdk.server.rest.quarkus.registry; + +import java.util.Map; +import org.a2aproject.sdk.transport.rest.handler.RestHandler; + +/** + * Registry for supporting multiple agents in a single Quarkus REST application. + * If a CDI bean implements this interface, the server will register routes for each + * agent in the registry under {@code //} and {@code //.well-known/agent-card.json}. + */ +public interface MultiAgentRegistry { + /** + * @return a map of agent ID (path segment) to their RestHandler + */ + Map getAgents(); +} diff --git a/reference/rest/src/test/java/org/a2aproject/sdk/server/rest/quarkus/A2AServerRoutesTest.java b/reference/rest/src/test/java/org/a2aproject/sdk/server/rest/quarkus/A2AServerRoutesTest.java index 6b908644b..c4a45a8aa 100644 --- a/reference/rest/src/test/java/org/a2aproject/sdk/server/rest/quarkus/A2AServerRoutesTest.java +++ b/reference/rest/src/test/java/org/a2aproject/sdk/server/rest/quarkus/A2AServerRoutesTest.java @@ -28,6 +28,7 @@ import jakarta.enterprise.inject.Instance; import org.a2aproject.sdk.server.ServerCallContext; +import org.a2aproject.sdk.server.rest.quarkus.registry.MultiAgentRegistry; import org.a2aproject.sdk.spec.ContentTypeNotSupportedError; import org.a2aproject.sdk.transport.rest.handler.RestHandler; import org.a2aproject.sdk.transport.rest.handler.RestHandler.HTTPRestResponse; @@ -71,8 +72,16 @@ public void setUp() { mockParams = MultiMap.caseInsensitiveMultiMap(); mockRequestBody = mock(RequestBody.class); + Instance mockJsonRestHandlerInstance = mock(Instance.class); + when(mockJsonRestHandlerInstance.isResolvable()).thenReturn(true); + when(mockJsonRestHandlerInstance.get()).thenReturn(mockRestHandler); + + Instance mockMultiAgentRegistry = mock(Instance.class); + when(mockMultiAgentRegistry.isResolvable()).thenReturn(false); + // Inject mocks via reflection since we can't use @InjectMocks - setField(routes, "jsonRestHandler", mockRestHandler); + setField(routes, "jsonRestHandler", mockJsonRestHandlerInstance); + setField(routes, "multiAgentRegistry", mockMultiAgentRegistry); setField(routes, "executor", mockExecutor); setField(routes, "callContextFactory", mockCallContextFactory); @@ -104,7 +113,7 @@ public void testSendMessage_MethodNameSetInContext() { ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.sendMessage("{}", mockRoutingContext); + routes.sendMessage("{}", mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).sendMessage(contextCaptor.capture(), anyString(), eq("{}")); @@ -126,7 +135,7 @@ public void testSendMessageStreaming_MethodNameSetInContext() { ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.sendMessageStreaming("{}", mockRoutingContext); + routes.sendMessageStreaming("{}", mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).sendStreamingMessage(contextCaptor.capture(), anyString(), eq("{}")); @@ -148,7 +157,7 @@ public void testGetTask_MethodNameSetInContext() { ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.getTask(mockRoutingContext); + routes.getTask(mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).getTask(contextCaptor.capture(), anyString(), eq("task123"), any()); @@ -170,7 +179,7 @@ public void testCancelTask_MethodNameSetInContext() { ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.cancelTask("{\"id\":\"task123\"}", mockRoutingContext); + routes.cancelTask("{\"id\":\"task123\"}", mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).cancelTask(contextCaptor.capture(), anyString(), eq("{\"id\":\"task123\"}"), eq("task123")); @@ -203,7 +212,7 @@ public void testCancelTask_WithMetadata() { .thenReturn(mockHttpResponse); // Act - routes.cancelTask(requestBody, mockRoutingContext); + routes.cancelTask(requestBody, mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).cancelTask(any(ServerCallContext.class), anyString(), bodyCaptor.capture(), eq("task456")); @@ -232,7 +241,7 @@ public void testCancelTask_WithEmptyMetadata() { .thenReturn(mockHttpResponse); // Act - routes.cancelTask(requestBody, mockRoutingContext); + routes.cancelTask(requestBody, mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).cancelTask(any(ServerCallContext.class), anyString(), bodyCaptor.capture(), eq("task789")); @@ -257,7 +266,7 @@ public void testCancelTask_WithNoMetadataField() { .thenReturn(mockHttpResponse); // Act - routes.cancelTask(requestBody, mockRoutingContext); + routes.cancelTask(requestBody, mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).cancelTask(any(ServerCallContext.class), anyString(), bodyCaptor.capture(), eq("task999")); @@ -280,7 +289,7 @@ public void testCancelTask_WithNullBody() { .thenReturn(mockHttpResponse); // Act - routes.cancelTask(null, mockRoutingContext); + routes.cancelTask(null, mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).cancelTask(any(ServerCallContext.class), anyString(), bodyCaptor.capture(), eq("task111")); @@ -317,7 +326,7 @@ public void testCancelTask_WithComplexMetadata() { .thenReturn(mockHttpResponse); // Act - routes.cancelTask(requestBody, mockRoutingContext); + routes.cancelTask(requestBody, mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).cancelTask(any(ServerCallContext.class), anyString(), bodyCaptor.capture(), eq("task222")); @@ -340,7 +349,7 @@ public void testSubscribeTask_MethodNameSetInContext() { ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.subscribeToTask(mockRoutingContext); + routes.subscribeToTask(mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).subscribeToTask(contextCaptor.capture(), anyString(), eq("task123")); @@ -362,7 +371,7 @@ public void testCreateTaskPushNotificationConfiguration_MethodNameSetInContext() ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.createTaskPushNotificationConfiguration("{}", mockRoutingContext); + routes.createTaskPushNotificationConfiguration("{}", mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).createTaskPushNotificationConfiguration(contextCaptor.capture(), anyString(), eq("{}"), eq("task123")); @@ -385,7 +394,7 @@ public void testGetTaskPushNotificationConfiguration_MethodNameSetInContext() { ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.getTaskPushNotificationConfiguration(mockRoutingContext); + routes.getTaskPushNotificationConfiguration(mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).getTaskPushNotificationConfiguration(contextCaptor.capture(), anyString(), eq("task123"), @@ -409,7 +418,7 @@ public void testListTaskPushNotificationConfigurations_MethodNameSetInContext() ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.listTaskPushNotificationConfigurations(mockRoutingContext); + routes.listTaskPushNotificationConfigurations(mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).listTaskPushNotificationConfigurations(contextCaptor.capture(), anyString(), eq("task123"), anyInt(), anyString()); @@ -432,7 +441,7 @@ public void testDeleteTaskPushNotificationConfiguration_MethodNameSetInContext() ArgumentCaptor contextCaptor = ArgumentCaptor.forClass(ServerCallContext.class); // Act - routes.deleteTaskPushNotificationConfiguration(mockRoutingContext); + routes.deleteTaskPushNotificationConfiguration(mockRoutingContext, mockRestHandler); // Assert verify(mockRestHandler).deleteTaskPushNotificationConfiguration(contextCaptor.capture(), anyString(), eq("task123"), @@ -453,7 +462,7 @@ public void testSendMessage_UnsupportedContentType_ReturnsContentTypeNotSupporte when(mockRequest.getHeader(any(CharSequence.class))).thenReturn("text/plain"); // Act - routes.sendMessage("{}", mockRoutingContext); + routes.sendMessage("{}", mockRoutingContext, mockRestHandler); // Assert: createErrorResponse called with ContentTypeNotSupportedError, sendMessage NOT called verify(mockRestHandler).createErrorResponse(any(ContentTypeNotSupportedError.class)); @@ -471,7 +480,7 @@ public void testSendMessageStreaming_UnsupportedContentType_ReturnsContentTypeNo when(mockRequest.getHeader(any(CharSequence.class))).thenReturn("text/plain"); // Act - routes.sendMessageStreaming("{}", mockRoutingContext); + routes.sendMessageStreaming("{}", mockRoutingContext, mockRestHandler); // Assert: createErrorResponse called with ContentTypeNotSupportedError, sendStreamingMessage NOT called verify(mockRestHandler).createErrorResponse(any(ContentTypeNotSupportedError.class)); @@ -490,7 +499,7 @@ public void testSendMessage_UnsupportedProtocolVersion_ReturnsVersionNotSupporte .thenReturn(mockErrorResponse); // Act - routes.sendMessage("{}", mockRoutingContext); + routes.sendMessage("{}", mockRoutingContext, mockRestHandler); // Assert: sendMessage was called and error response forwarded verify(mockRestHandler).sendMessage(any(ServerCallContext.class), anyString(), eq("{}"));