From 8dfa1f2ecc710f4dc51a8144ea3d0dbf1f5b2688 Mon Sep 17 00:00:00 2001 From: meraklbz Date: Mon, 10 Aug 2026 20:12:07 +0800 Subject: [PATCH 1/3] fix: validate A2A version and extensions on all gRPC handler methods --- .../transport/grpc/handler/GrpcHandler.java | 19 ++++++++ .../grpc/handler/GrpcHandlerTest.java | 46 +++++++++++++++++++ 2 files changed, 65 insertions(+) diff --git a/transport/grpc/src/main/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandler.java b/transport/grpc/src/main/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandler.java index 70185f32d..ea451b7aa 100644 --- a/transport/grpc/src/main/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandler.java +++ b/transport/grpc/src/main/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandler.java @@ -220,6 +220,8 @@ public void getTask(org.a2aproject.sdk.grpc.GetTaskRequest request, StreamObserver responseObserver) { try { ServerCallContext context = createCallContext(responseObserver); + A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); + A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); TaskQueryParams params = FromProto.taskQueryParams(request); Task task = getRequestHandler().onGetTask(params, context); if (task != null) { @@ -242,6 +244,8 @@ public void listTasks(org.a2aproject.sdk.grpc.ListTasksRequest request, StreamObserver responseObserver) { try { ServerCallContext context = createCallContext(responseObserver); + A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); + A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); org.a2aproject.sdk.spec.ListTasksParams params = FromProto.listTasksParams(request); ListTasksResult result = getRequestHandler().onListTasks(params, context); responseObserver.onNext(ToProto.listTasksResult(result)); @@ -260,6 +264,8 @@ public void cancelTask(org.a2aproject.sdk.grpc.CancelTaskRequest request, StreamObserver responseObserver) { try { ServerCallContext context = createCallContext(responseObserver); + A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); + A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); CancelTaskParams params = FromProto.cancelTaskParams(request); Task task = getRequestHandler().onCancelTask(params, context); if (task != null) { @@ -287,6 +293,8 @@ public void createTaskPushNotificationConfig(org.a2aproject.sdk.grpc.TaskPushNot try { ServerCallContext context = createCallContext(responseObserver); + A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); + A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); TaskPushNotificationConfig config = FromProto.createTaskPushNotificationConfig(request); TaskPushNotificationConfig responseConfig = getRequestHandler().onCreateTaskPushNotificationConfig(config, context); responseObserver.onNext(ToProto.taskPushNotificationConfig(responseConfig)); @@ -310,6 +318,8 @@ public void getTaskPushNotificationConfig(org.a2aproject.sdk.grpc.GetTaskPushNot try { ServerCallContext context = createCallContext(responseObserver); + A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); + A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); GetTaskPushNotificationConfigParams params = FromProto.getTaskPushNotificationConfigParams(request); TaskPushNotificationConfig config = getRequestHandler().onGetTaskPushNotificationConfig(params, context); responseObserver.onNext(ToProto.taskPushNotificationConfig(config)); @@ -333,6 +343,8 @@ public void listTaskPushNotificationConfigs(org.a2aproject.sdk.grpc.ListTaskPush try { ServerCallContext context = createCallContext(responseObserver); + A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); + A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); ListTaskPushNotificationConfigsParams params = FromProto.listTaskPushNotificationConfigsParams(request); ListTaskPushNotificationConfigsResult result = getRequestHandler().onListTaskPushNotificationConfigs(params, context); org.a2aproject.sdk.grpc.ListTaskPushNotificationConfigsResponse response = ToProto.listTaskPushNotificationConfigsResponse(result); @@ -420,6 +432,8 @@ public void subscribeToTask(org.a2aproject.sdk.grpc.SubscribeToTaskRequest reque try { ServerCallContext context = createCallContext(responseObserver); installForkedContextWrapper(context); + A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); + A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); TaskIdParams params = FromProto.taskIdParams(request); Flow.Publisher publisher = getRequestHandler().onSubscribeToTask(params, context); convertToStreamResponse(publisher, responseObserver, context); @@ -568,6 +582,9 @@ public void getExtendedAgentCard(org.a2aproject.sdk.grpc.GetExtendedAgentCardReq handleError(responseObserver, new UnsupportedOperationError()); return; } + ServerCallContext context = createCallContext(responseObserver); + A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); + A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); AgentCard extendedAgentCard = getExtendedAgentCard(); if (extendedAgentCard != null) { responseObserver.onNext(ToProto.agentCard(extendedAgentCard)); @@ -591,6 +608,8 @@ public void deleteTaskPushNotificationConfig(org.a2aproject.sdk.grpc.DeleteTaskP try { ServerCallContext context = createCallContext(responseObserver); + A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); + A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); DeleteTaskPushNotificationConfigParams params = FromProto.deleteTaskPushNotificationConfigParams(request); getRequestHandler().onDeleteTaskPushNotificationConfig(params, context); // void response diff --git a/transport/grpc/src/test/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandlerTest.java b/transport/grpc/src/test/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandlerTest.java index 66c620164..a4db08607 100644 --- a/transport/grpc/src/test/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandlerTest.java +++ b/transport/grpc/src/test/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandlerTest.java @@ -1073,6 +1073,52 @@ public ServerCallContext create(StreamObserver streamObserver) { assertGrpcError(streamRecorder, Status.Code.UNIMPLEMENTED); } + @Test + public void testVersionNotSupportedErrorOnGetTask() throws Exception { + // Regression test for BUG-33: getTask previously skipped A2A protocol version + // and extension validation, unlike sendMessage/sendStreamingMessage. + AgentCard agentCard = AgentCard.builder() + .name("test-card") + .description("Test card with version 1.0") + .supportedInterfaces(Collections.singletonList(new AgentInterface("GRPC", "http://localhost:9999"))) + .version("1.0.0") + .capabilities(AgentCapabilities.builder() + .streaming(true) + .pushNotifications(false) + .build()) + .defaultInputModes(List.of("text")) + .defaultOutputModes(List.of("text")) + .skills(List.of()) + .build(); + + // Create handler that provides incompatible version 2.0 in the context + GrpcHandler handler = new TestGrpcHandler(agentCard, requestHandler, internalExecutor) { + @Override + protected CallContextFactory getCallContextFactory() { + return new CallContextFactory() { + @Override + public ServerCallContext create(StreamObserver streamObserver) { + return new ServerCallContext( + UnauthenticatedUser.INSTANCE, + Map.of("grpc_response_observer", streamObserver), + new HashSet<>(), + "2.0" // Incompatible version + ); + } + }; + } + }; + + GetTaskRequest request = GetTaskRequest.newBuilder() + .setId(AbstractA2ARequestHandlerTest.MINIMAL_TASK.id()) + .build(); + StreamRecorder streamRecorder = StreamRecorder.create(); + handler.getTask(request, streamRecorder); + streamRecorder.awaitCompletion(5, TimeUnit.SECONDS); + + assertGrpcError(streamRecorder, Status.Code.UNIMPLEMENTED); + } + @Test public void testCompatibleVersionSuccess() throws Exception { // Create AgentCard with protocol version 1.0 From 972e4a786d1baae6be51b4c37b8c288a0e7c835a Mon Sep 17 00:00:00 2001 From: meraklbz Date: Tue, 11 Aug 2026 21:31:46 +0800 Subject: [PATCH 2/3] refactor: centralize A2A version/extension validation in createCallContext MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move validateProtocolVersion + validateRequiredExtensions into createCallContext() so every gRPC handler method validates through one path, per review suggestion. The streaming methods still install the forked context wrapper after createCallContext() returns; the wrapper is independent of the validation (it only attaches a forked gRPC Context around agent-executor runnables), so validating first is safe — and a failed validation now skips the fork entirely. --- .../transport/grpc/handler/GrpcHandler.java | 31 +++++-------------- 1 file changed, 7 insertions(+), 24 deletions(-) diff --git a/transport/grpc/src/main/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandler.java b/transport/grpc/src/main/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandler.java index ea451b7aa..d2ce6e720 100644 --- a/transport/grpc/src/main/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandler.java +++ b/transport/grpc/src/main/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandler.java @@ -199,8 +199,6 @@ public void sendMessage(org.a2aproject.sdk.grpc.SendMessageRequest request, StreamObserver responseObserver) { try { ServerCallContext context = createCallContext(responseObserver); - A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); - A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); MessageSendParams params = FromProto.messageSendParams(request); EventKind taskOrMessage = getRequestHandler().onMessageSend(params, context); org.a2aproject.sdk.grpc.SendMessageResponse response = ToProto.taskOrMessage(taskOrMessage); @@ -220,8 +218,6 @@ public void getTask(org.a2aproject.sdk.grpc.GetTaskRequest request, StreamObserver responseObserver) { try { ServerCallContext context = createCallContext(responseObserver); - A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); - A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); TaskQueryParams params = FromProto.taskQueryParams(request); Task task = getRequestHandler().onGetTask(params, context); if (task != null) { @@ -244,8 +240,6 @@ public void listTasks(org.a2aproject.sdk.grpc.ListTasksRequest request, StreamObserver responseObserver) { try { ServerCallContext context = createCallContext(responseObserver); - A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); - A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); org.a2aproject.sdk.spec.ListTasksParams params = FromProto.listTasksParams(request); ListTasksResult result = getRequestHandler().onListTasks(params, context); responseObserver.onNext(ToProto.listTasksResult(result)); @@ -264,8 +258,6 @@ public void cancelTask(org.a2aproject.sdk.grpc.CancelTaskRequest request, StreamObserver responseObserver) { try { ServerCallContext context = createCallContext(responseObserver); - A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); - A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); CancelTaskParams params = FromProto.cancelTaskParams(request); Task task = getRequestHandler().onCancelTask(params, context); if (task != null) { @@ -293,8 +285,6 @@ public void createTaskPushNotificationConfig(org.a2aproject.sdk.grpc.TaskPushNot try { ServerCallContext context = createCallContext(responseObserver); - A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); - A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); TaskPushNotificationConfig config = FromProto.createTaskPushNotificationConfig(request); TaskPushNotificationConfig responseConfig = getRequestHandler().onCreateTaskPushNotificationConfig(config, context); responseObserver.onNext(ToProto.taskPushNotificationConfig(responseConfig)); @@ -318,8 +308,6 @@ public void getTaskPushNotificationConfig(org.a2aproject.sdk.grpc.GetTaskPushNot try { ServerCallContext context = createCallContext(responseObserver); - A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); - A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); GetTaskPushNotificationConfigParams params = FromProto.getTaskPushNotificationConfigParams(request); TaskPushNotificationConfig config = getRequestHandler().onGetTaskPushNotificationConfig(params, context); responseObserver.onNext(ToProto.taskPushNotificationConfig(config)); @@ -343,8 +331,6 @@ public void listTaskPushNotificationConfigs(org.a2aproject.sdk.grpc.ListTaskPush try { ServerCallContext context = createCallContext(responseObserver); - A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); - A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); ListTaskPushNotificationConfigsParams params = FromProto.listTaskPushNotificationConfigsParams(request); ListTaskPushNotificationConfigsResult result = getRequestHandler().onListTaskPushNotificationConfigs(params, context); org.a2aproject.sdk.grpc.ListTaskPushNotificationConfigsResponse response = ToProto.listTaskPushNotificationConfigsResponse(result); @@ -407,8 +393,6 @@ public void sendStreamingMessage(org.a2aproject.sdk.grpc.SendMessageRequest requ try { ServerCallContext context = createCallContext(responseObserver); installForkedContextWrapper(context); - A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); - A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); MessageSendParams params = FromProto.messageSendParams(request); Flow.Publisher publisher = getRequestHandler().onMessageSendStream(params, context); convertToStreamResponse(publisher, responseObserver, context); @@ -432,8 +416,6 @@ public void subscribeToTask(org.a2aproject.sdk.grpc.SubscribeToTaskRequest reque try { ServerCallContext context = createCallContext(responseObserver); installForkedContextWrapper(context); - A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); - A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); TaskIdParams params = FromProto.taskIdParams(request); Flow.Publisher publisher = getRequestHandler().onSubscribeToTask(params, context); convertToStreamResponse(publisher, responseObserver, context); @@ -583,8 +565,6 @@ public void getExtendedAgentCard(org.a2aproject.sdk.grpc.GetExtendedAgentCardReq return; } ServerCallContext context = createCallContext(responseObserver); - A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); - A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); AgentCard extendedAgentCard = getExtendedAgentCard(); if (extendedAgentCard != null) { responseObserver.onNext(ToProto.agentCard(extendedAgentCard)); @@ -608,8 +588,6 @@ public void deleteTaskPushNotificationConfig(org.a2aproject.sdk.grpc.DeleteTaskP try { ServerCallContext context = createCallContext(responseObserver); - A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); - A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); DeleteTaskPushNotificationConfigParams params = FromProto.deleteTaskPushNotificationConfigParams(request); getRequestHandler().onDeleteTaskPushNotificationConfig(params, context); // void response @@ -662,6 +640,7 @@ public void deleteTaskPushNotificationConfig(org.a2aproject.sdk.grpc.DeleteTaskP */ private ServerCallContext createCallContext(StreamObserver responseObserver) { CallContextFactory factory = getCallContextFactory(); + ServerCallContext context; if (factory == null) { // Default implementation when no custom CallContextFactory is provided // This handles both CDI injection scenarios and test scenarios where callContextFactory is null @@ -718,12 +697,16 @@ private ServerCallContext createCallContext(StreamObserver responseObserv requestedExtensions = A2AExtensions.getRequestedExtensions(List.of(extensionsHeader)); } - return new ServerCallContext(user, state, requestedExtensions, requestedVersion); + context = new ServerCallContext(user, state, requestedExtensions, requestedVersion); } else { // TODO: CallContextFactory interface expects ServerCall + Metadata, but we only have StreamObserver // This is another manifestation of the architectural limitation mentioned above - return factory.create(responseObserver); // Fall back to basic create() method for now + context = factory.create(responseObserver); // Fall back to basic create() method for now } + + A2AVersionValidator.validateProtocolVersion(getAgentCardInternal(), context); + A2AExtensions.validateRequiredExtensions(getAgentCardInternal(), context); + return context; } /** From 1ab00f08734056939369858e8be683f5f6a9813e Mon Sep 17 00:00:00 2001 From: meraklbz Date: Tue, 11 Aug 2026 21:37:30 +0800 Subject: [PATCH 3/3] chore: remove internal tracking ids from comments --- .../a2aproject/sdk/transport/grpc/handler/GrpcHandlerTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/transport/grpc/src/test/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandlerTest.java b/transport/grpc/src/test/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandlerTest.java index a4db08607..2578340ee 100644 --- a/transport/grpc/src/test/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandlerTest.java +++ b/transport/grpc/src/test/java/org/a2aproject/sdk/transport/grpc/handler/GrpcHandlerTest.java @@ -1075,7 +1075,7 @@ public ServerCallContext create(StreamObserver streamObserver) { @Test public void testVersionNotSupportedErrorOnGetTask() throws Exception { - // Regression test for BUG-33: getTask previously skipped A2A protocol version + // Regression test: getTask previously skipped A2A protocol version // and extension validation, unlike sendMessage/sendStreamingMessage. AgentCard agentCard = AgentCard.builder() .name("test-card")