refactor(transport): improve MCP transport layer design

- Rename DefaultMcpTransport to AbstractMcpTransport and make it abstract
- Add graceful shutdown support via closeGracefully() method
- Enhance McpTransport interface with default close() implementation
- Improve documentation across transport layer classes

Author: tzolov

mcp/pom.xml

@@ -24,7 +24,6 @@
 	<properties>
 		<java.version>21</java.version>
 		<maven.compiler.release>21</maven.compiler.release>
-		<!-- <spring-ai.version>1.0.0-M4</spring-ai.version> -->
 	</properties>
 	<dependencies>
 		<dependency>

mcp/src/main/java/spring/ai/experimental/mcp/client/McpClient.java

@@ -1,18 +1,18 @@
 /*
-* Copyright 2024 - 2024 the original author or authors.
-*
-* Licensed under the Apache License, Version 2.0 (the "License");
-* you may not use this file except in compliance with the License.
-* You may obtain a copy of the License at
-*
-* https://www.apache.org/licenses/LICENSE-2.0
-*
-* Unless required by applicable law or agreed to in writing, software
-* distributed under the License is distributed on an "AS IS" BASIS,
-* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-* See the License for the specific language governing permissions and
-* limitations under the License.
-*/
+ * Copyright 2024 - 2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package spring.ai.experimental.mcp.client;
 
 import java.time.Duration;
@@ -21,41 +21,85 @@
 import spring.ai.experimental.mcp.spec.McpTransport;
 
 /**
- * The MCP client is the main entry point for interacting with the Model Context Protocol
- * (MCP) server.
+ * Factory class providing static methods for creating Model Context Protocol (MCP) clients.
+ * This class serves as the main entry point for establishing connections with MCP servers,
+ * offering both synchronous and asynchronous client implementations.
+ * 
+ * <p>The class provides factory methods to create either:
+ * <ul>
+ *   <li>{@link McpAsyncClient} for non-blocking operations
+ *   <li>{@link McpSyncClient} for blocking operations
+ * </ul>
+ * 
+ * <p>Each client type can be instantiated with default settings or with custom configuration
+ * including request timeout and JSON object mapping.
+ * 
+ * <p>Future implementations will introduce a builder pattern for more flexible client configuration.
  * 
  * @author Christian Tzolov
  * @author Dariusz Jędrzejczyk
  */
 public class McpClient {
 
-	private McpClient() {
-	}
-
-	// TODO: introduce a builder like:
-	//  McpClient.using(transport)
-	//           .withRequestTimeout(Duration.ofSeconds(5))
-	//           .withObjectMapper(objectMapper); <- or even a more sophisticated
-	//                                               JSONtoPOJOCodec type
-	//           .sync();
-
-	public static McpAsyncClient async(McpTransport transport) {
-		return new McpAsyncClient(transport);
-	}
+    /**
+     * Private constructor to prevent instantiation as this is a utility class
+     * containing only static factory methods.
+     */
+    private McpClient() {
+    }
 
-	public static McpAsyncClient async(McpTransport transport, Duration requestTimeout,
-			ObjectMapper objectMapper) {
-		return new McpAsyncClient(transport, requestTimeout, objectMapper);
-	}
+    // TODO: introduce a builder like:
+    //  McpClient.using(transport)
+    //           .withRequestTimeout(Duration.ofSeconds(5))
+    //           .withObjectMapper(objectMapper); <- or even a more sophisticated
+    //                                               JSONtoPOJOCodec type
+    //           .sync();
 
-	public static McpSyncClient sync(McpTransport transport) {
-		return new McpSyncClient(async(transport));
-	}
+    /**
+     * Creates an asynchronous MCP client with default configuration.
+     *
+     * @param transport The transport layer implementation for MCP communication
+     * @return A new instance of {@link McpAsyncClient}
+     */
+    public static McpAsyncClient async(McpTransport transport) {
+        return new McpAsyncClient(transport);
+    }
 
-	public static McpSyncClient sync(McpTransport transport, Duration requestTimeout,
-			ObjectMapper objectMapper) {
-		return new McpSyncClient(async(transport, requestTimeout, objectMapper));
-	}
+    /**
+     * Creates an asynchronous MCP client with custom configuration.
+     *
+     * @param transport The transport layer implementation for MCP communication
+     * @param requestTimeout The duration to wait before timing out requests
+     * @param objectMapper The Jackson object mapper for JSON serialization/deserialization
+     * @return A new instance of {@link McpAsyncClient}
+     */
+    public static McpAsyncClient async(McpTransport transport, Duration requestTimeout,
+            ObjectMapper objectMapper) {
+        return new McpAsyncClient(transport, requestTimeout, objectMapper);
+    }
 
+    /**
+     * Creates a synchronous MCP client with default configuration.
+     * This method wraps an asynchronous client to provide synchronous operations.
+     *
+     * @param transport The transport layer implementation for MCP communication
+     * @return A new instance of {@link McpSyncClient}
+     */
+    public static McpSyncClient sync(McpTransport transport) {
+        return new McpSyncClient(async(transport));
+    }
 
+    /**
+     * Creates a synchronous MCP client with custom configuration.
+     * This method wraps an asynchronous client to provide synchronous operations.
+     *
+     * @param transport The transport layer implementation for MCP communication
+     * @param requestTimeout The duration to wait before timing out requests
+     * @param objectMapper The Jackson object mapper for JSON serialization/deserialization
+     * @return A new instance of {@link McpSyncClient}
+     */
+    public static McpSyncClient sync(McpTransport transport, Duration requestTimeout,
+            ObjectMapper objectMapper) {
+        return new McpSyncClient(async(transport, requestTimeout, objectMapper));
+    }
 }

