modelcontextprotocol/java-sdk
Watch
1
0
Fork
You've already forked java-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: Add MCP server with Server stdio/sse transport improve MCP client

Browse source 
- Implement MCP server with async/sync support and builder pattern configuration
 - Add Server stdio/SSE transport layers with robust error handling
 - Convert McpClient and McpServer to interfaces for better extensibility
 - Add sampling capabilities for LLM interactions (text/image)
 - Implement dynamic root management and runtime tool management
 - Add comprehensive prompt and resource capabilities
 - Add support for dynamic root management (add/remove roots at runtime)

 Architecture Changes

 - Introduce registration records for tools, resources, and prompts
 - Improve thread safety with concurrent collections
 - Enhance resource template handling
 - Implement builder patterns for Server/Client capabilities
 - Add support for server capabilities and implementation info

Transport Layer:
 - Implement SseServerTransport with SSE endpoints
 - Add StdioServerTransport for process-based communication
 - Reorganize transport packages for better organization
 - Improve transport error handling and shutdown processes

Documentation:

 - Add detailed MCP reference documentation
 - Include architecture diagrams and API documentation
 - Add comprehensive test coverage for all components
 - Include sample projects demonstrating stdio/SSE transports

BREAKING CHANGES:

 Transport Package Reorganization

 - Move StdioClientTransport to o.s.ai.mcp.client.transport
 - Move SseClientTransport to o.s.ai.mcp.client.transport
 - Move ServerParameters to o.s.ai.mcp.client.transport
 - StdioClientTransport.errorHandler changed to private with
   setter method Migration: Use setErrorHandler() method
   instead of direct field access

 API Changes

 - Replace static factory methods with builder pattern
 - Update ServerCapabilities to use builder pattern
 - Make StdioClientTransport.errorHandler private with setter
 - Replace rootsListProvider with direct root management
 - Remove deprecated static factory methods

Resolves #27,#25,#24,#11
Related to #12
This commit is contained in:
Christian Tzolov 2024-12-19 19:38:38 +01:00
commit abd0facf4e
36 changed files with 5727 additions and 471 deletions
Download patch file Download diff file Expand all files Collapse all files

440
mcp/README.md
View file

