feat(server)!: add multitenancy support with per-tenant routing · a2aproject/a2a-java@3474697 · GitHub
Skip to content

Commit 3474697

Browse files
ehsavoieehsavoie
committed
feat(server)!: add multitenancy support with per-tenant routing
- Add AgentExecutorRouter and AgentCardRouter SPI in server-common - Add @tenant CDI qualifier and CDI router implementations in extras/multitenancy - Wire tenant-aware resolution in DefaultRequestHandler and all transport handlers - Add tenant field to RequestContext for execute/cancel flows - Update GetExtendedAgentCardRequest to carry GetExtendedAgentCardParams Signed-off-by: Emmanuel Hugonnet <ehugonne@redhat.com>
1 parent ef09aeb commit 3474697

62 files changed

Lines changed: 2492 additions & 222 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎boms/extras/pom.xml‎

Lines changed: 5 additions & 0 deletions

‎boms/extras/src/it/extras-usage-test/pom.xml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,10 @@
8888
<groupId>org.a2aproject.sdk</groupId>
8989
<artifactId>a2a-java-queue-manager-replication-mp-reactive</artifactId>
9090
</dependency>
91+
<dependency>
92+
<groupId>org.a2aproject.sdk</groupId>
93+
<artifactId>a2a-java-extras-multitenancy</artifactId>
94+
</dependency>
9195

9296
<!-- Core SDK modules (inherited from SDK BOM via Extras BOM) -->
9397
<dependency>

‎client/transport/grpc/src/main/java/org/a2aproject/sdk/client/transport/grpc/GrpcTransport.java‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@
4848
import org.a2aproject.sdk.spec.TaskIdParams;
4949
import org.a2aproject.sdk.spec.TaskPushNotificationConfig;
5050
import org.a2aproject.sdk.spec.TaskQueryParams;
51+
import org.a2aproject.sdk.spec.util.Utils;
5152
import io.grpc.Channel;
5253
import io.grpc.Metadata;
5354
import io.grpc.StatusException;
@@ -84,6 +85,9 @@ public GrpcTransport(Channel channel, AgentCard agentCard, @Nullable String agen
8485
this.blockingStub = A2AServiceGrpc.newBlockingV2Stub(channel);
8586
this.agentCard = agentCard;
8687
this.interceptors = interceptors;
88+
if (agentTenant != null && !agentTenant.isBlank()) {
89+
Utils.validateTenant(agentTenant);
90+
}
8791
this.agentTenant = agentTenant == null || agentTenant.isBlank() ? "" : agentTenant;
8892
}
8993

