refactor: rename artifacts and restructure modules

- Rename spring-ai-mcp-core to mcp for simpler artifact naming - Rename spring-ai-mcp-spring to spring-ai-mcp for clarity - Rename parent artifact from spring-ai-mcp to mcp-parent - Update artifact references in pom files and documentation - Update directory structure and paths in GitHub workflow Resolves #31 chore: improve module descriptions and update dependencies - Update parent pom description to better reflect overall project scope - Update mcp module description to detail core Java SDK capabilities - Update spring-ai-mcp module description to emphasize Spring integration features - Update dependency versions to stable releases: * JUnit: 5.11.3 -> 5.10.2 * Jackson: 2.18.2 -> 2.17.0
2024-12-20 22:38:42 +01:00 · 2024-12-20 22:38:42 +01:00 · 12fd8e51f3
commit 12fd8e51f3
parent 2ea5527ade
35 changed files with 92 additions and 109 deletions
--- a/.github/workflows/maven-central-release.yml
+++ b/.github/workflows/maven-central-release.yml
@ -17,9 +17,9 @@ jobs:

      - name: Prepare directory structure
        run: |
+          mkdir -p nexus/org/springframework/experimental/mcp-parent/$RELEASE_VERSION
+          mkdir -p nexus/org/springframework/experimental/mcp/$RELEASE_VERSION
          mkdir -p nexus/org/springframework/experimental/spring-ai-mcp/$RELEASE_VERSION
-          mkdir -p nexus/org/springframework/experimental/spring-ai-mcp-core/$RELEASE_VERSION
-          mkdir -p nexus/org/springframework/experimental/spring-ai-mcp-spring/$RELEASE_VERSION

      - name: Download release files from Artifactory
        env:
@ -33,19 +33,19 @@ jobs:
          
          echo "Downloading core artifacts"
          cd ../../../../../..
-          cd nexus/org/springframework/experimental/spring-ai-mcp-core/$RELEASE_VERSION
-          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/spring-ai-mcp-core/$RELEASE_VERSION/spring-ai-mcp-core-$RELEASE_VERSION.pom
-          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/spring-ai-mcp-core/$RELEASE_VERSION/spring-ai-mcp-core-$RELEASE_VERSION.jar
-          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/spring-ai-mcp-core/$RELEASE_VERSION/spring-ai-mcp-core-$RELEASE_VERSION-javadoc.jar
-          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/spring-ai-mcp-core/$RELEASE_VERSION/spring-ai-mcp-core-$RELEASE_VERSION-sources.jar
+          cd nexus/org/springframework/experimental/mcp/$RELEASE_VERSION
+          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/mcp/$RELEASE_VERSION/mcp-$RELEASE_VERSION.pom
+          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/mcp/$RELEASE_VERSION/mcp-$RELEASE_VERSION.jar
+          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/mcp/$RELEASE_VERSION/mcp-$RELEASE_VERSION-javadoc.jar
+          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/mcp/$RELEASE_VERSION/mcp-$RELEASE_VERSION-sources.jar

          echo "Downloading spring artifacts"
          cd ../../../../../..
-          cd nexus/org/springframework/experimental/spring-ai-mcp-spring/$RELEASE_VERSION
-          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/spring-ai-mcp-spring/$RELEASE_VERSION/spring-ai-mcp-spring-$RELEASE_VERSION.pom
-          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/spring-ai-mcp-spring/$RELEASE_VERSION/spring-ai-mcp-spring-$RELEASE_VERSION.jar
-          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/spring-ai-mcp-spring/$RELEASE_VERSION/spring-ai-mcp-spring-$RELEASE_VERSION-javadoc.jar
-          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/spring-ai-mcp-spring/$RELEASE_VERSION/spring-ai-mcp-spring-$RELEASE_VERSION-sources.jar
+          cd nexus/org/springframework/experimental/spring-ai-mcp/$RELEASE_VERSION
+          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/spring-ai-mcp/$RELEASE_VERSION/spring-ai-mcp-$RELEASE_VERSION.pom
+          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/spring-ai-mcp/$RELEASE_VERSION/spring-ai-mcp-$RELEASE_VERSION.jar
+          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/spring-ai-mcp/$RELEASE_VERSION/spring-ai-mcp-$RELEASE_VERSION-javadoc.jar
+          wget --user="$ARTIFACTORY_USERNAME" --password="$ARTIFACTORY_PASSWORD" $ARTIFACTORY_URL/spring-ai-mcp/$RELEASE_VERSION/spring-ai-mcp-$RELEASE_VERSION-sources.jar

      - name: Sign artifacts and release them to Maven Central
        uses: spring-io/nexus-sync-action@main
--- a/README.md
+++ b/README.md
@ -1,10 +1,10 @@
-# Spring AI MCP
+# Java & Spring MCP

 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 Java applications to interact with AI models and tools through a standardized interface, supporting both synchronous and asynchronous communication patterns.
+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="docs/spring-ai-mcp-architecture.jpg" width="600">

@ -12,15 +12,15 @@ Spring AI MCP is an experimental project that provides Java and Spring Framework

 The project consists of two main modules:

