Refactor the mcp-core module

- The core module has been refactored to remove unnecessary abstraction
  between the SSE transport and the STDIO transport.
- Tests have been improved and no longer complain about multiple
  subscriptions to single-subscriber Sink.
- SSE transport layer is now assuming non-blocking execution and no longer
  makes thread hops when not needed.
- Decouple ObjectMapper and error handling from AbstractMcpTransport
  and remove later.

Additional improvements:

- Rename transport classes to better reflect client-side role (*ServerTransport -> *ClientTransport)
- Introduce fluent builder pattern for client configuration
- Simplify DefaultMcpSession by removing ObjectMapper dependency
- Move unmarshal responsibility to transport layer
- Update documentation and diagrams to reflect new architecture
- Clean up test implementations

README.md

@@ -1,10 +1,10 @@
 # Spring AI MCP
 
-Java SDK for the Model Context Protocol (MCP), providing seamless integration between Java and Spring applications and MCP-compliant AI models and tools.
+Java SDK for the Model Context Protocol (MCP), providing seamless integration between Java and Spring applications and MCP-compliant AI resources and tools.
 
 ## Overview
 
-Spring AI MCP is an experimental project that provides Java and Spring Framework integration for the [Model Context Protocol](https://modelcontextprotocol.org/docs/concepts/architecture). It enables Spring applications to interact with AI models and tools through a standardized interface, supporting both synchronous and asynchronous communication patterns.
+Spring AI MCP is an experimental project that provides Java and Spring Framework integration for the [Model Context Protocol](https://modelcontextprotocol.org/docs/concepts/architecture). It enables Java applications to interact with AI models and tools through a standardized interface, supporting both synchronous and asynchronous communication patterns.
 
 <img src="spring-ai-mcp-architecture.jpg" width="600">
 
@@ -12,22 +12,20 @@ Spring AI MCP is an experimental project that provides Java and Spring Framework
 
 The project consists of two main modules:
 
-### [spring-ai-mcp-core](./spring-ai-mcp-core/README.md)
+### [mcp-core](./spring-ai-mcp-core/README.md)
 
 The core module provides a Java implementation of the Model Context Protocol specification. It includes:
 - Synchronous and asynchronous client implementations
-- Standard MCP operations support (tool discovery, resource management, prompt handling)
+- Standard MCP operations support (tool discovery, resource management, prompt handling). Support for request and notificaiotn handling.
-- Stdio-based server transport
+- [Stdio](https://spec.modelcontextprotocol.io/specification/basic/transports/#stdio) and [SSE](https://spec.modelcontextprotocol.io/specification/basic/transports/#http-with-sse) transport implementations. 
-- Reactive programming support using Project Reactor
+- [Find more](./spring-ai-mcp-core/README.md).
 
-[find more](./spring-ai-mcp-core/README.md)
+### [mcp-spring](./spring-ai-mcp-spring/README.md)
-
-### [spring-ai-mcp-spring](./spring-ai-mcp-spring/README.md)
 
 The Spring integration module provides Spring-specific functionality:
 - Integration with Spring AI's function calling system
 - Spring-friendly abstractions for MCP clients
-- Automatic conversion between JSON and Java objects for tool arguments
+- Auto-configurations (WIP)
 
 ## Requirements
 

spring-ai-mcp-core/README.md

@@ -20,7 +20,6 @@ This SDK implements the Model Context Protocol, enabling seamless integration wi
 - Multiple transport implementations:
   - Stdio-based transport for process-based communication
   - SSE-based transport for HTTP streaming
-- Reactive programming support using Project Reactor
 - Configurable request timeouts
 - Customizable JSON serialization/deserialization
 
@@ -32,43 +31,37 @@ Add the following dependency to your Maven project:
 <dependency>
     <groupId>org.springframework.experimental</groupId>
     <artifactId>spring-ai-mcp-core</artifactId>
-    <version>0.1.0</version>
+    <version>0.2.0-SNAPSHOT</version>
 </dependency>
 ```
 
-## Documentation
-
-Detailed UML class diagrams showing the relationships between components can be found in [docs/class-diagrams.puml](docs/class-diagrams.puml). The diagrams include:
-- Core Components: Shows the main interfaces, classes, and their relationships
-- Message Flow: Illustrates the message and resource type hierarchies
-
 ## Usage
 
 ### Transport Layer Options
 
 The SDK provides two transport implementations:
 
-#### StdioServerTransport
+#### StdioClientTransport
 Standard I/O based transport for process-based communication with MCP servers:
 
 ```java
 ServerParameters params = ServerParameters.builder("npx")
     .args("-y", "@modelcontextprotocol/server-everything", "dir")
     .build();
-McpTransport transport = new StdioServerTransport(params);
+McpTransport transport = new StdioClientTransport(params);
 ```
 
-#### SseServerTransport
+#### SseClientTransport
 Server-Sent Events (SSE) based transport following the MCP HTTP with SSE transport specification:
 
 ```java
 WebClient.Builder webClientBuilder = WebClient.builder()
     .baseUrl("http://your-mcp-server");
-McpTransport transport = new SseServerTransport(webClientBuilder);
+McpTransport transport = new SseClientTransport(webClientBuilder);
 
 // Or with custom ObjectMapper
 ObjectMapper mapper = new ObjectMapper();
-McpTransport transport = new SseServerTransport(webClientBuilder, mapper);
+McpTransport transport = new SseClientTransport(webClientBuilder, mapper);
 ```
 
 The SSE transport provides:
@@ -87,7 +80,7 @@ ServerParameters params = ServerParameters.builder("npx")
     .args("-y", "@modelcontextprotocol/server-everything", "dir")
     .build();
 
-try (McpSyncClient client = McpClient.sync(new StdioServerTransport(params))) {
+try (McpSyncClient client = McpClient.sync(new StdioClientTransport(params))) {
     // Initialize connection with protocol version and capabilities
     McpSchema.InitializeResult initResult = client.initialize();
 
@@ -123,7 +116,7 @@ ServerParameters params = ServerParameters.builder("npx")
 
 // Initialize async client with custom timeout and object mapper
 McpAsyncClient client = McpClient.async(
-    new StdioServerTransport(params),
+    new StdioClientTransport(params),
     Duration.ofSeconds(30),
     new ObjectMapper()
 );
@@ -172,11 +165,11 @@ The SDK follows a layered architecture with clear separation of concerns:
 - **McpAsyncClient**: Primary async implementation using Project Reactor for non-blocking operations
 - **McpSyncClient**: Synchronous wrapper around the async client for blocking operations
 - **McpSession**: Core session interface defining communication patterns
-- **McpTransport**: Transport layer interface for server communication
+- **McpTransport**: Transport layer interface for client/server communication
 - **McpSchema**: Comprehensive protocol schema definitions
-- **DefaultMcpSession**: Base implementation of the session management
+- **DefaultMcpSession**: Default implementation of the session management
-- **StdioServerTransport**: Standard I/O based server communication
+- **StdioClientTransport**: Standard I/O based client to server communication
-- **SseServerTransport**: HTTP-based transport using Server-Sent Events for bidirectional communication
+- **SseClientTransport**: HTTP-based transport using Server-Sent Events for bidirectional client-sever communication
 
 <img src="docs/spring-ai-mcp-uml-classdiagram.svg" width="600"/>
 

spring-ai-mcp-core/docs/CLIENT.SSE.TRANSPORT.md

@@ -2,7 +2,7 @@
 
 ## Overview
 
-The `SseServerTransport` class implements the Model Context Protocol (MCP) HTTP with SSE transport specification, providing a bidirectional communication channel between clients and servers. This implementation is part of the Spring AI MCP Core library and follows the [MCP HTTP with SSE Transport Specification](https://spec.modelcontextprotocol.io/specification/basic/transports/#http-with-sse).
+The `SseClientTransport` class implements the Model Context Protocol (MCP) HTTP with SSE transport specification, providing a bidirectional communication channel between clients and servers. This implementation is part of the Spring AI MCP Core library and follows the [MCP HTTP with SSE Transport Specification](https://spec.modelcontextprotocol.io/specification/basic/transports/#http-with-sse).
 
 ## Key Features
 
@@ -53,7 +53,7 @@ WebClient.Builder webClientBuilder = WebClient.builder()
     .baseUrl("http://localhost:3001");
 
 // Initialize transport
-SseServerTransport transport = new SseServerTransport(webClientBuilder);
+SseClientTransport transport = new SseClientTransport(webClientBuilder);
 
 // Create a JSON-RPC request
 JSONRPCRequest request = new JSONRPCRequest(
@@ -141,4 +141,4 @@ The implementation includes comprehensive tests covering:
 - Shutdown procedures
 - Edge cases and failure scenarios
 
-For detailed test examples, see `SseServerTransportTests.java`.
+For detailed test examples, see `SseClientTransportTests.java`.

spring-ai-mcp-core/docs/CLIENT.STDIO.TRANSPORT.md

@@ -1,6 +1,6 @@
-# StdioServerTransport
+# StdioClientTransport
 
-The `StdioServerTransport` class implements the [MCP Stdio Transport](https://spec.modelcontextprotocol.io/specification/basic/transports/#stdio) specification, providing communication with MCP servers over standard input/output streams.
+The `StdioClientTransport` class implements the [MCP Stdio Transport](https://spec.modelcontextprotocol.io/specification/basic/transports/#stdio) specification, providing communication with MCP servers over standard input/output streams.
 
 ## Overview
 
@@ -84,7 +84,7 @@ ServerParameters params = ServerParameters.builder()
     .build();
 
 // Create transport
-StdioServerTransport transport = new StdioServerTransport(params);
+StdioClientTransport transport = new StdioClientTransport(params);
 
 // Start transport
 transport.start();

spring-ai-mcp-core/docs/class-diagrams.puml

@@ -1,25 +1,18 @@
 @startuml Core Components
 
 interface McpTransport {
-  +void start()
+  +Mono<Void> connect(Function<Mono<JSONRPCMessage>, 
+                       Mono<JSONRPCMessage>> handler)
+  +Mono<Void> sendMessage()
   +void close()
   +Mono<Void> closeGracefully()
-  +void setInboudMessageHandler()
-  +void setInboundErrorHandler()
-  +Mono<Void> sendMessage()
-}
-
-abstract class AbstractMcpTransport {
-  #ObjectMapper objectMapper
-  #getInboundSink()
-  #getOutboundSink()
-  #getErrorSink()
 }
 
 interface McpSession {
-  +<T> Mono<T> sendRequest()
+  +<T> Mono<T> sendRequest(String method, Object requestParams, TypeReference<T> typeRef)
-  +Mono<Void> sendNotification()
+  +Mono<Void> sendNotification(String method, Map<String, Object> params)
   +Mono<Void> closeGracefully()
+  +close()
 }
 
 class DefaultMcpSession {
@@ -52,12 +45,12 @@ class McpSyncClient {
   +GetPromptResult getPrompt()
 }
 
-class StdioServerTransport {
+class StdioClientTransport {
   -Process serverProcess
   -ServerParameters parameters
 }
 
-class SseServerTransport {
+class SseClientTransport {
   -WebClient webClient
 }
 
@@ -74,11 +67,10 @@ class McpError {
   -String message
 }
 
-McpTransport <|.. AbstractMcpTransport
+McpTransport <|-- StdioClientTransport
-AbstractMcpTransport <|-- StdioServerTransport
+McpTransport <|-- SseClientTransport
-AbstractMcpTransport <|-- SseServerTransport
 McpSession <|-- DefaultMcpSession
-DefaultMcpSession <|-- McpAsyncClient
+DefaultMcpSession --o McpAsyncClient
 McpClient ..> McpAsyncClient : creates
 McpClient ..> McpSyncClient : creates
 McpSyncClient o-- McpAsyncClient

spring-ai-mcp-core/src/main/java/org/springframework/ai/mcp/client/McpAsyncClient.java

@@ -18,7 +18,6 @@ package org.springframework.ai.mcp.client;
 import java.time.Duration;
 
 import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
@@ -33,22 +32,41 @@ import org.springframework.ai.mcp.spec.McpSchema.PaginatedRequest;
 import org.springframework.ai.mcp.spec.McpTransport;
 
 /**
+ * The Model Context Protocol (MCP) client implementation that provides asynchronous
+ * communication with MCP servers.
+ *
  * @author Dariusz Jędrzejczyk
  * @author Christian Tzolov
  */
-public class McpAsyncClient extends DefaultMcpSession {
+public class McpAsyncClient {
 
 	private final static Logger logger = LoggerFactory.getLogger(McpAsyncClient.class);
 
 	private static TypeReference<Void> VOID_TYPE_REFERENCE = new TypeReference<>() {
 	};
 
+	/**
+	 * The MCP session implementation that manages bidirectional JSON-RPC communication
+	 * between clients and servers.
+	 */
+	private final DefaultMcpSession mcpSession;
+
+	/**
+	 * Create a new McpAsyncClient with the given transport.
+	 * @param transport the transport to use.
+	 */
 	public McpAsyncClient(McpTransport transport) {
-		this(transport, Duration.ofSeconds(20), new ObjectMapper());
+		this(transport, Duration.ofSeconds(20));
 	}
 
-	public McpAsyncClient(McpTransport transport, Duration requestTimeout, ObjectMapper objectMapper) {
+	/**
-		super(requestTimeout, objectMapper, transport);
+	 * Create a new McpAsyncClient with the given transport and session request-response
+	 * timeout.
+	 * @param transport the transport to use.
+	 * @param requestTimeout the session request-response timeout.
+	 */
+	public McpAsyncClient(McpTransport transport, Duration requestTimeout) {
+		this.mcpSession = new DefaultMcpSession(requestTimeout, transport);
 	}
 
 	/**
@@ -85,7 +103,7 @@ public class McpAsyncClient extends DefaultMcpSession {
 				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,
+		Mono<McpSchema.InitializeResult> result = this.mcpSession.sendRequest("initialize", initializeRequest,
 				new TypeReference<McpSchema.InitializeResult>() {
 				});
 
@@ -100,7 +118,7 @@ public class McpAsyncClient extends DefaultMcpSession {
 						"Unsupported protocol version from the server: " + initializeResult.protocolVersion()));
 			}
 			else {
-				return this.sendNotification("notifications/initialized", null).thenReturn(initializeResult);
+				return this.mcpSession.sendNotification("notifications/initialized", null).thenReturn(initializeResult);
 			}
 		});
 	}
@@ -109,14 +127,14 @@ public class McpAsyncClient extends DefaultMcpSession {
 	 * Send a roots/list_changed notification.
 	 */
 	public Mono<Void> sendRootsListChanged() {
-		return this.sendNotification("notifications/roots/list_changed");
+		return this.mcpSession.sendNotification("notifications/roots/list_changed");
 	}
 
 	/**
 	 * Send a synchronous ping request.
 	 */
 	public Mono<Void> ping() {
-		return this.sendRequest("ping", null, VOID_TYPE_REFERENCE);
+		return this.mcpSession.sendRequest("ping", null, VOID_TYPE_REFERENCE);
 	}
 
 	// --------------------------
@@ -134,7 +152,7 @@ public class McpAsyncClient extends DefaultMcpSession {
 	 * @return the call tool result.
 	 */
 	public Mono<McpSchema.CallToolResult> callTool(McpSchema.CallToolRequest callToolRequest) {
-		return this.sendRequest("tools/call", callToolRequest, CALL_TOOL_RESULT_TYPE_REF);
+		return this.mcpSession.sendRequest("tools/call", callToolRequest, CALL_TOOL_RESULT_TYPE_REF);
 	}
 
 	/**
@@ -151,7 +169,8 @@ public class McpAsyncClient extends DefaultMcpSession {
 	 * @return the list of tools result.
 	 */
 	public Mono<McpSchema.ListToolsResult> listTools(String cursor) {
-		return this.sendRequest("tools/list", new McpSchema.PaginatedRequest(cursor), LIST_TOOLS_RESULT_TYPE_REF);
+		return this.mcpSession.sendRequest("tools/list", new McpSchema.PaginatedRequest(cursor),
+				LIST_TOOLS_RESULT_TYPE_REF);
 	}
 
 	// --------------------------
@@ -181,7 +200,7 @@ public class McpAsyncClient extends DefaultMcpSession {
 	 * @return the list of resources result.
 	 */
 	public Mono<McpSchema.ListResourcesResult> listResources(String cursor) {
-		return this.sendRequest("resources/list", new McpSchema.PaginatedRequest(cursor),
+		return this.mcpSession.sendRequest("resources/list", new McpSchema.PaginatedRequest(cursor),
 				LIST_RESOURCES_RESULT_TYPE_REF);
 	}
 
@@ -200,7 +219,7 @@ public class McpAsyncClient extends DefaultMcpSession {
 	 * @return the resource content.
 	 */
 	public Mono<McpSchema.ReadResourceResult> readResource(McpSchema.ReadResourceRequest readResourceRequest) {
-		return this.sendRequest("resources/read", readResourceRequest, READ_RESOURCE_RESULT_TYPE_REF);
+		return this.mcpSession.sendRequest("resources/read", readResourceRequest, READ_RESOURCE_RESULT_TYPE_REF);
 	}
 
 	/**
@@ -223,7 +242,7 @@ public class McpAsyncClient extends DefaultMcpSession {
 	 * @return the list of resource templates result.
 	 */
 	public Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplates(String cursor) {
-		return this.sendRequest("resources/templates/list", new McpSchema.PaginatedRequest(cursor),
+		return this.mcpSession.sendRequest("resources/templates/list", new McpSchema.PaginatedRequest(cursor),
 				LIST_RESOURCE_TEMPLATES_RESULT_TYPE_REF);
 	}
 
@@ -232,7 +251,7 @@ public class McpAsyncClient extends DefaultMcpSession {
 	 * that declared the listChanged capability SHOULD send a notification:
 	 */
 	public Mono<Void> sendResourcesListChanged() {
-		return this.sendNotification("notifications/resources/list_changed");
+		return this.mcpSession.sendNotification("notifications/resources/list_changed");
 	}
 
 	/**
@@ -245,7 +264,7 @@ public class McpAsyncClient extends DefaultMcpSession {
 	 * subscribe to.
 	 */
 	public Mono<Void> subscribeResource(McpSchema.SubscribeRequest subscribeRequest) {
-		return this.sendRequest("resources/subscribe", subscribeRequest, VOID_TYPE_REFERENCE);
+		return this.mcpSession.sendRequest("resources/subscribe", subscribeRequest, VOID_TYPE_REFERENCE);
 	}
 
 	/**
@@ -254,7 +273,7 @@ public class McpAsyncClient extends DefaultMcpSession {
 	 * to unsubscribe from.
 	 */
 	public Mono<Void> unsubscribeResource(McpSchema.UnsubscribeRequest unsubscribeRequest) {
-		return this.sendRequest("resources/unsubscribe", unsubscribeRequest, VOID_TYPE_REFERENCE);
+		return this.mcpSession.sendRequest("resources/unsubscribe", unsubscribeRequest, VOID_TYPE_REFERENCE);
 	}
 
 	// --------------------------
@@ -280,7 +299,7 @@ public class McpAsyncClient extends DefaultMcpSession {
 	 * @return the list of prompts result.
 	 */
 	public Mono<ListPromptsResult> listPrompts(String cursor) {
-		return this.sendRequest("prompts/list", new PaginatedRequest(cursor), LIST_PROMPTS_RESULT_TYPE_REF);
+		return this.mcpSession.sendRequest("prompts/list", new PaginatedRequest(cursor), LIST_PROMPTS_RESULT_TYPE_REF);
 	}
 
 	/**
@@ -289,7 +308,7 @@ public class McpAsyncClient extends DefaultMcpSession {
 	 * @return the get prompt result.
 	 */
 	public Mono<GetPromptResult> getPrompt(GetPromptRequest getPromptRequest) {
-		return this.sendRequest("prompts/get", getPromptRequest, GET_PROMPT_RESULT_TYPE_REF);
+		return this.mcpSession.sendRequest("prompts/get", getPromptRequest, GET_PROMPT_RESULT_TYPE_REF);
 	}
 
 	/**
@@ -298,7 +317,15 @@ public class McpAsyncClient extends DefaultMcpSession {
 	 * any previous subscription from the client.
 	 */
 	public Mono<Void> promptListChangedNotification() {
-		return this.sendNotification("notifications/prompts/list_changed");
+		return this.mcpSession.sendNotification("notifications/prompts/list_changed");
+	}
+
+	public void close() {
+		this.mcpSession.close();
+	}
+
+	public Mono<Void> closeGracefully() {
+		return this.mcpSession.closeGracefully();
 	}
 
 }

spring-ai-mcp-core/src/main/java/org/springframework/ai/mcp/client/McpClient.java

@@ -18,9 +18,8 @@ package org.springframework.ai.mcp.client;
 
 import java.time.Duration;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-
 import org.springframework.ai.mcp.spec.McpTransport;
+import org.springframework.ai.mcp.util.Assert;
 
 /**
  * Factory class providing static methods for creating Model Context Protocol (MCP)
@@ -39,8 +38,11 @@ import org.springframework.ai.mcp.spec.McpTransport;
  * including request timeout and JSON object mapping.
  *
  * <p>
- * Future implementations will introduce a builder pattern for more flexible client
+ * Use the builder pattern for flexible client configuration: <pre>{@code
- * configuration.
+ * McpClient.using(transport)
+ *     .withRequestTimeout(Duration.ofSeconds(5))
+ *     .sync(); // or .async()
+ * }</pre>
  *
  * @author Christian Tzolov
  * @author Dariusz Jędrzejczyk
@@ -54,55 +56,88 @@ public class McpClient {
 	private McpClient() {
 	}
 
-	// TODO: introduce a builder like:
+	/**
-	// McpClient.using(transport)
+	 * Start building an MCP client with the specified transport.
-	// .withRequestTimeout(Duration.ofSeconds(5))
+	 * @param transport The transport layer implementation for MCP communication
-	// .withObjectMapper(objectMapper); <- or even a more sophisticated
+	 * @return A new builder instance
-	// JSONtoPOJOCodec type
+	 */
-	// .sync();
+	public static Builder using(McpTransport transport) {
+		return new Builder(transport);
+	}
 
 	/**
-	 * Creates an asynchronous MCP client with default configuration.
+	 * Builder class for creating MCP clients with custom configuration.
-	 * @param transport The transport layer implementation for MCP communication
-	 * @return A new instance of {@link McpAsyncClient}
 	 */
+	public static class Builder {
+
+		private final McpTransport transport;
+
+		private Duration requestTimeout = Duration.ofSeconds(20); // Default timeout
+
+		private Builder(McpTransport transport) {
+			Assert.notNull(transport, "Transport must not be null");
+			this.transport = transport;
+		}
+
+		/**
+		 * Set the request timeout duration.
+		 * @param requestTimeout The duration to wait before timing out requests
+		 * @return This builder instance
+		 */
+		public Builder withRequestTimeout(Duration requestTimeout) {
+			Assert.notNull(requestTimeout, "Request timeout must not be null");
+			this.requestTimeout = requestTimeout;
+			return this;
+		}
+
+		/**
+		 * Build a synchronous MCP client.
+		 * @return A new instance of {@link McpSyncClient}
+		 */
+		public McpSyncClient sync() {
+			return new McpSyncClient(async());
+		}
+
+		/**
+		 * Build an asynchronous MCP client.
+		 * @return A new instance of {@link McpAsyncClient}
+		 */
+		public McpAsyncClient async() {
+			return new McpAsyncClient(transport, requestTimeout);
+		}
+
+	}
+
+	/**
+	 * @deprecated Use {@link #using(McpTransport)} instead.
+	 */
+	@Deprecated
 	public static McpAsyncClient async(McpTransport transport) {
-		return new McpAsyncClient(transport);
+		return using(transport).async();
 	}
 
 	/**
-	 * Creates an asynchronous MCP client with custom configuration.
+	 * @deprecated Use {@link #using(McpTransport)} instead.
-	 * @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) {
+	@Deprecated
-		return new McpAsyncClient(transport, requestTimeout, objectMapper);
+	public static McpAsyncClient async(McpTransport transport, Duration requestTimeout) {
+		return using(transport).withRequestTimeout(requestTimeout).async();
 	}
 
 	/**
-	 * Creates a synchronous MCP client with default configuration. This method wraps an
+	 * @deprecated Use {@link #using(McpTransport)} instead.
-	 * asynchronous client to provide synchronous operations.
-	 * @param transport The transport layer implementation for MCP communication
-	 * @return A new instance of {@link McpSyncClient}
 	 */
+	@Deprecated
 	public static McpSyncClient sync(McpTransport transport) {
-		return new McpSyncClient(async(transport));
+		return using(transport).sync();
 	}
 
 	/**
-	 * Creates a synchronous MCP client with custom configuration. This method wraps an
+	 * @deprecated Use {@link #using(McpTransport)} instead.
-	 * 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) {
+	@Deprecated
-		return new McpSyncClient(async(transport, requestTimeout, objectMapper));
+	public static McpSyncClient sync(McpTransport transport, Duration requestTimeout) {
+		return using(transport).withRequestTimeout(requestTimeout).sync();
 	}
 
 }

spring-ai-mcp-core/src/main/java/org/springframework/ai/mcp/client/McpSyncClient.java

@@ -17,12 +17,16 @@
 package org.springframework.ai.mcp.client;
 
 import java.time.Duration;
+import java.util.concurrent.TimeoutException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import org.springframework.ai.mcp.client.util.Assert;
 import org.springframework.ai.mcp.spec.McpSchema;
 import org.springframework.ai.mcp.spec.McpSchema.GetPromptRequest;
 import org.springframework.ai.mcp.spec.McpSchema.GetPromptResult;
 import org.springframework.ai.mcp.spec.McpSchema.ListPromptsResult;
+import org.springframework.ai.mcp.util.Assert;
 
 /**
  * @author Dariusz Jędrzejczyk
@@ -30,6 +34,8 @@ import org.springframework.ai.mcp.spec.McpSchema.ListPromptsResult;
  */
 public class McpSyncClient implements AutoCloseable {
 
+	private static final Logger logger = LoggerFactory.getLogger(McpSyncClient.class);
+
 	// TODO: Consider providing a client config to set this properly
 	// this is currently a concern only because AutoCloseable is used - perhaps it
 	// is not a requirement?
@@ -44,7 +50,18 @@ public class McpSyncClient implements AutoCloseable {
 
 	@Override
 	public void close() {
-		this.delegate.closeGracefully().block(Duration.ofMillis(DEFAULT_CLOSE_TIMEOUT_MS));
+		this.delegate.close();
+	}
+
+	public boolean closeGracefully() {
+		try {
+			this.delegate.closeGracefully().block(Duration.ofMillis(DEFAULT_CLOSE_TIMEOUT_MS));
+		}
+		catch (RuntimeException e) {
+			logger.warn("Client didn't close within timeout of {} ms.", DEFAULT_CLOSE_TIMEOUT_MS, e);
+			return false;
+		}
+		return true;
 	}
 
 	/**

spring-ai-mcp-core/src/main/java/org/springframework/ai/mcp/client/sse/SseClientTransport.java

@@ -0,0 +1,289 @@
+/*
+ * 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 org.springframework.ai.mcp.client.sse;
+
+import java.io.IOException;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+import reactor.core.publisher.SynchronousSink;
+import reactor.core.scheduler.Schedulers;
+import reactor.util.retry.Retry;
+import reactor.util.retry.Retry.RetrySignal;
+
+import org.springframework.ai.mcp.spec.McpError;
+import org.springframework.ai.mcp.spec.McpSchema;
+import org.springframework.ai.mcp.spec.McpSchema.JSONRPCMessage;
+import org.springframework.ai.mcp.spec.McpTransport;
+import org.springframework.ai.mcp.util.Assert;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.ServerSentEvent;
+import org.springframework.web.reactive.function.client.WebClient;
+
+/**
+ * Server-Sent Events (SSE) implementation of the
+ * {@link org.springframework.ai.mcp.spec.McpTransport} that follows the MCP HTTP with SSE
+ * transport specification.
+ *
+ * <p>
+ * This transport establishes a bidirectional communication channel where:
+ * <ul>
+ * <li>Inbound messages are received through an SSE connection from the server</li>
+ * <li>Outbound messages are sent via HTTP POST requests to a server-provided
+ * endpoint</li>
+ * </ul>
+ *
+ * <p>
+ * The message flow follows these steps:
+ * <ol>
+ * <li>The client establishes an SSE connection to the server's /sse endpoint</li>
+ * <li>The server sends an 'endpoint' event containing the URI for sending messages</li>
+ *
+ * <p>
+ * This implementation handles automatic reconnection for transient failures and provides
+ * graceful shutdown capabilities. It uses {@link WebClient} for HTTP communications and
+ * supports JSON serialization/deserialization of messages.
+ *
+ * @author Christian Tzolov
+ * @see <a href=
+ * "https://spec.modelcontextprotocol.io/specification/basic/transports/#http-with-sse">MCP
+ * HTTP with SSE Transport Specification</a>
+ */
+public class SseClientTransport implements McpTransport {
+
+	private final static Logger logger = LoggerFactory.getLogger(SseClientTransport.class);
+
+	/**
+	 * Event type for JSON-RPC messages received through the SSE connection. The server
+	 * sends messages with this event type to transmit JSON-RPC protocol data.
+	 */
+	private final static String MESSAGE_EVENT_TYPE = "message";
+
+	/**
+	 * Event type for receiving the message endpoint URI from the server. The server MUST
+	 * send this event when a client connects, providing the URI where the client should
+	 * send its messages via HTTP POST.
+	 */
+	private final static String ENDPOINT_EVENT_TYPE = "endpoint";
+
+	/**
+	 * Default SSE endpoint path as specified by the MCP transport specification. This
+	 * endpoint is used to establish the SSE connection with the server.
+	 */
+	private final static String SSE_ENDPOINT = "/sse";
+
+	/**
+	 * Type reference for parsing SSE events containing string data.
+	 */
+	private final static ParameterizedTypeReference<ServerSentEvent<String>> SSE_TYPE = new ParameterizedTypeReference<>() {
+	};
+
+	/**
+	 * WebClient instance for handling both SSE connections and HTTP POST requests. Used
+	 * for establishing the SSE connection and sending outbound messages.
+	 */
+	private final WebClient webClient;
+
+	/**
+	 * ObjectMapper for serializing outbound messages and deserializing inbound messages.
+	 * Handles conversion between JSON-RPC messages and their string representation.
+	 */
+	protected ObjectMapper objectMapper;
+
+	/**
+	 * Subscription for the SSE connection handling inbound messages. Used for cleanup
+	 * during transport shutdown.
+	 */
+	private Disposable inboundSubscription;
+
+	/**
+	 * Flag indicating if the transport is in the process of shutting down. Used to
+	 * prevent new operations during shutdown and handle cleanup gracefully.
+	 */
+	private volatile boolean isClosing = false;
+
+	/**
+	 * Sink for managing the message endpoint URI provided by the server. Stores the most
+	 * recent endpoint URI and makes it available for outbound message processing.
+	 */
+	protected final Sinks.One<String> messageEndpointSink = Sinks.one();
+
+	/**
+	 * Constructs a new SseClientTransport with the specified WebClient builder. Uses a
+	 * default ObjectMapper instance for JSON processing.
+	 * @param webClientBuilder the WebClient.Builder to use for creating the WebClient
+	 * instance
+	 * @throws IllegalArgumentException if webClientBuilder is null
+	 */
+	public SseClientTransport(WebClient.Builder webClientBuilder) {
+		this(webClientBuilder, new ObjectMapper());
+	}
+
+	/**
+	 * Constructs a new SseClientTransport with the specified WebClient builder and
+	 * ObjectMapper. Initializes both inbound and outbound message processing pipelines.
+	 * @param webClientBuilder the WebClient.Builder to use for creating the WebClient
+	 * instance
+	 * @param objectMapper the ObjectMapper to use for JSON processing
+	 * @throws IllegalArgumentException if either parameter is null
+	 */
+	public SseClientTransport(WebClient.Builder webClientBuilder, ObjectMapper objectMapper) {
+		Assert.notNull(objectMapper, "ObjectMapper must not be null");
+		Assert.notNull(webClientBuilder, "WebClient.Builder must not be null");
+
+		this.objectMapper = objectMapper;
+		this.webClient = webClientBuilder.build();
+	}
+
+	@Override
+	public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> handler) {
+		Flux<ServerSentEvent<String>> events = eventStream();
+		this.inboundSubscription = events.concatMap(event -> Mono.just(event).<JSONRPCMessage>handle((e, s) -> {
+			if (ENDPOINT_EVENT_TYPE.equals(event.event())) {
+				String messageEndpointUri = event.data();
+				if (messageEndpointSink.tryEmitValue(messageEndpointUri).isSuccess()) {
+					s.complete();
+				}
+				else {
+					// TODO: clarify with the spec if multiple events can be
+					// received
+					s.error(new McpError("Failed to handle SSE endpoint event"));
+				}
+			}
+			else if (MESSAGE_EVENT_TYPE.equals(event.event())) {
+				try {
+					JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(this.objectMapper, event.data());
+					s.next(message);
+				}
+				catch (IOException ioException) {
+					s.error(ioException);
+				}
+			}
+			else {
+				s.error(new McpError("Received unrecognized SSE event type: " + event.event()));
+			}
+		}).transform(handler)).subscribe();
+
+		// The connection is established once the server sends the endpoint event
+		return messageEndpointSink.asMono().then();
+	}
+
+	@Override
+	public Mono<Void> sendMessage(JSONRPCMessage message) {
+		// The messageEndpoint is the endpoint URI to send the messages
+		// It is provided by the server as part of the endpoint event
+		return messageEndpointSink.asMono().flatMap(messageEndpointUri -> {
+			if (isClosing) {
+				return Mono.empty();
+			}
+			try {
+				String jsonText = this.objectMapper.writeValueAsString(message);
+				return webClient.post()
+					.uri(messageEndpointUri)
+					.contentType(MediaType.APPLICATION_JSON)
+					.bodyValue(jsonText)
+					.retrieve()
+					.toBodilessEntity()
+					.doOnSuccess(response -> {
+						logger.debug("Message sent successfully");
+					})
+					.doOnError(error -> {
+						if (!isClosing) {
+							logger.error("Error sending message: {}", error.getMessage());
+						}
+					});
+			}
+			catch (IOException e) {
+				if (!isClosing) {
+					return Mono.error(new RuntimeException("Failed to serialize message", e));
+				}
+				return Mono.empty();
+			}
+		}).then(); // TODO: Consider non-200-ok response
+	}
+
+	/**
+	 * Initializes and starts the inbound SSE event processing. Establishes the SSE
+	 * connection and sets up event handling for both message and endpoint events.
+	 * Includes automatic retry logic for handling transient connection failures.
+	 */
+	// visible for tests
+	protected Flux<ServerSentEvent<String>> eventStream() {// @formatter:off
+		return this.webClient
+			.get()
+			.uri(SSE_ENDPOINT)
+			.accept(MediaType.TEXT_EVENT_STREAM)
+			.retrieve()
+			.bodyToFlux(SSE_TYPE)
+			.retryWhen(Retry.from(retrySignal -> retrySignal.handle(inboundRetryHandler)));
+	} // @formatter:on
+
+	/**
+	 * Retry handler for the inbound SSE stream. Implements the retry logic for handling
+	 * connection failures and other errors.
+	 */
+	private BiConsumer<RetrySignal, SynchronousSink<Object>> inboundRetryHandler = (retrySpec, sink) -> {
+		if (isClosing) {
+			logger.debug("SSE connection closed during shutdown");
+			sink.error(retrySpec.failure());
+			return;
+		}
+		if (retrySpec.failure() instanceof IOException) {
+			logger.debug("Retrying SSE connection after IO error");
+			sink.next(retrySpec);
+			return;
+		}
+		logger.error("Fatal SSE error, not retrying: {}", retrySpec.failure().getMessage());
+		sink.error(retrySpec.failure());
+	};
+
+	/**
+	 * Implements graceful shutdown of the transport. Cleans up all resources including
+	 * subscriptions and schedulers. Ensures orderly shutdown of both inbound and outbound
+	 * message processing.
+	 * @return a Mono that completes when shutdown is finished
+	 */
+	@Override
+	public Mono<Void> closeGracefully() { // @formatter:off
+		return Mono.fromRunnable(() -> {
+			isClosing = true;
+			
+			// Dispose of subscriptions
+			
+			if (inboundSubscription != null) {
+				inboundSubscription.dispose();
+			}
+
+		})
+		.then()
+		.subscribeOn(Schedulers.boundedElastic());
+	} // @formatter:on
+
+	@Override
+	public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
+		return this.objectMapper.convertValue(data, typeRef);
+	}
+
+}

spring-ai-mcp-core/src/main/java/org/springframework/ai/mcp/client/stdio/ServerParameters.java

@@ -26,7 +26,7 @@ import java.util.stream.Collectors;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-import org.springframework.ai.mcp.client.util.Assert;
+import org.springframework.ai.mcp.util.Assert;
 
 /**
  * Server parameters for stdio client.

spring-ai-mcp-core/src/main/java/org/springframework/ai/mcp/client/stdio/StdioClientTransport.java

@@ -0,0 +1,345 @@
+/*
+ * 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 org.springframework.ai.mcp.client.stdio;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
+
+import org.springframework.ai.mcp.spec.McpSchema;
+import org.springframework.ai.mcp.spec.McpSchema.JSONRPCMessage;
+import org.springframework.ai.mcp.util.Assert;
+import org.springframework.ai.mcp.spec.McpTransport;
+
+/**
+ * Implementation of the MCP Stdio transport that communicates with a server process using
+ * standard input/output streams. Messages are exchanged as newline-delimited JSON-RPC
+ * messages over stdin/stdout, with errors and debug information sent to stderr.
+ *
+ * @author Christian Tzolov
+ * @author Dariusz Jędrzejczyk
+ */
+public class StdioClientTransport implements McpTransport {
+
+	private static final Logger logger = LoggerFactory.getLogger(StdioClientTransport.class);
+
+	private final Sinks.Many<JSONRPCMessage> inboundSink;
+
+	private final Sinks.Many<JSONRPCMessage> outboundSink;
+
+	/** The server process being communicated with */
+	private Process process;
+
+	private ObjectMapper objectMapper;
+
+	/** Scheduler for handling inbound messages from the server process */
+	private Scheduler inboundScheduler;
+
+	/** Scheduler for handling outbound messages to the server process */
+	private Scheduler outboundScheduler;
+
+	/** Scheduler for handling error messages from the server process */
+	private Scheduler errorScheduler;
+
+	/** Parameters for configuring and starting the server process */
+	private final ServerParameters params;
+
+	private final Sinks.Many<String> errorSink;
+
+	// visible for tests
+	Consumer<String> errorHandler = error -> logger.error("Error received: {}", error);
+
+	/**
+	 * Creates a new StdioClientTransport with the specified parameters and default
+	 * ObjectMapper.
+	 * @param params The parameters for configuring the server process
+	 */
+	public StdioClientTransport(ServerParameters params) {
+		this(params, new ObjectMapper());
+	}
+
+	/**
+	 * Creates a new StdioClientTransport with the specified parameters and ObjectMapper.
+	 * @param params The parameters for configuring the server process
+	 * @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
+	 */
+	public StdioClientTransport(ServerParameters params, ObjectMapper objectMapper) {
+		Assert.notNull(params, "The params can not be null");
+		Assert.notNull(objectMapper, "The ObjectMapper can not be null");
+
+		this.inboundSink = Sinks.many().unicast().onBackpressureBuffer();
+		this.outboundSink = Sinks.many().unicast().onBackpressureBuffer();
+
+		this.params = params;
+
+		this.objectMapper = objectMapper;
+
+		this.errorSink = Sinks.many().unicast().onBackpressureBuffer();
+
+		// Start threads
+		this.inboundScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(), "inbound");
+		this.outboundScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(), "outbound");
+		this.errorScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(), "error");
+	}
+
+	/**
+	 * Starts the server process and initializes the message processing streams. This
+	 * method sets up the process with the configured command, arguments, and environment,
+	 * then starts the inbound, outbound, and error processing threads.
+	 * @throws RuntimeException if the process fails to start or if the process streams
+	 * are null
+	 */
+	@Override
+	public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> handler) {
+		return Mono.<Void>fromRunnable(() -> {
+			handleIncomingMessages(handler);
+			handleIncomingErrors();
+
+			// Prepare command and environment
+			List<String> fullCommand = new ArrayList<>();
+			fullCommand.add(params.getCommand());
+			fullCommand.addAll(params.getArgs());
+
+			ProcessBuilder processBuilder = this.getProcessBuilder();
+			processBuilder.command(fullCommand);
+			processBuilder.environment().putAll(params.getEnv());
+
+			// Start the process
+			try {
+				this.process = processBuilder.start();
+			}
+			catch (IOException e) {
+				throw new RuntimeException("Failed to start process with command: " + fullCommand, e);
+			}
+
+			// Validate process streams
+			if (this.process.getInputStream() == null || process.getOutputStream() == null) {
+				this.process.destroy();
+				throw new RuntimeException("Process input or output stream is null");
+			}
+
+			// Start threads
+			startInboundProcessing();
+			startOutboundProcessing();
+			startErrorProcessing();
+		}).subscribeOn(Schedulers.boundedElastic());
+	}
+
+	/**
+	 * Creates and returns a new ProcessBuilder instance. Protected to allow overriding in
+	 * tests.
+	 * @return A new ProcessBuilder instance
+	 */
+	protected ProcessBuilder getProcessBuilder() {
+		return new ProcessBuilder();
+	}
+
+	/**
+	 * 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 errorHandler a consumer that processes error messages
+	 */
+	public void setInboundErrorHandler(Consumer<String> errorHandler) {
+		this.errorHandler = errorHandler;
+	}
+
+	/**
+	 * Waits for the server process to exit.
+	 * @throws RuntimeException if the process is interrupted while waiting
+	 */
+	public void awaitForExit() {
+		try {
+			this.process.waitFor();
+		}
+		catch (InterruptedException e) {
+			throw new RuntimeException("Process interrupted", e);
+		}
+	}
+
+	/**
+	 * Starts the error processing thread that reads from the process's error stream.
+	 * Error messages are logged and emitted to the error sink.
+	 */
+	private void startErrorProcessing() {
+		this.errorScheduler.schedule(() -> {
+			try (BufferedReader processErrorReader = new BufferedReader(
+					new InputStreamReader(process.getErrorStream()))) {
+				String line;
+				while ((line = processErrorReader.readLine()) != null) {
+					try {
+						logger.error("Received error line: {}", line);
+						// TODO: handle errors, etc.
+						this.errorSink.tryEmitNext(line);
+					}
+					catch (Exception e) {
+						throw new RuntimeException(e);
+					}
+				}
+			}
+			catch (IOException e) {
+				throw new RuntimeException(e);
+			}
+		});
+	}
+
+	private void handleIncomingMessages(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> inboundMessageHandler) {
+		this.inboundSink.asFlux()
+			.flatMap(message -> Mono.just(message)
+				.transform(inboundMessageHandler)
+				.contextWrite(ctx -> ctx.put("observation", "myObservation")))
+			.subscribe();
+	}
+
+	private void handleIncomingErrors() {
+		this.errorSink.asFlux().subscribe(e -> {
+			this.errorHandler.accept(e);
+		});
+	}
+
+	@Override
+	public Mono<Void> sendMessage(JSONRPCMessage message) {
+		if (this.outboundSink.tryEmitNext(message).isSuccess()) {
+			// TODO: essentially we could reschedule ourselves in some time and make
+			// another attempt with the already read data but pause reading until
+			// success
+			// In this approach we delegate the retry and the backpressure onto the
+			// caller. This might be enough for most cases.
+			return Mono.empty();
+		}
+		else {
+			return Mono.error(new RuntimeException("Failed to enqueue message"));
+		}
+	}
+
+	/**
+	 * Starts the inbound processing thread that reads JSON-RPC messages from the
+	 * process's input stream. Messages are deserialized and emitted to the inbound sink.
+	 */
+	private void startInboundProcessing() {
+		this.inboundScheduler.schedule(() -> {
+			try (BufferedReader processReader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
+				String line;
+				while ((line = processReader.readLine()) != null) {
+					try {
+						JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(this.objectMapper, line);
+						if (!this.inboundSink.tryEmitNext(message).isSuccess()) {
+							// TODO: Back off, reschedule, give up?
+							throw new RuntimeException("Failed to enqueue message");
+						}
+					}
+					catch (Exception e) {
+						throw new RuntimeException(e);
+					}
+				}
+			}
+			catch (IOException e) {
+				throw new RuntimeException(e);
+			}
+		});
+	}
+
+	/**
+	 * Starts the outbound processing thread that writes JSON-RPC messages to the
+	 * process's output stream. Messages are serialized to JSON and written with a newline
+	 * delimiter.
+	 */
+	private void startOutboundProcessing() {
+		this.handleOutbound(messages -> messages
+			// this bit is important since writes come from user threads and we
+			// want to ensure that the actual writing happens on a dedicated thread
+			.publishOn(outboundScheduler)
+			.handle((message, s) -> {
+				if (message != null) {
+					try {
+						String jsonMessage = objectMapper.writeValueAsString(message);
+						this.process.getOutputStream().write(jsonMessage.getBytes(StandardCharsets.UTF_8));
+						this.process.getOutputStream().write("\n".getBytes(StandardCharsets.UTF_8));
+						this.process.getOutputStream().flush();
+						s.next(message);
+					}
+					catch (IOException e) {
+						s.error(new RuntimeException(e));
+					}
+				}
+			}));
+	}
+
+	protected void handleOutbound(Function<Flux<JSONRPCMessage>, Flux<JSONRPCMessage>> outboundConsumer) {
+		outboundConsumer.apply(outboundSink.asFlux()).subscribe();
+	}
+
+	/**
+	 * Gracefully closes the transport by destroying the process and disposing of the
+	 * schedulers. This method sends a TERM signal to the process and waits for it to exit
+	 * before cleaning up resources.
+	 * @return A Mono that completes when the transport is closed
+	 */
+	@Override
+	public Mono<Void> closeGracefully() {
+		return Mono.fromFuture(() -> {
+			logger.info("Sending TERM to process");
+			if (this.process != null) {
+				this.process.destroy();
+				return process.onExit();
+			}
+			else {
+				return CompletableFuture.failedFuture(new RuntimeException("Process not started"));
+			}
+		}).doOnNext(process -> {
+			if (process.exitValue() != 0) {
+				logger.warn("Process terminated with code " + process.exitValue());
+			}
+		}).then(Mono.fromRunnable(() -> {
+			// The Threads are blocked on readLine so disposeGracefully would not
+			// interrupt them, therefore we issue an async hard dispose.
+			inboundScheduler.dispose();
+			errorScheduler.dispose();
+			outboundScheduler.dispose();
+		})).then().subscribeOn(Schedulers.boundedElastic());
+	}
+
+	public Sinks.Many<String> getErrorSink() {
+		return this.errorSink;
+	}
+
+	@Override
+	public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
+		return this.objectMapper.convertValue(data, typeRef);
+	}
+
+}

spring-ai-mcp-core/src/main/java/org/springframework/ai/mcp/spec/AbstractMcpTransport.java

@@ -1,122 +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 org.springframework.ai.mcp.spec;
-
-import java.util.function.Consumer;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import reactor.core.publisher.Mono;
-import reactor.core.publisher.Sinks;
-
-import org.springframework.ai.mcp.client.util.Assert;
-import org.springframework.ai.mcp.spec.McpSchema.JSONRPCMessage;
-
-/**
- * @author Christian Tzolov
- * @author Dariusz Jędrzejczyk
- */
-// TODO: The API is not finalized. Implementing SSE will dictate a more concrete
-// structure and relationship between the base and the actual implementation including
-// proper encapsulation and state management.
-public abstract class AbstractMcpTransport implements McpTransport {
-
-	private static final Logger logger = LoggerFactory.getLogger(AbstractMcpTransport.class);
-
-	protected final ObjectMapper objectMapper;
-
-	private final Sinks.Many<String> errorSink;
-
-	private final Sinks.Many<JSONRPCMessage> inboundSink;
-
-	private final Sinks.Many<JSONRPCMessage> outboundSink;
-
-	private Consumer<JSONRPCMessage> inboundMessageHandler = message -> logger.debug("Received message: {}", message);
-
-	private Consumer<String> errorHandler = error -> logger.error("Received error: {}", error);
-
-	public AbstractMcpTransport() {
-		this(new ObjectMapper());
-	}
-
-	public AbstractMcpTransport(ObjectMapper objectMapper) {
-
-		Assert.notNull(objectMapper, "ObjectMapper must not be null");
-		this.objectMapper = objectMapper;
-
-		this.errorSink = Sinks.many().unicast().onBackpressureBuffer();
-		this.inboundSink = Sinks.many().unicast().onBackpressureBuffer();
-		this.outboundSink = Sinks.many().unicast().onBackpressureBuffer();
-	}
-
-	public ObjectMapper getObjectMapper() {
-		return this.objectMapper;
-	}
-
-	private void handleIncomingMessages() {
-		this.inboundSink.asFlux().subscribe(message -> this.inboundMessageHandler.accept(message));
-	}
-
-	private void handleIncomingErrors() {
-		this.errorSink.asFlux().subscribe(e -> {
-			this.errorHandler.accept(e);
-		});
-	}
-
-	protected Sinks.Many<JSONRPCMessage> getInboundSink() {
-		return inboundSink;
-	}
-
-	protected Sinks.Many<JSONRPCMessage> getOutboundSink() {
-		return outboundSink;
-	}
-
-	protected Sinks.Many<String> getErrorSink() {
-		return errorSink;
-	}
-
-	public void setInboundMessageHandler(Consumer<JSONRPCMessage> inboundMessageHandler) {
-		this.inboundMessageHandler = inboundMessageHandler;
-	}
-
-	public void setInboundErrorHandler(Consumer<String> errorHandler) {
-		this.errorHandler = errorHandler;
-	}
-
-	// Start processing incoming messages
-	public void start() {
-		this.handleIncomingErrors();
-		this.handleIncomingMessages();
-	}
-
-	@Override
-	public Mono<Void> sendMessage(JSONRPCMessage message) {
-		if (this.outboundSink.tryEmitNext(message).isSuccess()) {
-			// TODO: essentially we could reschedule ourselves in some time and make
-			// another attempt with the already read data but pause reading until
-			// success
-			// In this approach we delegate the retry and the backpressure onto the
-			// caller. This might be enough for most cases.
-			return Mono.empty();
-		}
-		else {
-			return Mono.error(new RuntimeException("Failed to enqueue message"));
-		}
-	}
-
-}

spring-ai-mcp-core/src/main/java/org/springframework/ai/mcp/spec/DefaultMcpSession.java

@@ -23,13 +23,13 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 
 import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import reactor.core.Disposable;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoSink;
 
-import org.springframework.ai.mcp.client.util.Assert;
+import org.springframework.ai.mcp.util.Assert;
 
 /**
  * Default implementation of the MCP (Model Context Protocol) session that manages
@@ -56,9 +56,6 @@ public class DefaultMcpSession implements McpSession {
 	/** Duration to wait for request responses before timing out */
 	private final Duration requestTimeout;
 
-	/** JSON object mapper for serialization/deserialization */
-	private final ObjectMapper objectMapper;
-
 	/** Transport layer implementation for message exchange */
 	private final McpTransport transport;
 
@@ -77,6 +74,8 @@ public class DefaultMcpSession implements McpSession {
 	/** Atomic counter for generating unique request IDs */
 	private final AtomicLong requestCounter = new AtomicLong(0);
 
+	private final Disposable connection;
+
 	/**
 	 * Functional interface for handling incoming JSON-RPC requests. Implementations
 	 * should process the request parameters and return a response.
@@ -112,37 +111,36 @@ public class DefaultMcpSession implements McpSession {
 	/**
 	 * Creates a new DefaultMcpSession with the specified configuration.
 	 * @param requestTimeout Duration to wait for responses
-	 * @param objectMapper JSON object mapper for message serialization
 	 * @param transport Transport implementation for message exchange
 	 */
-	public DefaultMcpSession(Duration requestTimeout, ObjectMapper objectMapper, McpTransport transport) {
+	public DefaultMcpSession(Duration requestTimeout, McpTransport transport) {
-		this(requestTimeout, objectMapper, transport, Map.of(), Map.of());
+		this(requestTimeout, transport, Map.of(), Map.of());
 	}
 
 	/**
 	 * Creates a new DefaultMcpSession with the specified configuration and handlers.
 	 * @param requestTimeout Duration to wait for responses
-	 * @param objectMapper JSON object mapper for message serialization
 	 * @param transport Transport implementation for message exchange
 	 * @param requestHandlers Map of method names to request handlers
 	 * @param notificationHandlers Map of method names to notification handlers
 	 */
-	public DefaultMcpSession(Duration requestTimeout, ObjectMapper objectMapper, McpTransport transport,
+	public DefaultMcpSession(Duration requestTimeout, McpTransport transport,
 			Map<String, RequestHandler> requestHandlers, Map<String, NotificationHandler> notificationHandlers) {
 
-		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");
 		Assert.notNull(requestHandlers, "The requestHandlers can not be null");
 		Assert.notNull(notificationHandlers, "The notificationHandlers can not be null");
 
 		this.requestTimeout = requestTimeout;
-		this.objectMapper = objectMapper;
 		this.transport = transport;
 		this.requestHandlers.putAll(requestHandlers);
 		this.notificationHandlers.putAll(notificationHandlers);
 
-		this.transport.setInboundMessageHandler(message -> {
+		// TODO: consider mono.transformDeferredContextual where the Context contains the
+		// Observation associated with the individual message - it can be used to
+		// create child Observation and emit it together with the message to the consumer
+		this.connection = this.transport.connect(mono -> mono.doOnNext(message -> {
 			if (message instanceof McpSchema.JSONRPCResponse response) {
 				var sink = pendingResponses.remove(response.id());
 				if (sink == null) {
@@ -165,11 +163,7 @@ public class DefaultMcpSession implements McpSession {
 				handleIncomingNotification(notification).subscribe(null,
 						error -> logger.error("Error handling notification: {}", error.getMessage()));
 			}
-		});
+		})).subscribe();
-
-		this.transport.setInboundErrorHandler(error -> logger.error("Error received: {}", error));
-
-		this.transport.start();
 	}
 
 	/**
@@ -252,7 +246,7 @@ public class DefaultMcpSession implements McpSession {
 					sink.complete();
 				}
 				else {
-					sink.next(this.objectMapper.convertValue(jsonRpcResponse.result(), typeRef));
+					sink.next(this.transport.unmarshalFrom(jsonRpcResponse.result(), typeRef));
 				}
 			}
 		});
@@ -277,6 +271,7 @@ public class DefaultMcpSession implements McpSession {
 	 */
 	@Override
 	public Mono<Void> closeGracefully() {
+		this.connection.dispose();
 		return transport.closeGracefully();
 	}
 
@@ -285,6 +280,7 @@ public class DefaultMcpSession implements McpSession {
 	 */
 	@Override
 	public void close() {
+		this.connection.dispose();
 		transport.close();
 	}
 

spring-ai-mcp-core/src/main/java/org/springframework/ai/mcp/spec/McpTransport.java

@@ -16,8 +16,9 @@
 
 package org.springframework.ai.mcp.spec;
 
-import java.util.function.Consumer;
+import java.util.function.Function;
 
+import com.fasterxml.jackson.core.type.TypeReference;
 import reactor.core.publisher.Mono;
 
 import org.springframework.ai.mcp.spec.McpSchema.JSONRPCMessage;
@@ -59,7 +60,7 @@ public interface McpTransport {
 	 * necessary resources and establishes the connection to the server.
 	 * </p>
 	 */
-	void start();
+	Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> handler);
 
 	/**
 	 * Closes the transport connection and releases any associated resources.
@@ -80,30 +81,6 @@ public interface McpTransport {
 	 */
 	Mono<Void> closeGracefully();
 
-	/**
-	 * 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 setInboundMessageHandler(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.
 	 *
@@ -116,4 +93,6 @@ public interface McpTransport {
 	 */
 	Mono<Void> sendMessage(JSONRPCMessage message);
 
+	<T> T unmarshalFrom(Object data, TypeReference<T> typeRef);
+
 }

spring-ai-mcp-core/src/main/java/org/springframework/ai/mcp/util/Assert.java

@@ -0,0 +1,72 @@
+/*
+ * 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 org.springframework.ai.mcp.util;
+
+import reactor.util.annotation.Nullable;
+
+/**
+ * @author Christian Tzolov
+ */
+
+public class Assert {
+
+	/**
+	 * Assert that an object is not {@code null}.
+	 *
+	 * <pre class="code">
+	 * Assert.notNull(clazz, "The class must not be null");
+	 * </pre>
+	 * @param object the object to check
+	 * @param message the exception message to use if the assertion fails
+	 * @throws IllegalArgumentException if the object is {@code null}
+	 */
+	public static void notNull(@Nullable Object object, String message) {
+		if (object == null) {
+			throw new IllegalArgumentException(message);
+		}
+	}
+
+	/**
+	 * Assert that the given String contains valid text content; that is, it must not be
+	 * {@code null} and must contain at least one non-whitespace character.
+	 * <pre class="code">Assert.hasText(name, "'name' must not be empty");</pre>
+	 * @param text the String to check
+	 * @param message the exception message to use if the assertion fails
+	 * @throws IllegalArgumentException if the text does not contain valid text content
+	 */
+	public static void hasText(@Nullable String text, String message) {
+		if (!hasText(text)) {
+			throw new IllegalArgumentException(message);
+		}
+	}
+
+	/**
+	 * Check whether the given {@code String} contains actual <em>text</em>.
+	 * <p>
+	 * More specifically, this method returns {@code true} if the {@code String} is not
+	 * {@code null}, its length is greater than 0, and it contains at least one
+	 * non-whitespace character.
+	 * @param str the {@code String} to check (may be {@code null})
+	 * @return {@code true} if the {@code String} is not {@code null}, its length is
+	 * greater than 0, and it does not contain whitespace only
+	 * @see Character#isWhitespace
+	 */
+	public static boolean hasText(@Nullable String str) {
+		return (str != null && !str.isBlank());
+	}
+
+}

spring-ai-mcp-core/src/main/resources/logback.xml

@@ -20,7 +20,7 @@
     <logger name="org.springframework.ai.mcp.spec" level="INFO"/>
 
     <!-- Test packages -->
-    <logger name="org.springframework.ai.mcp.client.sse.SseServerTransportTests" level="DEBUG"/>
+    <logger name="org.springframework.ai.mcp.client.sse.SseClientTransportTests" level="DEBUG"/>
     <logger name="org.springframework.ai.mcp.client.stdio.McpAsyncClientTests" level="DEBUG"/>
     <logger name="org.springframework.ai.mcp.client.stdio.McpSyncClientTests" level="DEBUG"/>
 

spring-ai-mcp-core/src/test/java/org/springframework/ai/mcp/attic/ClientSessionTests2.java

@@ -18,12 +18,10 @@ package org.springframework.ai.mcp.attic;
 
 import java.time.Duration;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-
 import org.springframework.ai.mcp.client.McpClient;
 import org.springframework.ai.mcp.client.McpSyncClient;
 import org.springframework.ai.mcp.client.stdio.ServerParameters;
-import org.springframework.ai.mcp.client.stdio.StdioServerTransport;
+import org.springframework.ai.mcp.client.stdio.StdioClientTransport;
 import org.springframework.ai.mcp.spec.McpSchema.ListResourcesResult;
 import org.springframework.ai.mcp.spec.McpSchema.ListToolsResult;
 import org.springframework.ai.mcp.spec.McpSchema.Resource;
@@ -40,8 +38,9 @@ public class ClientSessionTests2 {
 			.args("--directory", "dir", "run", "mcp-server-sqlite", "--db-path", "~/test.db")
 			.build();
 
-		try (McpSyncClient clientSession = McpClient.sync(new StdioServerTransport(stdioParams), Duration.ofSeconds(10),
+		try (McpSyncClient clientSession = McpClient.using(new StdioClientTransport(stdioParams))
-				new ObjectMapper())) {
+			.withRequestTimeout(Duration.ofSeconds(10))
+			.sync()) {
 
 			clientSession.initialize();
 

spring-ai-mcp-core/src/test/java/org/springframework/ai/mcp/attic/ClientSessionTests3.java

@@ -18,12 +18,10 @@ package org.springframework.ai.mcp.attic;
 
 import java.time.Duration;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-
 import org.springframework.ai.mcp.client.McpClient;
 import org.springframework.ai.mcp.client.McpSyncClient;
 import org.springframework.ai.mcp.client.stdio.ServerParameters;
-import org.springframework.ai.mcp.client.stdio.StdioServerTransport;
+import org.springframework.ai.mcp.client.stdio.StdioClientTransport;
 import org.springframework.ai.mcp.spec.McpSchema.ListResourcesResult;
 import org.springframework.ai.mcp.spec.McpSchema.ListToolsResult;
 import org.springframework.ai.mcp.spec.McpSchema.Resource;
@@ -43,8 +41,9 @@ public class ClientSessionTests3 {
 		McpSyncClient clientSession = null;
 		try {
 
-			clientSession = McpClient.sync(new StdioServerTransport(stdioParams), Duration.ofSeconds(10),
+			clientSession = McpClient.using(new StdioClientTransport(stdioParams))
-					new ObjectMapper());
+				.withRequestTimeout(Duration.ofSeconds(10))
+				.sync();
 
 			clientSession.initialize();
 

spring-ai-mcp-core/src/test/java/org/springframework/ai/mcp/attic/MainSSE.java

@@ -18,11 +18,9 @@ package org.springframework.ai.mcp.attic;
 import java.time.Duration;
 import java.util.Map;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-
 import org.springframework.ai.mcp.client.McpClient;
 import org.springframework.ai.mcp.client.McpSyncClient;
-import org.springframework.ai.mcp.client.sse.SseServerTransport;
+import org.springframework.ai.mcp.client.sse.SseClientTransport;
 import org.springframework.ai.mcp.spec.McpSchema.CallToolRequest;
 import org.springframework.ai.mcp.spec.McpSchema.CallToolResult;
 import org.springframework.ai.mcp.spec.McpSchema.ListPromptsResult;
@@ -40,9 +38,10 @@ public class MainSSE {
 
 	public static void main(String[] args) {
 
-		try (McpSyncClient mcpClient = McpClient.sync(
+		try (McpSyncClient mcpClient = McpClient
-				new SseServerTransport(WebClient.builder().baseUrl("http://localhost:3001")), Duration.ofSeconds(1000),
+			.using(new SseClientTransport(WebClient.builder().baseUrl("http://localhost:3001")))
-				new ObjectMapper())) {
+			.withRequestTimeout(Duration.ofSeconds(1000))
+			.sync()) {
 
 			mcpClient.initialize();