Add option to disable standalone SSE stream

Signed-off-by: Arnab Nandy <arnab_nandy7@yahoo.com>

Author: arnabnandy7

docs/client.md

@@ -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

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java

@@ -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;
@@ -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;
@@ -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;
@@ -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);
@@ -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.
 
@@ -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;
@@ -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.
@@ -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);
 		}
 
 	}

mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportErrorHandlingTest.java

@@ -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 {