-### [mcp-core](./spring-ai-mcp-core/README.md)
+### [mcp](./mcp/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). Support for request and notificaiotn handling.
 - [Stdio](https://spec.modelcontextprotocol.io/specification/basic/transports/#stdio) and [SSE](https://spec.modelcontextprotocol.io/specification/basic/transports/#http-with-sse) transport implementations. 
- [Find more](./spring-ai-mcp-core/README.md).
+- [Find more](./mcp/README.md).

-### [mcp-spring](./spring-ai-mcp-spring/README.md)
+### [spring-ai-mcp](./spring-ai-mcp/README.md)

 The Spring integration module provides Spring-specific functionality:
 - Integration with Spring AI's function calling system
@ -41,14 +41,14 @@ Add the following dependencies to your Maven project:
 <!-- For core MCP functionality -->
 <dependency>
    <groupId>org.springframework.experimental</groupId>
-    <artifactId>spring-ai-mcp-core</artifactId>
+    <artifactId>mcp</artifactId>
    <version>0.2.0-SNAPSHOT</version>
 </dependency>

-<!-- For Spring integration -->
+<!-- For Spring AI integration -->
 <dependency>
    <groupId>org.springframework.experimental</groupId>
-    <artifactId>spring-ai-mcp-spring</artifactId>
+    <artifactId>spring-ai-mcp</artifactId>
    <version>0.2.0-SNAPSHOT</version>
 </dependency>
 ```
@ -64,9 +64,9 @@ Explore these MCP examples in the [spring-ai-examples/model-context-protocol](ht

 ## Documentation

- [Core Module Documentation](spring-ai-mcp-core/README.md)
- [Spring Integration Documentation](spring-ai-mcp-spring/README.md)
- [UML Class Diagrams](spring-ai-mcp-core/docs/spring-ai-mcp-uml-classdiagram.svg)
+- [Core Module Documentation](mcp/README.md)
+- [Spring Integration Documentation](spring-ai-mcp/README.md)
+- [UML Class Diagrams](spring-ai-mcp/docs/spring-ai-mcp-uml-classdiagram.svg)

 ## Development

--- a/mcp/README.md
+++ b/mcp/README.md
@ -0,0 +1,358 @@
+# Model Context Protocol (MCP) Java SDK
+
+A Java implementation of the [Model Context Protocol](https://modelcontextprotocol.org/docs/concepts/architecture) specification, providing both synchronous and asynchronous clients for MCP server interactions.
+
+## Overview
+
+This SDK implements the Model Context Protocol, enabling seamless integration with AI models and tools through a standardized interface. It supports both synchronous and asynchronous communication patterns, making it suitable for various use cases and integration scenarios.
+
+## Features
+
+- Synchronous and Asynchronous MCP Client implementations
+- Synchronous and Asynchronous MCP Server implementations (WIP: comming with 0.3.0)
+- Standard MCP operations support:
+  - Protocol version compatibility negotiation
+  - Client-server capability exchange
+  - Tool discovery and execution with change notifications
+  - Tool list change notifications with non-blocking consumer support
+  - Resource management with URI templates
+  - Resource subscription system
+  - Roots list management and notifications
+  - Prompt handling and management
+  - Server initialization and ping
+- Multiple transport implementations:
+  - Stdio-based transport for process-based communication
+  - SSE-based transport for HTTP streaming
+- Configurable request timeouts
+- Customizable JSON serialization/deserialization
+
+## Installation
+
+Add the following dependency to your Maven project:
+
+```xml
+<dependency>
+    <groupId>org.springframework.experimental</groupId>
+    <artifactId>mcp</artifactId>
+    <version>0.2.0-SNAPSHOT</version>
+</dependency>
+```
+
+## Usage
+
+### Transport Layer Options
+
+The SDK provides two transport implementations:
+
+#### 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 StdioClientTransport(params);
+```
+
+#### 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 SseClientTransport(webClientBuilder);
+```
+
+The SSE transport provides:
+- Bidirectional communication over HTTP
+- Automatic reconnection for transient failures
+- Inbound message streaming via SSE
+- Outbound message delivery via HTTP POST
+- Graceful shutdown handling
+- Configurable JSON serialization
+
+
+### Sync MCP Client Example
+
+```java
+// Create and initialize sync client
+McpTransport mcpTransport = ...
+
+try (McpSyncClient client = McpClient.using(mcpTransport).sync()) {
+    // Initialize connection with protocol version and capabilities
+    McpSchema.InitializeResult initResult = client.initialize();
+
+    // List available tools
+    McpSchema.ListToolsResult tools = client.listTools();
+
+    // Execute a tool
+    McpSchema.CallToolResult result = client.callTool(
+        new McpSchema.CallToolRequest("echo", Map.of("message", "Hello!"))
+    );
+
+    // Resource management
+    McpSchema.ListResourcesResult resources = client.listResources();
+    McpSchema.ReadResourceResult resource = client.readResource(
+        new McpSchema.ReadResourceRequest("resource-uri")
+    );
+
+    // List and retrieve prompts
+    ListPromptsResult prompts = client.listPrompts();
+    GetPromptResult prompt = client.getPrompt(
+        new McpSchema.GetPromptRequest("prompt-id", Map.of())
+    );
+}
+```
+
+### Async MCP Client Example with Custom Configuration
+
+```java
+McpTransport mcpTransport = ...
+
+// Initialize async client with custom timeout and object mapper
+McpAsyncClient client = McpClient.using(mcpTransport)
+    .requestTimeout(Duration.ofSeconds(30))
+    .async();
+
+// Initialize connection and chain operations
+var result = client.initialize()
+    .flatMap(initResult -> {
+        // Connection initialized with protocol version compatibility
+        return client.listTools();
+    })
+    .flatMap(tools -> {
+        // Process available tools
+        return client.callTool(new McpSchema.CallToolRequest("echo",
+            Map.of("message", "Hello MCP!")));
+    })
+    .flatMap(toolResult -> {
+        // Handle tool execution result
+        return client.listPrompts();
+    })
+    .flatMap(prompts -> {
+        // Process available prompts
+        return client.getPrompt(new McpSchema.GetPromptRequest("prompt-id", Map.of()));
+    });
+
+// Subscribe to resource changes
+var subscription = client.listResources()
+    .flatMap(resources -> {
+        return client.subscribeResource(new McpSchema.SubscribeRequest("resource-uri"));
+    });
+
+// Set up tools change notification handling
+List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumers = List.of(
+    tools -> {
+        // Handle tools list changes reactively
+        tools.forEach(tool -> {
+            System.out.println("Tool updated: " + tool.name());
+        });
+    }
+);
+
+McpAsyncClient clientWithToolsNotifications = McpClient.using(transport)
+    .toolsChangeConsumer(toolsChangeConsumers)
+    .async();
+
+// The client will now automatically handle tools/list_changed notifications
+// and invoke the consumers on the boundedElastic scheduler to avoid blocking
+
+// Handle results reactively or block if needed
+McpSchema.GetPromptResult promptResult = result.block();
+subscription.block();
+
+// Cleanup
+client.closeGracefully().block();
+```
+
+## Architecture
+
+The SDK follows a layered architecture with clear separation of concerns:
+
+### Core Components
+
+- **McpClient**: Factory class for creating sync and async clients with optional custom configuration
+- **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 client/server communication
+- **McpSchema**: Comprehensive protocol schema definitions
+- **DefaultMcpSession**: Default implementation of the session management
+- **StdioClientTransport**: Standard I/O based client to server 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"/>
+
+### Key Interactions
+
+1. Client Initialization
+   - Transport setup and connection establishment
+   - Protocol version compatibility check
+   - Capability negotiation
+   - Implementation details exchange
+
+2. Message Flow
+   - JSON-RPC message creation and validation
+   - Transport layer handling
+   - Response processing with type safety
+   - Error handling with specific error codes
+
+3. Resource Management
+   - Resource discovery and listing
+   - URI template-based resource access
+   - Subscription system for change notifications
+   - Resource content retrieval
+
+4. Prompt System
+   - Prompt discovery and listing
+   - Parameter-based prompt retrieval
+   - Change notifications support
+   - Prompt content management
+
+5. Tool Execution
+   - Tool discovery and capability checking
+   - Parameter validation and processing
+   - Execution handling with timeout support
+   - Result processing with error handling
+
+### Roots List Support
+
+The SDK supports the MCP roots list capability, which allows servers to understand which directories and files they have access to. Clients can provide a list of root directories/files and notify servers when this list changes.
+
+#### Features
+- Define root providers that supply filesystem access boundaries
+- Support for roots list changed notifications
+- Automatic roots list request handling
+
+#### Example with Roots List Configuration
+
+```java
+// Create root providers
+List<Supplier<List<Root>>> rootProviders = List.of(
+    () -> List.of(new Root("file:///workspace/project", "Project Root")),
+    () -> List.of(new Root("file:///workspace/docs", "Documentation"))
+);
+
+// Create async client with roots list support
+McpAsyncClient client = new McpAsyncClient(
+    new StdioClientTransport(params),
+    Duration.ofSeconds(30),
+    rootProviders,  // Configure root providers
+    true           // Enable roots list changed notifications
+);
+
+// Initialize connection
+client.initialize()
+    .doOnSuccess(result -> {
+        // Connection initialized with roots list capability
+        
+        // Notify server when roots list changes
+        return client.sendRootsListChanged();
+    })
+    .subscribe();
+```
+
+### Change Notifications Support
+
+The SDK supports automatic handling of changes through a non-blocking notification system for tools, resources, and prompts:
+
+#### Features
+- Register multiple consumers for tools, resources, and prompts changes
+- Non-blocking execution using Project Reactor's boundedElastic scheduler
+- Automatic list request handling when notifications are received
+- Error resilient with proper error handling and logging
+
+#### Example with Change Notifications
+
+```java
+// Create tool change consumers
+List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumers = List.of(
+    tools -> {
+        // First consumer - e.g., update UI
+        tools.forEach(tool -> updateToolsUI(tool));
+    },
+    tools -> {
+        // Second consumer - e.g., update cache
+        toolsCache.updateTools(tools);
+    }
+);
+
+// Create resource change consumers
+List<Consumer<List<McpSchema.Resource>>> resourcesChangeConsumers = List.of(
+    resources -> {
+        // Handle resource changes
+        resources.forEach(resource -> {
+            System.out.println("Resource updated: " + resource.uri());
+            updateResourcesUI(resource);
+        });
+    }
+);
+
+// Create prompt change consumers
+List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumers = List.of(
+    prompts -> {
+        // Handle prompt changes
+        prompts.forEach(prompt -> {
+            System.out.println("Prompt updated: " + prompt.name());
+            updatePromptsCache(prompt);
+        });
+    }
+);
+
+// Create client with change notification support
+McpAsyncClient client = McpClient.using(transport)
+    .toolsChangeConsumer(toolsChangeConsumer)
+    .resourcesChangeConsumer(resourcesChangeConsumer)
+    .promptsChangeConsumer(promptsChangeConsumer)
+    .async();
+
+// Initialize client
+client.initialize()
+    .doOnSuccess(result -> {
+        // Client will automatically handle all change notifications
+        // and invoke consumers non-blockingly on boundedElastic scheduler
+    })
+    .subscribe();
+```
+
+The change notification system ensures that all consumers are executed non-blockingly, preventing any potential performance impact from blocking implementations. All consumers are executed on Project Reactor's boundedElastic scheduler, making it safe to perform potentially blocking operations within the consumers.
+
+Each type of notification handler operates independently:
+- Tools: Handles notifications when available tools change
+- Resources: Handles notifications when available resources change
+- Prompts: Handles notifications when available prompts change
+
+## Error Handling
+
+The SDK provides comprehensive error handling through the McpError class:
+
+- Protocol version incompatibility
+- Transport-level communication errors
+- JSON-RPC protocol violations
+- Tool execution failures
+- Resource access and subscription errors
+- Prompt management errors
+- Request timeout handling
+- Server capability mismatches
+- SSE connection failures and retry handling
+- HTTP request/response errors
+
+## Contributing
+
+Contributions are welcome! Please feel free to submit a Pull Request.
+
+## License
+
+Copyright 2024 - 2024 the original author or authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+https://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
--- a/mcp/pom.xml
+++ b/mcp/pom.xml
@ -0,0 +1,111 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.springframework.experimental</groupId>
+		<artifactId>mcp-parent</artifactId>
+		<version>0.2.0-SNAPSHOT</version>
+	</parent>
+	<artifactId>mcp</artifactId>
+	<packaging>jar</packaging>
+	<name>Java MCP SDK</name>
+	<description>Core Java implementation of the Model Context Protocol (MCP) specification, providing synchronous and asynchronous clients, server implementations (coming in 0.3.0), with support for tool discovery, resource management, and multiple transport implementations</description>
+	<url>https://github.com/spring-projects-experimental/spring-ai-mcp</url>
+
+	<scm>
+		<url>https://github.com/spring-projects-experimental/spring-ai-mcp</url>
+		<connection>git://github.com/spring-projects-experimental/spring-ai-mcp.git</connection>
+		<developerConnection>git@github.com:spring-projects-experimental/spring-ai-mcp.git</developerConnection>
+	</scm>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>slf4j-api</artifactId>
+			<version>${slf4j-api.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>ch.qos.logback</groupId>
+			<artifactId>logback-classic</artifactId>
+			<version>${logback.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>com.fasterxml.jackson.core</groupId>
+			<artifactId>jackson-databind</artifactId>
+			<version>${jackson.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-core</artifactId>
+		</dependency>
+
+
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-webflux</artifactId>
+			<version>${springframework.version}</version>
+		</dependency>
+
+
+		<dependency>
+			<groupId>io.projectreactor.netty</groupId>
+			<artifactId>reactor-netty-http</artifactId>
+			<scope>test</scope>
+		</dependency>
+
+		<!-- The Spring Context is required due to the reactor-netty connector being dependant on
+		the Spring Lifecycle, as discussed here:
+		https://github.com/spring-projects/spring-framework/issues/31180 -->
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-context</artifactId>
+			<version>${springframework.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.assertj</groupId>
+			<artifactId>assertj-core</artifactId>
+			<version>${assert4j.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-api</artifactId>
+			<version>${junit.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.mockito</groupId>
+			<artifactId>mockito-core</artifactId>
+			<version>${mockito.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>junit-jupiter</artifactId>
+			<version>${testcontainers.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.awaitility</groupId>
+			<artifactId>awaitility</artifactId>
+			<version>4.2.0</version>
+			<scope>test</scope>
+		</dependency>
+
+	</dependencies>
+
+
+</project>
--- a/mcp/src/main/java/org/springframework/ai/mcp/client/McpAsyncClient.java
+++ b/mcp/src/main/java/org/springframework/ai/mcp/client/McpAsyncClient.java
@ -0,0 +1,473 @@
+/*
+ * 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;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
+import org.springframework.ai.mcp.spec.DefaultMcpSession;
+import org.springframework.ai.mcp.spec.DefaultMcpSession.NotificationHandler;
+import org.springframework.ai.mcp.spec.DefaultMcpSession.RequestHandler;
+import org.springframework.ai.mcp.spec.McpError;
+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.spec.McpSchema.PaginatedRequest;
+import org.springframework.ai.mcp.spec.McpSchema.Root;
+import org.springframework.ai.mcp.spec.McpTransport;
+import org.springframework.ai.mcp.util.Utils;
+
+/**
+ * The Model Context Protocol (MCP) client implementation that provides asynchronous
+ * communication with MCP servers.
+ *
+ * @author Dariusz Jędrzejczyk
+ * @author Christian Tzolov
+ */
+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;
+
+	/**
+	 * Roots define the boundaries of where servers can operate within the filesystem,
+	 * allowing them to understand which directories and files they have access to.
+	 * Servers can request the list of roots from supporting clients and receive
+	 * notifications when that list changes.
+	 */
+	private McpSchema.ClientCapabilities.RootCapabilities rootCapabilities;
+
+	/**
+	 * 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.
+	 * @param rootsListProviders the list of suppliers that provide the list of roots
+	 * backing the roots list request.
+	 * @param rootsListChangedNotification whether the client supports roots/list_changed
+	 * notification.
+	 */
+	public McpAsyncClient(McpTransport transport, Duration requestTimeout,
+			List<Supplier<List<Root>>> rootsListProviders, boolean rootsListChangedNotification,
+			List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumers,
+			List<Consumer<List<McpSchema.Resource>>> resourcesChangeConsumers,
+			List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumers) {
+
+		// Request Handlers
+		Map<String, RequestHandler> requestHanlers = new HashMap<>();
+
+		// Roots List Request Handler
+		if (rootsListProviders != null && !rootsListProviders.isEmpty()) {
+			requestHanlers.put("roots/list", rootsListRequestHandler(rootsListProviders));
+			this.rootCapabilities = new McpSchema.ClientCapabilities.RootCapabilities(rootsListChangedNotification);
+		}
+
+		// Notification Handlers
+		Map<String, NotificationHandler> notificationHandlers = new HashMap<>();
+
+		// Tools Change Notification
+		List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumersFinal = new ArrayList<>();
+		toolsChangeConsumersFinal.add((notification) -> logger.info("Tools changed: {}", notification));
+		if (!Utils.isEmpty(toolsChangeConsumers)) {
+			toolsChangeConsumersFinal.addAll(toolsChangeConsumers);
+		}
+		notificationHandlers.put("notifications/tools/list_changed",
+				toolsChangeNotificationHandler(toolsChangeConsumersFinal));
+
+		// Resources Change Notification
+		List<Consumer<List<McpSchema.Resource>>> resourcesChangeConsumersFinal = new ArrayList<>();
+		resourcesChangeConsumersFinal.add((notification) -> logger.info("Resources changed: {}", notification));
+		if (!Utils.isEmpty(resourcesChangeConsumers)) {
+			resourcesChangeConsumersFinal.addAll(resourcesChangeConsumers);
+		}
+		notificationHandlers.put("notifications/resources/list_changed",
+				resourcesChangeNotificationHandler(resourcesChangeConsumersFinal));
+
+		// Prompts Change Notification
+		List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumersFinal = new ArrayList<>();
+		promptsChangeConsumersFinal.add((notification) -> logger.info("Prompts changed: {}", notification));
+		if (!Utils.isEmpty(promptsChangeConsumers)) {
+			promptsChangeConsumersFinal.addAll(promptsChangeConsumers);
+		}
+		notificationHandlers.put("notifications/prompts/list_changed",
+				promptsChangeNotificationHandler(promptsChangeConsumersFinal));
+
+		this.mcpSession = new DefaultMcpSession(requestTimeout, transport, requestHanlers, notificationHandlers);
+
+	}
+
+	private RequestHandler rootsListRequestHandler(List<Supplier<List<Root>>> rootsListProviders) {
+		return new RequestHandler() {
+			@Override
+			public Mono<Object> handle(Object params) {
+				if (rootsListProviders == null || rootsListProviders.isEmpty()) {
+					return Mono.just(new ArrayList<Root>());
+				}
+
+				List<Mono<List<Root>>> monos = rootsListProviders.stream()
+					.map(supplier -> Mono.fromSupplier(supplier).subscribeOn(Schedulers.boundedElastic()))
+					.toList();
+
+				return Mono.zip(monos, arrays -> {
+					List<Root> combinedList = new ArrayList<>();
+					for (Object array : arrays) {
+						@SuppressWarnings("unchecked")
+						List<Root> roots = (List<Root>) array;
+						combinedList.addAll(roots);
+					}
+					return combinedList;
+				});
+			}
+		};
+	};
+
+	private NotificationHandler toolsChangeNotificationHandler(
+			List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumers) {
+
+		return new NotificationHandler() {
+			@Override
+			public Mono<Void> handle(Object params) {
+				// TODO: add support for cursor/pagination
+				return listTools().flatMap(listToolsResult -> Mono.fromRunnable(() -> {
+					for (Consumer<List<McpSchema.Tool>> toolsChangeConsumer : toolsChangeConsumers) {
+						toolsChangeConsumer.accept(listToolsResult.tools());
+					}
+				}).subscribeOn(Schedulers.boundedElastic())).onErrorResume(error -> {
+					logger.error("Error handling tools list change notification", error);
+					return Mono.empty();
+				}).then(); // Convert to Mono<Void>
+			}
+		};
+	};
+
+	private NotificationHandler resourcesChangeNotificationHandler(
+			List<Consumer<List<McpSchema.Resource>>> resourcesChangeConsumers) {
+
+		return new NotificationHandler() {
+			@Override
+			public Mono<Void> handle(Object params) {
+				// TODO: add support for cursor/pagination
+				return listResources().flatMap(listResourcesResult -> Mono.fromRunnable(() -> {
+					for (Consumer<List<McpSchema.Resource>> resourceChangeConsumer : resourcesChangeConsumers) {
+						resourceChangeConsumer.accept(listResourcesResult.resources());
+					}
+				}).subscribeOn(Schedulers.boundedElastic())).onErrorResume(error -> {
+					logger.error("Error handling resources list change notification", error);
+					return Mono.empty();
+				}).then(); // Convert to Mono<Void>
+			}
+		};
+	};
+
+	private NotificationHandler promptsChangeNotificationHandler(
+			List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumers) {
+
+		return new NotificationHandler() {
+			@Override
+			public Mono<Void> handle(Object params) {
+				// TODO: add support for cursor/pagination
+				return listPrompts().flatMap(listPromptsResult -> Mono.fromRunnable(() -> {
+					for (Consumer<List<McpSchema.Prompt>> promptChangeConsumer : promptsChangeConsumers) {
+						promptChangeConsumer.accept(listPromptsResult.prompts());
+					}
+				}).subscribeOn(Schedulers.boundedElastic())).onErrorResume(error -> {
+					logger.error("Error handling prompts list change notification", error);
+					return Mono.empty();
+				}).then(); // Convert to Mono<Void>
+			}
+		};
+	};
+
+	/**
+	 * 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>
+	 * <li>Exchange and negotiate capabilities</li>
+	 * <li>Share implementation details</li>
+	 * </ul>
+	 * <br/>
+	 * 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>
+	 * <li>Client implementation information</li>
+	 * </ul>
+	 *
+	 * The server MUST respond with its own capabilities and information:
+	 * {@link McpSchema.ServerCapabilities}. <br/>
+	 * After successful initialization, the client MUST send an initialized notification
+	 * to indicate it is ready to begin normal operations.
+	 *
+	 * <br/>
+	 *
+	 * <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() {
+		McpSchema.InitializeRequest initializeRequest = new McpSchema.InitializeRequest(// @formatter:off
+				McpSchema.LATEST_PROTOCOL_VERSION,
+				new McpSchema.ClientCapabilities(null, rootCapabilities, null),
+				new McpSchema.Implementation("mcp-java-client", "0.2.0")); // @formatter:on
+
+		Mono<McpSchema.InitializeResult> result = this.mcpSession.sendRequest("initialize", initializeRequest,
+				new TypeReference<McpSchema.InitializeResult>() {
+				});
+
+		return result.flatMap(initializeResult -> {
+
+			logger.info("Server response with Protocol: {}, Capabilities: {}, Info: {} and Instructions {}",
+					initializeResult.protocolVersion(), initializeResult.capabilities(), initializeResult.serverInfo(),
+					initializeResult.instructions());
+
+			if (!McpSchema.LATEST_PROTOCOL_VERSION.equals(initializeResult.protocolVersion())) {
+				return Mono.error(new McpError(
+						"Unsupported protocol version from the server: " + initializeResult.protocolVersion()));
+			}
+			else {
+				return this.mcpSession.sendNotification("notifications/initialized", null).thenReturn(initializeResult);
+			}
+		});
+	}
+
+	/**
+	 * Send a roots/list_changed notification.
+	 */
+	public Mono<Void> sendRootsListChanged() {
+		return this.mcpSession.sendNotification("notifications/roots/list_changed");
+	}
+
+	/**
+	 * Send a synchronous ping request.
+	 */
+	public Mono<Void> ping() {
+		return this.mcpSession.sendRequest("ping", null, VOID_TYPE_REFERENCE);
+	}
+
+	// --------------------------
+	// Tools
+	// --------------------------
+	private static TypeReference<McpSchema.CallToolResult> CALL_TOOL_RESULT_TYPE_REF = new TypeReference<>() {
+	};
+
+	private static TypeReference<McpSchema.ListToolsResult> LIST_TOOLS_RESULT_TYPE_REF = new TypeReference<>() {
+	};
+
+	/**
+	 * 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.mcpSession.sendRequest("tools/call", callToolRequest, CALL_TOOL_RESULT_TYPE_REF);
+	}
+
+	/**
+	 * Send a tools/list request.
+	 * @return the list of tools result.
+	 */
+	public Mono<McpSchema.ListToolsResult> listTools() {
+		return this.listTools(null);
+	}
+
+	/**
+	 * Send a tools/list request.
+	 * @param cursor the cursor
+	 * @return the list of tools result.
+	 */
+	public Mono<McpSchema.ListToolsResult> listTools(String cursor) {
+		return this.mcpSession.sendRequest("tools/list", new McpSchema.PaginatedRequest(cursor),
+				LIST_TOOLS_RESULT_TYPE_REF);
+	}
+
+	// --------------------------
+	// Resources
+	// --------------------------
+
+	private static TypeReference<McpSchema.ListResourcesResult> LIST_RESOURCES_RESULT_TYPE_REF = new TypeReference<>() {
+	};
+
+	private static TypeReference<McpSchema.ReadResourceResult> READ_RESOURCE_RESULT_TYPE_REF = new TypeReference<>() {
+	};
+
+	private static TypeReference<McpSchema.ListResourceTemplatesResult> LIST_RESOURCE_TEMPLATES_RESULT_TYPE_REF = new TypeReference<>() {
+	};
+
+	/**
+	 * Send a resources/list request.
+	 * @return the list of resources result.
+	 */
+	public Mono<McpSchema.ListResourcesResult> listResources() {
+		return this.listResources(null);
+	}
+
+	/**
+	 * Send a resources/list request.
+	 * @param cursor the cursor
+	 * @return the list of resources result.
+	 */
+	public Mono<McpSchema.ListResourcesResult> listResources(String cursor) {
+		return this.mcpSession.sendRequest("resources/list", new McpSchema.PaginatedRequest(cursor),
+				LIST_RESOURCES_RESULT_TYPE_REF);
+	}
+
+	/**
+	 * Send a resources/read request.
+	 * @param resource the resource to read
+	 * @return the resource content.
+	 */
+	public Mono<McpSchema.ReadResourceResult> readResource(McpSchema.Resource resource) {
+		return this.readResource(new McpSchema.ReadResourceRequest(resource.uri()));
+	}
+
+	/**
+	 * 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.mcpSession.sendRequest("resources/read", readResourceRequest, READ_RESOURCE_RESULT_TYPE_REF);
+	}
+
+	/**
+	 * Resource templates allow servers to expose parameterized resources using URI
+	 * templates. Arguments may be auto-completed through the completion API.
+	 *
+	 * Request a list of resource templates the server has.
+	 * @return the list of resource templates result.
+	 */
+	public Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplates() {
+		return this.listResourceTemplates(null);
+	}
+
+	/**
+	 * Resource templates allow servers to expose parameterized resources using URI
+	 * templates. Arguments may be auto-completed through the completion API.
+	 *
+	 * Request a list of resource templates the server has.
+	 * @param cursor the cursor
+	 * @return the list of resource templates result.
+	 */
+	public Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplates(String cursor) {
+		return this.mcpSession.sendRequest("resources/templates/list", new McpSchema.PaginatedRequest(cursor),
+				LIST_RESOURCE_TEMPLATES_RESULT_TYPE_REF);
+	}
+
+	/**
+	 * List Changed Notification. When the list of available resources changes, servers
+	 * that declared the listChanged capability SHOULD send a notification:
+	 */
+	public Mono<Void> sendResourcesListChanged() {
+		return this.mcpSession.sendNotification("notifications/resources/list_changed");
+	}
+
+	/**
+	 * 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.
+	 */
+	public Mono<Void> subscribeResource(McpSchema.SubscribeRequest subscribeRequest) {
+		return this.mcpSession.sendRequest("resources/subscribe", subscribeRequest, VOID_TYPE_REFERENCE);
+	}
+
+	/**
+	 * Send a resources/unsubscribe request.
+	 * @param unsubscribeRequest the unsubscribe request contains the uri of the resource
+	 * to unsubscribe from.
+	 */
+	public Mono<Void> unsubscribeResource(McpSchema.UnsubscribeRequest unsubscribeRequest) {
+		return this.mcpSession.sendRequest("resources/unsubscribe", unsubscribeRequest, VOID_TYPE_REFERENCE);
+	}
+
+	// --------------------------
+	// Prompts
+	// --------------------------
+	private static TypeReference<McpSchema.ListPromptsResult> LIST_PROMPTS_RESULT_TYPE_REF = new TypeReference<>() {
+	};
+
+	private static TypeReference<McpSchema.GetPromptResult> GET_PROMPT_RESULT_TYPE_REF = new TypeReference<>() {
+	};
+
+	/**
+	 * List all available prompts.
+	 * @return the list of prompts result.
+	 */
+	public Mono<ListPromptsResult> listPrompts() {
+		return this.listPrompts(null);
+	}
+
+	/**
+	 * List all available prompts.
+	 * @param cursor the cursor
+	 * @return the list of prompts result.
+	 */
+	public Mono<ListPromptsResult> listPrompts(String cursor) {
+		return this.mcpSession.sendRequest("prompts/list", new PaginatedRequest(cursor), LIST_PROMPTS_RESULT_TYPE_REF);
+	}
+
+	/**
+	 * Get a prompt by its id.
+	 * @param getPromptRequest the get prompt request.
+	 * @return the get prompt result.
+	 */
+	public Mono<GetPromptResult> getPrompt(GetPromptRequest getPromptRequest) {
+		return this.mcpSession.sendRequest("prompts/get", getPromptRequest, GET_PROMPT_RESULT_TYPE_REF);
+	}
+
+	/**
+	 * (Server) An optional notification from the server to the client, informing it that
+	 * the list of prompts it offers has changed. This may be issued by servers without
+	 * any previous subscription from the client.
+	 */
+	public Mono<Void> promptListChangedNotification() {
+		return this.mcpSession.sendNotification("notifications/prompts/list_changed");
+	}
+
+	public void close() {
+		this.mcpSession.close();
+	}
+
+	public Mono<Void> closeGracefully() {
+		return this.mcpSession.closeGracefully();
+	}
+
+}
--- a/mcp/src/main/java/org/springframework/ai/mcp/client/McpClient.java
+++ b/mcp/src/main/java/org/springframework/ai/mcp/client/McpClient.java
@ -0,0 +1,183 @@
+/*
+ * 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;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import org.springframework.ai.mcp.spec.McpSchema;
+import org.springframework.ai.mcp.spec.McpSchema.Root;
+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)
+ * clients. This class serves as the main entry point for establishing connections with
+ * MCP servers, offering both synchronous and asynchronous client implementations.
+ *
+ * <p>
+ * The class provides factory methods to create either:
+ * <ul>
+ * <li>{@link McpAsyncClient} for non-blocking operations
+ * <li>{@link McpSyncClient} for blocking operations
+ * </ul>
+ *
+ * <p>
+ * Use the builder pattern for flexible client configuration:
+ *
+ * <pre>{@code
+ * McpClient.using(transport)
+ * 		.withRequestTimeout(Duration.ofSeconds(5))
+ * 		.sync(); // or .async()
+ * }</pre>
+ *
+ * @author Christian Tzolov
+ * @author Dariusz Jędrzejczyk
+ */
+public class McpClient {
+
+	/**
+	 * Private constructor to prevent instantiation as this is a utility class containing
+	 * only static factory methods.
+	 */
+	private McpClient() {
+	}
+
+	/**
+	 * Start building an MCP client with the specified transport.
+	 * @param transport The transport layer implementation for MCP communication
+	 * @return A new builder instance
+	 */
+	public static Builder using(McpTransport transport) {
+		return new Builder(transport);
+	}
+
+	/**
+	 * Builder class for creating MCP clients with custom configuration.
+	 */
+	public static class Builder {
+
+		private final McpTransport transport;
+
+		private Duration requestTimeout = Duration.ofSeconds(20); // Default timeout
+
+		private boolean rootsListChangedNotification = false;
+
+		private List<Supplier<List<Root>>> rootsListProviders = new ArrayList<>();
+
+		private List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumers = new ArrayList<>();
+
+		private List<Consumer<List<McpSchema.Resource>>> resourcesChangeConsumers = new ArrayList<>();
+
+		private List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumers = new ArrayList<>();
+
+		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 requestTimeout(Duration requestTimeout) {
+			Assert.notNull(requestTimeout, "Request timeout must not be null");
+			this.requestTimeout = requestTimeout;
+			return this;
+		}
+
+		public Builder rootsListChangedNotification(boolean rootsListChangedNotification) {
+			this.rootsListChangedNotification = rootsListChangedNotification;
+			return this;
+		}
+
+		public Builder rootsListProvider(Supplier<List<Root>> rootsListProvider) {
+			this.rootsListProviders.add(rootsListProvider);
+			return this;
+		}
+
+		public Builder toolsChangeConsumer(Consumer<List<McpSchema.Tool>> toolsChangeConsumer) {
+			this.toolsChangeConsumers.add(toolsChangeConsumer);
+			return this;
+		}
+
+		public Builder resourcesChangeConsumer(Consumer<List<McpSchema.Resource>> resourcesChangeConsumer) {
+			this.resourcesChangeConsumers.add(resourcesChangeConsumer);
+			return this;
+		}
+
+		public Builder promptsChangeConsumer(Consumer<List<McpSchema.Prompt>> promptsChangeConsumer) {
+			this.promptsChangeConsumers.add(promptsChangeConsumer);
+			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, rootsListProviders, rootsListChangedNotification,
+					toolsChangeConsumers, resourcesChangeConsumers, promptsChangeConsumers);
+		}
+
+	}
+
+	/**
+	 * @deprecated Use {@link #using(McpTransport)} instead.
+	 */
+	@Deprecated
+	public static McpAsyncClient async(McpTransport transport) {
+		return using(transport).async();
+	}
+
+	/**
+	 * @deprecated Use {@link #using(McpTransport)} instead.
+	 */
+	@Deprecated
+	public static McpAsyncClient async(McpTransport transport, Duration requestTimeout) {
+		return using(transport).requestTimeout(requestTimeout).async();
+	}
+
+	/**
+	 * @deprecated Use {@link #using(McpTransport)} instead.
+	 */
+	@Deprecated
+	public static McpSyncClient sync(McpTransport transport) {
+		return using(transport).sync();
+	}
+
+	/**
+	 * @deprecated Use {@link #using(McpTransport)} instead.
+	 */
+	@Deprecated
+	public static McpSyncClient sync(McpTransport transport, Duration requestTimeout) {
+		return using(transport).requestTimeout(requestTimeout).sync();
+	}
+
+}
--- a/mcp/src/main/java/org/springframework/ai/mcp/client/McpSyncClient.java
+++ b/mcp/src/main/java/org/springframework/ai/mcp/client/McpSyncClient.java
@ -0,0 +1,257 @@
+/*
+ * 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;
+
+import java.time.Duration;
+import java.util.concurrent.TimeoutException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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
+ * @author Christian Tzolov
+ */
+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?
+	private static final long DEFAULT_CLOSE_TIMEOUT_MS = 10_000L;
+
+	private final McpAsyncClient delegate;
+
+	public McpSyncClient(McpAsyncClient delegate) {
+		Assert.notNull(delegate, "The delegate can not be null");
+		this.delegate = delegate;
+	}
+
+	@Override
+	public void close() {
+		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;
+	}
+
+	/**
+	 * 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>
+	 * <li>Exchange and negotiate capabilities</li>
+	 * <li>Share implementation details</li>
+	 * </ul>
+	 * <br/>
+	 * 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>
+	 * <li>Client implementation information</li>
+	 * </ul>
+	 *
+	 * The server MUST respond with its own capabilities and information:
+	 * {@link McpSchema.ServerCapabilities}. <br/>
+	 * After successful initialization, the client MUST send an initialized notification
+	 * to indicate it is ready to begin normal operations.
+	 *
+	 * <br/>
+	 *
+	 * <a href=
+	 * "https://github.com/modelcontextprotocol/specification/blob/main/docs/specification/basic/lifecycle.md#initialization">Initialization
+	 * Spec</a>
+	 * @return the initialize result.
+	 */
+	public McpSchema.InitializeResult initialize() {
+		// TODO: block takes no argument here as we assume the async client is
+		// configured with a requestTimeout at all times
+		return this.delegate.initialize().block();
+	}
+
+	/**
+	 * Send a roots/list_changed notification.
+	 */
+	public void sendRootsListChanged() {
+		this.delegate.sendRootsListChanged().block();
+	}
+
+	/**
+	 * Send a synchronous ping request.
+	 */
+	public void ping() {
+		this.delegate.ping().block();
+	}
+
+	// --------------------------
+	// Tools
+	// --------------------------
+	/**
+	 * Send a tools/call request.
+	 * @param callToolRequest the call tool request.
+	 * @return the call tool result.
+	 */
+	public McpSchema.CallToolResult callTool(McpSchema.CallToolRequest callToolRequest) {
+		return this.delegate.callTool(callToolRequest).block();
+	}
+
+	/**
+	 * Send a tools/list request.
+	 * @return the list of tools result.
+	 */
+	public McpSchema.ListToolsResult listTools() {
+		return this.delegate.listTools().block();
+	}
+
+	/**
+	 * Send a tools/list request.
+	 * @param cursor the cursor
+	 * @return the list of tools result.
+	 */
+	public McpSchema.ListToolsResult listTools(String cursor) {
+		return this.delegate.listTools(cursor).block();
+	}
+
+	// --------------------------
+	// Resources
+	// --------------------------
+
+	/**
+	 * Send a resources/list request.
+	 * @param cursor the cursor
+	 * @return the list of resources result.
+	 */
+	public McpSchema.ListResourcesResult listResources(String cursor) {
+		return this.delegate.listResources(cursor).block();
+	}
+
+	/**
+	 * Send a resources/list request.
+	 * @return the list of resources result.
+	 */
+	public McpSchema.ListResourcesResult listResources() {
+		return this.delegate.listResources().block();
+	}
+
+	/**
+	 * Send a resources/read request.
+	 * @param resource the resource to read
+	 * @return the resource content.
+	 */
+	public McpSchema.ReadResourceResult readResource(McpSchema.Resource resource) {
+		return this.delegate.readResource(resource).block();
+	}
+
+	/**
+	 * Send a resources/read request.
+	 * @param readResourceRequest the read resource request.
+	 * @return the resource content.
+	 */
+	public McpSchema.ReadResourceResult readResource(McpSchema.ReadResourceRequest readResourceRequest) {
+		return this.delegate.readResource(readResourceRequest).block();
+	}
+
+	/**
+	 * Resource templates allow servers to expose parameterized resources using URI
+	 * templates. Arguments may be auto-completed through the completion API.
+	 *
+	 * Request a list of resource templates the server has.
+	 * @param cursor the cursor
+	 * @return the list of resource templates result.
+	 */
+	public McpSchema.ListResourceTemplatesResult listResourceTemplates(String cursor) {
+		return this.delegate.listResourceTemplates(cursor).block();
+	}
+
+	/**
+	 * Request a list of resource templates the server has.
+	 * @return the list of resource templates result.
+	 */
+	public McpSchema.ListResourceTemplatesResult listResourceTemplates() {
+		return this.delegate.listResourceTemplates().block();
+	}
+
+	/**
+	 * List Changed Notification. When the list of available resources changes, servers
+	 * that declared the listChanged capability SHOULD send a notification:
+	 */
+	public void sendResourcesListChanged() {
+		this.delegate.sendResourcesListChanged().block();
+	}
+
+	/**
+	 * 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.
+	 */
+	public void subscribeResource(McpSchema.SubscribeRequest subscribeRequest) {
+		this.delegate.subscribeResource(subscribeRequest).block();
+	}
+
+	/**
+	 * Send a resources/unsubscribe request.
+	 * @param unsubscribeRequest the unsubscribe request contains the uri of the resource
+	 * to unsubscribe from.
+	 */
+	public void unsubscribeResource(McpSchema.UnsubscribeRequest unsubscribeRequest) {
+		this.delegate.unsubscribeResource(unsubscribeRequest).block();
+	}
+
+	// --------------------------
+	// Prompts
+	// --------------------------
+	public ListPromptsResult listPrompts(String cursor) {
+		return this.delegate.listPrompts(cursor).block();
+	}
+
+	public ListPromptsResult listPrompts() {
+		return this.delegate.listPrompts().block();
+	}
+
+	public GetPromptResult getPrompt(GetPromptRequest getPromptRequest) {
+		return this.delegate.getPrompt(getPromptRequest).block();
+	}
+
+	/**
+	 * (Server) An optional notification from the server to the client, informing it that
+	 * the list of prompts it offers has changed. This may be issued by servers without
+	 * any previous subscription from the client.
+	 */
+	public void promptListChangedNotification() {
+		this.delegate.promptListChangedNotification().block();
+	}
+
+}
--- a/mcp/src/main/java/org/springframework/ai/mcp/client/sse/SseClientTransport.java
+++ b/mcp/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);
+	}
+
+}
--- a/mcp/src/main/java/org/springframework/ai/mcp/client/stdio/ServerParameters.java
+++ b/mcp/src/main/java/org/springframework/ai/mcp/client/stdio/ServerParameters.java
@ -0,0 +1,150 @@
+/*
+ * 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.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import org.springframework.ai.mcp.util.Assert;
+
+/**
+ * Server parameters for stdio client.
+ *
+ * @author Christian Tzolov
+ * @author Dariusz Jędrzejczyk
+ */
+@JsonInclude(JsonInclude.Include.NON_ABSENT)
+public class ServerParameters {
+
+	// Environment variables to inherit by default
+	private static final List<String> DEFAULT_INHERITED_ENV_VARS = System.getProperty("os.name")
+		.toLowerCase()
+		.contains("win")
+				? Arrays.asList("APPDATA", "HOMEDRIVE", "HOMEPATH", "LOCALAPPDATA", "PATH", "PROCESSOR_ARCHITECTURE",
+						"SYSTEMDRIVE", "SYSTEMROOT", "TEMP", "USERNAME", "USERPROFILE")
+				: Arrays.asList("HOME", "LOGNAME", "PATH", "SHELL", "TERM", "USER");
+
+	@JsonProperty("command")
+	private String command;
+
+	@JsonProperty("args")
+	private List<String> args = new ArrayList<>();
+
+	@JsonProperty("env")
+	private Map<String, String> env;
+
+	private ServerParameters(String command, List<String> args, Map<String, String> env) {
+		Assert.notNull(command, "The command can not be null");
+		Assert.notNull(args, "The args can not be null");
+
+		this.command = command;
+		this.args = args;
+		this.env = new HashMap<>(getDefaultEnvironment());
+		if (env != null && !env.isEmpty()) {
+			this.env.putAll(env);
+		}
+	}
+
+	public String getCommand() {
+		return this.command;
+	}
+
+	public List<String> getArgs() {
+		return this.args;
+	}
+
+	public Map<String, String> getEnv() {
+		return this.env;
+	}
+
+	public static Builder builder(String command) {
+		return new Builder(command);
+	}
+
+	public static class Builder {
+
+		private String command;
+
+		private List<String> args = new ArrayList<>();
+
+		private Map<String, String> env = new HashMap<>();
+
+		public Builder(String command) {
+			Assert.notNull(command, "The command can not be null");
+			this.command = command;
+		}
+
+		public Builder args(String... args) {
+			Assert.notNull(args, "The args can not be null");
+			this.args = Arrays.asList(args);
+			return this;
+		}
+
+		public Builder args(List<String> args) {
+			Assert.notNull(args, "The args can not be null");
+			this.args = new ArrayList<>(args);
+			return this;
+		}
+
+		public Builder arg(String arg) {
+			Assert.notNull(arg, "The arg can not be null");
+			this.args.add(arg);
+			return this;
+		}
+
+		public Builder env(Map<String, String> env) {
+			if (env != null && !env.isEmpty()) {
+				this.env.putAll(env);
+			}
+			return this;
+		}
+
+		public Builder addEnvVar(String key, String value) {
+			Assert.notNull(key, "The key can not be null");
+			Assert.notNull(value, "The value can not be null");
+			this.env.put(key, value);
+			return this;
+		}
+
+		public ServerParameters build() {
+			return new ServerParameters(command, args, env);
+		}
+
+	}
+
+	/**
+	 * Returns a default environment object including only environment variables deemed
+	 * safe to inherit.
+	 */
+	private static Map<String, String> getDefaultEnvironment() {
+		return System.getenv()
+			.entrySet()
+			.stream()
+			.filter(entry -> DEFAULT_INHERITED_ENV_VARS.contains(entry.getKey()))
+			.filter(entry -> entry.getValue() != null)
+			.filter(entry -> !entry.getValue().startsWith("()"))
+			.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+	}
+
+}
--- a/mcp/src/main/java/org/springframework/ai/mcp/client/stdio/StdioClientTransport.java
+++ b/mcp/src/main/java/org/springframework/ai/mcp/client/stdio/StdioClientTransport.java
@ -0,0 +1,351 @@
+/*
+ * 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);
+						// Escape any embedded newlines in the JSON message as per spec:
+						// https://spec.modelcontextprotocol.io/specification/basic/transports/#stdio
+						// - Messages are delimited by newlines, and MUST NOT contain
+						// embedded newlines.
+						jsonMessage = jsonMessage.replace("\r\n", "\\n").replace("\n", "\\n").replace("\r", "\\n");
+
+						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);
+	}
+
+}
--- a/mcp/src/main/java/org/springframework/ai/mcp/spec/DefaultMcpSession.java
+++ b/mcp/src/main/java/org/springframework/ai/mcp/spec/DefaultMcpSession.java
@ -0,0 +1,278 @@
+/*
+ * 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.time.Duration;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+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.util.Assert;
+
+/**
+ * Default implementation of the MCP (Model Context Protocol) session that manages
+ * bidirectional JSON-RPC communication between clients and servers. This implementation
+ * follows the MCP specification for message exchange and transport handling.
+ *
+ * <p>
+ * The session manages:
+ * <ul>
+ * <li>Request/response handling with unique message IDs</li>
+ * <li>Notification processing</li>
+ * <li>Message timeout management</li>
+ * <li>Transport layer abstraction</li>
+ * </ul>
+ *
+ * @author Christian Tzolov
+ * @author Dariusz Jędrzejczyk
+ */
+public class DefaultMcpSession implements McpSession {
+
+	/** Logger for this class */
+	private static final Logger logger = LoggerFactory.getLogger(DefaultMcpSession.class);
+
+	/** Duration to wait for request responses before timing out */
+	private final Duration requestTimeout;
+
+	/** Transport layer implementation for message exchange */
+	private final McpTransport transport;
+
+	/** Map of pending responses keyed by request ID */
+	private final ConcurrentHashMap<Object, MonoSink<McpSchema.JSONRPCResponse>> pendingResponses = new ConcurrentHashMap<>();
+
+	/** Map of request handlers keyed by method name */
+	private final ConcurrentHashMap<String, RequestHandler> requestHandlers = new ConcurrentHashMap<>();
+
+	/** Map of notification handlers keyed by method name */
+	private final ConcurrentHashMap<String, NotificationHandler> notificationHandlers = new ConcurrentHashMap<>();
+
+	/** Session-specific prefix for request IDs */
+	private final String sessionPrefix = UUID.randomUUID().toString().substring(0, 8);
+
+	/** 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.
+	 */
+	@FunctionalInterface
+	public interface RequestHandler {
+
+		/**
+		 * Handles an incoming request with the given parameters.
+		 * @param params The request parameters
+		 * @return A Mono containing the response object
+		 */
+		Mono<Object> handle(Object params);
+
+	}
+
+	/**
+	 * Functional interface for handling incoming JSON-RPC notifications. Implementations
+	 * should process the notification parameters without returning a response.
+	 */
+	@FunctionalInterface
+	public interface NotificationHandler {
+
+		/**
+		 * Handles an incoming notification with the given parameters.
+		 * @param params The notification parameters
+		 * @return A Mono that completes when the notification is processed
+		 */
+		Mono<Void> handle(Object params);
+
+	}
+
+	/**
+	 * Creates a new DefaultMcpSession with the specified configuration and handlers.
+	 * @param requestTimeout Duration to wait for responses
+	 * @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, McpTransport transport,
+			Map<String, RequestHandler> requestHandlers, Map<String, NotificationHandler> notificationHandlers) {
+
+		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.transport = transport;
+		this.requestHandlers.putAll(requestHandlers);
+		this.notificationHandlers.putAll(notificationHandlers);
+
+		// 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) {
+					logger.warn("Unexpected response for unkown id {}", response.id());
+				}
+				else {
+					sink.success(response);
+				}
+			}
+			else if (message instanceof McpSchema.JSONRPCRequest request) {
+				handleIncomingRequest(request).subscribe(response -> transport.sendMessage(response).subscribe(),
+						error -> {
+							var errorResponse = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
+									null, new McpSchema.JSONRPCResponse.JSONRPCError(
+											McpSchema.ErrorCodes.INTERNAL_ERROR, error.getMessage(), null));
+							transport.sendMessage(errorResponse).subscribe();
+						});
+			}
+			else if (message instanceof McpSchema.JSONRPCNotification notification) {
+				handleIncomingNotification(notification).subscribe(null,
+						error -> logger.error("Error handling notification: {}", error.getMessage()));
+			}
+		})).subscribe();
+	}
+
+	/**
+	 * Handles an incoming JSON-RPC request by routing it to the appropriate handler.
+	 * @param request The incoming JSON-RPC request
+	 * @return A Mono containing the JSON-RPC response
+	 */
+	private Mono<McpSchema.JSONRPCResponse> handleIncomingRequest(McpSchema.JSONRPCRequest request) {
+		return Mono.defer(() -> {
+			var handler = this.requestHandlers.get(request.method());
+			if (handler == null) {
+				return Mono.just(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), null,
+						new McpSchema.JSONRPCResponse.JSONRPCError(McpSchema.ErrorCodes.METHOD_NOT_FOUND,
+								"Method not found: " + request.method(), null)));
+			}
+
+			return handler.handle(request.params())
+				.map(result -> new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), result, null))
+				.onErrorResume(error -> Mono.just(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
+						null, new McpSchema.JSONRPCResponse.JSONRPCError(McpSchema.ErrorCodes.INTERNAL_ERROR,
+								error.getMessage(), null)))); // TODO: add error message
+																// through the data field
+		});
+	}
+
+	/**
+	 * Handles an incoming JSON-RPC notification by routing it to the appropriate handler.
+	 * @param notification The incoming JSON-RPC notification
+	 * @return A Mono that completes when the notification is processed
+	 */
+	private Mono<Void> handleIncomingNotification(McpSchema.JSONRPCNotification notification) {
+		return Mono.defer(() -> {
+			var handler = notificationHandlers.get(notification.method());
+			if (handler == null) {
+				logger.error("No handler registered for notification method: {}", notification.method());
+				return Mono.empty();
+			}
+			return handler.handle(notification.params());
+		});
+	}
+
+	/**
+	 * Generates a unique request ID in a non-blocking way. Combines a session-specific
+	 * prefix with an atomic counter to ensure uniqueness.
+	 * @return A unique request ID string
+	 */
+	private String generateRequestId() {
+		return this.sessionPrefix + "-" + this.requestCounter.getAndIncrement();
+	}
+
+	/**
+	 * Sends a JSON-RPC request and returns the response.
+	 * @param <T> The expected response type
+	 * @param method The method name to call
+	 * @param requestParams The request parameters
+	 * @param typeRef Type reference for response deserialization
+	 * @return A Mono containing the response
+	 */
+	@Override
+	public <T> Mono<T> sendRequest(String method, Object requestParams, TypeReference<T> typeRef) {
+		String requestId = this.generateRequestId();
+
+		return Mono.<McpSchema.JSONRPCResponse>create(sink -> {
+			this.pendingResponses.put(requestId, sink);
+			McpSchema.JSONRPCRequest jsonrpcRequest = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, method,
+					requestId, requestParams);
+			this.transport.sendMessage(jsonrpcRequest)
+				// TODO: It's most efficient to create a dedicated Subscriber here
+				.subscribe(v -> {
+				}, error -> {
+					this.pendingResponses.remove(requestId);
+					sink.error(error);
+				});
+		}).timeout(this.requestTimeout).handle((jsonRpcResponse, sink) -> {
+			if (jsonRpcResponse.error() != null) {
+				sink.error(new McpError(jsonRpcResponse.error()));
+			}
+			else {
+				if (typeRef.getType().equals(Void.class)) {
+					sink.complete();
+				}
+				else {
+					sink.next(this.transport.unmarshalFrom(jsonRpcResponse.result(), typeRef));
+				}
+			}
+		});
+	}
+
+	/**
+	 * Sends a JSON-RPC notification.
+	 * @param method The method name for the notification
+	 * @param params The notification parameters
+	 * @return A Mono that completes when the notification is sent
+	 */
+	@Override
+	public Mono<Void> sendNotification(String method, Map<String, Object> params) {
+		McpSchema.JSONRPCNotification jsonrpcNotification = new McpSchema.JSONRPCNotification(McpSchema.JSONRPC_VERSION,
+				method, params);
+		return this.transport.sendMessage(jsonrpcNotification);
+	}
+
+	/**
+	 * Closes the session gracefully, allowing pending operations to complete.
+	 * @return A Mono that completes when the session is closed
+	 */
+	@Override
+	public Mono<Void> closeGracefully() {
+		this.connection.dispose();
+		return transport.closeGracefully();
+	}
+
+	/**
+	 * Closes the session immediately, potentially interrupting pending operations.
+	 */
+	@Override
+	public void close() {
+		this.connection.dispose();
+		transport.close();
+	}
+
+}
--- a/mcp/src/main/java/org/springframework/ai/mcp/spec/McpError.java
+++ b/mcp/src/main/java/org/springframework/ai/mcp/spec/McpError.java
@ -0,0 +1,24 @@
+/*
+* 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;
+
+public class McpError extends RuntimeException {
+
+	public McpError(Object error) {
+		super(error.toString());
+	}
+
+}
--- a/mcp/src/main/java/org/springframework/ai/mcp/spec/McpSchema.java
+++ b/mcp/src/main/java/org/springframework/ai/mcp/spec/McpSchema.java
@ -0,0 +1,719 @@
+/*
+ * 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.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.fasterxml.jackson.annotation.JsonTypeInfo.As;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * Based on the <a href="http://www.jsonrpc.org/specification">JSON-RPC 2.0
+ * specification<a/> and the <a href=
+ * "https://github.com/modelcontextprotocol/specification/blob/main/schema/schema.ts">Model
+ * Context Protocol Schema</a>.
+ *
+ * @author Christian Tzolov
+ */
+public class McpSchema {
+
+	public static final String LATEST_PROTOCOL_VERSION = "2024-11-05";
+
+	public static final String JSONRPC_VERSION = "2.0";
+
+	// ---------------------------
+	// JSON-RPC Error Codes
+	// ---------------------------
+	public final class ErrorCodes {
+
+		public static final int PARSE_ERROR = -32700;
+
+		public static final int INVALID_REQUEST = -32600;
+
+		public static final int METHOD_NOT_FOUND = -32601;
+
+		public static final int INVALID_PARAMS = -32602;
+
+		public static final int INTERNAL_ERROR = -32603;
+
+	}
+
+	public sealed interface Request
+			permits InitializeRequest, CallToolRequest, CreateMessageRequest, CompleteRequest, GetPromptRequest {
+
+	}
+
+	private final static TypeReference<HashMap<String, Object>> MAP_TYPE_REF = new TypeReference<>() {
+	};
+
+	/**
+	 * Deserializes a JSON string into a JSONRPCMessage object.
+	 * @param objectMapper The ObjectMapper instance to use for deserialization
+	 * @param jsonText The JSON string to deserialize
+	 * @return A JSONRPCMessage instance using either the {@link JSONRPCRequest},
+	 * {@link JSONRPCNotification}, or {@link JSONRPCResponse} classes.
+	 * @throws IOException If there's an error during deserialization
+	 * @throws IllegalArgumentException If the JSON structure doesn't match any known
+	 * message type
+	 */
+	public static JSONRPCMessage deserializeJsonRpcMessage(ObjectMapper objectMapper, String jsonText)
+			throws IOException {
+
+		var map = objectMapper.readValue(jsonText, MAP_TYPE_REF);
+
+		// Determine message type based on specific JSON structure
+		if (map.containsKey("method") && map.containsKey("id")) {
+			return objectMapper.convertValue(map, JSONRPCRequest.class);
+		}
+		else if (map.containsKey("method") && !map.containsKey("id")) {
+			return objectMapper.convertValue(map, JSONRPCNotification.class);
+		}
+		else if (map.containsKey("result") || map.containsKey("error")) {
+			return objectMapper.convertValue(map, JSONRPCResponse.class);
+		}
+
+		throw new IllegalArgumentException("Cannot deserialize JSONRPCMessage: " + jsonText);
+	}
+
+	// ---------------------------
+	// JSON-RPC Message Types
+	// ---------------------------
+	public sealed interface JSONRPCMessage permits JSONRPCRequest, JSONRPCNotification, JSONRPCResponse {
+
+		String jsonrpc();
+
+	}
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record JSONRPCRequest( // @formatter:off
+			@JsonProperty("jsonrpc") String jsonrpc,
+			@JsonProperty("method") String method,
+			@JsonProperty("id") Object id,
+			@JsonProperty("params") Object params) implements JSONRPCMessage {
+	} // @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record JSONRPCNotification( // @formatter:off
+			@JsonProperty("jsonrpc") String jsonrpc,
+			@JsonProperty("method") String method,
+			@JsonProperty("params") Map<String, Object> params) implements JSONRPCMessage {
+	} // @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record JSONRPCResponse( // @formatter:off
+			@JsonProperty("jsonrpc") String jsonrpc,
+			@JsonProperty("id") Object id,
+			@JsonProperty("result") Object result,
+			@JsonProperty("error") JSONRPCError error) implements JSONRPCMessage {
+
+		@JsonInclude(JsonInclude.Include.NON_ABSENT)
+		public record JSONRPCError(
+			@JsonProperty("code") int code,
+			@JsonProperty("message") String message,
+			@JsonProperty("data") Object data) {
+		}
+	}// @formatter:on
+
+	// ---------------------------
+	// Initialization
+	// ---------------------------
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record InitializeRequest( // @formatter:off
+		@JsonProperty("protocolVersion") String protocolVersion,
+		@JsonProperty("capabilities") ClientCapabilities capabilities,
+		@JsonProperty("clientInfo") Implementation clientInfo) implements Request {		
+	} // @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record InitializeResult( // @formatter:off
+		@JsonProperty("protocolVersion") String protocolVersion,
+		@JsonProperty("capabilities") ServerCapabilities capabilities,
+		@JsonProperty("serverInfo") Implementation serverInfo,
+		@JsonProperty("instructions") String instructions) {
+	} // @formatter:on
+
+	/**
+	 * Clients can implement additional features to enrich connected MCP servers with
+	 * additional capabilities. These capabilities can be used to extend the functionality
+	 * of the server, or to provide additional information to the server about the
+	 * client's capabilities.
+	 *
+	 * @param experimental WIP
+	 * @param roots define the boundaries of where servers can operate within the
+	 * filesystem, allowing them to understand which directories and files they have
+	 * access to.
+	 * @param sampling Provides a standardized way for servers to request LLM sampling
+	 * (“completions” or “generations”) from language models via clients.
+	 *
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ClientCapabilities( // @formatter:off
+		@JsonProperty("experimental") Map<String, Object> experimental,
+		@JsonProperty("roots") RootCapabilities roots,
+		@JsonProperty("sampling") Sampling sampling) {
+
+		/**
+		 * Roots define the boundaries of where servers can operate within the filesystem,
+		 * allowing them to understand which directories and files they have access to.
+		 * Servers can request the list of roots from supporting clients and
+		 * receive notifications when that list changes.
+		 *
+		 * @param listChanged Whether the client would send notification about roots
+		 * 		  has changed since the last time the server checked.
+		 */
+		@JsonInclude(JsonInclude.Include.NON_ABSENT)			
+		public record RootCapabilities(
+			@JsonProperty("listChanged") Boolean listChanged) {
+		}
+
+		/**
+		 * Provides a standardized way for servers to request LLM
+	 	 * sampling (“completions” or “generations”) from language
+		 * models via clients. This flow allows clients to maintain
+		 * control over model access, selection, and permissions
+		 * while enabling servers to leverage AI capabilities—with
+		 * no server API keys necessary. Servers can request text or
+		 * image-based interactions and optionally include context
+		 * from MCP servers in their prompts.
+		 */
+		@JsonInclude(JsonInclude.Include.NON_ABSENT)			
+		public record Sampling() {
+		}
+	}// @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ServerCapabilities( // @formatter:off
+		@JsonProperty("experimental") Map<String, Object> experimental,
+		@JsonProperty("logging") Object logging,
+		@JsonProperty("prompts") PromptCapabilities prompts,
+		@JsonProperty("resources") ResourceCapabilities resources,
+		@JsonProperty("tools") ToolCapabilities tools) {
+		
+		@JsonInclude(JsonInclude.Include.NON_ABSENT)
+		public record PromptCapabilities(
+			@JsonProperty("listChanged") Boolean listChanged) {
+		}
+
+		@JsonInclude(JsonInclude.Include.NON_ABSENT)
+		public record ResourceCapabilities(
+			@JsonProperty("subscribe") Boolean subscribe,
+			@JsonProperty("listChanged") Boolean listChanged) {
+		}
+
+		@JsonInclude(JsonInclude.Include.NON_ABSENT)
+		public record ToolCapabilities(
+			@JsonProperty("listChanged") Boolean listChanged) {
+		}
+	} // @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record Implementation(// @formatter:off
+		@JsonProperty("name") String name,
+		@JsonProperty("version") String version) {
+	} // @formatter:on
+
+	// Existing Enums and Base Types (from previous implementation)
+	public enum Role {// @formatter:off
+
+		@JsonProperty("user") USER,
+		@JsonProperty("assistant") ASSISTANT
+	}// @formatter:on
+
+	public enum LoggingLevel {// @formatter:off
+
+		@JsonProperty("debug") DEBUG,
+		@JsonProperty("info") INFO,
+		@JsonProperty("notice") NOTICE,
+		@JsonProperty("warning") WARNING,
+		@JsonProperty("error") ERROR,
+		@JsonProperty("critical") CRITICAL,
+		@JsonProperty("alert") ALERT,
+		@JsonProperty("emergency") EMERGENCY
+
+	} // @formatter:on
+
+	// ---------------------------
+	// Resource Interfaces
+	// ---------------------------
+	/**
+	 * Base for objects that include optional annotations for the client. The client can
+	 * use annotations to inform how objects are used or displayed
+	 */
+	public interface Annotated {
+
+		Annotations annotations();
+
+	}
+
+	/**
+	 * @param audience Describes who the intended customer of this object or data is. It
+	 * can include multiple entries to indicate content useful for multiple audiences
+	 * (e.g., `["user", "assistant"]`).
+	 * @param priority Describes how important this data is for operating the server. A
+	 * value of 1 means "most important," and indicates that the data is effectively
+	 * required, while 0 means "least important," and indicates that the data is entirely
+	 * optional. It is a number between 0 and 1.
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record Annotations( // @formatter:off
+		@JsonProperty("audience") List<Role> audience,
+		@JsonProperty("priority") Double priority) {
+	} // @formatter:on
+
+	/**
+	 * A known resource that the server is capable of reading.
+	 *
+	 * @param uri the URI of the resource.
+	 * @param name A human-readable name for this resource. This can be used by clients to
+	 * populate UI elements.
+	 * @param description A description of what this resource represents. This can be used
+	 * by clients to improve the LLM's understanding of available resources. It can be
+	 * thought of like a "hint" to the model.
+	 * @param mimeType The MIME type of this resource, if known.
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	@JsonIgnoreProperties(ignoreUnknown = true)
+	public record Resource( // @formatter:off
+		@JsonProperty("uri") String uri,
+		@JsonProperty("name") String name,
+		@JsonProperty("description") String description,
+		@JsonProperty("mimeType") String mimeType,
+		@JsonProperty("annotations") Annotations annotations) implements Annotated {
+	} // @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ResourceTemplate( // @formatter:off
+		@JsonProperty("uriTemplate") String uriTemplate,
+		@JsonProperty("name") String name,
+		@JsonProperty("description") String description,
+		@JsonProperty("mimeType") String mimeType,
+		@JsonProperty("annotations") Annotations annotations) implements Annotated {
+	} // @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ListResourcesResult( // @formatter:off
+		@JsonProperty("resources") List<Resource> resources,
+		@JsonProperty("nextCursor") String nextCursor) {
+	} // @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ListResourceTemplatesResult( // @formatter:off
+		@JsonProperty("resourceTemplates") List<ResourceTemplate> resourceTemplates,
+		@JsonProperty("nextCursor") String nextCursor) {
+	} // @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ReadResourceRequest( // @formatter:off
+		@JsonProperty("uri") String uri){
+	} // @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ReadResourceResult( // @formatter:off
+		@JsonProperty("contents") List<ResourceContents> contents){
+	} // @formatter:on
+
+	/**
+	 * Sent from the client to request resources/updated notifications from the server
+	 * whenever a particular resource changes.
+	 *
+	 * @param uri the URI of the resource to subscribe to. The URI can use any protocol;
+	 * it is up to the server how to interpret it.
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record SubscribeRequest( // @formatter:off
+		@JsonProperty("uri") String uri){
+	} // @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record UnsubscribeRequest( // @formatter:off
+		@JsonProperty("uri") String uri){
+	} // @formatter:on
+
+	/**
+	 * The contents of a specific resource or sub-resource.
+	 */
+	@JsonTypeInfo(use = JsonTypeInfo.Id.DEDUCTION, include = As.PROPERTY)
+	@JsonSubTypes({ @JsonSubTypes.Type(value = TextResourceContents.class, name = "text"),
+			@JsonSubTypes.Type(value = BlobResourceContents.class, name = "blob") })
+	public sealed interface ResourceContents permits TextResourceContents, BlobResourceContents {
+
+		/**
+		 * The URI of this resource.
+		 * @return the URI of this resource.
+		 */
+		String uri();
+
+		/**
+		 * The MIME type of this resource.
+		 * @return the MIME type of this resource.
+		 */
+		String mimeType();
+
+	}
+
+	/**
+	 * Text contents of a resource.
+	 *
+	 * @param uri the URI of this resource.
+	 * @param mimeType the MIME type of this resource.
+	 * @param text the text of the resource. This must only be set if the resource can
+	 * actually be represented as text (not binary data).
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	@JsonIgnoreProperties(ignoreUnknown = true)
+	public record TextResourceContents( // @formatter:off
+		@JsonProperty("uri") String uri,
+		@JsonProperty("mimeType") String mimeType,
+		@JsonProperty("text") String text) implements ResourceContents {
+	} // @formatter:on
+
+	/**
+	 * Binary contents of a resource.
+	 *
+	 * @param uri the URI of this resource.
+	 * @param mimeType the MIME type of this resource.
+	 * @param blob a base64-encoded string representing the binary data of the resource.
+	 * This must only be set if the resource can actually be represented as binary data
+	 * (not text).
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	@JsonIgnoreProperties(ignoreUnknown = true)
+	public record BlobResourceContents( // @formatter:off
+		@JsonProperty("uri") String uri,
+		@JsonProperty("mimeType") String mimeType,
+		@JsonProperty("blob") String blob) implements ResourceContents {
+	} // @formatter:on
+
+	// ---------------------------
+	// Prompt Interfaces
+	// ---------------------------
+	/**
+	 * A prompt or prompt template that the server offers.
+	 *
+	 * @param name The name of the prompt or prompt template.
+	 * @param description An optional description of what this prompt provides.
+	 * @param arguments A list of arguments to use for templating the prompt.
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record Prompt( // @formatter:off
+		@JsonProperty("name") String name,
+		@JsonProperty("description") String description,
+		@JsonProperty("arguments") List<PromptArgument> arguments) {
+	} // @formatter:on
+
+	/**
+	 * Describes an argument that a prompt can accept.
+	 *
+	 * @param name The name of the argument.
+	 * @param description A human-readable description of the argument.
+	 * @param required Whether this argument must be provided.
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record PromptArgument( // @formatter:off
+		@JsonProperty("name") String name,
+		@JsonProperty("description") String description,
+		@JsonProperty("required") Boolean required) {
+	}// @formatter:on
+
+	/**
+	 * Describes a message returned as part of a prompt.
+	 *
+	 * This is similar to `SamplingMessage`, but also supports the embedding of resources
+	 * from the MCP server.
+	 *
+	 * @param role The sender or recipient of messages and data in a conversation.
+	 * @param content The content of the message of type {@link Content}.
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record PromptMessage( // @formatter:off
+		@JsonProperty("role") Role role,
+		@JsonProperty("content") Content content) {
+	} // @formatter:on
+
+	/**
+	 * The server's response to a prompts/list request from the client.
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ListPromptsResult( // @formatter:off
+		@JsonProperty("prompts") List<Prompt> prompts,
+		@JsonProperty("nextCursor") String nextCursor) {
+	}// @formatter:on
+
+	/**
+	 * Used by the client to get a prompt provided by the server.
+	 *
+	 * @param name The name of the prompt or prompt template.
+	 * @param arguments Arguments to use for templating the prompt.
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record GetPromptRequest(// @formatter:off
+		@JsonProperty("name") String name,
+		@JsonProperty("arguments") Map<String, Object> arguments) implements Request {
+	}// @formatter:off
+
+	/**
+	 * The server's response to a prompts/get request from the client.
+	 *
+	 * @param description An optional description for the prompt.
+	 * @param messages A list of messages to display as part of the prompt.
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record GetPromptResult( // @formatter:off
+		@JsonProperty("description") String description,
+		@JsonProperty("messages") List<PromptMessage> messages) {
+	} // @formatter:on
+
+	// ---------------------------
+	// Tool Interfaces
+	// ---------------------------
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ListToolsResult( // @formatter:off
+		@JsonProperty("tools") List<Tool> tools,
+		@JsonProperty("nextCursor") String nextCursor) {
+	}// @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record Tool( // @formatter:off
+		@JsonProperty("name") String name,
+		@JsonProperty("description") String description,
+		@JsonProperty("inputSchema") Map<String, Object> inputSchema) {
+	} // @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record CallToolRequest(// @formatter:off
+		@JsonProperty("name") String name,
+		@JsonProperty("arguments") Map<String, Object> arguments) implements Request {
+	}// @formatter:off
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record CallToolResult( // @formatter:off
+		@JsonProperty("content") List<Content> content,
+		@JsonProperty("isError") Boolean isError) {
+	} // @formatter:on
+
+	// ---------------------------
+	// Sampling Interfaces
+	// ---------------------------
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record SamplingMessage(// @formatter:off
+		@JsonProperty("role") Role role,
+		@JsonProperty("content") Content content) {
+	} // @formatter:on
+
+	// Sampling and Message Creation
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record CreateMessageRequest(// @formatter:off
+		@JsonProperty("messages") List<SamplingMessage> messages,
+		@JsonProperty("modelPreferences") ModelPreferences modelPreferences,
+		@JsonProperty("systemPrompt") String systemPrompt,
+		@JsonProperty("includeContext") ContextInclusionStrategy includeContext,
+		@JsonProperty("temperature") Double temperature,
+		@JsonProperty("maxTokens") int maxTokens,
+		@JsonProperty("stopSequences") List<String> stopSequences, 			
+		@JsonProperty("metadata") Map<String, Object> metadata) implements Request {
+
+		public enum ContextInclusionStrategy {
+			@JsonProperty("none") NONE,
+			@JsonProperty("this_server") THIS_SERVER,
+			@JsonProperty("all_server") ALL_SERVERS
+		}
+	}// @formatter:on
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record CreateMessageResult(Role role, Content content, String model, StopReason stopReason) {
+		public enum StopReason {
+
+		// @formatter:off
+			@JsonProperty("end_turn") END_TURN,
+			@JsonProperty("stop_sequence") STOP_SEQUENCE,
+			@JsonProperty("max_tokens") MAX_TOKENS
+		} // @formatter:on
+	}
+
+	// ---------------------------
+	// Model Preferences
+	// ---------------------------
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ModelPreferences(List<ModelHint> hints, Double costPriority, Double speedPriority,
+			Double intelligencePriority) {
+	}
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ModelHint(String name) {
+	}
+
+	// ---------------------------
+	// Pagination Interfaces
+	// ---------------------------
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record PaginatedRequest(@JsonProperty("cursor") String cursor) {
+	}
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record PaginatedResult(@JsonProperty("nextCursor") String nextCursor) {
+	}
+
+	// ---------------------------
+	// Progress and Logging
+	// ---------------------------
+	public record ProgressNotification(// @formatter:off
+		@JsonProperty("progressToken") String progressToken,
+		@JsonProperty("progress") double progress,
+		@JsonProperty("total") Double total) {
+	}// @formatter:on
+
+	public record LoggingMessageNotification(// @formatter:off
+		@JsonProperty("level") LoggingLevel level,
+		@JsonProperty("logger") String logger,
+		@JsonProperty("data") Object data) {
+	}// @formatter:on
+
+	// ---------------------------
+	// Autocomplete
+	// ---------------------------
+	public record CompleteRequest(PromptOrResourceReference ref, CompleteArgument argument) implements Request {
+		public sealed interface PromptOrResourceReference permits PromptReference, ResourceReference {
+
+			String type();
+
+		}
+
+		public record PromptReference(// @formatter:off
+			@JsonProperty("type") String type,
+			@JsonProperty("name") String name) implements PromptOrResourceReference {
+		}// @formatter:on
+
+		public record ResourceReference(// @formatter:off
+			@JsonProperty("type") String type,
+			@JsonProperty("uri") String uri) implements PromptOrResourceReference {
+		}// @formatter:on
+
+		public record CompleteArgument(// @formatter:off
+			@JsonProperty("name") String name,
+			@JsonProperty("value") String value) {
+		}// @formatter:on
+	}
+
+	public record CompleteResult(CompleteCompletion completion) {
+		public record CompleteCompletion(// @formatter:off
+			@JsonProperty("values") List<String> values,
+			@JsonProperty("total") Integer total,
+			@JsonProperty("hasMore") Boolean hasMore) {
+		}// @formatter:on
+	}
+
+	// ---------------------------
+	// Content Types
+	// ---------------------------
+	@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
+	@JsonSubTypes({ @JsonSubTypes.Type(value = TextContent.class, name = "text"),
+			@JsonSubTypes.Type(value = ImageContent.class, name = "image"),
+			@JsonSubTypes.Type(value = EmbeddedResource.class, name = "resource") })
+	public sealed interface Content permits TextContent, ImageContent, EmbeddedResource {
+
+		String type();
+
+	}
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record TextContent( // @formatter:off
+		@JsonProperty("audience") List<Role> audience,
+		@JsonProperty("priority") Double priority,
+		@JsonProperty("type") String type,
+		@JsonProperty("text") String text) implements Content { // @formatter:on
+
+		public TextContent {
+			type = "text";
+		}
+
+		public String type() {
+			return type;
+		}
+	}
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ImageContent( // @formatter:off
+		@JsonProperty("audience") List<Role> audience,
+		@JsonProperty("priority") Double priority,
+		@JsonProperty("type") String type,
+		@JsonProperty("data") String data,
+		@JsonProperty("mimeType") String mimeType) implements Content { // @formatter:on
+
+		public ImageContent {
+			type = "image";
+		}
+
+		public String type() {
+			return type;
+		}
+	}
+
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record EmbeddedResource( // @formatter:off
+		@JsonProperty("audience") List<Role> audience,
+		@JsonProperty("priority") Double priority,
+		@JsonProperty("type") String type,
+		@JsonProperty("resource") ResourceContents resource) implements Content { // @formatter:on
+
+		public EmbeddedResource {
+			type = "resource";
+		}
+
+		public String type() {
+			return type;
+		}
+	}
+
+	// ---------------------------
+	// Roots
+	// ---------------------------
+	/**
+	 * Represents a root directory or file that the server can operate on.
+	 *
+	 * @param uri The URI identifying the root. This *must* start with file:// for now.
+	 * This restriction may be relaxed in future versions of the protocol to allow other
+	 * URI schemes.
+	 * @param name An optional name for the root. This can be used to provide a
+	 * human-readable identifier for the root, which may be useful for display purposes or
+	 * for referencing the root in other parts of the application.
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record Root( // @formatter:off
+		@JsonProperty("uri") String uri,
+		@JsonProperty("name") String name) {
+	} // @formatter:on
+
+	/**
+	 * The client's response to a roots/list request from the server. This result contains
+	 * an array of Root objects, each representing a root directory or file that the
+	 * server can operate on.
+	 */
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	public record ListRootsResult( // @formatter:off
+		@JsonProperty("roots") List<Root> roots) {
+	} // @formatter:on
+
+}
--- a/mcp/src/main/java/org/springframework/ai/mcp/spec/McpSession.java
+++ b/mcp/src/main/java/org/springframework/ai/mcp/spec/McpSession.java
@ -0,0 +1,93 @@
+/*
+ * 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.Map;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import reactor.core.publisher.Mono;
+
+/**
+ * Represents a Model Control Protocol (MCP) session that handles communication between
+ * clients and the server. This interface provides methods for sending requests and
+ * notifications, as well as managing the session lifecycle.
+ *
+ * <p>
+ * The session operates asynchronously using Project Reactor's {@link Mono} type for
+ * non-blocking operations. It supports both request-response patterns and one-way
+ * notifications.
+ * </p>
+ *
+ * @author Christian Tzolov
+ * @author Dariusz Jędrzejczyk
+ */
+public interface McpSession {
+
+	/**
+	 * Sends a request to the model server and expects a response of type T.
+	 *
+	 * <p>
+	 * This method handles the request-response pattern where a response is expected from
+	 * the server. The response type is determined by the provided TypeReference.
+	 * </p>
+	 * @param <T> the type of the expected response
+	 * @param method the name of the method to be called on the server
+	 * @param requestParams the parameters to be sent with the request
+	 * @param typeRef the TypeReference describing the expected response type
+	 * @return a Mono that will emit the response when received
+	 */
+	<T> Mono<T> sendRequest(String method, Object requestParams, TypeReference<T> typeRef);
+
+	/**
+	 * Sends a notification to the model server without parameters.
+	 *
+	 * <p>
+	 * This method implements the notification pattern where no response is expected from
+	 * the server. It's useful for fire-and-forget scenarios.
+	 * </p>
+	 * @param method the name of the notification method to be called on the server
+	 * @return a Mono that completes when the notification has been sent
+	 */
+	default Mono<Void> sendNotification(String method) {
+		return sendNotification(method, null);
+	}
+
+	/**
+	 * Sends a notification to the model server with parameters.
+	 *
+	 * <p>
+	 * Similar to {@link #sendNotification(String)} but allows sending additional
+	 * parameters with the notification.
+	 * </p>
+	 * @param method the name of the notification method to be called on the server
+	 * @param params a map of parameters to be sent with the notification
+	 * @return a Mono that completes when the notification has been sent
+	 */
+	Mono<Void> sendNotification(String method, Map<String, Object> params);
+
+	/**
+	 * Closes the session and releases any associated resources asynchronously.
+	 * @return a {@link Mono<Void>} that completes when the session has been closed.
+	 */
+	Mono<Void> closeGracefully();
+
+	/**
+	 * Closes the session and releases any associated resources.
+	 */
+	void close();
+
+}
--- a/mcp/src/main/java/org/springframework/ai/mcp/spec/McpTransport.java
+++ b/mcp/src/main/java/org/springframework/ai/mcp/spec/McpTransport.java
@ -0,0 +1,98 @@
+/*
+ * 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.Function;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import reactor.core.publisher.Mono;
+
+import org.springframework.ai.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 {
+
+	/**
+	 * 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>
+	 */
+	Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> handler);
+
+	/**
+	 * 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>
+	 */
+	default void close() {
+		this.closeGracefully().subscribe();
+	}
+
+	/**
+	 * Closes the transport connection and releases any associated resources
+	 * asynchronously.
+	 * @return a {@link Mono<Void>} that completes when the connection has been closed.
+	 */
+	Mono<Void> closeGracefully();
+
+	/**
+	 * 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);
+
+	<T> T unmarshalFrom(Object data, TypeReference<T> typeRef);
+
+}
--- a/mcp/src/main/java/org/springframework/ai/mcp/util/Assert.java
+++ b/mcp/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());
+	}
+
+}
--- a/mcp/src/main/java/org/springframework/ai/mcp/util/Utils.java
+++ b/mcp/src/main/java/org/springframework/ai/mcp/util/Utils.java
@ -0,0 +1,69 @@
+/*
+ * 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 java.util.Collection;
+import java.util.Map;
+
+import reactor.util.annotation.Nullable;
+
+import org.springframework.lang.Contract;
+
+/**
+ * @author Christian Tzolov
+ */
+
+public class Utils {
+
+	/**
+	 * 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());
+	}
+
+	/**
+	 * Return {@code true} if the supplied Collection is {@code null} or empty. Otherwise,
+	 * return {@code false}.
+	 * @param collection the Collection to check
+	 * @return whether the given Collection is empty
+	 */
+	@Contract("null -> true")
+	public static boolean isEmpty(@Nullable Collection<?> collection) {
+		return (collection == null || collection.isEmpty());
+	}
+
+	/**
+	 * Return {@code true} if the supplied Map is {@code null} or empty. Otherwise, return
+	 * {@code false}.
+	 * @param map the Map to check
+	 * @return whether the given Map is empty
+	 */
+	@Contract("null -> true")
+	public static boolean isEmpty(@Nullable Map<?, ?> map) {
+		return (map == null || map.isEmpty());
+	}
+
+}
--- a/mcp/src/main/resources/logback.xml
+++ b/mcp/src/main/resources/logback.xml
@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE configuration>
+
+<configuration>
+    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <!-- Main MCP package -->
+    <logger name="org.springframework.ai.mcp" level="INFO"/>
+
+    <!-- Client packages -->
+    <logger name="org.springframework.ai.mcp.client" level="INFO"/>
+    <logger name="org.springframework.ai.mcp.client.sse" level="INFO"/>
+    <logger name="org.springframework.ai.mcp.client.stdio" level="INFO"/>
+
+    <!-- Spec package -->
+    <logger name="org.springframework.ai.mcp.spec" level="INFO"/>
+
+    <!-- Test packages -->
+    <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"/>
+
+    <!-- Root logger -->
+    <root level="INFO">
+        <appender-ref ref="CONSOLE"/>
+    </root>
+</configuration>
--- a/mcp/src/test/java/org/springframework/ai/mcp/attic/ClientSessionTests2.java
+++ b/mcp/src/test/java/org/springframework/ai/mcp/attic/ClientSessionTests2.java
@ -0,0 +1,79 @@
+/*
+ * 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.attic;
+
+import java.time.Duration;
+
+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.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;
+
+/**
+ * @author Christian Tzolov
+ * @since 1.0.0
+ */
+public class ClientSessionTests2 {
+
+	public static void main(String[] args) {
+
+		var stdioParams = ServerParameters.builder("uv")
+			.args("--directory", "dir", "run", "mcp-server-sqlite", "--db-path", "~/test.db")
+			.build();
+
+		try (McpSyncClient clientSession = McpClient.using(new StdioClientTransport(stdioParams))
+			.requestTimeout(Duration.ofSeconds(10))
+			.sync()) {
+
+			clientSession.initialize();
+
+			ListToolsResult tools = clientSession.listTools(null);
+			System.out.println("Tools: " + tools);
+
+			clientSession.ping();
+
+			// CallToolRequest callToolRequest = new CallToolRequest("echo",
+			// Map.of("message", "Hello MCP Spring AI!"));
+			// CallToolResult callToolResult =
+			// clientSession.callTool(callToolRequest);
+			// System.out.println("Call Tool Result: " + callToolResult);
+			// //
+			// clientSession.sendRootsListChanged();
+
+			// Resources
+			ListResourcesResult resources = clientSession.listResources(null);
+			System.out.println("Resources Size: " + resources.resources().size());
+			System.out.println("Resources: " + resources);
+			for (Resource resource : resources.resources()) {
+				System.out.println(clientSession.readResource(resource));
+
+			}
+
+			// var resourceTemplate = clientSession.listResourceTemplates(null);
+			// System.out.println("Resource Templates: " + resourceTemplate);
+
+			// stdioClient.awaitForExit();
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+		}
+	}
+
+}
--- a/mcp/src/test/java/org/springframework/ai/mcp/attic/ClientSessionTests3.java
+++ b/mcp/src/test/java/org/springframework/ai/mcp/attic/ClientSessionTests3.java
@ -0,0 +1,74 @@
+/*
+ * 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.attic;
+
+import java.time.Duration;
+
+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.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;
+
+/**
+ * @author Christian Tzolov
+ * @since 1.0.0
+ */
+public class ClientSessionTests3 {
+
+	public static void main(String[] args) {
+
+		var stdioParams = ServerParameters.builder("uv")
+			.args("--directory", "dir", "run", "mcp-server-sqlite", "--db-path", "~/test.db")
+			.build();
+
+		McpSyncClient clientSession = null;
+		try {
+
+			clientSession = McpClient.using(new StdioClientTransport(stdioParams))
+				.requestTimeout(Duration.ofSeconds(10))
+				.sync();
+
+			clientSession.initialize();
+
+			ListToolsResult tools = clientSession.listTools(null);
+			System.out.println("Tools: " + tools);
+
+			clientSession.ping();
+
+			// Resources
+			ListResourcesResult resources = clientSession.listResources(null);
+			System.out.println("Resources Size: " + resources.resources().size());
+			System.out.println("Resources: " + resources);
+			for (Resource resource : resources.resources()) {
+				System.out.println(clientSession.readResource(resource));
+
+			}
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+		}
+		finally {
+			if (clientSession != null) {
+				clientSession.close();
+			}
+		}
+	}
+
+}
--- a/Show more
+++ b/Show more