Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/client.md
Original file line number Diff line number Diff line change
Expand Up @@ -156,12 +156,14 @@ McpTransport transport = new StdioClientTransport(params, McpJsonDefaults.getMap
McpTransport transport = HttpClientStreamableHttpTransport
.builder("http://your-mcp-server")
.endpoint("/mcp")
.openSseStream(false) // Optional: use POST request-response mode only
.build();
```

The Streamable HTTP transport supports:

- Resumable streams for connection recovery
- Optional standalone GET SSE stream for server-initiated messages
- Configurable connect timeout
- Custom HTTP request customization
- Multiple protocol version negotiation
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,8 @@ static boolean isMessageEvent(String eventName) {

private final boolean openConnectionOnStartup;

private final boolean openSseStream;

private final McpHttpClientTransportAuthorizationErrorHandler authorizationErrorHandler;

private final boolean resumableStreams;
Expand All @@ -163,7 +165,8 @@ static boolean isMessageEvent(String eventName) {

private HttpClientStreamableHttpTransport(McpJsonMapper jsonMapper, HttpClient httpClient,
HttpRequest.Builder requestBuilder, String baseUri, String endpoint, boolean resumableStreams,
boolean openConnectionOnStartup, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer,
boolean openConnectionOnStartup, boolean openSseStream,
McpAsyncHttpClientRequestCustomizer httpRequestCustomizer,
McpHttpClientTransportAuthorizationErrorHandler authorizationErrorHandler,
List<String> supportedProtocolVersions) {
this.jsonMapper = jsonMapper;
Expand All @@ -173,6 +176,7 @@ private HttpClientStreamableHttpTransport(McpJsonMapper jsonMapper, HttpClient h
this.endpoint = endpoint;
this.resumableStreams = resumableStreams;
this.openConnectionOnStartup = openConnectionOnStartup;
this.openSseStream = openSseStream;
this.authorizationErrorHandler = authorizationErrorHandler;
this.activeSession.set(createTransportSession());
this.httpRequestCustomizer = httpRequestCustomizer;
Expand All @@ -196,7 +200,7 @@ public static Builder builder(String baseUri) {
public Mono<Void> connect(Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
return Mono.deferContextual(ctx -> {
this.handler.set(handler);
if (this.openConnectionOnStartup) {
if (this.openConnectionOnStartup && this.openSseStream) {
logger.debug("Eagerly opening connection on startup");
return this.reconnect(null).onErrorComplete(t -> {
logger.warn("Eager connect failed ", t);
Expand Down Expand Up @@ -560,8 +564,10 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sentMessage) {
"Authorization error when sending message", requestSnapshot, responseEvent.responseInfo()));
}

if (transportSession.markInitialized(
responseEvent.responseInfo().headers().firstValue("mcp-session-id").orElseGet(() -> null))) {
if (transportSession.markInitialized(responseEvent.responseInfo()
.headers()
.firstValue("mcp-session-id")
.orElseGet(() -> null)) && this.openSseStream) {
// Once we have a session, we try to open an async stream for
// the server to send notifications and requests out-of-band.

Expand Down Expand Up @@ -739,6 +745,8 @@ public static class Builder {

private boolean openConnectionOnStartup = false;

private boolean openSseStream = true;

private HttpRequest.Builder requestBuilder = HttpRequest.newBuilder();

private McpAsyncHttpClientRequestCustomizer httpRequestCustomizer = McpAsyncHttpClientRequestCustomizer.NOOP;
Expand Down Expand Up @@ -841,6 +849,20 @@ public Builder openConnectionOnStartup(boolean openConnectionOnStartup) {
return this;
}

/**
* Configure whether the client should open a standalone SSE stream using an HTTP
* GET request. By default, this value is {@code true}. When disabled, the client
* operates without a standalone SSE stream, but can still process SSE responses
* returned by HTTP POST requests.
* @param openSseStream if {@code true}, the client may open a standalone SSE
* stream
* @return the builder instance
*/
public Builder openSseStream(boolean openSseStream) {
this.openSseStream = openSseStream;
return this;
}

/**
* Sets the customizer for {@link HttpRequest.Builder}, to modify requests before
* executing them.
Expand Down Expand Up @@ -957,7 +979,7 @@ public HttpClientStreamableHttpTransport build() {
HttpClient httpClient = this.clientBuilder.connectTimeout(this.connectTimeout).build();
return new HttpClientStreamableHttpTransport(jsonMapper == null ? McpJsonDefaults.getMapper() : jsonMapper,
httpClient, requestBuilder, baseUri, endpoint, resumableStreams, openConnectionOnStartup,
httpRequestCustomizer, authorizationErrorHandler, supportedProtocolVersions);
openSseStream, httpRequestCustomizer, authorizationErrorHandler, supportedProtocolVersions);
}

}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -389,6 +389,29 @@ void test405OnConnectReturnsEmptyFlux() {
StepVerifier.create(transport.closeGracefully()).verifyComplete();
}

@Test
void shouldNotOpenSseStreamWhenDisabled() {
currentServerSessionId.set("test-session-123");
var getRequestCount = new AtomicInteger();
transport = HttpClientStreamableHttpTransport.builder(HOST)
.openConnectionOnStartup(true)
.openSseStream(false)
.asyncHttpRequestCustomizer((builder, method, uri, body, context) -> {
if ("GET".equals(method)) {
getRequestCount.incrementAndGet();
}
return Mono.just(builder);
})
.build();

StepVerifier.create(transport.connect(msg -> msg)).verifyComplete();
StepVerifier.create(transport.sendMessage(createTestRequestMessage())).verifyComplete();

assertThat(processedMessagesCount.get()).isEqualTo(1);
assertThat(getRequestCount.get()).isZero();
assertThat(processedSseConnectCount.get()).isZero();
}

@Nested
class AuthorizationError {

Expand Down