Skip to content

Commit 3e8a927

Browse files
kabirkabir
committed
Self review and fixes
1 parent eae0a18 commit 3e8a927

9 files changed

Lines changed: 212 additions & 37 deletions

File tree

‎client/base/src/main/java/org/a2aproject/sdk/A2A.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -290,6 +290,11 @@ public static AgentCard getAgentCard(String agentUrl) throws A2AClientError, A2A
290290
return getAgentCard(A2AHttpClientFactory.create(), agentUrl);
291291
}
292292

293+
/**
294+
* Retrieves an agent card while explicitly selecting the protocol versions to accept.
295+
* The set must be non-empty; v0.3 discovery additionally requires the optional compatibility
296+
* parser and a matching binding adapter.
297+
*/
293298
public static AgentCard getAgentCard(String agentUrl, Set<String> supportedProtocolVersions)
294299
throws A2AClientError, A2AClientJSONError {
295300
return getAgentCard(A2AHttpClientFactory.create(), agentUrl, supportedProtocolVersions);
@@ -321,6 +326,9 @@ public static AgentCard getAgentCard(A2AHttpClient httpClient, String agentUrl)
321326
return getAgentCard(httpClient, agentUrl, null, null);
322327
}
323328

329+
/**
330+
* Retrieves an agent card with a custom HTTP client and explicit protocol-version policy.
331+
*/
324332
public static AgentCard getAgentCard(A2AHttpClient httpClient, String agentUrl,
325333
Set<String> supportedProtocolVersions) throws A2AClientError, A2AClientJSONError {
326334
return getAgentCard(httpClient, agentUrl, null, null, supportedProtocolVersions);
@@ -371,6 +379,9 @@ public static AgentCard getAgentCard(String agentUrl, String relativeCardPath, M
371379
return getAgentCard(A2AHttpClientFactory.create(), agentUrl, relativeCardPath, authHeaders);
372380
}
373381

382+
/**
383+
* Retrieves an agent card with custom endpoint/authentication settings and explicit protocol versions.
384+
*/
374385
public static AgentCard getAgentCard(String agentUrl, String relativeCardPath,
375386
Map<String, String> authHeaders, Set<String> supportedProtocolVersions)
376387
throws A2AClientError, A2AClientJSONError {
@@ -414,6 +425,9 @@ public static AgentCard getAgentCard(A2AHttpClient httpClient, String agentUrl,
414425
return getAgentCard(httpClient, agentUrl, relativeCardPath, authHeaders, Set.of("1.0"));
415426
}
416427

428+
/**
429+
* Retrieves an agent card with full HTTP, endpoint, authentication, and protocol-version settings.
430+
*/
417431
public static AgentCard getAgentCard(A2AHttpClient httpClient, String agentUrl, String relativeCardPath,
418432
Map<String, String> authHeaders, Set<String> supportedProtocolVersions)
419433
throws A2AClientError, A2AClientJSONError {

‎compat-0.3/client/adapter-rest/src/main/java/org/a2aproject/sdk/compat03/client/adapter/rest/RestCompat03ClientTransport.java‎

Lines changed: 35 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -41,10 +41,11 @@ public RestCompat03ClientTransport(RestTransport_v0_3 delegate, AgentCard card,
4141
public EventKind sendMessage(MessageSendParams request, @Nullable ClientCallContext context) {
4242
Compat03ClientTransportSupport.validateMessageSend(request);
4343
PayloadAndHeaders payload = apply(A2AMethods.SEND_MESSAGE_METHOD,
44-
ProtoUtils.ToProto.sendMessageRequest(request), org.a2aproject.sdk.grpc.SendMessageRequest.class, context);
44+
org.a2aproject.sdk.grpc.SendMessageRequest.newBuilder(ProtoUtils.ToProto.sendMessageRequest(request)),
45+
org.a2aproject.sdk.grpc.SendMessageRequest.Builder.class, context);
4546
return Compat03ClientTransportSupport.call(() -> Compat03ClientTransportSupport.toV10(delegate.sendMessage(
4647
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.messageSendParams(
47-
(org.a2aproject.sdk.grpc.SendMessageRequest) payload.getPayload())),
48+
((org.a2aproject.sdk.grpc.SendMessageRequest.Builder) payload.getPayload()).build())),
4849
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload)))));
4950
}
5051

