code cleaning

mcp/README.md

@@ -101,7 +101,7 @@ The SDK follows a layered architecture:
 - **McpClient**: Main interface defining the synchronous operations
 - **McpAsyncClient**: Async implementation using Project Reactor
 - **McpSyncClient**: Synchronous wrapper around the async client
-- **McpAsyncTransport**: Transport layer interface
+- **McpTransport**: Transport layer interface
 - **DefaultMcpTransport**: Base transport implementation
 - **StdioServerTransport**: Stdio-based server communication
 

mcp/docs/mcp-class-diagram.puml

@@ -8,7 +8,7 @@ skinparam class {
 }
 
 ' Interfaces
-interface McpAsyncTransport {
+interface McpTransport {
     +start(): void
     +close(): void
     +setInboudMessageHandler(Consumer<JSONRPCMessage>): void
@@ -42,16 +42,16 @@ class DefaultMcpTransport {
 }
 
 class McpAsyncClient {
-    -transport: McpAsyncTransport
+    -transport: McpTransport
     -session: McpAsyncSession
-    +McpAsyncClient(McpAsyncTransport)
+    +McpAsyncClient(McpTransport)
     +initialize(): Mono<InitializeResult>
     +close(): void
 }
 
 class McpSyncClient {
     -asyncClient: McpAsyncClient
-    +McpSyncClient(McpAsyncTransport)
+    +McpSyncClient(McpTransport)
     +initialize(): InitializeResult
     +close(): void
 }
@@ -79,14 +79,14 @@ class Assert {
 }
 
 ' Relationships
-McpAsyncTransport <|.. DefaultMcpTransport
+McpTransport <|.. DefaultMcpTransport
 DefaultMcpTransport <|-- StdioServerTransport
 
 McpClient <|.. McpSyncClient
 McpAsyncSession <|.. McpAsyncClient
 
 McpSyncClient o-- McpAsyncClient
-McpAsyncClient o-- McpAsyncTransport
+McpAsyncClient o-- McpTransport
 McpAsyncClient o-- McpAsyncSession
 
 StdioServerTransport o-- ServerParameters
@@ -94,7 +94,7 @@ ServerParameters <.. ServerParametersParser : creates
 
 ' Package organization
 package "spring.ai.mcp.spec" {
-    ' McpAsyncTransport
+    ' McpTransport
     ' McpAsyncSession
     ' DefaultMcpTransport
 }

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

@@ -5,23 +5,27 @@ import java.time.Duration;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import reactor.core.publisher.Mono;
-import spring.ai.experimental.mcp.spec.McpAsyncSession;
-import spring.ai.experimental.mcp.spec.McpAsyncTransport;
+import spring.ai.experimental.mcp.spec.McpSession;
+import spring.ai.experimental.mcp.spec.McpTransport;
 import spring.ai.experimental.mcp.spec.McpSchema;
 
-public class McpAsyncClient extends McpAsyncSession {
+/**
+ * @author Dariusz Jędrzejczyk
+ * @author Christian Tzolov
+ */
+public class McpAsyncClient extends McpSession {
 
-	public McpAsyncClient(McpAsyncTransport transport) {
+	public McpAsyncClient(McpTransport transport) {
 		this(transport, Duration.ofSeconds(10), new ObjectMapper());
 	}
 
-	public McpAsyncClient(McpAsyncTransport transport, Duration requestTimeout, ObjectMapper objectMapper) {		
+	public McpAsyncClient(McpTransport transport, Duration requestTimeout, ObjectMapper objectMapper) {
 		super(requestTimeout, objectMapper, transport);
 	}
 
-
 	/**
-	 * The initialization phase MUST be the first interaction between client and server.
+	 * The initialization phase MUST be the first interaction between client and
+	 * server.
 	 * During this phase, the client and server:
 	 * <ul>
 	 * <li>Establish protocol version compatibility</li>
@@ -29,7 +33,8 @@ public class McpAsyncClient extends McpAsyncSession {
 	 * <li>Share implementation details</li>
 	 * </ul>
 	 * <br/>
-	 * The client MUST initiate this phase by sending an initialize request containing:
+	 * The client MUST initiate this phase by sending an initialize request
+	 * containing:
 	 * <ul>
 	 * <li>The protocol version the client supports</li>
 	 * <li>The client's capabilities</li>
@@ -38,7 +43,8 @@ public class McpAsyncClient extends McpAsyncSession {
 	 *
 	 * The server MUST respond with its own capabilities and information:
 	 * {@link McpSchema.ServerCapabilities}. <br/>
-	 * After successful initialization, the client MUST send an initialized notification
+	 * After successful initialization, the client MUST send an initialized
+	 * notification
 	 * to indicate it is ready to begin normal operations.
 	 *
 	 * <br/>
@@ -46,6 +52,7 @@ public class McpAsyncClient extends McpAsyncSession {
 	 * <a href=
 	 * "https://github.com/modelcontextprotocol/specification/blob/main/docs/specification/basic/lifecycle.md#initialization">Initialization
 	 * Spec</a>
+	 * 
 	 * @return the initialize result.
 	 */
 	public Mono<McpSchema.InitializeResult> initialize() {
@@ -54,8 +61,9 @@ public class McpAsyncClient extends McpAsyncSession {
 				new McpSchema.ClientCapabilities(null, new McpSchema.ClientCapabilities.RootCapabilities(true), null),
 				new McpSchema.Implementation("mcp-java-client", "0.0.1")); // @formatter:on
 
-		Mono<McpSchema.InitializeResult> result =
-				this.sendRequest("initialize", initializeRequest, new TypeReference<McpSchema.InitializeResult>() {});
+		Mono<McpSchema.InitializeResult> result = this.sendRequest("initialize", initializeRequest,
+				new TypeReference<McpSchema.InitializeResult>() {
+				});
 
 		return result.flatMap(initializeResult -> {
 			if (!McpSchema.LATEST_PROTOCOL_VERSION.equals(initializeResult.protocolVersion())) {
@@ -64,7 +72,7 @@ public class McpAsyncClient extends McpAsyncSession {
 								+ initializeResult.protocolVersion()));
 			} else {
 				return this.sendNotification("notifications/initialized", null)
-				           .thenReturn(initializeResult);
+						.thenReturn(initializeResult);
 			}
 		});
 	}
@@ -81,7 +89,8 @@ public class McpAsyncClient extends McpAsyncSession {
 	 */
 	public Mono<Void> ping() {
 		return this.sendRequest("ping", null,
-				new TypeReference<Void>() {});
+				new TypeReference<Void>() {
+				});
 	}
 
 	// --------------------------
@@ -89,21 +98,25 @@ public class McpAsyncClient extends McpAsyncSession {
 	// --------------------------
 	/**
 	 * Send a tools/call request.
+	 * 
 	 * @param callToolRequest the call tool request.
 	 * @return the call tool result.
 	 */
 	public Mono<McpSchema.CallToolResult> callTool(McpSchema.CallToolRequest callToolRequest) {
 		return this.sendRequest("tools/call", callToolRequest,
-				new TypeReference<McpSchema.CallToolResult>() {});
+				new TypeReference<McpSchema.CallToolResult>() {
+				});
 	}
 
 	/**
 	 * Send a tools/list request.
+	 * 
 	 * @return the list of tools result.
 	 */
 	public Mono<McpSchema.ListToolsResult> listTools(String cursor) {
 		return this.sendRequest("tools/list", new McpSchema.PaginatedRequest(cursor),
-				new TypeReference<McpSchema.ListToolsResult>() {});
+				new TypeReference<McpSchema.ListToolsResult>() {
+				});
 	}
 
 	// --------------------------
@@ -112,16 +125,19 @@ public class McpAsyncClient extends McpAsyncSession {
 
 	/**
 	 * Send a resources/list request.
+	 * 
 	 * @param cursor the cursor
 	 * @return the list of resources result.
 	 */
 	public Mono<McpSchema.ListResourcesResult> listResources(String cursor) {
 		return this.sendRequest("resources/list", new McpSchema.PaginatedRequest(cursor),
-				new TypeReference<McpSchema.ListResourcesResult>() {});
+				new TypeReference<McpSchema.ListResourcesResult>() {
+				});
 	}
 
 	/**
 	 * Send a resources/read request.
+	 * 
 	 * @param resource the resource to read
 	 * @return the resource content.
 	 */
@@ -131,12 +147,14 @@ public class McpAsyncClient extends McpAsyncSession {
 
 	/**
 	 * Send a resources/read request.
+	 * 
 	 * @param readResourceRequest the read resource request.
 	 * @return the resource content.
 	 */
 	public Mono<McpSchema.ReadResourceResult> readResource(McpSchema.ReadResourceRequest readResourceRequest) {
 		return this.sendRequest("resources/read", readResourceRequest,
-				new TypeReference<McpSchema.ReadResourceResult>() {});
+				new TypeReference<McpSchema.ReadResourceResult>() {
+				});
 	}
 
 	/**
@@ -144,16 +162,19 @@ public class McpAsyncClient extends McpAsyncSession {
 	 * 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.
 	 */
 	public Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplates(String cursor) {
 		return this.sendRequest("resources/templates/list", new McpSchema.PaginatedRequest(cursor),
-				new TypeReference<McpSchema.ListResourceTemplatesResult>() {});
+				new TypeReference<McpSchema.ListResourceTemplatesResult>() {
+				});
 	}
 
 	/**
-	 * List Changed Notification. When the list of available resources changes, servers
+	 * List Changed Notification. When the list of available resources changes,
+	 * servers
 	 * that declared the listChanged capability SHOULD send a notification:
 	 */
 	public Mono<Void> sendResourcesListChanged() {
@@ -161,26 +182,34 @@ public class McpAsyncClient extends McpAsyncSession {
 	}
 
 	/**
-	 * Subscriptions. The protocol supports optional subscriptions to resource changes.
-	 * Clients can subscribe to specific resources and receive notifications when they
+	 * 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.
-	 * @param subscribeRequest the subscribe request contains the uri of the resource to
-	 * subscribe to.
+	 * 
+	 * @param subscribeRequest the subscribe request contains the uri of the
+	 *                         resource to
+	 *                         subscribe to.
 	 */
 	public Mono<Void> subscribeResource(McpSchema.SubscribeRequest subscribeRequest) {
 		return this.sendRequest("resources/subscribe", subscribeRequest,
-				new TypeReference<Void>() {});
+				new TypeReference<Void>() {
+				});
 	}
 
 	/**
 	 * Send a resources/unsubscribe request.
-	 * @param unsubscribeRequest the unsubscribe request contains the uri of the resource
-	 * to unsubscribe from.
+	 * 
+	 * @param unsubscribeRequest the unsubscribe request contains the uri of the
+	 *                           resource
+	 *                           to unsubscribe from.
 	 */
 	public Mono<Void> unsubscribeResource(McpSchema.UnsubscribeRequest unsubscribeRequest) {
 		return this.sendRequest("resources/unsubscribe", unsubscribeRequest,
-				new TypeReference<Void>() {});
+				new TypeReference<Void>() {
+				});
 	}
 }

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

@@ -18,15 +18,14 @@ package spring.ai.experimental.mcp.client;
 import java.time.Duration;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import spring.ai.experimental.mcp.spec.McpAsyncTransport;
+import spring.ai.experimental.mcp.spec.McpTransport;
 
 /**
  * The MCP client is the main entry point for interacting with the Model Context Protocol
  * (MCP) server.
  * 
- * 
  * @author Christian Tzolov
- * @since 1.0.0
+ * @author Dariusz Jędrzejczyk
  */
 public class McpClient {
 
@@ -40,20 +39,20 @@ public class McpClient {
 	//                                               JSONtoPOJOCodec type
 	//           .sync();
 
-	public static McpAsyncClient async(McpAsyncTransport transport) {
+	public static McpAsyncClient async(McpTransport transport) {
 		return new McpAsyncClient(transport);
 	}
 
-	public static McpAsyncClient async(McpAsyncTransport transport, Duration requestTimeout,
+	public static McpAsyncClient async(McpTransport transport, Duration requestTimeout,
 			ObjectMapper objectMapper) {
 		return new McpAsyncClient(transport, requestTimeout, objectMapper);
 	}
 
-	public static McpSyncClient sync(McpAsyncTransport transport) {
+	public static McpSyncClient sync(McpTransport transport) {
 		return new McpSyncClient(async(transport));
 	}
 
-	public static McpSyncClient sync(McpAsyncTransport transport, Duration requestTimeout,
+	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/McpSyncClient.java

@@ -5,6 +5,10 @@ import java.time.Duration;
 import spring.ai.experimental.mcp.client.util.Assert;
 import spring.ai.experimental.mcp.spec.McpSchema;
 
+/**
+ * @author Dariusz Jędrzejczyk
+ * @author Christian Tzolov
+ */
 public class McpSyncClient implements AutoCloseable {
 
 	// TODO: Consider providing a client config to set this properly

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

@@ -13,6 +13,9 @@ import spring.ai.experimental.mcp.client.util.Assert;
 
 /**
  * Server parameters for stdio client.
+ * 
+ * @author Christian Tzolov
+ * @author Dariusz Jędrzejczyk
  */
 @JsonInclude(JsonInclude.Include.NON_ABSENT)
 public class ServerParameters {

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

@@ -25,7 +25,7 @@ import spring.ai.experimental.mcp.client.stdio.ServerParametersParser.McpServerC
 
 /**
  * @author Christian Tzolov
- * @since 1.0.0
+ * @author Dariusz Jędrzejczyk
  */
 
 public class ServerParametersParser {

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

@@ -21,6 +21,9 @@ import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCResponse;
 
 /**
  * Stdio client for communicating with a server process.
+ * 
+ * @author Christian Tzolov
+ * @author Dariusz Jędrzejczyk
  */
 public class StdioServerTransport extends DefaultMcpTransport {
 

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

@@ -25,9 +25,9 @@ import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCMessage;
 
 /**
  * @author Christian Tzolov
- * @since 1.0.0
+ * @author Dariusz Jędrzejczyk
  */
-public class DefaultMcpTransport implements McpAsyncTransport {
+public class DefaultMcpTransport implements McpTransport {
 
 	protected final ObjectMapper objectMapper;
 
@@ -93,7 +93,7 @@ public class DefaultMcpTransport implements McpAsyncTransport {
 		this.inboundMessageHandler = inboundMessageHandler;
 	}
 
-	public void setErrorHandler(Consumer<String> errorHandler) {
+	public void setInboundErrorHandler(Consumer<String> errorHandler) {
 		this.errorHandler = errorHandler;
 	}
 

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

@@ -1,39 +0,0 @@
-/* 
-* 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.spec;
-
-import java.util.function.Consumer;
-
-import reactor.core.publisher.Mono;
-import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCMessage;
-
-/**
- * @author Christian Tzolov
- * @since 1.0.0
- */
-
-public interface McpAsyncTransport extends AutoCloseable{
-    
-    void start();
-
-	void close();
-
-    void setInboudMessageHandler(Consumer<JSONRPCMessage> messageReader);
-
-	void setErrorHandler(Consumer<String> errorReader);
-
-	Mono<Void> sendMessage(JSONRPCMessage message);    
-}

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

@@ -0,0 +1,127 @@
+package spring.ai.experimental.mcp.spec;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoSink;
+import spring.ai.experimental.mcp.client.util.Assert;
+
+/**
+ * Implementation of the MCP client session.
+ * 
+ * @author Christian Tzolov
+ * @author Dariusz Jędrzejczyk
+ */
+public class McpSession {
+
+	private final ConcurrentHashMap<Object, MonoSink<McpSchema.JSONRPCResponse>> pendingResponses = new ConcurrentHashMap<>();
+
+	private final Duration requestTimeout;
+
+	private final ObjectMapper objectMapper;
+
+	private final McpTransport transport;
+
+	public McpSession(Duration requestTimeout, ObjectMapper objectMapper, McpTransport transport) {
+
+		Assert.notNull(objectMapper, "The ObjectMapper can not be null");
+		Assert.notNull(requestTimeout, "The requstTimeout can not be null");
+		Assert.notNull(transport, "The transport can not be null");
+
+		this.requestTimeout = requestTimeout;
+		this.objectMapper = objectMapper;
+		this.transport = transport;
+
+		this.transport.setInboudMessageHandler(message -> {
+			switch (message) {
+				case McpSchema.JSONRPCResponse response -> {
+					var sink = pendingResponses.remove(response.id());
+					if (sink == null) {
+						System.out.println("Unexpected response for unkown id " + response.id());
+					} else {
+						sink.success(response);
+					}
+				}
+				case McpSchema.JSONRPCRequest request -> {
+					System.out.println("Client does not yet support server requests");
+				}
+				case McpSchema.JSONRPCNotification notification -> {
+					System.out.println("Notifications not yet supported");
+				}
+			}
+		});
+
+		this.transport.setInboundErrorHandler(error -> System.out.println("Error received " + error));
+
+		this.transport.start();
+	}
+
+	public <T> Mono<T> sendRequest(String method, Object requestParams, TypeReference<T> typeRef) {
+		String requestId = UUID.randomUUID().toString();
+
+		return Mono.<McpSchema.JSONRPCResponse>create(sink -> {
+			this.pendingResponses.put(requestId, sink);
+			McpSchema.JSONRPCRequest jsonrpcRequest = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, method,
+					requestId, requestParams);
+			try {
+				// TODO: This is non-blocking, but it's actually a synchronous call,
+				// perhaps there's no need to make it return Mono?
+				this.transport.sendMessage(jsonrpcRequest)
+						// TODO: It's most efficient to create a dedicated
+						// Subscriber here
+						.subscribe(v -> {
+						}, e -> {
+							this.pendingResponses.remove(requestId);
+							sink.error(e);
+						});
+			} catch (Exception e) {
+				sink.error(e);
+			}
+		}).timeout(this.requestTimeout).handle((jsonRpcResponse, s) -> {
+			if (jsonRpcResponse.error() != null) {
+				s.error(new McpError(jsonRpcResponse.error()));
+			} else {
+				if (typeRef.getType().getTypeName().equals("java.lang.Void")) {
+					s.complete();
+				} else {
+					s.next(this.objectMapper.convertValue(jsonRpcResponse.result(), typeRef));
+				}
+			}
+		});
+	}
+
+	public static class McpError extends RuntimeException {
+
+		public McpError(Object error) {
+			super(error.toString());
+		}
+
+	}
+
+	public Mono<Void> sendNotification(String method) {
+		return sendNotification(method, null);
+	}
+
+	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));
+		}
+		return Mono.empty();
+	}
+
+	public Mono<Void> closeGracefully(Duration timeout) {
+		// TODO handle the timeout in transport
+		return Mono.fromRunnable(this.transport::close);
+	}
+
+}

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

@@ -0,0 +1,91 @@
+/* 
+ * 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.spec;
+
+import java.util.function.Consumer;
+
+import reactor.core.publisher.Mono;
+import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCMessage;
+
+/**
+ * Defines the asynchronous transport layer for the Model Context Protocol (MCP).
+ * 
+ * <p>The McpTransport interface provides the foundation for implementing custom transport
+ * mechanisms in the Model Context Protocol. It handles the bidirectional communication between
+ * the client and server components, supporting asynchronous message exchange using JSON-RPC format.</p>
+ * 
+ * <p>Implementations of this interface are responsible for:</p>
+ * <ul>
+ *   <li>Managing the lifecycle of the transport connection</li>
+ *   <li>Handling incoming messages and errors from the server</li>
+ *   <li>Sending outbound messages to the server</li>
+ * </ul>
+ * 
+ * <p>The transport layer is designed to be protocol-agnostic, allowing for various implementations
+ * such as WebSocket, HTTP, or custom protocols.</p>
+ *
+ * @author Christian Tzolov
+ * @author Dariusz Jędrzejczyk
+ */
+public interface McpTransport extends AutoCloseable {
+    
+    /**
+     * Initializes and starts the transport connection.
+     * 
+     * <p>This method should be called before any message exchange can occur. It sets up
+     * the necessary resources and establishes the connection to the server.</p>
+     */
+    void start();
+
+    /**
+     * Closes the transport connection and releases any associated resources.
+     * 
+     * <p>This method ensures proper cleanup of resources when the transport is no longer needed.
+     * It should handle the graceful shutdown of any active connections.</p>
+     */
+    void close();
+
+    /**
+     * Sets the handler for processing incoming messages from the server.
+     * 
+     * <p>The provided handler will be called whenever a new message is received from the server.
+     * Messages are expected to follow the JSON-RPC format as defined in the MCP specification.</p>
+     *
+     * @param inboundMessageHandler a consumer that processes incoming {@link JSONRPCMessage}s
+     */
+    void setInboudMessageHandler(Consumer<JSONRPCMessage> inboundMessageHandler);
+
+    /**
+     * Sets the handler for processing transport-level errors.
+     * 
+     * <p>The provided handler will be called when errors occur during transport operations,
+     * such as connection failures or protocol violations.</p>
+     *
+     * @param inboundErrorHandler a consumer that processes error messages
+     */
+    void setInboundErrorHandler(Consumer<String> inboundErrorHandler);
+
+    /**
+     * Sends a message to the server asynchronously.
+     * 
+     * <p>This method handles the transmission of messages to the server in an asynchronous manner.
+     * Messages are sent in JSON-RPC format as specified by the MCP protocol.</p>
+     *
+     * @param message the {@link JSONRPCMessage} to be sent to the server
+     * @return a {@link Mono<Void>} that completes when the message has been sent
+     */
+    Mono<Void> sendMessage(JSONRPCMessage message);    
+}

mcp/src/test/java/spring/ai/experimental/mcp/client/McpAsyncClientTests.java

@@ -23,8 +23,6 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
-import spring.ai.experimental.mcp.client.McpAsyncClient;
-import spring.ai.experimental.mcp.client.McpClient;
 import spring.ai.experimental.mcp.client.stdio.ServerParameters;
 import spring.ai.experimental.mcp.client.stdio.StdioServerTransport;
 import spring.ai.experimental.mcp.spec.McpSchema.CallToolRequest;

mcp/src/test/java/spring/ai/experimental/mcp/client/McpSyncClientTests.java

@@ -16,29 +16,26 @@
 package spring.ai.experimental.mcp.client;
 
 import java.time.Duration;
-import java.util.List;
 import java.util.Map;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
-import spring.ai.experimental.mcp.client.McpClient;
-import spring.ai.experimental.mcp.client.McpSyncClient;
 import spring.ai.experimental.mcp.client.stdio.ServerParameters;
 import spring.ai.experimental.mcp.client.stdio.StdioServerTransport;
 import spring.ai.experimental.mcp.spec.McpSchema.CallToolRequest;
 import spring.ai.experimental.mcp.spec.McpSchema.CallToolResult;
-import spring.ai.experimental.mcp.spec.McpSchema.Content;
 import spring.ai.experimental.mcp.spec.McpSchema.ListResourcesResult;
 import spring.ai.experimental.mcp.spec.McpSchema.ListToolsResult;
 import spring.ai.experimental.mcp.spec.McpSchema.Resource;
 import spring.ai.experimental.mcp.spec.McpSchema.TextContent;
 import spring.ai.experimental.mcp.spec.McpSchema.Tool;
 
-import static org.assertj.core.api.Assertions.*;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /**
  * Unit tests for MCP Client Session functionality.

mcp/src/test/java/spring/ai/experimental/mcp/spec/DefaultMcpTransportTests.java

@@ -107,7 +107,7 @@ class DefaultMcpTransportTests {
     @Test
     void customErrorHandlerShouldReceiveErrors() {
         AtomicReference<String> receivedError = new AtomicReference<>();
-        transport.setErrorHandler(receivedError::set);
+        transport.setInboundErrorHandler(receivedError::set);
         transport.start();
 
         String errorMessage = "Test error";
@@ -124,7 +124,7 @@ class DefaultMcpTransportTests {
         AtomicReference<String> receivedError = new AtomicReference<>();
 
         transport.setInboudMessageHandler(receivedMessage::set);
-        transport.setErrorHandler(receivedError::set);
+        transport.setInboundErrorHandler(receivedError::set);
         transport.start();
 
         JSONRPCRequest message = new JSONRPCRequest(