@ -8,23 +8,19 @@ This SDK implements the Model Context Protocol, enabling seamless integration wi
## Features
- Synchronous and Asynchronous MCP Client implementations
- Synchronous and Asynchronous MCP Server implementations (WIP: comming with 0.3.0)
- Synchronous and Asynchronous MCP Client and MCP Server implementations
- 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
@ -34,17 +30,116 @@ Add the following dependency to your Maven project:
<dependency>
<groupId>org.springframework.experimental</groupId>
<artifactId>mcp</artifactId>
<version>0.2.0-SNAPSHOT</version>
<version>0.3.0-SNAPSHOT</version>
</dependency>
```
## Usage
## Architecture
### Transport Layer Options
The SDK follows a layered architecture with clear separation of concerns:
The SDK provides two transport implementations:
### Core Components Hierarchy
#### StdioClientTransport
<img src="../docs/MCP-layers.svg" width="300" align="right"/>
- **Transport Layer (McpTransport)**: Handles JSON-RPC message serialization/deserialization via StdioTransport (stdin/stdout) and SseTransport (HTTP streaming).
- **Session Layer (McpSession)**: Manages communication patterns and state using DefaultMcpSession implementation.
- **Client/Server Layer**: Both use McpSession for sync/async operations, with McpClient handling client-side protocol operations and McpServer managing server-side protocol operations.
Following class diagram illustrates the layered architecture of the MCP SDK, showing the relationships between core interfaces (McpTransport, McpSession), their implementations, and the client/server components. It highlights how the transport layer connects to sessions, which in turn support both synchronous and asynchronous client/server implementations.
<img src="../docs/spring-ai-mcp-uml-classdiagram.svg" width="600"/>
### Key Interactions
- **Client/Server Initialization**: Transport setup, protocol compatibility check, capability negotiation, and implementation details exchange.
- **Message Flow**: JSON-RPC message handling with validation, type-safe response processing, and error handling.
- **Resource Management**: Resource discovery, URI template-based access, subscription system, and content retrieval.
- **Prompt System**: Discovery, parameter-based retrieval, change notifications, and content management.
- **Tool Execution**: Discovery, parameter validation, timeout-aware execution, and result processing.
### [Reference Documentation](../docs/ref-index.md)
Detailed reference documentation covering the Spring AI implementation of the Model Context Protocol (MCP) specification.
## Quick Start
### McpClient
Factory for creating `sync` and `async` variants MCP clients. Uses McpSession for communication and handles client-side protocol operations.
#### Sync MCP Client
```java
// Create and initialize sync client
McpTransport mcpTransport = ...
try (McpSyncClient client = McpClient.using(mcpTransport).sync()) {
// Initialize connection with protocol version and capabilities
InitializeResult initResult = client.initialize();
// List available tools
ListToolsResult tools = client.listTools();
// Execute a tool
CallToolResult result = client.callTool(
new CallToolRequest("echo", Map.of("message", "Hello!"))
);
// Resource management
ListResourcesResult resources = client.listResources();
ReadResourceResult resource = client.readResource(
new ReadResourceRequest("resource-uri")
);
// List and retrieve prompts
ListPromptsResult prompts = client.listPrompts();
GetPromptResult prompt = client.getPrompt(
new GetPromptRequest("prompt-id", Map.of())
);
}
```
#### Async MCP Client
```java
McpTransport mcpTransport = ...
// Initialize async client with custom timeout
McpAsyncClient client = McpClient.using(mcpTransport)
.requestTimeout(Duration.ofSeconds(30))
.async();
// Initialize connection and chain operations
var result = client.initialize()
// Connection initialized with protocol version compatibility
.flatMap(initResult -> client.listTools())
// Process available tools
.flatMap(tools -> client.callTool(
new CallToolRequest("echo", Map.of("message", "Hello MCP!"))))
// Handle tool execution result
.flatMap(toolResult -> client.listPrompts())
// Process available prompts
.flatMap(prompts -> client.getPrompt(new GetPromptRequest("prompt-id", Map.of()));
// Subscribe to resource changes
var subscription = client.listResources()
.flatMap(resources -> client.subscribeResource(new SubscribeRequest("resource-uri")));
// Handle results reactively or block if needed
GetPromptResult promptResult = result.block();
subscription.block();
// Cleanup
client.closeGracefully().block();
```
#### Client Transports
Clients can use the following transport options:
##### StdioClientTransport
Standard I/O based transport for process-based communication with MCP servers:
```java
@ -54,7 +149,7 @@ ServerParameters params = ServerParameters.builder("npx")
McpTransport transport = new StdioClientTransport(params);
```
#### SseClientTransport
##### SseClientTransport
Server-Sent Events (SSE) based transport following the MCP HTTP with SSE transport specification:
```java
@ -63,279 +158,102 @@ WebClient.Builder webClientBuilder = WebClient.builder()
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
### MCP Server
### Sync MCP Client Example
Factory for creating `sync` and `async` variants MCP servers. Uses McpSession for communication and handles server-side protocol operations.
#### MCP Sync Server
```java
// Create and initialize sync client
McpTransport mcpTransport = ...
// Create a sync server with custom configuration
McpSyncServer syncServer = McpServer.using(transport)
.serverInfo("my-server", "1.0.0")
.serverCapabilities(new ServerCapabilities(...))
.addTool(new CalculatorTool())
.resourcesProvider(cursor -> resources)
.promptsProvider(cursor -> prompts)
.sync();
try (McpSyncClient client = McpClient.using(mcpTransport).sync()) {
// Initialize connection with protocol version and capabilities
McpSchema.InitializeResult initResult = client.initialize();
// Add a tool handler at runtime
syncServer.addTool(new CalculatorTool());
logger.info("Tool added");
// List available tools
McpSchema.ListToolsResult tools = client.listTools();
// Remove a tool handler
syncServer.removeTool("calculator");
logger.info("Tool removed");
// Execute a tool
McpSchema.CallToolResult result = client.callTool(
new McpSchema.CallToolRequest("echo", Map.of("message", "Hello!"))
);
// Notify clients of changes
syncServer.notifyToolsListChanged();
syncServer.notifyResourcesListChanged();
syncServer.notifyPromptsListChanged();
// 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())
);
}
// Graceful shutdown
syncServer.closeGracefully();
```
### Async MCP Client Example with Custom Configuration
#### MCP Async Server
```java
McpTransport mcpTransport = ...
// Initialize async client with custom timeout and object mapper
McpAsyncClient client = McpClient.using(mcpTransport)
.requestTimeout(Duration.ofSeconds(30))
// Create an async server with custom configuration
McpAsyncServer asyncServer = McpServer.using(transport)
.serverInfo("my-server", "1.0.0")
.serverCapabilities(new ServerCapabilities(...))
.addTool(new CalculatorTool())
.resourcesProvider(cursor -> resources)
.promptsProvider(cursor -> prompts)
.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();
})
// Add a tool handler at runtime
asyncServer.addTool(new CalculatorTool())
.doOnSuccess(v -> logger.info("Tool added"))
.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
})
// Remove a tool handler
asyncServer.removeTool("calculator")
.doOnSuccess(v -> logger.info("Tool removed"))
.subscribe();
// Notify clients of changes
asyncServer.notifyToolsListChanged().subscribe();
asyncServer.notifyResourcesListChanged().subscribe();
asyncServer.notifyPromptsListChanged().subscribe();
// Graceful shutdown
asyncServer.closeGracefully().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.
#### Server Transports
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
Servers can use the following transport options:
##### StdioServerTransport
Standard I/O based transport for process-based communication:
```java
// Create transport with custom ObjectMapper
ObjectMapper mapper = new ObjectMapper();
StdioServerTransport transport = new StdioServerTransport(mapper);
```
Provides bidirectional JSON-RPC message handling over standard input/output streams with non-blocking message processing, serialization/deserialization, and graceful shutdown support.
##### SseServerTransport
Server-Sent Events based transport for HTTP streaming:
```java
// Create SSE transport
ObjectMapper mapper = new ObjectMapper();
String messageEndpoint = "/mcp/message";
SseServerTransport transport = new SseServerTransport(mapper, messageEndpoint);
// Get router function for web server configuration
RouterFunction<?> router = transport.getRouterFunction();
```
Implements the MCP HTTP with SSE transport specification, providing concurrent client connections through SSE endpoints with message routing, session management, and graceful shutdown capabilities.
## 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
The SDK provides comprehensive error handling through the McpError class, covering protocol compatibility, transport communication, JSON-RPC messaging, tool execution, resource management, prompt handling, timeouts, and connection issues. This unified error handling approach ensures consistent and reliable error management across both synchronous and asynchronous operations.
## Contributing

7
mcp/pom.xml
View file

@ -68,6 +68,13 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${springframework.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>

160
mcp/src/main/java/org/springframework/ai/mcp/client/McpAsyncClient.java
View file

@ -20,8 +20,9 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.function.Function;
import com.fasterxml.jackson.core.type.TypeReference;
import org.slf4j.Logger;
@ -34,8 +35,14 @@ 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.ClientCapabilities;
import org.springframework.ai.mcp.spec.McpSchema.ClientCapabilities.RootCapabilities;
import org.springframework.ai.mcp.spec.McpSchema.ClientCapabilities.Sampling;
import org.springframework.ai.mcp.spec.McpSchema.CreateMessageRequest;
import org.springframework.ai.mcp.spec.McpSchema.CreateMessageResult;
import org.springframework.ai.mcp.spec.McpSchema.GetPromptRequest;
import org.springframework.ai.mcp.spec.McpSchema.GetPromptResult;
import org.springframework.ai.mcp.spec.McpSchema.Implementation;
import org.springframework.ai.mcp.spec.McpSchema.ListPromptsResult;
import org.springframework.ai.mcp.spec.McpSchema.PaginatedRequest;
import org.springframework.ai.mcp.spec.McpSchema.Root;
@ -62,13 +69,38 @@ public class McpAsyncClient {
*/
private final DefaultMcpSession mcpSession;
/**
* Client capabilities.
*/
private final McpSchema.ClientCapabilities clientCapabilities;
/**
* 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;
private final ConcurrentHashMap<String, Root> roots;
/**
* MCP 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.
*/
private Function<CreateMessageRequest, CreateMessageResult> samplingHandler;
/**
* Client transport implementation.
*/
private final McpTransport transport;
/**
* Client implementation information.
*/
private Implementation clientInfo;
/**
* Create a new McpAsyncClient with the given transport and session request-response
@ -80,19 +112,35 @@ public class McpAsyncClient {
* @param rootsListChangedNotification whether the client supports roots/list_changed
* notification.
*/
public McpAsyncClient(McpTransport transport, Duration requestTimeout,
List<Supplier<List<Root>>> rootsListProviders, boolean rootsListChangedNotification,
public McpAsyncClient(McpTransport transport, Duration requestTimeout, Implementation clientInfo,
ClientCapabilities clientCapabilities, Map<String, Root> roots,
List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumers,
List<Consumer<List<McpSchema.Resource>>> resourcesChangeConsumers,
List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumers) {
List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumers,
Function<CreateMessageRequest, CreateMessageResult> samplingHandler) {
this.clientInfo = clientInfo;
this.clientCapabilities = (clientCapabilities != null) ? clientCapabilities
: new McpSchema.ClientCapabilities(null, !Utils.isEmpty(roots) ? new RootCapabilities(false) : null,
samplingHandler != null ? new Sampling() : null);
this.transport = transport;
this.roots = !Utils.isEmpty(roots) ? new ConcurrentHashMap<>(roots) : new ConcurrentHashMap<>();
// 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);
if (!Utils.isEmpty(this.roots) && this.clientCapabilities.roots() != null) {
requestHanlers.put("roots/list", rootsListRequestHandler());
}
// Sampling Handler
if (samplingHandler != null && this.clientCapabilities.sampling() != null) {
this.samplingHandler = samplingHandler;
requestHanlers.put("sampling/createMessage", samplingCreateMessageHandler());
}
// Notification Handlers
@ -129,30 +177,75 @@ public class McpAsyncClient {
}
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>());
}
public Mono<Void> addRoot(Root root) {
List<Mono<List<Root>>> monos = rootsListProviders.stream()
.map(supplier -> Mono.fromSupplier(supplier).subscribeOn(Schedulers.boundedElastic()))
.toList();
if (root == null) {
return Mono.error(new McpError("Root must not be null"));
}
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;
});
if (this.clientCapabilities.roots() == null) {
return Mono.error(new McpError("Client must be configured with roots capabilities"));
}
if (this.roots.containsKey(root.uri())) {
return Mono.error(new McpError("Root with uri '" + root.uri() + "' already exists"));
}
this.roots.put(root.uri(), root);
logger.info("Added root: {}", root);
if (this.clientCapabilities.roots().listChanged()) {
return this.rootsListChangedNotification();
}
return Mono.empty();
}
public Mono<Void> removeRoot(String rootUri) {
if (rootUri == null) {
return Mono.error(new McpError("Root uri must not be null"));
}
if (this.clientCapabilities.roots() == null) {
return Mono.error(new McpError("Client must be configured with roots capabilities"));
}
Root removed = this.roots.remove(rootUri);
if (removed != null) {
logger.info("Removed Root: {}", rootUri);
if (this.clientCapabilities.roots().listChanged()) {
return this.rootsListChangedNotification();
}
return Mono.empty();
}
return Mono.error(new McpError("Root with uri '" + rootUri + "' not found"));
}
private RequestHandler samplingCreateMessageHandler() {
return params -> {
McpSchema.CreateMessageRequest request = transport.unmarshalFrom(params,
new TypeReference<McpSchema.CreateMessageRequest>() {
});
CreateMessageResult response = this.samplingHandler.apply(request);
return Mono.just(response);
};
};
}
private RequestHandler rootsListRequestHandler() {
return params -> {
McpSchema.PaginatedRequest request = transport.unmarshalFrom(params,
new TypeReference<McpSchema.PaginatedRequest>() {
});
List<Root> roots = this.roots.values().stream().toList();
return Mono.just(roots);
};
}
private NotificationHandler toolsChangeNotificationHandler(
List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumers) {
@ -242,8 +335,8 @@ public class McpAsyncClient {
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
this.clientCapabilities,
this.clientInfo); // @formatter:on
Mono<McpSchema.InitializeResult> result = this.mcpSession.sendRequest("initialize", initializeRequest,
new TypeReference<McpSchema.InitializeResult>() {
@ -268,15 +361,16 @@ public class McpAsyncClient {
/**
* Send a roots/list_changed notification.
*/
public Mono<Void> sendRootsListChanged() {
public Mono<Void> rootsListChangedNotification() {
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);
public Mono<Object> ping() {
return this.mcpSession.sendRequest("ping", null, new TypeReference<Object>() {
});
}
// --------------------------

88
mcp/src/main/java/org/springframework/ai/mcp/client/McpClient.java
View file

@ -18,11 +18,17 @@ 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 java.util.function.Function;
import org.springframework.ai.mcp.spec.McpSchema;
import org.springframework.ai.mcp.spec.McpSchema.ClientCapabilities;
import org.springframework.ai.mcp.spec.McpSchema.CreateMessageRequest;
import org.springframework.ai.mcp.spec.McpSchema.CreateMessageResult;
import org.springframework.ai.mcp.spec.McpSchema.Implementation;
import org.springframework.ai.mcp.spec.McpSchema.Root;
import org.springframework.ai.mcp.spec.McpTransport;
import org.springframework.ai.mcp.util.Assert;
@ -51,14 +57,7 @@ import org.springframework.ai.mcp.util.Assert;
* @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() {
}
public interface McpClient {
/**
* Start building an MCP client with the specified transport.
@ -78,9 +77,11 @@ public class McpClient {
private Duration requestTimeout = Duration.ofSeconds(20); // Default timeout
private boolean rootsListChangedNotification = false;
private ClientCapabilities capabilities;
private List<Supplier<List<Root>>> rootsListProviders = new ArrayList<>();
private Implementation clientInfo = new Implementation("Spring AI MCP Client", "0.3.0");
private Map<String, Root> roots = new HashMap<>();
private List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumers = new ArrayList<>();
@ -88,6 +89,8 @@ public class McpClient {
private List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumers = new ArrayList<>();
private Function<CreateMessageRequest, CreateMessageResult> samplingHandler;
private Builder(McpTransport transport) {
Assert.notNull(transport, "Transport must not be null");
this.transport = transport;
@ -104,13 +107,32 @@ public class McpClient {
return this;
}
public Builder rootsListChangedNotification(boolean rootsListChangedNotification) {
this.rootsListChangedNotification = rootsListChangedNotification;
public Builder capabilities(ClientCapabilities capabilities) {
this.capabilities = capabilities;
return this;
}
public Builder rootsListProvider(Supplier<List<Root>> rootsListProvider) {
this.rootsListProviders.add(rootsListProvider);
public Builder clientInfo(Implementation clientInfo) {
this.clientInfo = clientInfo;
return this;
}
public Builder roots(List<Root> roots) {
for (Root root : roots) {
this.roots.put(root.uri(), root);
}
return this;
}
public Builder roots(Root... roots) {
for (Root root : roots) {
this.roots.put(root.uri(), root);
}
return this;
}
public Builder sampling(Function<CreateMessageRequest, CreateMessageResult> samplingHandler) {
this.samplingHandler = samplingHandler;
return this;
}
@ -142,42 +164,10 @@ public class McpClient {
* @return A new instance of {@link McpAsyncClient}
*/
public McpAsyncClient async() {
return new McpAsyncClient(transport, requestTimeout, rootsListProviders, rootsListChangedNotification,
toolsChangeConsumers, resourcesChangeConsumers, promptsChangeConsumers);
return new McpAsyncClient(transport, requestTimeout, clientInfo, capabilities, roots, toolsChangeConsumers,
resourcesChangeConsumers, promptsChangeConsumers, samplingHandler);
}
}
/**
* @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();
}
}

24
mcp/src/main/java/org/springframework/ai/mcp/client/McpSyncClient.java
View file

@ -17,7 +17,6 @@
package org.springframework.ai.mcp.client;
import java.time.Duration;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -101,15 +100,30 @@ public class McpSyncClient implements AutoCloseable {
/**
* Send a roots/list_changed notification.
*/
public void sendRootsListChanged() {
this.delegate.sendRootsListChanged().block();
public void rootsListChangedNotification() {
this.delegate.rootsListChangedNotification().block();
}
/**
* Add a roots dynamically.
*/
public void addRoot(McpSchema.Root root) {
this.delegate.addRoot(root).block();
}
/**
* Remove a root dynamically.
*/
public void removeRoot(String rootUri) {
this.delegate.removeRoot(rootUri).block();
}
/**
* Send a synchronous ping request.
* @return
*/
public void ping() {
this.delegate.ping().block();
public Object ping() {
return this.delegate.ping().block();
}
// --------------------------

150
mcp/src/main/java/org/springframework/ai/mcp/client/transport/ServerParameters.java Normal file
View file

@ -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.transport;
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));
}
}

289
mcp/src/main/java/org/springframework/ai/mcp/client/transport/SseClientTransport.java Normal file
View file

@ -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.transport;
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);
}
}

351
mcp/src/main/java/org/springframework/ai/mcp/client/transport/StdioClientTransport.java Normal file
View file

@ -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.transport;
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
private 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 setErrorHandler(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);
}
}

444
mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java Normal file
View file

@ -0,0 +1,444 @@
package org.springframework.ai.mcp.server;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
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.server.McpServer.PromptRegistration;
import org.springframework.ai.mcp.server.McpServer.ResourceRegistration;
import org.springframework.ai.mcp.server.McpServer.ToolRegistration;
import org.springframework.ai.mcp.spec.DefaultMcpSession;
import org.springframework.ai.mcp.spec.DefaultMcpSession.NotificationHandler;
import org.springframework.ai.mcp.spec.McpError;
import org.springframework.ai.mcp.spec.McpSchema;
import org.springframework.ai.mcp.spec.McpSchema.CallToolResult;
import org.springframework.ai.mcp.spec.McpSchema.Tool;
import org.springframework.ai.mcp.spec.McpTransport;
import org.springframework.ai.mcp.util.Utils;
/**
* The Model Context Protocol (MCP) server implementation that provides asynchronous
* communication.
*
* @author Christian Tzolov
*/
public class McpAsyncServer {
private final static Logger logger = LoggerFactory.getLogger(McpAsyncServer.class);
/**
* The MCP session implementation that manages bidirectional JSON-RPC communication
* between clients and servers.
*/
private final DefaultMcpSession mcpSession;
private final McpTransport transport;
private final McpSchema.ServerCapabilities serverCapabilities;
private final McpSchema.Implementation serverInfo;
/**
* Thread-safe list of tool handlers that can be modified at runtime.
*/
private final CopyOnWriteArrayList<ToolRegistration> tools;
private final CopyOnWriteArrayList<McpSchema.ResourceTemplate> resourceTemplates;
private final ConcurrentHashMap<String, ResourceRegistration> resources;
private final ConcurrentHashMap<String, PromptRegistration> prompts;
/**
* Create a new McpAsyncServer with the given transport and capabilities.
* @param transport The transport to use for communication
* @param tools List of tool registrations
* @param resources Map of resource registrations
* @param prompts Map of prompt registrations
*/
public McpAsyncServer(McpTransport mcpTransport, McpSchema.Implementation serverInfo,
McpSchema.ServerCapabilities serverCapabilities, List<ToolRegistration> tools,
Map<String, ResourceRegistration> resources, List<McpSchema.ResourceTemplate> resourceTemplates,
Map<String, PromptRegistration> prompts) {
this.serverInfo = serverInfo;
this.tools = new CopyOnWriteArrayList<>(tools != null ? tools : List.of());
this.resources = !Utils.isEmpty(resources) ? new ConcurrentHashMap<>(resources) : new ConcurrentHashMap<>();
this.resourceTemplates = !Utils.isEmpty(resourceTemplates) ? new CopyOnWriteArrayList<>(resourceTemplates)
: new CopyOnWriteArrayList<>();
this.prompts = !Utils.isEmpty(prompts) ? new ConcurrentHashMap<>(prompts) : new ConcurrentHashMap<>();
this.serverCapabilities = (serverCapabilities != null) ? serverCapabilities : new McpSchema.ServerCapabilities(
null, // experimental
null, // logging
!Utils.isEmpty(this.prompts) ? new McpSchema.ServerCapabilities.PromptCapabilities(false) : null,
!Utils.isEmpty(this.resources) ? new McpSchema.ServerCapabilities.ResourceCapabilities(false, false)
: null,
!Utils.isEmpty(this.tools) ? new McpSchema.ServerCapabilities.ToolCapabilities(false) : null);
Map<String, DefaultMcpSession.RequestHandler> requestHandlers = new HashMap<>();
// Initialize request handlers for standard MCP methods
requestHandlers.put("initialize", initializeRequestHandler());
// Ping MUST respond with an empty data, but not NULL response.
requestHandlers.put("ping", (params) -> Mono.<Object>just(""));
// Add tools API handlers if the tool capability is enabled
if (this.serverCapabilities.tools() != null) {
requestHandlers.put("tools/list", toolsListRequestHandler());
requestHandlers.put("tools/call", toolsCallRequestHandler());
}
// Add resources API handlers if provided
if (!Utils.isEmpty(this.resources)) {
requestHandlers.put("resources/list", resourcesListRequestHandler());
requestHandlers.put("resources/read", resourcesReadRequestHandler());
}
// Add resource templates API handlers if provided.
if (!Utils.isEmpty(this.resourceTemplates)) {
requestHandlers.put("resources/templates/list", resourceTemplateListRequestHandler());
}
// Add prompts API handlers if provider exists
if (!Utils.isEmpty(this.prompts)) {
requestHandlers.put("prompts/list", promptsListRequestHandler());
requestHandlers.put("prompts/get", promptsGetRequestHandler());
}
Map<String, NotificationHandler> notificationHandlers = new HashMap<>();
notificationHandlers.put("notifications/initialized", (params) -> Mono.empty());
this.transport = mcpTransport;
this.mcpSession = new DefaultMcpSession(Duration.ofSeconds(10), mcpTransport, requestHandlers,
notificationHandlers);
}
/**
* Add a new tool registration at runtime.
* @param toolRegistration The tool registration to add
* @return Mono that completes when clients have been notified of the change
*/
public Mono<Void> addTool(ToolRegistration toolRegistration) {
if (toolRegistration == null) {
return Mono.error(new McpError("Tool registration must not be null"));
}
if (toolRegistration.tool() == null) {
return Mono.error(new McpError("Tool must not be null"));
}
if (toolRegistration.call() == null) {
return Mono.error(new McpError("Tool call handler must not be null"));
}
if (this.serverCapabilities.tools() == null) {
return Mono.error(new McpError("Server must be configured with tool capabilities"));
}
// Check for duplicate tool names
if (this.tools.stream().anyMatch(th -> th.tool().name().equals(toolRegistration.tool().name()))) {
return Mono.error(new McpError("Tool with name '" + toolRegistration.tool().name() + "' already exists"));
}
this.tools.add(toolRegistration);
logger.info("Added tool handler: {}", toolRegistration.tool().name());
if (this.serverCapabilities.tools().listChanged()) {
return notifyToolsListChanged();
}
return Mono.empty();
}
/**
* Remove a tool handler at runtime.
* @param toolName The name of the tool handler to remove
* @return Mono that completes when clients have been notified of the change
*/
public Mono<Void> removeTool(String toolName) {
if (toolName == null) {
return Mono.error(new McpError("Tool name must not be null"));
}
if (this.serverCapabilities.tools() == null) {
return Mono.error(new McpError("Server must be configured with tool capabilities"));
}
boolean removed = this.tools.removeIf(toolRegistration -> toolRegistration.tool().name().equals(toolName));
if (removed) {
logger.info("Removed tool handler: {}", toolName);
if (this.serverCapabilities.tools().listChanged()) {
return notifyToolsListChanged();
}
return Mono.empty();
}
return Mono.error(new McpError("Tool with name '" + toolName + "' not found"));
}
/**
* Add a new resource handler at runtime.
* @param resourceHandler The resource handler to add
* @return Mono that completes when clients have been notified of the change
*/
public Mono<Void> addResource(ResourceRegistration resourceHandler) {
if (resourceHandler == null || resourceHandler.resource() == null) {
return Mono.error(new McpError("Resource must not be null"));
}
if (this.serverCapabilities.resources() == null) {
return Mono.error(new McpError("Server must be configured with resource capabilities"));
}
if (this.resources.containsKey(resourceHandler.resource().uri())) {
return Mono
.error(new McpError("Resource with URI '" + resourceHandler.resource().uri() + "' already exists"));
}
this.resources.put(resourceHandler.resource().uri(), resourceHandler);
logger.info("Added resource handler: {}", resourceHandler.resource().uri());
if (this.serverCapabilities.resources().listChanged()) {
return notifyResourcesListChanged();
}
return Mono.empty();
}
/**
* Remove a resource handler at runtime.
* @param resourceUri The URI of the resource handler to remove
* @return Mono that completes when clients have been notified of the change
*/
public Mono<Void> removeResource(String resourceUri) {
if (resourceUri == null) {
return Mono.error(new McpError("Resource URI must not be null"));
}
if (this.serverCapabilities.resources() == null) {
return Mono.error(new McpError("Server must be configured with resource capabilities"));
}
ResourceRegistration removed = this.resources.remove(resourceUri);
if (removed != null) {
logger.info("Removed resource handler: {}", resourceUri);
if (this.serverCapabilities.resources().listChanged()) {
return notifyResourcesListChanged();
}
return Mono.empty();
}
return Mono.error(new McpError("Resource with URI '" + resourceUri + "' not found"));
}
/**
* Add a new prompt handler at runtime.
* @param promptRegistration The prompt handler to add
* @return Mono that completes when clients have been notified of the change
*/
public Mono<Void> addPrompt(PromptRegistration promptRegistration) {
if (promptRegistration == null) {
return Mono.error(new McpError("Prompt registration must not be null"));
}
if (this.serverCapabilities.prompts() == null) {
return Mono.error(new McpError("Server must be configured with prompt capabilities"));
}
if (this.prompts.containsKey(promptRegistration.propmpt().name())) {
return Mono
.error(new McpError("Prompt with name '" + promptRegistration.propmpt().name() + "' already exists"));
}
this.prompts.put(promptRegistration.propmpt().name(), promptRegistration);
logger.info("Added prompt handler: {}", promptRegistration.propmpt().name());
if (this.serverCapabilities.prompts().listChanged()) {
return notifyPromptsListChanged();
}
return Mono.empty();
}
/**
* Remove a prompt handler at runtime.
* @param promptName The name of the prompt handler to remove
* @return Mono that completes when clients have been notified of the change
*/
public Mono<Void> removePrompt(String promptName) {
if (promptName == null) {
return Mono.error(new McpError("Prompt name must not be null"));
}
if (this.serverCapabilities.prompts() == null) {
return Mono.error(new McpError("Server must be configured with prompt capabilities"));
}
PromptRegistration removed = this.prompts.remove(promptName);
if (removed != null) {
logger.info("Removed prompt handler: {}", promptName);
if (this.serverCapabilities.prompts().listChanged()) {
return this.notifyPromptsListChanged();
}
return Mono.empty();
}
return Mono.error(new McpError("Prompt with name '" + promptName + "' not found"));
}
// ---------------------------------------
// Request Handlers
// ---------------------------------------
private DefaultMcpSession.RequestHandler initializeRequestHandler() {
return params -> {
McpSchema.InitializeRequest request = transport.unmarshalFrom(params,
new TypeReference<McpSchema.InitializeRequest>() {
});
logger.info("Client initialize request - Protocol: {}, Capabilities: {}, Info: {}",
request.protocolVersion(), request.capabilities(), request.clientInfo());
if (!McpSchema.LATEST_PROTOCOL_VERSION.equals(request.protocolVersion())) {
return Mono
.<Object>error(
new McpError("Unsupported protocol version from client: " + request.protocolVersion()))
.publishOn(Schedulers.boundedElastic());
}
return Mono
.<Object>just(new McpSchema.InitializeResult(McpSchema.LATEST_PROTOCOL_VERSION, this.serverCapabilities,
this.serverInfo, null))
.publishOn(Schedulers.boundedElastic());
};
}
private DefaultMcpSession.RequestHandler toolsListRequestHandler() {
return params -> {
McpSchema.PaginatedRequest request = transport.unmarshalFrom(params,
new TypeReference<McpSchema.PaginatedRequest>() {
});
List<Tool> tools = this.tools.stream().map(toolRegistration -> {
return toolRegistration.tool();
}).toList();
logger.info("Client tools list request - Cursor: {}", request.cursor());
return Mono.just(new McpSchema.ListToolsResult(tools, null));
};
}
private DefaultMcpSession.RequestHandler toolsCallRequestHandler() {
return params -> {
McpSchema.CallToolRequest callToolRequest = transport.unmarshalFrom(params,
new TypeReference<McpSchema.CallToolRequest>() {
});
Optional<ToolRegistration> toolRegistration = this.tools.stream()
.filter(tr -> callToolRequest.name().equals(tr.tool().name()))
.findAny();
if (toolRegistration.isEmpty()) {
return Mono.<Object>error(new McpError("Tool not found: " + callToolRequest.name()));
}
CallToolResult callResponse = toolRegistration.get().call().apply(callToolRequest.arguments());
return Mono.just(callResponse);
};
}
private DefaultMcpSession.RequestHandler resourcesListRequestHandler() {
return params -> {
McpSchema.PaginatedRequest request = transport.unmarshalFrom(params,
new TypeReference<McpSchema.PaginatedRequest>() {
});
var resourceList = this.resources.values().stream().map(ResourceRegistration::resource).toList();
return Mono.just(new McpSchema.ListResourcesResult(resourceList, null));
};
}
private DefaultMcpSession.RequestHandler resourceTemplateListRequestHandler() {
return params -> {
McpSchema.PaginatedRequest request = transport.unmarshalFrom(params,
new TypeReference<McpSchema.PaginatedRequest>() {
});
return Mono.just(new McpSchema.ListResourceTemplatesResult(this.resourceTemplates, null));
};
}
private DefaultMcpSession.RequestHandler resourcesReadRequestHandler() {
return params -> {
McpSchema.ReadResourceRequest request = transport.unmarshalFrom(params,
new TypeReference<McpSchema.ReadResourceRequest>() {
});
var resourceUri = request.uri();
if (this.resources.containsKey(resourceUri)) {
return Mono.just(this.resources.get(resourceUri).readHandler().apply(request));
}
return Mono.error(new McpError("Resource not found: " + resourceUri));
};
}
private DefaultMcpSession.RequestHandler promptsListRequestHandler() {
return params -> {
McpSchema.PaginatedRequest request = transport.unmarshalFrom(params,
new TypeReference<McpSchema.PaginatedRequest>() {
});
var promptList = this.prompts.values().stream().map(PromptRegistration::propmpt).toList();
return Mono.just(new McpSchema.ListPromptsResult(promptList, null));
};
}
private DefaultMcpSession.RequestHandler promptsGetRequestHandler() {
return params -> {
McpSchema.GetPromptRequest request = transport.unmarshalFrom(params,
new TypeReference<McpSchema.GetPromptRequest>() {
});
// Implement prompt retrieval logic here
if (this.prompts.containsKey(request.name())) {
return Mono.just(this.prompts.get(request.name()).promptHandler().apply(request));
}
return Mono.error(new McpError("Prompt not found: " + request.name()));
};
}
/**
* Notify clients that the list of available tools has changed.
*/
public Mono<Void> notifyToolsListChanged() {
return this.mcpSession.sendNotification("notifications/tools/list_changed", null);
}
/**
* Notify clients that the list of available resources has changed.
*/
public Mono<Void> notifyResourcesListChanged() {
return this.mcpSession.sendNotification("notifications/resources/list_changed", null);
}
/**
* Notify clients that the list of available prompts has changed.
*/
public Mono<Void> notifyPromptsListChanged() {
return this.mcpSession.sendNotification("notifications/prompts/list_changed", null);
}
/**
* Close the server gracefully.
*/
public Mono<Void> closeGracefully() {
return this.mcpSession.closeGracefully();
}
/**
* Close the server immediately.
*/
public void close() {
this.mcpSession.close();
}
}

330
mcp/src/main/java/org/springframework/ai/mcp/server/McpServer.java Normal file
View file

@ -0,0 +1,330 @@
/*
* 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.server;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import org.springframework.ai.mcp.spec.McpSchema;
import org.springframework.ai.mcp.spec.McpSchema.CallToolResult;
import org.springframework.ai.mcp.spec.McpSchema.ResourceTemplate;
import org.springframework.ai.mcp.spec.McpSchema.Tool;
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)
* servers. This class serves as the main entry point for establishing MCP servers,
* offering both synchronous and asynchronous server implementations.
*
* <p>
* The class provides factory methods to create either:
* <ul>
* <li>{@link McpAsyncServer} for non-blocking operations
* <li>{@link McpSyncServer} for blocking operations
* </ul>
*
* <p>
* Use the builder pattern for flexible server configuration:
*
* <pre>{@code
* McpServer.using(transport)
* .serverInfo(new McpSchema.Implementation("my-server", "1.0.0"))
* .tool(new MyToolHandler())
* .async(); // or .sync()
* }</pre>
*
* @author Christian Tzolov
*/
public interface McpServer {
/**
* Registration of a tool with its handler function. MCP allows servers to expose
* tools that can be invoked by language models. Tools enable models to interact with
* external systems, such as querying databases, calling APIs, or performing
* computations. Each tool is uniquely identified by a name and includes metadata
* describing its schema.
*
* @param tool The tool definition
* @param call The function to handle tool execution
*/
public static record ToolRegistration(Tool tool, Function<Map<String, Object>, CallToolResult> call) {
}
public static record ResourceRegistration(McpSchema.Resource resource,
Function<McpSchema.ReadResourceRequest, McpSchema.ReadResourceResult> readHandler) {
}
public static record PromptRegistration(McpSchema.Prompt propmpt,
Function<McpSchema.GetPromptRequest, McpSchema.GetPromptResult> promptHandler) {
}
/**
* Start building an MCP server 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 servers with custom configuration.
*/
public static class Builder {
private final McpTransport transport;
private McpSchema.Implementation serverInfo;
private McpSchema.ServerCapabilities serverCapabilities;
/**
* The Model Context Protocol (MCP) allows servers to expose tools that can be
* invoked by language models. Tools enable models to interact with external
* systems, such as querying databases, calling APIs, or performing computations.
* Each tool is uniquely identified by a name and includes metadata describing its
* schema.
*/
private final List<ToolRegistration> tools = new ArrayList<>();
/**
* The Model Context Protocol (MCP) provides a standardized way for servers to
* expose resources to clients. Resources allow servers to share data that
* provides context to language models, such as files, database schemas, or
* application-specific information. Each resource is uniquely identified by a
* URI.
*/
private Map<String, ResourceRegistration> resources = new HashMap<>();
private List<ResourceTemplate> resourceTemplates = new ArrayList<>();
/**
* The Model Context Protocol (MCP) provides a standardized way for servers to
* expose prompt templates to clients. Prompts allow servers to provide structured
* messages and instructions for interacting with language models. Clients can
* discover available prompts, retrieve their contents, and provide arguments to
* customize them.
*/
private Map<String, PromptRegistration> prompts = new HashMap<>();
private Builder(McpTransport transport) {
Assert.notNull(transport, "Transport must not be null");
this.transport = transport;
}
/**
* Set the server implementation information.
* @param serverInfo The server implementation details
* @return This builder instance
*/
public Builder info(McpSchema.Implementation serverInfo) {
Assert.notNull(serverInfo, "Server info must not be null");
this.serverInfo = serverInfo;
return this;
}
/**
* Set the server implementation information using name and version.
* @param name The server name
* @param version The server version
* @return This builder instance
*/
public Builder info(String name, String version) {
Assert.hasText(name, "Name must not be null or empty");
Assert.hasText(version, "Version must not be null or empty");
this.serverInfo = new McpSchema.Implementation(name, version);
return this;
}
/**
* Set the server implementation information using name and version.
* @param name The server name
* @param version The server version
* @return This builder instance
*/
public Builder serverInfo(String name, String version) {
Assert.notNull(name, "Server name must not be null");
Assert.notNull(version, "Server version must not be null");
this.serverInfo = new McpSchema.Implementation(name, version);
return this;
}
/**
* Set the server capabilities.
* @param serverCapabilities The server capabilities configuration
* @return This builder instance
*/
public Builder capabilities(McpSchema.ServerCapabilities serverCapabilities) {
this.serverCapabilities = serverCapabilities;
return this;
}
/**
* Add a tool with its handler function.
* @param tool The tool definition
* @param handler The function to handle tool execution
* @return This builder instance
*/
public Builder tool(McpSchema.Tool tool, Function<Map<String, Object>, McpSchema.CallToolResult> handler) {
Assert.notNull(tool, "Tool must not be null");
Assert.notNull(handler, "Handler must not be null");
this.tools.add(new ToolRegistration(tool, handler));
return this;
}
/**
* Add multiple tool handlers to the server.
* @param toolRegistrations The list of tool handlers to add
* @return This builder instance
*/
public Builder tools(List<ToolRegistration> toolRegistrations) {
Assert.notNull(toolRegistrations, "Tool handlers list must not be null");
this.tools.addAll(toolRegistrations);
return this;
}
/**
* Register tools with the server.
* @param toolRegistrations The map of tools to register
* @return This builder instance
*/
public Builder tools(ToolRegistration... toolRegistrations) {
for (ToolRegistration tool : toolRegistrations) {
this.tools.add(tool);
}
return this;
}
/**
* Register resources with the server.
* @param resourceRegsitrations The map of resources to register
* @return This builder instance
*/
public Builder resources(Map<String, ResourceRegistration> resourceRegsitrations) {
Assert.notNull(resourceRegsitrations, "Resource handlers map must not be null");
this.resources.putAll(resourceRegsitrations);
return this;
}
/**
* Register resources with the server.
* @param resourceRegsitrations The list of resources to register
* @return This builder instance
*/
public Builder resources(List<ResourceRegistration> resourceRegsitrations) {
Assert.notNull(resourceRegsitrations, "Resource handlers list must not be null");
for (ResourceRegistration resource : resourceRegsitrations) {
this.resources.put(resource.resource().name(), resource);
}
return this;
}
/**
* Register resources with the server.
* @param resourceRegistrations The list of resources to register
* @return This builder instance
*/
public Builder resources(ResourceRegistration... resourceRegistrations) {
Assert.notNull(resourceRegistrations, "Resource handlers list must not be null");
for (ResourceRegistration resource : resourceRegistrations) {
this.resources.put(resource.resource().name(), resource);
}
return this;
}
/**
* Set the resource templates.
* @param resourceTemplates The list of resource templates
* @return This builder instance
*/
public Builder resourceTemplates(List<ResourceTemplate> resourceTemplates) {
this.resourceTemplates = resourceTemplates;
return this;
}
public Builder resourceTemplates(ResourceTemplate... resourceTemplates) {
for (ResourceTemplate resourceTemplate : resourceTemplates) {
this.resourceTemplates.add(resourceTemplate);
}
return this;
}
/**
* Register prompts with the server.
* @param prompts The map of prompts to register
* @return This builder instance
*/
public Builder prompts(Map<String, PromptRegistration> prompts) {
this.prompts.putAll(prompts);
return this;
}
/**
* Register prompts with the server.
* @param prompts The list of prompts to register
* @return This builder instance.
*/
public Builder prompts(List<PromptRegistration> prompts) {
for (PromptRegistration prompt : prompts) {
this.prompts.put(prompt.propmpt().name(), prompt);
}
return this;
}
/**
* Register prompts with the server.
* @param prompts The list of prompts to register
* @return This builder instance
*/
public Builder prompts(PromptRegistration... prompts) {
for (PromptRegistration prompt : prompts) {
this.prompts.put(prompt.propmpt().name(), prompt);
}
return this;
}
/**
* Build a synchronous MCP server.
* @return A new instance of {@link McpSyncServer}
*/
public McpSyncServer sync() {
return new McpSyncServer(async());
}
/**
* Build an asynchronous MCP server.
* @return A new instance of {@link McpAsyncServer}
*/
public McpAsyncServer async() {
if (serverInfo == null) {
serverInfo = new McpSchema.Implementation("mcp-server", "1.0.0");
}
return new McpAsyncServer(transport, serverInfo, serverCapabilities, tools, resources, resourceTemplates,
prompts);
}
}
}

138
mcp/src/main/java/org/springframework/ai/mcp/server/McpSyncServer.java Normal file
View file

@ -0,0 +1,138 @@
/*
* 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.server;
import org.springframework.ai.mcp.server.McpServer.PromptRegistration;
import org.springframework.ai.mcp.server.McpServer.ResourceRegistration;
import org.springframework.ai.mcp.server.McpServer.ToolRegistration;
import org.springframework.ai.mcp.util.Assert;
/**
* Synchronous wrapper around {@link McpAsyncServer} that provides blocking operations.
* This class delegates all operations to an underlying async server instance while
* providing a synchronous API.
*
* @author Christian Tzolov
*/
public class McpSyncServer {
/**
* The async server to wrap.
*/
private final McpAsyncServer asyncServer;
/**
* Creates a new synchronous server that wraps the provided async server.
* @param asyncServer The async server to wrap
*/
public McpSyncServer(McpAsyncServer asyncServer) {
Assert.notNull(asyncServer, "Async server must not be null");
this.asyncServer = asyncServer;
}
/**
* Add a new tool handler.
* @param toolHandler The tool handler to add
*/
public void addTool(ToolRegistration toolHandler) {
this.asyncServer.addTool(toolHandler).block();
}
/**
* Remove a tool handler.
* @param toolName The name of the tool handler to remove
*/
public void removeTool(String toolName) {
this.asyncServer.removeTool(toolName).block();
}
/**
* Notify clients that the list of available tools has changed.
*/
public void notifyToolsListChanged() {
this.asyncServer.notifyToolsListChanged().block();
}
/**
* Notify clients that the list of available resources has changed.
*/
public void notifyResourcesListChanged() {
this.asyncServer.notifyResourcesListChanged().block();
}
/**
* Notify clients that the list of available prompts has changed.
*/
public void notifyPromptsListChanged() {
this.asyncServer.notifyPromptsListChanged().block();
}
/**
* Add a new resource handler.
* @param resourceHandler The resource handler to add
*/
public void addResource(ResourceRegistration resourceHandler) {
this.asyncServer.addResource(resourceHandler).block();
}
/**
* Remove a resource handler.
* @param resourceUri The URI of the resource handler to remove
*/
public void removeResource(String resourceUri) {
this.asyncServer.removeResource(resourceUri).block();
}
/**
* Add a new prompt handler.
* @param promptRegistration The prompt registration to add
*/
public void addPrompt(PromptRegistration promptRegistration) {
this.asyncServer.addPrompt(promptRegistration).block();
}
/**
* Remove a prompt handler.
* @param promptName The name of the prompt handler to remove
*/
public void removePrompt(String promptName) {
this.asyncServer.removePrompt(promptName).block();
}
/**
* Close the server gracefully.
*/
public void closeGracefully() {
this.asyncServer.closeGracefully().block();
}
/**
* Close the server immediately.
*/
public void close() {
this.asyncServer.close();
}
/**
* Get the underlying async server instance.
* @return The wrapped async server
*/
public McpAsyncServer getAsyncServer() {
return this.asyncServer;
}
}

271
mcp/src/main/java/org/springframework/ai/mcp/server/transport/SseServerTransport.java Normal file
View file

@ -0,0 +1,271 @@
package org.springframework.ai.mcp.server.transport;
import java.io.IOException;
import java.time.Duration;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.stream.Collectors;
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 org.springframework.ai.mcp.spec.McpError;
import org.springframework.ai.mcp.spec.McpSchema;
import org.springframework.ai.mcp.spec.McpTransport;
import org.springframework.ai.mcp.util.Assert;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.RouterFunctions;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
/**
* Server-side implementation of the MCP HTTP with SSE transport specification. Provides
* endpoints for SSE connections and message handling.
*
* @author Christian Tzolov
*/
public class SseServerTransport implements McpTransport {
private final static Logger logger = LoggerFactory.getLogger(SseServerTransport.class);
/**
* Event type for JSON-RPC messages sent through the SSE connection.
*/
public final static String MESSAGE_EVENT_TYPE = "message";
/**
* Event type for sending the message endpoint URI to clients.
*/
public final static String ENDPOINT_EVENT_TYPE = "endpoint";
/**
* Default SSE endpoint path as specified by the MCP transport specification.
*/
public final static String SSE_ENDPOINT = "/sse";
private final ObjectMapper objectMapper;
private final String messageEndpoint;
private final RouterFunction<?> routerFunction;
/**
* Map of active client sessions, keyed by session ID.
*/
private final ConcurrentHashMap<String, ClientSession> sessions = new ConcurrentHashMap<>();
/**
* Flag indicating if the transport is shutting down.
*/
private volatile boolean isClosing = false;
private Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> connectHandler;
/**
* Constructs a new SseServerTransport.
* @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
* @param messageEndpoint The endpoint URI where clients should send messages
*/
public SseServerTransport(ObjectMapper objectMapper, String messageEndpoint) {
Assert.notNull(objectMapper, "ObjectMapper must not be null");
Assert.notNull(messageEndpoint, "Message endpoint must not be null");
this.objectMapper = objectMapper;
this.messageEndpoint = messageEndpoint;
this.routerFunction = RouterFunctions.route()
.GET(SSE_ENDPOINT, this::handleSseConnection)
.POST(messageEndpoint, this::handleMessage)
.build();
}
@Override
public Mono<Void> connect(Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
this.connectHandler = handler;
// Server-side transport doesn't initiate connections
return Mono.empty().then();
}
@Override
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
if (sessions.isEmpty()) {
logger.debug("No active sessions to broadcast message to");
return Mono.empty();
}
return Mono.<Void>create(sink -> {
try {// @formatter:off
String jsonText = objectMapper.writeValueAsString(message);
ServerSentEvent<Object> event = ServerSentEvent.builder()
.event(MESSAGE_EVENT_TYPE)
.data(jsonText)
.build();
logger.debug("Attempting to broadcast message to {} active sessions", sessions.size());
List<String> failedSessions = sessions.values().stream()
.filter(session -> session.messageSink.tryEmitNext(event).isFailure())
.map(session -> session.id)
.toList();
if (failedSessions.isEmpty()) {
logger.debug("Successfully broadcast message to all sessions");
sink.success();
}
else {
String error = "Failed to broadcast message to sessions: " + String.join(", ", failedSessions);
logger.error(error);
sink.error(new RuntimeException(error));
} // @formatter:on
}
catch (IOException e) {
logger.error("Failed to serialize message: {}", e.getMessage());
sink.error(e);
}
});
}
@Override
public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
return this.objectMapper.convertValue(data, typeRef);
}
@Override
public Mono<Void> closeGracefully() {
return Mono.fromRunnable(() -> {
isClosing = true;
logger.debug("Initiating graceful shutdown with {} active sessions", sessions.size());
}).then(Mono.when(sessions.values().stream().map(session -> {
String sessionId = session.id;
return Mono.fromRunnable(() -> session.close())
.then(Mono.delay(Duration.ofMillis(100)))
.then(Mono.fromRunnable(() -> sessions.remove(sessionId)));
}).toList()))
.timeout(Duration.ofSeconds(5))
.doOnSuccess(v -> logger.info("Graceful shutdown completed"))
.doOnError(e -> logger.error("Error during graceful shutdown: {}", e.getMessage()));
}
/**
* Get the router function for configuring the web server.
*/
public RouterFunction<?> getRouterFunction() {
return this.routerFunction;
}
/**
* Handles new SSE connection requests from clients.
*/
private Mono<ServerResponse> handleSseConnection(ServerRequest request) {
if (isClosing) {
return ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE).bodyValue("Server is shutting down");
}
String sessionId = UUID.randomUUID().toString();
logger.debug("Creating new SSE connection for session: {}", sessionId);
ClientSession session = new ClientSession(sessionId);
this.sessions.put(sessionId, session);
return ServerResponse.ok()
.contentType(MediaType.TEXT_EVENT_STREAM)
.body(Flux.<ServerSentEvent<?>>create(sink -> {
// Send initial endpoint event
logger.debug("Sending initial endpoint event to session: {}", sessionId);
sink.next(ServerSentEvent.builder().event(ENDPOINT_EVENT_TYPE).data(messageEndpoint).build());
// Subscribe to session messages
session.messageSink.asFlux()
.doOnSubscribe(s -> logger.debug("Session {} subscribed to message sink", sessionId))
.doOnComplete(() -> {
logger.debug("Session {} completed", sessionId);
sessions.remove(sessionId);
})
.doOnError(error -> {
logger.error("Error in session {}: {}", sessionId, error.getMessage());
sessions.remove(sessionId);
})
.doOnCancel(() -> {
logger.debug("Session {} cancelled", sessionId);
sessions.remove(sessionId);
})
.subscribe(event -> {
logger.debug("Forwarding event to session {}: {}", sessionId, event);
sink.next(event);
}, sink::error, sink::complete);
sink.onCancel(() -> {
logger.debug("Session {} cancelled", sessionId);
sessions.remove(sessionId);
});
}), ServerSentEvent.class);
}
/**
* Handles incoming messages from clients.
*/
private Mono<ServerResponse> handleMessage(ServerRequest request) {
if (isClosing) {
return ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE).bodyValue("Server is shutting down");
}
return request.bodyToMono(String.class).flatMap(body -> {
try {
McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(objectMapper, body);
return Mono.just(message)
.transform(this.connectHandler)
.flatMap(response -> ServerResponse.ok().build())
.onErrorResume(error -> {
logger.error("Error processing message: {}", error.getMessage());
return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR)
.bodyValue(new McpError(error.getMessage()));
});
}
catch (IllegalArgumentException e) {
logger.error("Failed to deserialize message: {}", e.getMessage());
return ServerResponse.badRequest().bodyValue(new McpError("Invalid message format"));
}
catch (IOException e) {
logger.error("Failed to deserialize message: {}", e.getMessage());
return ServerResponse.badRequest().bodyValue(new McpError("Invalid message format"));
}
});
}
/**
* Represents an active client session.
*/
private static class ClientSession {
private final String id;
private final Sinks.Many<ServerSentEvent<?>> messageSink;
ClientSession(String id) {
this.id = id;
logger.debug("Creating new session: {}", id);
this.messageSink = Sinks.many().replay().latest();
logger.debug("Session {} initialized with replay sink", id);
}
void close() {
logger.debug("Closing session: {}", id);
Sinks.EmitResult result = messageSink.tryEmitComplete();
if (result.isFailure()) {
logger.warn("Failed to complete message sink for session {}: {}", id, result);
}
else {
logger.debug("Successfully completed message sink for session {}", id);
}
}
}
}

273
mcp/src/main/java/org/springframework/ai/mcp/server/transport/StdioServerTransport.java Normal file
View file

@ -0,0 +1,273 @@
/*
* 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.server.transport;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
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.spec.McpTransport;
import org.springframework.ai.mcp.util.Assert;
/**
* Implementation of the MCP Stdio transport for servers that communicates 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
*/
public class StdioServerTransport implements McpTransport {
private static final Logger logger = LoggerFactory.getLogger(StdioServerTransport.class);
private final Sinks.Many<JSONRPCMessage> inboundSink;
private final Sinks.Many<JSONRPCMessage> outboundSink;
private ObjectMapper objectMapper;
/** Scheduler for handling inbound messages */
private Scheduler inboundScheduler;
/** Scheduler for handling outbound messages */
private Scheduler outboundScheduler;
private volatile boolean isClosing = false;
private final InputStream inputStream;
private final OutputStream outputStream;
private final Sinks.One<Void> inboundReady = Sinks.one();
private final Sinks.One<Void> outboundReady = Sinks.one();
/**
* Creates a new StdioServerTransport with a default ObjectMapper and System streams.
*/
public StdioServerTransport() {
this(new ObjectMapper(), System.in, System.out);
}
/**
* Creates a new StdioServerTransport with the specified ObjectMapper and System
* streams.
* @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
*/
public StdioServerTransport(ObjectMapper objectMapper) {
this(objectMapper, System.in, System.out);
}
/**
* Creates a new StdioServerTransport with the specified ObjectMapper and streams.
* @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
* @param inputStream The input stream to read from
* @param outputStream The output stream to write to
*/
public StdioServerTransport(ObjectMapper objectMapper, InputStream inputStream, OutputStream outputStream) {
Assert.notNull(objectMapper, "The ObjectMapper can not be null");
Assert.notNull(inputStream, "The InputStream can not be null");
Assert.notNull(outputStream, "The OutputStream can not be null");
this.inboundSink = Sinks.many().unicast().onBackpressureBuffer();
this.outboundSink = Sinks.many().unicast().onBackpressureBuffer();
this.objectMapper = objectMapper;
this.inputStream = inputStream;
this.outputStream = outputStream;
// Use bounded schedulers for better resource management
this.inboundScheduler = Schedulers.newBoundedElastic(1, 1, "inbound");
this.outboundScheduler = Schedulers.newBoundedElastic(1, 1, "outbound");
}
@Override
public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> handler) {
return Mono.<Void>fromRunnable(() -> {
handleIncomingMessages(handler);
// Start threads
startInboundProcessing();
startOutboundProcessing();
}).subscribeOn(Schedulers.boundedElastic());
}
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();
}
@Override
public Mono<Void> sendMessage(JSONRPCMessage message) {
return Mono.zip(inboundReady.asMono(), outboundReady.asMono()).then(Mono.defer(() -> {
if (this.outboundSink.tryEmitNext(message).isSuccess()) {
return Mono.empty();
}
else {
return Mono.error(new RuntimeException("Failed to enqueue message"));
}
}));
}
/**
* Starts the inbound processing thread that reads JSON-RPC messages from stdin.
* Messages are deserialized and emitted to the inbound sink.
*/
private void startInboundProcessing() {
this.inboundScheduler.schedule(() -> {
inboundReady.tryEmitValue(null);
try (BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
String line;
while (!isClosing && (line = reader.readLine()) != null) {
try {
JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(this.objectMapper, line);
if (!this.inboundSink.tryEmitNext(message).isSuccess()) {
if (!isClosing) {
logger.error("Failed to enqueue message");
}
break;
}
}
catch (Exception e) {
if (!isClosing) {
logger.error("Error processing inbound message", e);
}
break;
}
}
}
catch (IOException e) {
if (!isClosing) {
logger.error("Error reading from stdin", e);
}
}
finally {
if (!isClosing) {
isClosing = true;
}
inboundSink.tryEmitComplete();
}
});
}
/**
* Starts the outbound processing thread that writes JSON-RPC messages to stdout.
* Messages are serialized to JSON and written with a newline delimiter.
*/
private void startOutboundProcessing() {
Function<Flux<JSONRPCMessage>, Flux<JSONRPCMessage>> outboundConsumer = messages -> messages // @formatter:off
.doOnSubscribe(subscription -> outboundReady.tryEmitValue(null))
.publishOn(outboundScheduler)
.handle((message, sink) -> {
if (message != null && !isClosing) {
try {
String jsonMessage = objectMapper.writeValueAsString(message);
// Escape any embedded newlines in the JSON message as per spec
jsonMessage = jsonMessage.replace("\r\n", "\\n").replace("\n", "\\n").replace("\r", "\\n");
synchronized (outputStream) {
outputStream.write(jsonMessage.getBytes(StandardCharsets.UTF_8));
outputStream.write("\n".getBytes(StandardCharsets.UTF_8));
outputStream.flush();
}
sink.next(message);
}
catch (IOException e) {
if (!isClosing) {
logger.error("Error writing message", e);
sink.error(new RuntimeException(e));
}
}
}
else if (isClosing) {
sink.complete();
}
})
.doOnComplete(() -> {
if (!isClosing) {
isClosing = true;
}
outboundSink.tryEmitComplete();
})
.doOnError(e -> {
if (!isClosing) {
logger.error("Error in outbound processing", e);
isClosing = true;
outboundSink.tryEmitComplete();
}
})
.map(msg -> (JSONRPCMessage) msg);
outboundConsumer.apply(outboundSink.asFlux()).subscribe();
} // @formatter:on
@Override
public Mono<Void> closeGracefully() {
return Mono.fromRunnable(() -> {
isClosing = true;
logger.debug("Initiating graceful shutdown");
// }).then(Mono.delay(Duration.ofMillis(100))).then(Mono.fromRunnable(() -> {
}).then(Mono.fromRunnable(() -> {
try {
// inboundSink.tryEmitComplete();
// outboundSink.tryEmitComplete();
inboundScheduler.dispose();
outboundScheduler.dispose();
// Wait for schedulers to terminate
if (!inboundScheduler.isDisposed()) {
inboundScheduler.disposeGracefully().block(Duration.ofSeconds(5));
}
if (!outboundScheduler.isDisposed()) {
outboundScheduler.disposeGracefully().block(Duration.ofSeconds(5));
}
logger.info("Graceful shutdown completed");
}
catch (Exception e) {
logger.error("Error during graceful shutdown", e);
}
})).then().subscribeOn(Schedulers.boundedElastic());
}
@Override
public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
return this.objectMapper.convertValue(data, typeRef);
}
}

3
mcp/src/main/java/org/springframework/ai/mcp/spec/DefaultMcpSession.java
View file

@ -133,6 +133,7 @@ public class DefaultMcpSession implements McpSession {
// 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) {
logger.info("Received Response: {}", response);
var sink = pendingResponses.remove(response.id());
if (sink == null) {
logger.warn("Unexpected response for unkown id {}", response.id());
@ -142,6 +143,7 @@ public class DefaultMcpSession implements McpSession {
}
}
else if (message instanceof McpSchema.JSONRPCRequest request) {
logger.info("Received request: {}", request);
handleIncomingRequest(request).subscribe(response -> transport.sendMessage(response).subscribe(),
error -> {
var errorResponse = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
@ -151,6 +153,7 @@ public class DefaultMcpSession implements McpSession {
});
}
else if (message instanceof McpSchema.JSONRPCNotification notification) {
logger.info("Received notification: {}", notification);
handleIncomingNotification(notification).subscribe(null,
error -> logger.error("Error handling notification: {}", error.getMessage()));
}

120
mcp/src/main/java/org/springframework/ai/mcp/spec/McpSchema.java
View file

@ -191,7 +191,7 @@ public class McpSchema {
/**
* Provides a standardized way for servers to request LLM
* sampling (“completions” or “generations”) from language
* 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
@ -202,6 +202,35 @@ public class McpSchema {
@JsonInclude(JsonInclude.Include.NON_ABSENT)
public record Sampling() {
}
public static Builder builder() {
return new Builder();
}
public static class Builder {
private Map<String, Object> experimental;
private RootCapabilities roots;
private Sampling sampling;
public Builder experimental(Map<String, Object> experimental) {
this.experimental = experimental;
return this;
}
public Builder roots(Boolean listChanged) {
this.roots = new RootCapabilities(listChanged);
return this;
}
public Builder sampling() {
this.sampling = new Sampling();
return this;
}
public ClientCapabilities build() {
return new ClientCapabilities(experimental, roots, sampling);
}
}
}// @formatter:on
@JsonInclude(JsonInclude.Include.NON_ABSENT)
@ -227,6 +256,48 @@ public class McpSchema {
public record ToolCapabilities(
@JsonProperty("listChanged") Boolean listChanged) {
}
public static Builder builder() {
return new Builder();
}
public static class Builder {
private Map<String, Object> experimental;
private Object logging;
private PromptCapabilities prompts;
private ResourceCapabilities resources;
private ToolCapabilities tools;
public Builder experimental(Map<String, Object> experimental) {
this.experimental = experimental;
return this;
}
public Builder logging(Object logging) {
this.logging = logging;
return this;
}
public Builder prompts(Boolean listChanged) {
this.prompts = new PromptCapabilities(listChanged);
return this;
}
public Builder resources(Boolean subscribe, Boolean listChanged) {
this.resources = new ResourceCapabilities(subscribe, listChanged);
return this;
}
public Builder tools(Boolean listChanged) {
this.tools = new ToolCapabilities(listChanged);
return this;
}
public ServerCapabilities build() {
return new ServerCapabilities(experimental, logging, prompts, resources, tools);
}
}
} // @formatter:on
@JsonInclude(JsonInclude.Include.NON_ABSENT)
@ -304,6 +375,20 @@ public class McpSchema {
@JsonProperty("annotations") Annotations annotations) implements Annotated {
} // @formatter:on
/**
* Resource templates allow servers to expose parameterized resources using URI
* templates.
*
* @param uriTemplate A URI template that can be used to generate URIs for this
* 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.
* @see <a href="https://datatracker.ietf.org/doc/html/rfc6570">RFC 6570</a>
*/
@JsonInclude(JsonInclude.Include.NON_ABSENT)
public record ResourceTemplate( // @formatter:off
@JsonProperty("uriTemplate") String uriTemplate,
@ -500,8 +585,22 @@ public class McpSchema {
@JsonProperty("name") String name,
@JsonProperty("description") String description,
@JsonProperty("inputSchema") Map<String, Object> inputSchema) {
public Tool(String name, String description, String schema) {
this(name, description, parseSchema(schema));
}
} // @formatter:on
public static Map<String, Object> parseSchema(String schema) {
try {
return new ObjectMapper().readValue(schema, MAP_TYPE_REF);
}
catch (IOException e) {
throw new IllegalArgumentException("Invalid schema: " + schema, e);
}
}
@JsonInclude(JsonInclude.Include.NON_ABSENT)
public record CallToolRequest(// @formatter:off
@JsonProperty("name") String name,
@ -543,15 +642,18 @@ public class McpSchema {
}// @formatter:on
@JsonInclude(JsonInclude.Include.NON_ABSENT)
public record CreateMessageResult(Role role, Content content, String model, StopReason stopReason) {
public record CreateMessageResult(// @formatter:off
@JsonProperty("role") Role role,
@JsonProperty("content") Content content,
@JsonProperty("model") String model,
@JsonProperty("stopReason") StopReason stopReason) {
public enum StopReason {
// @formatter:off
@JsonProperty("end_turn") END_TURN,
@JsonProperty("stop_sequence") STOP_SEQUENCE,
@JsonProperty("max_tokens") MAX_TOKENS
} // @formatter:on
}
}
}// @formatter:on
// ---------------------------
// Model Preferences
@ -652,6 +754,10 @@ public class McpSchema {
public String type() {
return type;
}
public TextContent(String content) {
this(null, null, "text", content);
}
}
@JsonInclude(JsonInclude.Include.NON_ABSENT)
@ -716,4 +822,4 @@ public class McpSchema {
@JsonProperty("roots") List<Root> roots) {
} // @formatter:on
}
}

4
mcp/src/test/java/org/springframework/ai/mcp/attic/ClientSessionTests2.java
View file

@ -20,8 +20,8 @@ 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.client.transport.ServerParameters;
import org.springframework.ai.mcp.client.transport.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;

4
mcp/src/test/java/org/springframework/ai/mcp/attic/ClientSessionTests3.java
View file

@ -20,8 +20,8 @@ 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.client.transport.ServerParameters;
import org.springframework.ai.mcp.client.transport.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;

61
mcp/src/test/java/org/springframework/ai/mcp/attic/DemoServer.java Normal file
View file

@ -0,0 +1,61 @@
/*
* 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.util.List;
import java.util.Map;
import com.fasterxml.jackson.databind.ObjectMapper;
import reactor.netty.http.server.HttpServer;
import org.springframework.ai.mcp.server.McpServer;
import org.springframework.ai.mcp.server.transport.SseServerTransport;
import org.springframework.ai.mcp.spec.McpSchema;
import org.springframework.ai.mcp.spec.McpSchema.CallToolResult;
import org.springframework.http.server.reactive.HttpHandler;
import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
import org.springframework.web.reactive.function.server.RouterFunctions;
/**
*
* https://docs.spring.io/spring-framework/reference/web/webflux-functional.html#webflux-fn-running
* https://docs.spring.io/spring-framework/reference/web/webflux/reactive-spring.html#webflux-httphandler
*
* @author Christian Tzolov
*/
public class DemoServer {
public static void main(String[] args) {
SseServerTransport transport = new SseServerTransport(new ObjectMapper(), "/mcp/message");
var mcpServer = McpServer.using(transport)
.serverInfo("Weather Forecast", "1.0.0")
.tool(new McpSchema.Tool("weather", "Weather forecast tool by location", Map.of("city", "String")),
(arguments) -> {
String city = (String) arguments.get("city");
return new CallToolResult(List.of(), false);
})
.async();
HttpHandler httpHandler = RouterFunctions.toHttpHandler(transport.getRouterFunction());
ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(httpHandler);
HttpServer httpServer = HttpServer.create().port(8080).handle(adapter);
httpServer.bindNow().onDispose().block();
}
}

2
mcp/src/test/java/org/springframework/ai/mcp/attic/MainSSE.java
View file

@ -20,7 +20,7 @@ import java.util.Map;
import org.springframework.ai.mcp.client.McpClient;
import org.springframework.ai.mcp.client.McpSyncClient;
import org.springframework.ai.mcp.client.sse.SseClientTransport;
import org.springframework.ai.mcp.client.transport.SseClientTransport;
import org.springframework.ai.mcp.spec.McpSchema.CallToolRequest;
import org.springframework.ai.mcp.spec.McpSchema.CallToolResult;
import org.springframework.ai.mcp.spec.McpSchema.ListPromptsResult;

146
mcp/src/test/java/org/springframework/ai/mcp/client/AbstractMcpAsyncClientTests.java
View file

@ -17,9 +17,8 @@
package org.springframework.ai.mcp.client;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@ -27,11 +26,14 @@ import org.junit.jupiter.api.Test;
import reactor.test.StepVerifier;
import org.springframework.ai.mcp.spec.McpSchema.CallToolRequest;
import org.springframework.ai.mcp.spec.McpSchema.ClientCapabilities;
import org.springframework.ai.mcp.spec.McpSchema.GetPromptRequest;
import org.springframework.ai.mcp.spec.McpSchema.Prompt;
import org.springframework.ai.mcp.spec.McpSchema.Resource;
import org.springframework.ai.mcp.spec.McpSchema.Root;
import org.springframework.ai.mcp.spec.McpSchema.SubscribeRequest;
import org.springframework.ai.mcp.spec.McpSchema.Tool;
import org.springframework.ai.mcp.spec.McpSchema.UnsubscribeRequest;
import org.springframework.ai.mcp.spec.McpTransport;
import static org.assertj.core.api.Assertions.assertThat;
@ -57,9 +59,11 @@ public abstract class AbstractMcpAsyncClientTests {
abstract protected McpTransport createMcpTransport();
abstract protected void onStart();
protected void onStart() {
}
abstract protected void onClose();
protected void onClose() {
}
@BeforeEach
void setUp() {
@ -67,7 +71,10 @@ public abstract class AbstractMcpAsyncClientTests {
this.mcpTransport = createMcpTransport();
assertThatCode(() -> {
mcpAsyncClient = McpClient.using(mcpTransport).requestTimeout(TIMEOUT).async();
mcpAsyncClient = McpClient.using(mcpTransport)
.requestTimeout(TIMEOUT)
.capabilities(ClientCapabilities.builder().roots(true).build())
.async();
mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
}).doesNotThrowAnyException();
}
@ -175,19 +182,142 @@ public abstract class AbstractMcpAsyncClientTests {
@Test
void testRootsListChanged() {
assertThatCode(() -> mcpAsyncClient.sendRootsListChanged().block()).doesNotThrowAnyException();
assertThatCode(() -> mcpAsyncClient.rootsListChangedNotification().block()).doesNotThrowAnyException();
}
@Test
void testInitializeWithRootsListProviders() {
var transport = createMcpTransport();
List<Supplier<List<Root>>> providers = List.of(() -> List.of(new Root("file:///test/path", "test-root")));
var client = McpClient.using(transport).requestTimeout(TIMEOUT).rootsListProvider(providers.get(0)).async();
var client = McpClient.using(transport)
.requestTimeout(TIMEOUT)
.roots(new Root("file:///test/path", "test-root"))
.async();
assertThatCode(() -> client.initialize().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
assertThatCode(() -> client.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
@Test
void testAddRoot() {
Root newRoot = new Root("file:///new/test/path", "new-test-root");
assertThatCode(() -> mcpAsyncClient.addRoot(newRoot).block()).doesNotThrowAnyException();
}
@Test
void testAddRootWithNullValue() {
assertThatThrownBy(() -> mcpAsyncClient.addRoot(null).block()).hasMessageContaining("Root must not be null");
}
@Test
void testRemoveRoot() {
Root root = new Root("file:///test/path/to/remove", "root-to-remove");
assertThatCode(() -> {
mcpAsyncClient.addRoot(root).block();
mcpAsyncClient.removeRoot(root.uri()).block();
}).doesNotThrowAnyException();
}
@Test
void testRemoveNonExistentRoot() {
assertThatThrownBy(() -> mcpAsyncClient.removeRoot("nonexistent-uri").block())
.hasMessageContaining("Root with uri 'nonexistent-uri' not found");
}
// @Test
void testReadResource() {
StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> {
if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0);
StepVerifier.create(mcpAsyncClient.readResource(firstResource)).consumeNextWith(result -> {
assertThat(result).isNotNull();
assertThat(result.contents()).isNotNull();
}).verifyComplete();
}
}).verifyComplete();
}
@Test
void testListResourceTemplates() {
StepVerifier.create(mcpAsyncClient.listResourceTemplates()).consumeNextWith(result -> {
assertThat(result).isNotNull();
assertThat(result.resourceTemplates()).isNotNull();
}).verifyComplete();
}
// @Test
void testResourceSubscription() {
StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> {
if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0);
// Test subscribe
StepVerifier.create(mcpAsyncClient.subscribeResource(new SubscribeRequest(firstResource.uri())))
.verifyComplete();
// Test unsubscribe
StepVerifier.create(mcpAsyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())))
.verifyComplete();
}
}).verifyComplete();
}
@Test
void testNotificationHandlers() {
AtomicBoolean toolsNotificationReceived = new AtomicBoolean(false);
AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false);
AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false);
var transport = createMcpTransport();
var client = McpClient.using(transport)
.requestTimeout(TIMEOUT)
.toolsChangeConsumer(tools -> toolsNotificationReceived.set(true))
.resourcesChangeConsumer(resources -> resourcesNotificationReceived.set(true))
.promptsChangeConsumer(prompts -> promptsNotificationReceived.set(true))
.async();
assertThatCode(() -> {
client.initialize().block();
// Trigger notifications
client.sendResourcesListChanged().block();
client.promptListChangedNotification().block();
client.closeGracefully().block();
}).doesNotThrowAnyException();
}
@Test
void testInitializeWithSamplingCapability() {
var transport = createMcpTransport();
var capabilities = ClientCapabilities.builder().sampling().build();
var client = McpClient.using(transport).requestTimeout(TIMEOUT).capabilities(capabilities).async();
assertThatCode(() -> {
client.initialize().block(Duration.ofSeconds(10));
client.closeGracefully().block(Duration.ofSeconds(10));
}).doesNotThrowAnyException();
}
@Test
void testInitializeWithAllCapabilities() {
var transport = createMcpTransport();
var capabilities = ClientCapabilities.builder()
.experimental(Map.of("feature", "test"))
.roots(true)
.sampling()
.build();
var client = McpClient.using(transport).requestTimeout(TIMEOUT).capabilities(capabilities).async();
assertThatCode(() -> {
var result = client.initialize().block(Duration.ofSeconds(10));
assertThat(result).isNotNull();
assertThat(result.capabilities()).isNotNull();
client.closeGracefully().block(Duration.ofSeconds(10));
}).doesNotThrowAnyException();
}
}

Some files were not shown because too many files have changed in this diff Show more