@@ -53,10 +54,11 @@ public void sendMessageStreaming(MessageSendParams request, Consumer<StreamingEv
5354
Consumer<Throwable> errors, @Nullable ClientCallContext context) {
5455
Compat03ClientTransportSupport.validateMessageSend(request);
5556
PayloadAndHeaders payload = apply(A2AMethods.SEND_STREAMING_MESSAGE_METHOD,
56-
ProtoUtils.ToProto.sendMessageRequest(request), org.a2aproject.sdk.grpc.SendMessageRequest.class, context);
57+
org.a2aproject.sdk.grpc.SendMessageRequest.newBuilder(ProtoUtils.ToProto.sendMessageRequest(request)),
58+
org.a2aproject.sdk.grpc.SendMessageRequest.Builder.class, context);
5759
Compat03ClientTransportSupport.run(() -> delegate.sendMessageStreaming(
5860
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.messageSendParams(
59-
(org.a2aproject.sdk.grpc.SendMessageRequest) payload.getPayload())),
61+
((org.a2aproject.sdk.grpc.SendMessageRequest.Builder) payload.getPayload()).build())),
6062
event -> events.accept(Compat03ClientTransportSupport.toV10(event)),
6163
Compat03ClientTransportSupport.mapAsyncError(errors),
6264
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload))));
@@ -65,22 +67,24 @@ public void sendMessageStreaming(MessageSendParams request, Consumer<StreamingEv
6567
@Override
6668
public Task getTask(TaskQueryParams request, @Nullable ClientCallContext context) {
6769
Compat03ClientTransportSupport.validateTaskQuery(request);
68-
PayloadAndHeaders payload = apply(A2AMethods.GET_TASK_METHOD, ProtoUtils.ToProto.getTaskRequest(request),
69-
org.a2aproject.sdk.grpc.GetTaskRequest.class, context);
70+
PayloadAndHeaders payload = apply(A2AMethods.GET_TASK_METHOD,
71+
org.a2aproject.sdk.grpc.GetTaskRequest.newBuilder(ProtoUtils.ToProto.getTaskRequest(request)),
72+
org.a2aproject.sdk.grpc.GetTaskRequest.Builder.class, context);
7073
return Compat03ClientTransportSupport.call(() -> TaskMapper_v0_3.INSTANCE.toV10(delegate.getTask(
7174
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.taskQueryParams(
72-
(org.a2aproject.sdk.grpc.GetTaskRequest) payload.getPayload())),
75+
((org.a2aproject.sdk.grpc.GetTaskRequest.Builder) payload.getPayload()).build())),
7376
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload)))));
7477
}
7578

7679
@Override
7780
public Task cancelTask(CancelTaskParams request, @Nullable ClientCallContext context) {
7881
Compat03ClientTransportSupport.validateCancel(request);
79-
PayloadAndHeaders payload = apply(A2AMethods.CANCEL_TASK_METHOD, ProtoUtils.ToProto.cancelTaskRequest(request),
80-
org.a2aproject.sdk.grpc.CancelTaskRequest.class, context);
82+
PayloadAndHeaders payload = apply(A2AMethods.CANCEL_TASK_METHOD,
83+
org.a2aproject.sdk.grpc.CancelTaskRequest.newBuilder(ProtoUtils.ToProto.cancelTaskRequest(request)),
84+
org.a2aproject.sdk.grpc.CancelTaskRequest.Builder.class, context);
8185
return Compat03ClientTransportSupport.call(() -> TaskMapper_v0_3.INSTANCE.toV10(delegate.cancelTask(
8286
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.cancelTaskParams(
83-
(org.a2aproject.sdk.grpc.CancelTaskRequest) payload.getPayload())),
87+
((org.a2aproject.sdk.grpc.CancelTaskRequest.Builder) payload.getPayload()).build())),
8488
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload)))));
8589
}
8690

@@ -89,10 +93,12 @@ public TaskPushNotificationConfig createTaskPushNotificationConfiguration(TaskPu
8993
@Nullable ClientCallContext context) {
9094
Compat03ClientTransportSupport.validatePushConfig(request);
9195
PayloadAndHeaders payload = apply(A2AMethods.SET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD,
92-
ProtoUtils.ToProto.taskPushNotificationConfig(request), org.a2aproject.sdk.grpc.TaskPushNotificationConfig.class, context);
96+
org.a2aproject.sdk.grpc.TaskPushNotificationConfig.newBuilder(ProtoUtils.ToProto.taskPushNotificationConfig(request)),
97+
org.a2aproject.sdk.grpc.TaskPushNotificationConfig.Builder.class, context);
9398
return Compat03ClientTransportSupport.call(() -> Compat03ClientTransportSupport.toV10(
9499
delegate.setTaskPushNotificationConfiguration(Compat03ClientTransportSupport.toV03(
95-
ProtoUtils.FromProto.taskPushNotificationConfig((org.a2aproject.sdk.grpc.TaskPushNotificationConfig) payload.getPayload())),
100+
ProtoUtils.FromProto.taskPushNotificationConfig(
101+
((org.a2aproject.sdk.grpc.TaskPushNotificationConfig.Builder) payload.getPayload()).build())),
96102
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload)))));
97103
}
98104