@@ -352,11 +356,9 @@ private MessageSendParams createRequestWithTenant(MessageSendParams request) {
352356

353357
@Override
354358
public AgentCard getExtendedAgentCard(GetExtendedAgentCardParams params, @Nullable ClientCallContext context) throws A2AClientException {
355-
GetExtendedAgentCardRequest.Builder builder = GetExtendedAgentCardRequest.newBuilder();
356-
if (params.tenant() != null) {
357-
builder.setTenant(params.tenant());
358-
}
359-
GetExtendedAgentCardRequest request = builder.build();
359+
GetExtendedAgentCardRequest request = GetExtendedAgentCardRequest.newBuilder()
360+
.setTenant(resolveTenant(params.tenant()))
361+
.build();
360362
PayloadAndHeaders payloadAndHeaders = applyInterceptors(GET_EXTENDED_AGENT_CARD_METHOD, request, agentCard, context);
361363

362364
try {

‎client/transport/jsonrpc/src/main/java/org/a2aproject/sdk/client/transport/jsonrpc/JSONRPCTransport.java‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -98,7 +98,7 @@ public EventKind sendMessage(MessageSendParams request, @Nullable ClientCallCont
9898
agentCard, context);
9999

100100
try {
101-
String httpResponseBody = sendPostRequest(Utils.buildBaseUrl(agentInterface, request.tenant()), payloadAndHeaders, SEND_MESSAGE_METHOD);
101+
String httpResponseBody = sendPostRequest(agentInterface.url(), payloadAndHeaders, SEND_MESSAGE_METHOD);
102102
SendMessageResponse response = unmarshalResponse(httpResponseBody, SEND_MESSAGE_METHOD);
103103
return response.getResult();
104104
} catch (A2AClientException e) {
@@ -120,7 +120,7 @@ public void sendMessageStreaming(MessageSendParams request, Consumer<StreamingEv
120120
SSEEventListener sseEventListener = new SSEEventListener(eventConsumer, errorConsumer);
121121

122122
try {
123-
A2AHttpClient.PostBuilder builder = createPostBuilder(Utils.buildBaseUrl(agentInterface, request.tenant()), payloadAndHeaders, SEND_STREAMING_MESSAGE_METHOD);
123+
A2AHttpClient.PostBuilder builder = createPostBuilder(agentInterface.url(), payloadAndHeaders, SEND_STREAMING_MESSAGE_METHOD);
124124
ref.set(builder.postAsyncSSE(
125125
event -> sseEventListener.onMessage(event, ref.get()),
126126
throwable -> sseEventListener.onError(throwable, ref.get()),
@@ -144,7 +144,7 @@ public Task getTask(TaskQueryParams request, @Nullable ClientCallContext context
144144
agentCard, context);
145145

146146
try {
147-
String httpResponseBody = sendPostRequest(Utils.buildBaseUrl(agentInterface, request.tenant()), payloadAndHeaders, GET_TASK_METHOD);
147+
String httpResponseBody = sendPostRequest(agentInterface.url(), payloadAndHeaders, GET_TASK_METHOD);
148148
GetTaskResponse response = unmarshalResponse(httpResponseBody, GET_TASK_METHOD);
149149
return response.getResult();
150150
} catch (A2AClientException e) {
@@ -161,7 +161,7 @@ public Task cancelTask(CancelTaskParams request, @Nullable ClientCallContext con
161161
agentCard, context);
162162

163163
try {
164-
String httpResponseBody = sendPostRequest(Utils.buildBaseUrl(agentInterface, request.tenant()), payloadAndHeaders, CANCEL_TASK_METHOD);
164+
String httpResponseBody = sendPostRequest(agentInterface.url(), payloadAndHeaders, CANCEL_TASK_METHOD);
165165
CancelTaskResponse response = unmarshalResponse(httpResponseBody, CANCEL_TASK_METHOD);
166166
return response.getResult();
167167
} catch (A2AClientException e) {
@@ -177,7 +177,7 @@ public ListTasksResult listTasks(ListTasksParams request, @Nullable ClientCallCo
177177
PayloadAndHeaders payloadAndHeaders = applyInterceptors(LIST_TASK_METHOD, ProtoUtils.ToProto.listTasksParams(request),
178178
agentCard, context);
179179
try {
180-
String httpResponseBody = sendPostRequest(Utils.buildBaseUrl(agentInterface, request.tenant()), payloadAndHeaders, LIST_TASK_METHOD);
180+
String httpResponseBody = sendPostRequest(agentInterface.url(), payloadAndHeaders, LIST_TASK_METHOD);
181181
ListTasksResponse response = unmarshalResponse(httpResponseBody, LIST_TASK_METHOD);
182182
return response.getResult();
183183
} catch (IOException | InterruptedException | JsonProcessingException e) {
@@ -193,7 +193,7 @@ public TaskPushNotificationConfig createTaskPushNotificationConfiguration(TaskPu
193193
ProtoUtils.ToProto.createTaskPushNotificationConfigRequest(request), agentCard, context);
194194

195195
try {
196-
String httpResponseBody = sendPostRequest(Utils.buildBaseUrl(agentInterface, request.tenant()), payloadAndHeaders, SET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
196+
String httpResponseBody = sendPostRequest(agentInterface.url(), payloadAndHeaders, SET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
197197
CreateTaskPushNotificationConfigResponse response = unmarshalResponse(httpResponseBody, SET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
198198
return response.getResult();
199199
} catch (A2AClientException e) {
@@ -211,7 +211,7 @@ public TaskPushNotificationConfig getTaskPushNotificationConfiguration(GetTaskPu
211211
ProtoUtils.ToProto.getTaskPushNotificationConfigRequest(request), agentCard, context);
212212

213213
try {
214-
String httpResponseBody = sendPostRequest(Utils.buildBaseUrl(agentInterface, request.tenant()), payloadAndHeaders, GET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
214+
String httpResponseBody = sendPostRequest(agentInterface.url(), payloadAndHeaders, GET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
215215
GetTaskPushNotificationConfigResponse response = unmarshalResponse(httpResponseBody, GET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
216216
return response.getResult();
217217
} catch (A2AClientException e) {
@@ -230,7 +230,7 @@ public ListTaskPushNotificationConfigsResult listTaskPushNotificationConfigurati
230230
ProtoUtils.ToProto.listTaskPushNotificationConfigsRequest(request), agentCard, context);
231231

232232
try {
233-
String httpResponseBody = sendPostRequest(Utils.buildBaseUrl(agentInterface, request.tenant()), payloadAndHeaders, LIST_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
233+
String httpResponseBody = sendPostRequest(agentInterface.url(), payloadAndHeaders, LIST_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
234234
ListTaskPushNotificationConfigsResponse response = unmarshalResponse(httpResponseBody, LIST_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
235235
return response.getResult();
236236
} catch (A2AClientException e) {
@@ -248,7 +248,7 @@ public void deleteTaskPushNotificationConfigurations(DeleteTaskPushNotificationC
248248
ProtoUtils.ToProto.deleteTaskPushNotificationConfigRequest(request), agentCard, context);
249249

250250
try {
251-
String httpResponseBody = sendPostRequest(Utils.buildBaseUrl(agentInterface, request.tenant()), payloadAndHeaders, DELETE_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
251+
String httpResponseBody = sendPostRequest(agentInterface.url(), payloadAndHeaders, DELETE_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
252252
DeleteTaskPushNotificationConfigResponse response = unmarshalResponse(httpResponseBody, DELETE_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
253253
// Response validated (no error), but no result to return
254254
} catch (A2AClientException e) {
@@ -270,7 +270,7 @@ public void subscribeToTask(TaskIdParams request, Consumer<StreamingEventKind> e
270270
SSEEventListener sseEventListener = new SSEEventListener(eventConsumer, errorConsumer);
271271

272272
try {
273-
A2AHttpClient.PostBuilder builder = createPostBuilder(Utils.buildBaseUrl(agentInterface, request.tenant()), payloadAndHeaders, SUBSCRIBE_TO_TASK_METHOD);
273+
A2AHttpClient.PostBuilder builder = createPostBuilder(agentInterface.url(), payloadAndHeaders, SUBSCRIBE_TO_TASK_METHOD);
274274
ref.set(builder.postAsyncSSE(
275275
event -> sseEventListener.onMessage(event, ref.get()),
276276
throwable -> sseEventListener.onError(throwable, ref.get()),
@@ -294,7 +294,7 @@ public AgentCard getExtendedAgentCard(GetExtendedAgentCardParams params, @Nullab
294294
ProtoUtils.ToProto.extendedAgentCard(params), agentCard, context);
295295

296296
try {
297-
String httpResponseBody = sendPostRequest(Utils.buildBaseUrl(agentInterface, params.tenant()), payloadAndHeaders, GET_EXTENDED_AGENT_CARD_METHOD);
297+
String httpResponseBody = sendPostRequest(agentInterface.url(), payloadAndHeaders, GET_EXTENDED_AGENT_CARD_METHOD);
298298
GetExtendedAgentCardResponse response = unmarshalResponse(httpResponseBody, GET_EXTENDED_AGENT_CARD_METHOD);
299299
return response.getResult();
300300
} catch (IOException | InterruptedException | JsonProcessingException e) {

‎docs/content/dev/extras.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,10 @@ JPA-backed replacements for the in-memory stores, providing database persistence
4040

4141
Replaces the default `InMemoryQueueManager` with event replication across multiple A2A server instances via message brokers. Required for multi-instance deployments. The core module (`a2a-java-queue-manager-replicated-core`) pairs with a MicroProfile Reactive Messaging strategy (`a2a-java-queue-manager-replication-mp-reactive`) supporting Apache Kafka, Pulsar, or AMQP. You can also write your own `ReplicationStrategy`.
4242

43+
## [Multi-Tenancy](../extra/multi-tenancy)
44+
45+
Serve multiple tenants from a single A2A server with per-tenant `AgentExecutor` and `AgentCard` routing. The module (`a2a-java-extras-multitenancy`) provides a `@Tenant` CDI qualifier and automatic routing — requests are dispatched to tenant-specific beans based on the `tenant` field in the request payload, with fallback to the default beans for unknown tenants.
46+
4347
## [OpenTelemetry](../extra/opentelemetry)
4448

4549
Distributed tracing, metrics, and context propagation for A2A servers and clients using OpenTelemetry. The server module (`a2a-java-sdk-opentelemetry-server`) adds automatic span creation for all protocol methods with context propagation across async boundaries. Client modules (`a2a-java-sdk-opentelemetry-client`, `a2a-java-sdk-opentelemetry-client-propagation`) instrument A2A client operations.

‎docs/content/dev/multi-tenancy.md‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
---
2+
title: Multi-Tenancy
3+
description: Serve multiple tenants from a single A2A server — per-tenant AgentExecutor and AgentCard routing with CDI qualifiers.
4+
layout: page
5+
---
6+
7+
# Multi-Tenancy
8+
9+
Multi-tenancy lets a single A2A server provide different agent behavior per tenant. Each tenant can have its own `AgentExecutor` (business logic) and `AgentCard` (capabilities, skills, metadata). Requests without a recognized tenant automatically fall back to the default beans.
10+
11+
This feature is provided by the `a2a-java-extras-multitenancy` extras module, which registers two CDI routers — `CdiAgentExecutorRouter` and `CdiAgentCardRouter` — that dispatch each request to the matching `@Tenant`-qualified bean.
12+
13+
For setup instructions, configuration reference, and code examples, see the **[Multi-Tenancy extras page](../extra/multi-tenancy)**.

‎docs/content/dev/server.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -223,6 +223,10 @@ public class CloseStreamsHook implements TaskStreamLifecycleHook {
223223

224224
See the [`examples/stream-lifecycle`](https://github.com/a2aproject/a2a-java/tree/main/examples/stream-lifecycle) directory for a complete working example with server, client, and integration tests for all three transports.
225225

226+
## Multi-Tenancy (Optional)
227+
228+
Serve multiple tenants from a single server with per-tenant `AgentExecutor` and `AgentCard` beans. See [Multi-Tenancy](multi-tenancy) for setup and configuration.
229+
226230
## Observability (Optional)
227231

228232
Add distributed tracing to your server with the [OpenTelemetry extras module](extra/opentelemetry). It decorates the request handler to create spans for every A2A protocol method, with automatic error tracking and optional request/response extraction.

‎docs/data/versions/dev.yml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,9 @@ menu:
2222
- title: "Compatibility"
2323
path: "/compatibility"
2424
icon: "fa-solid fa-code-branch"
25+
- title: "Multi-Tenancy"
26+
path: "/multi-tenancy"
27+
icon: "fa-solid fa-building"
2528
- title: "Extras"
2629
path: "/extras"
2730
icon: "fa-solid fa-puzzle-piece"

‎docs/src/main/resources/templates/partials/roq-default/head-scripts.html‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,11 @@
77
{#bundle /}
88
{#search-script /}
99

10+
<script type="module">
11+
import mermaid from 'https://cdn.jsdelivr.net/npm/mermaid@11/dist/mermaid.esm.min.mjs';
12+
mermaid.initialize({ startOnLoad: true, theme: 'neutral' });
13+
</script>
14+
1015
{#if site.data.containsKey("analytics")}
1116
{#ga4 tag=site.data.getJsonObject('analytics').getString("ga4", "") /}
1217
{/if}

‎extras/multitenancy/pom.xml‎

Lines changed: 69 additions & 0 deletions

0 commit comments

Comments
 (0)