makes possible to use StreamableHttpClientTransport in McpClient

Author: Aliaksie

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

@@ -159,7 +159,7 @@ public class McpAsyncClient {
 	 * @param features the MCP Client supported features.
 	 */
 	McpAsyncClient(McpClientTransport transport, Duration requestTimeout, Duration initializationTimeout,
-			McpClientFeatures.Async features) {
+			McpClientFeatures.Async features, boolean connectOnInit) {
 
 		Assert.notNull(transport, "Transport must not be null");
 		Assert.notNull(requestTimeout, "Request timeout must not be null");
@@ -234,7 +234,9 @@ public class McpAsyncClient {
 				asyncLoggingNotificationHandler(loggingConsumersFinal));
 
 		this.mcpSession = new McpClientSession(requestTimeout, transport, requestHandlers, notificationHandlers);
-
+		if (connectOnInit) {
+			this.mcpSession.openSSE();
+		}
 	}
 
 	/**
@@ -301,6 +303,18 @@ public Mono<Void> closeGracefully() {
 		return this.mcpSession.closeGracefully();
 	}
 
+	// ---------------------------
+	// open an SSE stream
+	// ---------------------------
+	/**
+	 * The client may issue an HTTP GET to the MCP endpoint. This can be used to open an
+	 * SSE stream, allowing the server to communicate to the client, without the client
+	 * first sending data via HTTP POST.
+	 */
+	public void openSSE() {
+		this.mcpSession.openSSE();
+	}
+
 	// --------------------------
 	// Initialization
 	// --------------------------

mcp/src/main/java/io/modelcontextprotocol/client/McpClient.java

@@ -157,11 +157,13 @@ class SyncSpec {
 
 		private Duration requestTimeout = Duration.ofSeconds(20); // Default timeout
 
+		private boolean connectOnInit = true; // Default true, for backward compatibility
+
 		private Duration initializationTimeout = Duration.ofSeconds(20);
 
 		private ClientCapabilities capabilities;
 
-		private Implementation clientInfo = new Implementation("Java SDK MCP Client", "1.0.0");
+		private Implementation clientInfo = new Implementation("Java SDK MCP Sync Client", "0.10.0");
 
 		private final Map<String, Root> roots = new HashMap<>();
 
@@ -195,6 +197,17 @@ public SyncSpec requestTimeout(Duration requestTimeout) {
 			return this;
 		}
 
+		/**
+		 * Sets whether to connect to the server during the initialization phase (open an
+		 * SSE stream).
+		 * @param connectOnInit true to open an SSE stream during the initialization
+		 * @return This builder instance for method chaining
+		 */
+		public SyncSpec withConnectOnInit(final boolean connectOnInit) {
+			this.connectOnInit = connectOnInit;
+			return this;
+		}
+
 		/**
 		 * @param initializationTimeout The duration to wait for the initialization
 		 * lifecycle step to complete.
@@ -368,8 +381,8 @@ public McpSyncClient build() {
 
 			McpClientFeatures.Async asyncFeatures = McpClientFeatures.Async.fromSync(syncFeatures);
 
-			return new McpSyncClient(
-					new McpAsyncClient(transport, this.requestTimeout, this.initializationTimeout, asyncFeatures));
+			return new McpSyncClient(new McpAsyncClient(transport, this.requestTimeout, this.initializationTimeout,
+					asyncFeatures, this.connectOnInit));
 		}
 
 	}
@@ -396,11 +409,13 @@ class AsyncSpec {
 
 		private Duration requestTimeout = Duration.ofSeconds(20); // Default timeout
 
+		private boolean connectOnInit = true; // Default true, for backward compatibility
+
 		private Duration initializationTimeout = Duration.ofSeconds(20);
 
 		private ClientCapabilities capabilities;
 
-		private Implementation clientInfo = new Implementation("Spring AI MCP Client", "0.3.1");
+		private Implementation clientInfo = new Implementation("Java SDK MCP Async Client", "0.10.0");
 
 		private final Map<String, Root> roots = new HashMap<>();
 
@@ -434,6 +449,17 @@ public AsyncSpec requestTimeout(Duration requestTimeout) {
 			return this;
 		}
 
+		/**
+		 * Sets whether to connect to the server during the initialization phase (open an
+		 * SSE stream).
+		 * @param connectOnInit true to open an SSE stream during the initialization
+		 * @return This builder instance for method chaining
+		 */
+		public AsyncSpec withConnectOnInit(final boolean connectOnInit) {
+			this.connectOnInit = connectOnInit;
+			return this;
+		}
+
 		/**
 		 * @param initializationTimeout The duration to wait for the initialization
 		 * lifecycle step to complete.
@@ -606,7 +632,8 @@ public McpAsyncClient build() {
 			return new McpAsyncClient(this.transport, this.requestTimeout, this.initializationTimeout,
 					new McpClientFeatures.Async(this.clientInfo, this.capabilities, this.roots,
 							this.toolsChangeConsumers, this.resourcesChangeConsumers, this.promptsChangeConsumers,
-							this.loggingConsumers, this.samplingHandler));
+							this.loggingConsumers, this.samplingHandler),
+					this.connectOnInit);
 		}
 
 	}

mcp/src/main/java/io/modelcontextprotocol/client/McpSyncClient.java

@@ -128,6 +128,18 @@ public boolean closeGracefully() {
 		return true;
 	}
 
+	// ---------------------------
+	// open an SSE stream
+	// ---------------------------
+	/**
+	 * The client may issue an HTTP GET to the MCP endpoint. This can be used to open an
+	 * SSE stream, allowing the server to communicate to the client, without the client
+	 * first sending data via HTTP POST.
+	 */
+	public void openSSE() {
+		this.delegate.openSSE();
+	}
+
 	/**
 	 * The initialization phase MUST be the first interaction between client and server.
 	 * During this phase, the client and server:
@@ -147,9 +159,7 @@ public boolean closeGracefully() {
 	 * The server MUST respond with its own capabilities and information:
 	 * {@link McpSchema.ServerCapabilities}. <br/>
 	 * After successful initialization, the client MUST send an initialized notification
-	 * to indicate it is ready to begin normal operations.
-	 *
-	 * <br/>
+	 * to indicate it is ready to begin normal operations. <br/>
 	 *
 	 * <a href=
 	 * "https://github.com/modelcontextprotocol/specification/blob/main/docs/specification/basic/lifecycle.md#initialization">Initialization
@@ -271,9 +281,8 @@ public McpSchema.ReadResourceResult readResource(McpSchema.ReadResourceRequest r
 
 	/**
 	 * Resource templates allow servers to expose parameterized resources using URI
-	 * templates. Arguments may be auto-completed through the completion API.
-	 *
-	 * Request a list of resource templates the server has.
+	 * templates. Arguments may be auto-completed through the completion API. Request a
+	 * list of resource templates the server has.
 	 * @param cursor the cursor
 	 * @return the list of resource templates result.
 	 */
@@ -292,9 +301,7 @@ public McpSchema.ListResourceTemplatesResult listResourceTemplates() {
 	/**
 	 * Subscriptions. The protocol supports optional subscriptions to resource changes.
 	 * Clients can subscribe to specific resources and receive notifications when they
-	 * change.
-	 *
-	 * Send a resources/subscribe request.
+	 * change. Send a resources/subscribe request.
 	 * @param subscribeRequest the subscribe request contains the uri of the resource to
 	 * subscribe to.
 	 */

mcp/src/main/java/io/modelcontextprotocol/client/transport/StreamableHttpClientTransport.java

@@ -72,8 +72,6 @@ public class StreamableHttpClientTransport implements McpClientTransport {
 
 	private final URI uri;
 
-	private final AtomicReference<TransportState> state = new AtomicReference<>(TransportState.DISCONNECTED);
-
 	private final AtomicReference<String> lastEventId = new AtomicReference<>();
 
 	private final AtomicReference<String> mcpSessionId = new AtomicReference<>();
@@ -99,15 +97,6 @@ public static Builder builder(final String uri) {
 		return new Builder().withBaseUri(uri);
 	}
 
-	/**
-	 * The state of the Transport connection.
-	 */
-	public enum TransportState {
-
-		DISCONNECTED, CONNECTING, CONNECTED, CLOSED
-
-	}
-
 	/**
 	 * A builder for creating instances of WebSocketClientTransport.
 	 */
@@ -188,10 +177,6 @@ public Mono<Void> connect(final Function<Mono<McpSchema.JSONRPCMessage>, Mono<Mc
 			return sseClientTransport.connect(handler);
 		}
 
-		if (!state.compareAndSet(TransportState.DISCONNECTED, TransportState.CONNECTING)) {
-			return Mono.error(new IllegalStateException("Already connected or connecting"));
-		}
-
 		return Mono.defer(() -> Mono.fromFuture(() -> {
 			final HttpRequest.Builder request = requestBuilder.copy().GET().header(ACCEPT, TEXT_EVENT_STREAM).uri(uri);
 			final String lastId = lastEventId.get();
@@ -219,13 +204,10 @@ public Mono<Void> connect(final Function<Mono<McpSchema.JSONRPCMessage>, Mono<Mc
 			return handleStreamingResponse(response, handler);
 		})
 			.retryWhen(Retry.backoff(3, Duration.ofSeconds(3)).filter(err -> err instanceof IllegalStateException))
-			.doOnSuccess(v -> state.set(TransportState.CONNECTED))
-			.doOnTerminate(() -> state.set(TransportState.CLOSED))
 			.onErrorResume(e -> {
 				LOGGER.error("Streamable transport connection error", e);
-				state.set(TransportState.DISCONNECTED);
 				return Mono.error(e);
-			}));
+			})).doOnTerminate(this::closeGracefully);
 	}
 
 	@Override
@@ -239,10 +221,6 @@ public Mono<Void> sendMessage(final McpSchema.JSONRPCMessage message,
 			return fallbackToSse(message);
 		}
 
-		if (state.get() == TransportState.CLOSED) {
-			return Mono.empty();
-		}
-
 		return serializeJson(message).flatMap(json -> {
 			final HttpRequest.Builder request = requestBuilder.copy()
 				.POST(HttpRequest.BodyPublishers.ofString(json))
@@ -427,7 +405,8 @@ else if (node.isObject()) {
 
 	@Override
 	public Mono<Void> closeGracefully() {
-		state.set(TransportState.CLOSED);
+		mcpSessionId.set(null);
+		lastEventId.set(null);
 		if (fallbackToSse.get()) {
 			return sseClientTransport.closeGracefully();
 		}
@@ -439,8 +418,4 @@ public <T> T unmarshalFrom(final Object data, final TypeReference<T> typeRef) {
 		return objectMapper.convertValue(data, typeRef);
 	}
 
-	public TransportState getState() {
-		return state.get();
-	}
-
 }

mcp/src/main/java/io/modelcontextprotocol/spec/McpClientSession.java

@@ -61,7 +61,7 @@ public class McpClientSession implements McpSession {
 	/** Atomic counter for generating unique request IDs */
 	private final AtomicLong requestCounter = new AtomicLong(0);
 
-	private final Disposable connection;
+	private Disposable connection;
 
 	/**
 	 * Functional interface for handling incoming JSON-RPC requests. Implementations
@@ -116,6 +116,17 @@ public McpClientSession(Duration requestTimeout, McpClientTransport transport,
 		this.transport = transport;
 		this.requestHandlers.putAll(requestHandlers);
 		this.notificationHandlers.putAll(notificationHandlers);
+	}
+
+	/**
+	 * The client may issue an HTTP GET to the MCP endpoint. This can be used to open an
+	 * SSE stream, allowing the server to communicate to the client, without the client
+	 * first sending data via HTTP POST.
+	 */
+	public void openSSE() {
+		if (this.connection != null && !this.connection.isDisposed()) {
+			return; // already connected and still active
+		}
 
 		// TODO: consider mono.transformDeferredContextual where the Context contains
 		// the
@@ -281,7 +292,9 @@ public Mono<Void> closeGracefully() {
 	 */
 	@Override
 	public void close() {
-		this.connection.dispose();
+		if (this.connection != null) {
+			this.connection.dispose();
+		}
 		transport.close();
 	}
 

mcp/src/test/java/io/modelcontextprotocol/spec/McpClientSessionTests.java

@@ -48,6 +48,7 @@ void setUp() {
 		transport = new MockMcpClientTransport();
 		session = new McpClientSession(TIMEOUT, transport, Map.of(),
 				Map.of(TEST_NOTIFICATION, params -> Mono.fromRunnable(() -> logger.info("Status update: " + params))));
+		session.openSSE();
 	}
 
 	@AfterEach
@@ -141,6 +142,7 @@ void testRequestHandling() {
 				params -> Mono.just(params));
 		transport = new MockMcpClientTransport();
 		session = new McpClientSession(TIMEOUT, transport, requestHandlers, Map.of());
+		session.openSSE();
 
 		// Simulate incoming request
 		McpSchema.JSONRPCRequest request = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, ECHO_METHOD,
@@ -162,7 +164,7 @@ void testNotificationHandling() {
 		transport = new MockMcpClientTransport();
 		session = new McpClientSession(TIMEOUT, transport, Map.of(),
 				Map.of(TEST_NOTIFICATION, params -> Mono.fromRunnable(() -> receivedParams.tryEmitValue(params))));
-
+		session.openSSE();
 		// Simulate incoming notification from the server
 		Map<String, Object> notificationParams = Map.of("status", "ready");