@@ -101,10 +107,13 @@ public TaskPushNotificationConfig getTaskPushNotificationConfiguration(GetTaskPu
101107
@Nullable ClientCallContext context) {
102108
Compat03ClientTransportSupport.validateGetPush(request);
103109
PayloadAndHeaders payload = apply(A2AMethods.GET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD,
104-
ProtoUtils.ToProto.getTaskPushNotificationConfigRequest(request), org.a2aproject.sdk.grpc.GetTaskPushNotificationConfigRequest.class, context);
110+
org.a2aproject.sdk.grpc.GetTaskPushNotificationConfigRequest.newBuilder(
111+
ProtoUtils.ToProto.getTaskPushNotificationConfigRequest(request)),
112+
org.a2aproject.sdk.grpc.GetTaskPushNotificationConfigRequest.Builder.class, context);
105113
return Compat03ClientTransportSupport.call(() -> Compat03ClientTransportSupport.toV10(
106114
delegate.getTaskPushNotificationConfiguration(Compat03ClientTransportSupport.toV03(
107-
ProtoUtils.FromProto.getTaskPushNotificationConfigParams((org.a2aproject.sdk.grpc.GetTaskPushNotificationConfigRequest) payload.getPayload())),
115+
ProtoUtils.FromProto.getTaskPushNotificationConfigParams(
116+
((org.a2aproject.sdk.grpc.GetTaskPushNotificationConfigRequest.Builder) payload.getPayload()).build())),
108117
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload)))));
109118
}
110119

@@ -113,10 +122,13 @@ public ListTaskPushNotificationConfigsResult listTaskPushNotificationConfigurati
113122
ListTaskPushNotificationConfigsParams request, @Nullable ClientCallContext context) {
114123
Compat03ClientTransportSupport.validatePushList(request);
115124
PayloadAndHeaders payload = apply(A2AMethods.LIST_TASK_PUSH_NOTIFICATION_CONFIG_METHOD,
116-
ProtoUtils.ToProto.listTaskPushNotificationConfigsRequest(request), org.a2aproject.sdk.grpc.ListTaskPushNotificationConfigsRequest.class, context);
125+
org.a2aproject.sdk.grpc.ListTaskPushNotificationConfigsRequest.newBuilder(
126+
ProtoUtils.ToProto.listTaskPushNotificationConfigsRequest(request)),
127+
org.a2aproject.sdk.grpc.ListTaskPushNotificationConfigsRequest.Builder.class, context);
117128
return Compat03ClientTransportSupport.call(() -> Compat03ClientTransportSupport.toV10PushList(
118129
delegate.listTaskPushNotificationConfigurations(Compat03ClientTransportSupport.toV03(
119-
ProtoUtils.FromProto.listTaskPushNotificationConfigsParams((org.a2aproject.sdk.grpc.ListTaskPushNotificationConfigsRequest) payload.getPayload())),
130+
ProtoUtils.FromProto.listTaskPushNotificationConfigsParams(
131+
((org.a2aproject.sdk.grpc.ListTaskPushNotificationConfigsRequest.Builder) payload.getPayload()).build())),
120132
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload)))));
121133
}
122134

@@ -125,10 +137,12 @@ public void deleteTaskPushNotificationConfigurations(DeleteTaskPushNotificationC
125137
@Nullable ClientCallContext context) {
126138
Compat03ClientTransportSupport.validateDeletePush(request);
127139
PayloadAndHeaders payload = apply(A2AMethods.DELETE_TASK_PUSH_NOTIFICATION_CONFIG_METHOD,
128-
ProtoUtils.ToProto.deleteTaskPushNotificationConfigRequest(request), org.a2aproject.sdk.grpc.DeleteTaskPushNotificationConfigRequest.class, context);
140+
org.a2aproject.sdk.grpc.DeleteTaskPushNotificationConfigRequest.newBuilder(
141+
ProtoUtils.ToProto.deleteTaskPushNotificationConfigRequest(request)),
142+
org.a2aproject.sdk.grpc.DeleteTaskPushNotificationConfigRequest.Builder.class, context);
129143
Compat03ClientTransportSupport.run(() -> delegate.deleteTaskPushNotificationConfigurations(
130144
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.deleteTaskPushNotificationConfigParams(
131-
(org.a2aproject.sdk.grpc.DeleteTaskPushNotificationConfigRequest) payload.getPayload())),
145+
((org.a2aproject.sdk.grpc.DeleteTaskPushNotificationConfigRequest.Builder) payload.getPayload()).build())),
132146
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload))));
133147
}
134148

