Skip to content

Commit 9d58621

Browse files
tzolovtzolov
authored
feat: implement MCP protocol version per transport support (#404)
- Add protocolVersion() method to transport and transport provider interfaces and implementations - Replace hardcoded HTTP header strings with HttpHeaders constants - Update protocol versions to be transport-specific rather than global - Deprecate McpSchema.LATEST_PROTOCOL_VERSION in favor of transport-specific versions - Standardize header names: MCP-Protocol-Version, mcp-session-id, last-event-id - Update clients and servers to use transport.protocolVersion() for initialization - Refactor tests to use transport-specific protocol versions - Include MCP-Protocol-Version header in all GET/POST/DELETE requests - Update WebClientStreamableHttpTransport, WebFluxSseClientTransport, HttpClientSseClientTransport, and HttpClientStreamableHttpTransport Related to #398 , #363 , #250 Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
1 parent 67de84b commit 9d58621

27 files changed

+169
-34
lines changed

‎mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/client/transport/WebClientStreamableHttpTransport.java‎

Lines changed: 28 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323

2424
import io.modelcontextprotocol.spec.DefaultMcpTransportSession;
2525
import io.modelcontextprotocol.spec.DefaultMcpTransportStream;
26+
import io.modelcontextprotocol.spec.HttpHeaders;
2627
import io.modelcontextprotocol.spec.McpClientTransport;
2728
import io.modelcontextprotocol.spec.McpError;
2829
import io.modelcontextprotocol.spec.McpSchema;
@@ -66,6 +67,8 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
6667

6768
private static final Logger logger = LoggerFactory.getLogger(WebClientStreamableHttpTransport.class);
6869

70+
private static final String MCP_PROTOCOL_VERSION = "2025-03-26";
71+
6972
private static final String DEFAULT_ENDPOINT = "/mcp";
7073

7174
/**
@@ -103,6 +106,11 @@ private WebClientStreamableHttpTransport(ObjectMapper objectMapper, WebClient.Bu
103106
this.activeSession.set(createTransportSession());
104107
}
105108

109+
@Override
110+
public String protocolVersion() {
111+
return MCP_PROTOCOL_VERSION;
112+
}
113+
106114
/**
107115
* Create a stateful builder for creating {@link WebClientStreamableHttpTransport}
108116
* instances.
@@ -128,12 +136,20 @@ public Mono<Void> connect(Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchem
128136

129137
private DefaultMcpTransportSession createTransportSession() {
130138
Function<String, Publisher<Void>> onClose = sessionId -> sessionId == null ? Mono.empty()
131-
: webClient.delete().uri(this.endpoint).headers(httpHeaders -> {
132-
httpHeaders.add("mcp-session-id", sessionId);
133-
}).retrieve().toBodilessEntity().onErrorComplete(e -> {
134-
logger.warn("Got error when closing transport", e);
135-
return true;
136-
}).then();
139+
: webClient.delete()
140+
.uri(this.endpoint)
141+
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
142+
.headers(httpHeaders -> {
143+
httpHeaders.add(HttpHeaders.MCP_SESSION_ID, sessionId);
144+
httpHeaders.add(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION);
145+
})
146+
.retrieve()
147+
.toBodilessEntity()
148+
.onErrorComplete(e -> {
149+
logger.warn("Got error when closing transport", e);
150+
return true;
151+
})
152+
.then();
137153
return new DefaultMcpTransportSession(onClose);
138154
}
139155

@@ -186,10 +202,11 @@ private Mono<Disposable> reconnect(McpTransportStream<Disposable> stream) {
186202
Disposable connection = webClient.get()
187203
.uri(this.endpoint)
188204
.accept(MediaType.TEXT_EVENT_STREAM)
205+
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
189206
.headers(httpHeaders -> {
190-
transportSession.sessionId().ifPresent(id -> httpHeaders.add("mcp-session-id", id));
207+
transportSession.sessionId().ifPresent(id -> httpHeaders.add(HttpHeaders.MCP_SESSION_ID, id));
191208
if (stream != null) {
192-
stream.lastId().ifPresent(id -> httpHeaders.add("last-event-id", id));
209+
stream.lastId().ifPresent(id -> httpHeaders.add(HttpHeaders.LAST_EVENT_ID, id));
193210
}
194211
})
195212
.exchangeToFlux(response -> {
@@ -246,13 +263,14 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
246263
Disposable connection = webClient.post()
247264
.uri(this.endpoint)
248265
.accept(MediaType.APPLICATION_JSON, MediaType.TEXT_EVENT_STREAM)
266+
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
249267
.headers(httpHeaders -> {
250-
transportSession.sessionId().ifPresent(id -> httpHeaders.add("mcp-session-id", id));
268+
transportSession.sessionId().ifPresent(id -> httpHeaders.add(HttpHeaders.MCP_SESSION_ID, id));
251269
})
252270
.bodyValue(message)
253271
.exchangeToFlux(response -> {
254272
if (transportSession
255-
.markInitialized(response.headers().asHttpHeaders().getFirst("mcp-session-id"))) {
273+
.markInitialized(response.headers().asHttpHeaders().getFirst(HttpHeaders.MCP_SESSION_ID))) {
256274
// Once we have a session, we try to open an async stream for
257275
// the server to send notifications and requests out-of-band.
258276
reconnect(null).contextWrite(sink.contextView()).subscribe();

‎mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/client/transport/WebFluxSseClientTransport.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@
99

1010
import com.fasterxml.jackson.core.type.TypeReference;
1111
import com.fasterxml.jackson.databind.ObjectMapper;
12+
13+
import io.modelcontextprotocol.spec.HttpHeaders;
1214
import io.modelcontextprotocol.spec.McpClientTransport;
1315
import io.modelcontextprotocol.spec.McpError;
1416
import io.modelcontextprotocol.spec.McpSchema;
@@ -62,6 +64,8 @@ public class WebFluxSseClientTransport implements McpClientTransport {
6264

6365
private static final Logger logger = LoggerFactory.getLogger(WebFluxSseClientTransport.class);
6466

67+
private static final String MCP_PROTOCOL_VERSION = "2024-11-05";
68+
6569
/**
6670
* Event type for JSON-RPC messages received through the SSE connection. The server
6771
* sends messages with this event type to transmit JSON-RPC protocol data.
@@ -166,6 +170,11 @@ public WebFluxSseClientTransport(WebClient.Builder webClientBuilder, ObjectMappe
166170
this.sseEndpoint = sseEndpoint;
167171
}
168172

173+
@Override
174+
public String protocolVersion() {
175+
return MCP_PROTOCOL_VERSION;
176+
}
177+
169178
/**
170179
* Establishes a connection to the MCP server using Server-Sent Events (SSE). This
171180
* method initiates the SSE connection and sets up the message processing pipeline.
@@ -250,6 +259,7 @@ public Mono<Void> sendMessage(JSONRPCMessage message) {
250259
return webClient.post()
251260
.uri(messageEndpointUri)
252261
.contentType(MediaType.APPLICATION_JSON)
262+
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
253263
.bodyValue(jsonText)
254264
.retrieve()
255265
.toBodilessEntity()
@@ -282,6 +292,7 @@ protected Flux<ServerSentEvent<String>> eventStream() {// @formatter:off
282292
.get()
283293
.uri(this.sseEndpoint)
284294
.accept(MediaType.TEXT_EVENT_STREAM)
295+
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
285296
.retrieve()
286297
.bodyToFlux(SSE_TYPE)
287298
.retryWhen(Retry.from(retrySignal -> retrySignal.handle(inboundRetryHandler)));

‎mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/server/transport/WebFluxSseServerTransportProvider.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,8 @@ public class WebFluxSseServerTransportProvider implements McpServerTransportProv
7979
*/
8080
public static final String ENDPOINT_EVENT_TYPE = "endpoint";
8181

82+
private static final String MCP_PROTOCOL_VERSION = "2025-06-18";
83+
8284
/**
8385
* Default SSE endpoint path as specified by the MCP transport specification.
8486
*/
@@ -212,6 +214,11 @@ public WebFluxSseServerTransportProvider(ObjectMapper objectMapper, String baseU
212214
}
213215
}
214216