mcp/src/main/java/spring/ai/experimental/mcp/client/stdio/StdioServerTransport.java

@@ -10,10 +10,11 @@
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 import spring.ai.experimental.mcp.client.util.Assert;
-import spring.ai.experimental.mcp.spec.DefaultMcpTransport;
+import spring.ai.experimental.mcp.spec.AbstractMcpTransport;
 import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCMessage;
 import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCNotification;
 import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCRequest;
@@ -25,7 +26,7 @@
  * @author Christian Tzolov
  * @author Dariusz Jędrzejczyk
  */
-public class StdioServerTransport extends DefaultMcpTransport {
+public class StdioServerTransport extends AbstractMcpTransport {
 
 	private final static TypeReference<HashMap<String, Object>> MAP_TYPE_REF = new TypeReference<HashMap<String, Object>>() {
 	};
@@ -194,47 +195,46 @@ private void startOutboundProcessing() {
 				.subscribe();
 	}
 
-	// TODO: provide a non-blocking variant with graceful option
-	public void stop() {
-		this.outboundScheduler.dispose();
-		this.inboundScheduler.dispose();
-		this.errorScheduler.dispose();
-
-		// Destroy process
-		if (this.process != null) {
-			this.process.destroyForcibly();
-		}
-	}
-
 	@Override
-	public void close() {
+	public Mono<Void> closeGracefully() {
 
-		this.stop();
+		this.isRunning = false;
 
-		super.close(); // Do we need this?
-
-		// Close resources
-		if (this.processReader != null) {
-			try {
-				this.processReader.close();
-			} catch (IOException e) {
-				e.printStackTrace();
-			}
-		}
-		if (this.processWriter != null) {
-			try {
-				this.processWriter.close();
-			} catch (IOException e) {
-				e.printStackTrace();
-			}
-		}
-		if (this.processErrorReader != null) {
-			try {
-				this.processErrorReader.close();
-			} catch (IOException e) {
-				e.printStackTrace();
-			}
-		}
+		return Mono.whenDelayError(
+				Mono.fromRunnable(() -> {
+					try {
+						this.processErrorReader.close();
+					} catch (IOException e) {
+						throw new RuntimeException(e);
+					}
+				}).subscribeOn(errorScheduler),
+				Mono.fromRunnable(() -> {
+					try {
+						this.processReader.close();
+					} catch (IOException e) {
+						throw new RuntimeException(e);
+					}
+				}).subscribeOn(inboundScheduler),
+				Mono.fromRunnable(() -> {
+					try {
+						this.processWriter.close();
+					} catch (IOException e) {
+						throw new RuntimeException(e);
+					}
+				}).subscribeOn(outboundScheduler))
+				.then(Mono.whenDelayError(inboundScheduler.disposeGracefully(),
+						outboundScheduler.disposeGracefully(),
+						errorScheduler.disposeGracefully()))
+				.then(Mono.fromRunnable(() -> {
+					if (this.process != null) {
+						var process = this.process.destroyForcibly();
+						if (process.exitValue() != 0) {
+							throw new RuntimeException("Failed to destroy process");
+						}
+					}
+				}))
+				.then()
+				.subscribeOn(Schedulers.boundedElastic());
 	}
 
 }

mcp/src/main/java/spring/ai/experimental/mcp/spec/DefaultMcpTransport.java renamed to mcp/src/main/java/spring/ai/experimental/mcp/spec/AbstractMcpTransport.java

@@ -27,7 +27,7 @@
  * @author Christian Tzolov
  * @author Dariusz Jędrzejczyk
  */
-public class DefaultMcpTransport implements McpTransport {
+public abstract class AbstractMcpTransport implements McpTransport {
 
 	protected final ObjectMapper objectMapper;
 
@@ -39,11 +39,11 @@ public class DefaultMcpTransport implements McpTransport {
 			.println("Received message: " + message);
 	private Consumer<String> errorHandler = error -> System.err.println("Received error: " + error);
 
-	public DefaultMcpTransport() {		
+	public AbstractMcpTransport() {		
 		this(new ObjectMapper());
 	}
 
-	public DefaultMcpTransport(ObjectMapper objectMapper) {
+	public AbstractMcpTransport(ObjectMapper objectMapper) {
 
 		Assert.notNull(objectMapper, "ObjectMapper must not be null");
 		this.objectMapper = objectMapper;
@@ -73,7 +73,7 @@ private void handleIncomingMessages() {
 	private void handleIncomingErrors() {
 		this.errorSink.asFlux().subscribe(e -> {
 			this.errorHandler.accept(e);
-			System.err.println(e);
+			// System.err.println(e); TODO: log the error
 		});
 	}
 
@@ -103,12 +103,6 @@ public void start() {
 		this.handleIncomingMessages();
 	}
 
-	// Close the transport
-	@Override
-	public void close() {
-		// this.onClose();
-	}
-
 	@Override
 	public Mono<Void> sendMessage(JSONRPCMessage message) {
 		if (this.outboundSink.tryEmitNext(message).isSuccess()) {

mcp/src/main/java/spring/ai/experimental/mcp/spec/McpSession.java

@@ -111,7 +111,6 @@ public Mono<Void> sendNotification(String method, Map<String, Object> params) {
 		McpSchema.JSONRPCNotification jsonrpcNotification = new McpSchema.JSONRPCNotification(McpSchema.JSONRPC_VERSION,
 				method, params);
 		try {
-			// TODO: make it non-blocking
 			this.transport.sendMessage(jsonrpcNotification);
 		} catch (Exception e) {
 			return Mono.error(new McpError(e));