@@ -137,10 +151,11 @@ public void subscribeToTask(TaskIdParams request, Consumer<StreamingEventKind> e
137151
Consumer<Throwable> errors, @Nullable ClientCallContext context) {
138152
Compat03ClientTransportSupport.validateTenant("subscribeToTask", request.tenant());
139153
PayloadAndHeaders payload = apply(A2AMethods.SUBSCRIBE_TO_TASK_METHOD,
140-
ProtoUtils.ToProto.subscribeToTaskRequest(request), org.a2aproject.sdk.grpc.SubscribeToTaskRequest.class, context);
154+
org.a2aproject.sdk.grpc.SubscribeToTaskRequest.newBuilder(ProtoUtils.ToProto.subscribeToTaskRequest(request)),
155+
org.a2aproject.sdk.grpc.SubscribeToTaskRequest.Builder.class, context);
141156
Compat03ClientTransportSupport.run(() -> delegate.resubscribe(
142157
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.taskIdParams(
143-
(org.a2aproject.sdk.grpc.SubscribeToTaskRequest) payload.getPayload())),
158+
((org.a2aproject.sdk.grpc.SubscribeToTaskRequest.Builder) payload.getPayload()).build())),
144159
event -> events.accept(Compat03ClientTransportSupport.toV10(event)),
145160
Compat03ClientTransportSupport.mapAsyncError(errors),
146161
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload))));

‎compat-0.3/client/transport/grpc/src/main/java/org/a2aproject/sdk/compat03/client/transport/grpc/GrpcTransport_v0_3.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import static org.a2aproject.sdk.util.Assert.checkNotNullParam;
44

55
import java.util.List;
6+
import java.util.Locale;
67
import java.util.Map;
78
import java.util.function.Consumer;
89
import java.util.stream.Collectors;
@@ -319,7 +320,8 @@ private Metadata createGrpcMetadata(@Nullable ClientCallContext_v0_3 context, @N
319320
metadata.put(AUTHORIZATION_METADATA_KEY, headerValue);
320321
} else {
321322
// Create a metadata key dynamically for API keys and other custom headers
322-
Metadata.Key<String> metadataKey = Metadata.Key.of(headerName, Metadata.ASCII_STRING_MARSHALLER);
323+
Metadata.Key<String> metadataKey = Metadata.Key.of(
324+
headerName.toLowerCase(Locale.ROOT), Metadata.ASCII_STRING_MARSHALLER);
323325
metadata.put(metadataKey, headerValue);
324326
}
325327
}
@@ -386,4 +388,4 @@ private PayloadAndHeaders_v0_3 applyInterceptors(String methodName, Object paylo
386388
return payloadAndHeaders;
387389
}
388390

389-
}
391+
}

‎compat-0.3/client/transport/jsonrpc/src/main/java/org/a2aproject/sdk/compat03/client/transport/jsonrpc/sse/SSEEventListener_v0_3.java‎

Lines changed: 16 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -29,15 +29,11 @@ public void onMessage(String message, Future<Void> completableFuture) {
2929
try {
3030
handleMessage(JsonParser.parseString(message).getAsJsonObject(), completableFuture);
3131
} catch (JsonSyntaxException e) {
32-
LOGGER.warning("Failed to parse JSON message: " + message);
32+
fail(e, completableFuture);
3333
} catch (JsonProcessingException_v0_3 e) {
34-
LOGGER.warning("Failed to process JSON message: " + message);
34+
fail(e, completableFuture);
3535
} catch (IllegalArgumentException e) {
36-
LOGGER.warning("Invalid message format: " + message);
37-
if (errorHandler != null) {
38-
errorHandler.accept(e);
39-
}
40-
completableFuture.cancel(true); // close SSE channel
36+
fail(e, completableFuture);
4137
}
4238
}
4339