217+
@Override
218+
public String protocolVersion() {
219+
return "2024-11-05";
220+
}
221+
215222
@Override
216223
public void setSessionFactory(McpServerSession.Factory sessionFactory) {
217224
this.sessionFactory = sessionFactory;

‎mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/server/transport/WebFluxStatelessServerTransport.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@
2121

2222
import java.io.IOException;
2323
import java.util.List;
24-
import java.util.function.Function;
2524

2625
/**
2726
* Implementation of a WebFlux based {@link McpStatelessServerTransport}.

‎mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/server/transport/WebFluxStreamableServerTransportProvider.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,11 @@ private WebFluxStreamableServerTransportProvider(ObjectMapper objectMapper, Stri
9595

9696
}
9797

98+
@Override
99+
public String protocolVersion() {
100+
return "2025-03-26";
101+
}
102+
98103
@Override
99104
public void setSessionFactory(McpStreamableServerSession.Factory sessionFactory) {
100105
this.sessionFactory = sessionFactory;

‎mcp-spring/mcp-spring-webmvc/src/main/java/io/modelcontextprotocol/server/transport/WebMvcSseServerTransportProvider.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -209,6 +209,11 @@ public WebMvcSseServerTransportProvider(ObjectMapper objectMapper, String baseUr
209209
}
210210
}
211211

212+
@Override
213+
public String protocolVersion() {
214+
return "2024-11-05";
215+
}
216+
212217
@Override
213218
public void setSessionFactory(McpServerSession.Factory sessionFactory) {
214219
this.sessionFactory = sessionFactory;

‎mcp-spring/mcp-spring-webmvc/src/main/java/io/modelcontextprotocol/server/transport/WebMvcStreamableServerTransportProvider.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,11 @@ private WebMvcStreamableServerTransportProvider(ObjectMapper objectMapper, Strin
147147
}
148148
}
149149

150+
@Override
151+
public String protocolVersion() {
152+
return "2025-03-26";
153+
}
154+
150155
@Override
151156
public void setSessionFactory(McpStreamableServerSession.Factory sessionFactory) {
152157
this.sessionFactory = sessionFactory;

‎mcp-spring/mcp-spring-webmvc/src/test/java/io/modelcontextprotocol/server/WebMvcSseCustomContextPathTests.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -91,8 +91,15 @@ static class TestConfig {
9191
@Bean
9292
public WebMvcSseServerTransportProvider webMvcSseServerTransportProvider() {
9393

94-
return new WebMvcSseServerTransportProvider(new ObjectMapper(), CUSTOM_CONTEXT_PATH, MESSAGE_ENDPOINT,
95-
WebMvcSseServerTransportProvider.DEFAULT_SSE_ENDPOINT);
94+
return WebMvcSseServerTransportProvider.builder()
95+
.objectMapper(new ObjectMapper())
96+
.baseUrl(CUSTOM_CONTEXT_PATH)
97+
.messageEndpoint(MESSAGE_ENDPOINT)
98+
.sseEndpoint(WebMvcSseServerTransportProvider.DEFAULT_SSE_ENDPOINT)
99+
.build();
100+
// return new WebMvcSseServerTransportProvider(new ObjectMapper(),
101+
// CUSTOM_CONTEXT_PATH, MESSAGE_ENDPOINT,
102+
// WebMvcSseServerTransportProvider.DEFAULT_SSE_ENDPOINT);
96103
}
97104

98105
@Bean

‎mcp/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -272,9 +272,8 @@ public class McpAsyncClient {
272272
asyncProgressNotificationHandler(progressConsumersFinal));
273273

274274
this.initializer = new LifecycleInitializer(clientCapabilities, clientInfo,
275-
List.of(McpSchema.LATEST_PROTOCOL_VERSION), initializationTimeout,
276-
ctx -> new McpClientSession(requestTimeout, transport, requestHandlers, notificationHandlers,
277-
con -> con.contextWrite(ctx)));
275+
List.of(transport.protocolVersion()), initializationTimeout, ctx -> new McpClientSession(requestTimeout,
276+
transport, requestHandlers, notificationHandlers, con -> con.contextWrite(ctx)));
278277
this.transport.setExceptionHandler(this.initializer::handleException);
279278
}
280279

‎mcp/src/main/java/io/modelcontextprotocol/client/transport/HttpClientSseClientTransport.java‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,10 @@
6161
*/
6262
public class HttpClientSseClientTransport implements McpClientTransport {
6363

64+
private static final String MCP_PROTOCOL_VERSION = "2024-11-05";
65+
66+
private static final String MCP_PROTOCOL_VERSION_HEADER_NAME = "MCP-Protocol-Version";
67+
6468
private static final Logger logger = LoggerFactory.getLogger(HttpClientSseClientTransport.class);
6569

6670
/** SSE event type for JSON-RPC messages */
@@ -211,6 +215,11 @@ public HttpClientSseClientTransport(HttpClient.Builder clientBuilder, HttpReques
211215
this.httpRequestCustomizer = httpRequestCustomizer;
212216
}
213217

218+
@Override
219+
public String protocolVersion() {
220+
return MCP_PROTOCOL_VERSION;
221+
}
222+
214223
/**
215224
* Creates a new builder for {@link HttpClientSseClientTransport}.
216225
* @param baseUri the base URI of the MCP server
@@ -391,6 +400,7 @@ public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> h
391400
.uri(uri)
392401
.header("Accept", "text/event-stream")
393402
.header("Cache-Control", "no-cache")
403+
.header(MCP_PROTOCOL_VERSION_HEADER_NAME, MCP_PROTOCOL_VERSION)
394404
.GET();
395405
return Mono.from(this.httpRequestCustomizer.customize(builder, "GET", uri, null));
396406
}).flatMap(requestBuilder -> Mono.create(sink -> {
@@ -516,7 +526,10 @@ private Mono<String> serializeMessage(final JSONRPCMessage message) {
516526
private Mono<HttpResponse<String>> sendHttpPost(final String endpoint, final String body) {
517527
final URI requestUri = Utils.resolveUri(baseUri, endpoint);
518528
return Mono.defer(() -> {
519-
var builder = this.requestBuilder.copy().uri(requestUri).POST(HttpRequest.BodyPublishers.ofString(body));
529+
var builder = this.requestBuilder.copy()
530+
.uri(requestUri)
531+
.header(MCP_PROTOCOL_VERSION_HEADER_NAME, MCP_PROTOCOL_VERSION)
532+
.POST(HttpRequest.BodyPublishers.ofString(body));
520533
return Mono.from(this.httpRequestCustomizer.customize(builder, "POST", requestUri, body));
521534
}).flatMap(customizedBuilder -> {
522535
var request = customizedBuilder.build();

0 commit comments

Comments
 (0)