@@ -74,6 +70,9 @@ private void handleMessage(JsonObject jsonObject, Future<Void> future) throws Js
7470
if (errorHandler != null) {
7571
errorHandler.accept(error);
7672
}
73+
if (future != null) {
74+
future.cancel(true); // close SSE channel
75+
}
7776
} else if (jsonObject.has("result")) {
7877
// result can be a Task, Message, TaskStatusUpdateEvent, or TaskArtifactUpdateEvent
7978
String resultJson = jsonObject.get("result").toString();
@@ -87,4 +86,14 @@ private void handleMessage(JsonObject jsonObject, Future<Void> future) throws Js
8786
}
8887
}
8988

89+
private void fail(Throwable throwable, Future<Void> future) {
90+
completed = true;
91+
if (errorHandler != null) {
92+
errorHandler.accept(throwable);
93+
}
94+
if (future != null) {
95+
future.cancel(true);
96+
}
97+
}
98+
9099
}

‎compat-0.3/client/transport/jsonrpc/src/test/java/org/a2aproject/sdk/compat03/client/transport/jsonrpc/sse/SSEEventListener_v0_3_Test.java‎

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -154,7 +154,8 @@ public void testOnEventWithError() throws Exception {
154154
JsonStreamingMessages_v0_3.STREAMING_ERROR_EVENT.indexOf("{"));
155155

156156
// Call onEvent method
157-
listener.onMessage(eventData, null);
157+
CancelCapturingFuture future = new CancelCapturingFuture();
158+
listener.onMessage(eventData, future);
158159

159160
// Verify the error was processed correctly
160161
assertNotNull(receivedError.get());
@@ -163,6 +164,20 @@ public void testOnEventWithError() throws Exception {
163164
assertEquals(-32602, jsonrpcError.getCode());
164165
assertEquals("Invalid parameters", jsonrpcError.getMessage());
165166
assertEquals("Missing required field", jsonrpcError.getData());
167+
assertTrue(future.cancelHandlerCalled);
168+
}
169+
170+
@Test
171+
public void testMalformedEventReportsAndCancels() {
172+
AtomicReference<Throwable> receivedError = new AtomicReference<>();
173+
SSEEventListener_v0_3 listener = new SSEEventListener_v0_3(
174+
event -> {}, receivedError::set);
175+
CancelCapturingFuture future = new CancelCapturingFuture();
176+
177+
listener.onMessage("{not-json", future);
178+
179+
assertNotNull(receivedError.get());
180+
assertTrue(future.cancelHandlerCalled);
166181
}
167182

168183
@Test
@@ -264,4 +279,4 @@ public Void get(long timeout, TimeUnit unit) throws InterruptedException, Execut
264279
return null;
265280
}
266281
}
267-
}
282+
}

‎compat-0.3/client/transport/rest/src/main/java/org/a2aproject/sdk/compat03/client/transport/rest/sse/RestSSEEventListener_v0_3.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,9 +29,10 @@ public void onMessage(String message, @Nullable Future<Void> completableFuture)
2929
LOGGER.fine("Streaming message received: " + message);
3030
org.a2aproject.sdk.compat03.grpc.StreamResponse.Builder builder = org.a2aproject.sdk.compat03.grpc.StreamResponse.newBuilder();
3131
JsonFormat.parser().merge(message, builder);
32-
handleMessage(builder.build());
32+
handleMessage(builder.build(), completableFuture);
3333
} catch (InvalidProtocolBufferException e) {
3434
errorHandler.accept(RestErrorMapper_v0_3.mapRestError(message, 500));
35+
cancel(completableFuture);
3536
}
3637
}
3738

@@ -44,7 +45,7 @@ public void onError(Throwable throwable, @Nullable Future<Void> future) {
4445
}
4546
}
4647

47-
private void handleMessage(StreamResponse response) {
48+
private void handleMessage(StreamResponse response, @Nullable Future<Void> future) {
4849
StreamingEventKind_v0_3 event;
4950
switch (response.getPayloadCase()) {
5051
case MSG ->
@@ -58,10 +59,17 @@ private void handleMessage(StreamResponse response) {
5859
default -> {
5960
LOGGER.warning("Invalid stream response " + response.getPayloadCase());
6061
errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase()));
62+
cancel(future);
6163
return;
6264
}
6365
}
6466
eventHandler.accept(event);
6567
}
6668

69+
private static void cancel(@Nullable Future<Void> future) {
70+
if (future != null) {
71+
future.cancel(true);
72+
}
73+
}
74+
6775
}

0 commit comments

Comments
 (0)