feat: add SSE transport and improve documentation
- Add Server-Sent Events (SSE) transport implementation with test coverage - Add detailed documentation for both SSE and STDIO transports - Extract common test logic into abstract base classes - Add testcontainers support for SSE integration tests - Replace System.out with SLF4J logging - Externalize dependency versions to properties - Implement endpoint event handling per MCP spec - Add message endpoint sink for reliable message routing - Improve error handling and graceful shutdown - Update core module README with expanded features and examples - Add dependency version properties and logging configuration - Update architecture and class diagrams Resolves #4
This commit is contained in:
Christian Tzolov
parent
be31c6119c
commit
bc281dee54
25 changed files with 1583 additions and 162 deletions
|
|
@ -12,7 +12,7 @@ Spring AI MCP is an experimental project that provides Java and Spring Framework
|
|||
|
||||
The project consists of two main modules:
|
||||
|
||||
### spring-ai-mcp-core
|
||||
### [spring-ai-mcp-core](./spring-ai-mcp-core/README.md)
|
||||
|
||||
The core module provides a Java implementation of the Model Context Protocol specification. It includes:
|
||||
- Synchronous and asynchronous client implementations
|
||||
|
|
@ -20,7 +20,9 @@ The core module provides a Java implementation of the Model Context Protocol spe
|
|||
- Stdio-based server transport
|
||||
- Reactive programming support using Project Reactor
|
||||
|
||||
### spring-ai-mcp-spring
|
||||
[find more](./spring-ai-mcp-core/README.md)
|
||||
|
||||
### [spring-ai-mcp-spring](./spring-ai-mcp-spring/README.md)
|
||||
|
||||
The Spring integration module provides Spring-specific functionality:
|
||||
- Integration with Spring AI's function calling system
|
||||
|
|
|
|||
7
pom.xml
7
pom.xml
|
|
@ -70,7 +70,12 @@
|
|||
<assert4j.version>3.26.3</assert4j.version>
|
||||
<junit.version>5.11.3</junit.version>
|
||||
<mockito.version>5.11.0</mockito.version>
|
||||
<testcontainers.version>1.20.4</testcontainers.version>
|
||||
|
||||
<slf4j-api.version>2.0.16</slf4j-api.version>
|
||||
<logback.version>1.5.12</logback.version>
|
||||
<jackson.version>2.18.2</jackson.version>
|
||||
<springframework.version>6.1.13</springframework.version>
|
||||
|
||||
<!-- plugin versions -->
|
||||
<maven-compiler-plugin.version>3.11.0</maven-compiler-plugin.version>
|
||||
|
|
@ -347,4 +352,4 @@
|
|||
</repository>
|
||||
</repositories>
|
||||
|
||||
</project>
|
||||
</project>
|
||||
|
|
|
|||
Binary file not shown.
|
Before Width: | Height: | Size: 581 KiB After Width: | Height: | Size: 736 KiB |
|
|
@ -10,13 +10,19 @@ This SDK implements the Model Context Protocol, enabling seamless integration wi
|
|||
|
||||
- Synchronous and Asynchronous client implementations
|
||||
- Standard MCP operations support:
|
||||
- Protocol version compatibility negotiation
|
||||
- Client-server capability exchange
|
||||
- Tool discovery and execution
|
||||
- Resource management and templates
|
||||
- Prompt handling and management
|
||||
- Resource management with URI templates
|
||||
- Resource subscription system
|
||||
- Prompt handling and management
|
||||
- Server initialization and ping
|
||||
- Stdio-based server transport
|
||||
- Multiple transport implementations:
|
||||
- Stdio-based transport for process-based communication
|
||||
- SSE-based transport for HTTP streaming
|
||||
- Reactive programming support using Project Reactor
|
||||
- Configurable request timeouts
|
||||
- Customizable JSON serialization/deserialization
|
||||
|
||||
## Installation
|
||||
|
||||
|
|
@ -38,6 +44,41 @@ Detailed UML class diagrams showing the relationships between components can be
|
|||
|
||||
## Usage
|
||||
|
||||
### Transport Layer Options
|
||||
|
||||
The SDK provides two transport implementations:
|
||||
|
||||
#### StdioServerTransport
|
||||
Standard I/O based transport for process-based communication with MCP servers:
|
||||
|
||||
```java
|
||||
ServerParameters params = ServerParameters.builder("npx")
|
||||
.args("-y", "@modelcontextprotocol/server-everything", "dir")
|
||||
.build();
|
||||
McpTransport transport = new StdioServerTransport(params);
|
||||
```
|
||||
|
||||
#### SseServerTransport
|
||||
Server-Sent Events (SSE) based transport following the MCP HTTP with SSE transport specification:
|
||||
|
||||
```java
|
||||
WebClient.Builder webClientBuilder = WebClient.builder()
|
||||
.baseUrl("http://your-mcp-server");
|
||||
McpTransport transport = new SseServerTransport(webClientBuilder);
|
||||
|
||||
// Or with custom ObjectMapper
|
||||
ObjectMapper mapper = new ObjectMapper();
|
||||
McpTransport transport = new SseServerTransport(webClientBuilder, mapper);
|
||||
```
|
||||
|
||||
The SSE transport provides:
|
||||
- Bidirectional communication over HTTP
|
||||
- Automatic reconnection for transient failures
|
||||
- Inbound message streaming via SSE
|
||||
- Outbound message delivery via HTTP POST
|
||||
- Graceful shutdown handling
|
||||
- Configurable JSON serialization
|
||||
|
||||
### Sync Client Example
|
||||
|
||||
```java
|
||||
|
|
@ -47,32 +88,32 @@ ServerParameters params = ServerParameters.builder("npx")
|
|||
.build();
|
||||
|
||||
try (McpSyncClient client = McpClient.sync(new StdioServerTransport(params))) {
|
||||
// Initialize connection
|
||||
// Initialize connection with protocol version and capabilities
|
||||
McpSchema.InitializeResult initResult = client.initialize();
|
||||
|
||||
// List tools synchronously
|
||||
McpSchema.ListToolsResult tools = client.listTools(null);
|
||||
// List available tools
|
||||
McpSchema.ListToolsResult tools = client.listTools();
|
||||
|
||||
// Call tool synchronously
|
||||
// Execute a tool
|
||||
McpSchema.CallToolResult result = client.callTool(
|
||||
new McpSchema.CallToolRequest("echo", Map.of("message", "Hello!"))
|
||||
);
|
||||
|
||||
// Resource management
|
||||
McpSchema.ListResourcesResult resources = client.listResources(null);
|
||||
McpSchema.ListResourcesResult resources = client.listResources();
|
||||
McpSchema.ReadResourceResult resource = client.readResource(
|
||||
new McpSchema.ReadResourceRequest("resource-uri")
|
||||
);
|
||||
|
||||
// Prompt management
|
||||
ListPromptsResult prompts = client.listPrompts(null);
|
||||
// List and retrieve prompts
|
||||
ListPromptsResult prompts = client.listPrompts();
|
||||
GetPromptResult prompt = client.getPrompt(
|
||||
new McpSchema.GetPromptRequest("prompt-id", Map.of())
|
||||
);
|
||||
}
|
||||
```
|
||||
|
||||
### Async Client Example
|
||||
### Async Client Example with Custom Configuration
|
||||
|
||||
```java
|
||||
// Create server parameters
|
||||
|
|
@ -80,43 +121,42 @@ ServerParameters params = ServerParameters.builder("npx")
|
|||
.args("-y", "@modelcontextprotocol/server-everything", "dir")
|
||||
.build();
|
||||
|
||||
// Initialize the async client
|
||||
// Initialize async client with custom timeout and object mapper
|
||||
McpAsyncClient client = McpClient.async(
|
||||
new StdioServerTransport(params)
|
||||
new StdioServerTransport(params),
|
||||
Duration.ofSeconds(30),
|
||||
new ObjectMapper()
|
||||
);
|
||||
|
||||
// Initialize the connection
|
||||
var promptResult = client.initialize()
|
||||
flatMap(result -> {
|
||||
// Connection initialized
|
||||
return client.listTools(null);
|
||||
// Initialize connection and chain operations
|
||||
var result = client.initialize()
|
||||
.flatMap(initResult -> {
|
||||
// Connection initialized with protocol version compatibility
|
||||
return client.listTools();
|
||||
})
|
||||
.flatMap(tools -> {
|
||||
// Process tools
|
||||
// Process available tools
|
||||
return client.callTool(new McpSchema.CallToolRequest("echo",
|
||||
Map.of("message", "Hello MCP!")));
|
||||
})
|
||||
.flatMap(result -> {
|
||||
// Handle tool result
|
||||
return client.listPrompts(null);
|
||||
.flatMap(toolResult -> {
|
||||
// Handle tool execution result
|
||||
return client.listPrompts();
|
||||
})
|
||||
.flatMap(prompts -> {
|
||||
// Process available prompts
|
||||
return client.getPrompt(new McpSchema.GetPromptRequest("prompt-id", Map.of()));
|
||||
});
|
||||
|
||||
// Handle prompt result, e.g. by blocking on it
|
||||
McpSchema.GetPromptResult result = promptResult.block();
|
||||
|
||||
// Resource management example
|
||||
var resourcesResult = client.listResources(null)
|
||||
// Subscribe to resource changes
|
||||
var subscription = client.listResources()
|
||||
.flatMap(resources -> {
|
||||
// Subscribe to resource changes
|
||||
return client.subscribeResource(new McpSchema.SubscribeRequest("resource-uri"));
|
||||
});
|
||||
|
||||
// Handle resources result
|
||||
resourcesResult.block();
|
||||
// Handle results reactively or block if needed
|
||||
McpSchema.GetPromptResult promptResult = result.block();
|
||||
subscription.block();
|
||||
|
||||
// Cleanup
|
||||
client.closeGracefully().block();
|
||||
|
|
@ -124,62 +164,68 @@ client.closeGracefully().block();
|
|||
|
||||
## Architecture
|
||||
|
||||
The SDK follows a layered architecture:
|
||||
The SDK follows a layered architecture with clear separation of concerns:
|
||||
|
||||
### Core Components
|
||||
|
||||
- **McpClient**: Factory class for creating sync and async clients
|
||||
- **McpAsyncClient**: Primary async implementation using Project Reactor
|
||||
- **McpSyncClient**: Synchronous wrapper around the async client
|
||||
- **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
|
||||
- **McpTransport**: Transport layer interface for server communication
|
||||
- **McpSchema**: Comprehensive protocol schema definitions
|
||||
- **AbstractMcpTransport**: Base transport implementation
|
||||
- **StdioServerTransport**: Stdio-based server communication
|
||||
- **DefaultMcpSession**: Base implementation of the session management
|
||||
- **StdioServerTransport**: Standard I/O based server communication
|
||||
- **SseServerTransport**: HTTP-based transport using Server-Sent Events for bidirectional communication
|
||||
|
||||
<img src="docs/spring-ai-mcp-uml-classdiagram.svg" width="600"/>
|
||||
|
||||
### Key Interactions
|
||||
|
||||
1. Client initialization
|
||||
- Transport setup
|
||||
- Server connection establishment
|
||||
- Protocol handshake
|
||||
1. Client Initialization
|
||||
- Transport setup and connection establishment
|
||||
- Protocol version compatibility check
|
||||
- Capability negotiation
|
||||
- Implementation details exchange
|
||||
|
||||
2. Message Flow
|
||||
- JSON-RPC message creation
|
||||
- JSON-RPC message creation and validation
|
||||
- Transport layer handling
|
||||
- Response processing
|
||||
- Error handling
|
||||
- Response processing with type safety
|
||||
- Error handling with specific error codes
|
||||
|
||||
3. Resource Management
|
||||
- Resource listing and reading
|
||||
- Template management
|
||||
- Subscription handling
|
||||
- Change notifications
|
||||
- Resource discovery and listing
|
||||
- URI template-based resource access
|
||||
- Subscription system for change notifications
|
||||
- Resource content retrieval
|
||||
|
||||
4. Prompt System
|
||||
- Prompt discovery
|
||||
- Prompt retrieval
|
||||
- Change notifications
|
||||
- Prompt discovery and listing
|
||||
- Parameter-based prompt retrieval
|
||||
- Change notifications support
|
||||
- Prompt content management
|
||||
|
||||
5. Tool Execution
|
||||
- Tool discovery
|
||||
- Parameter validation
|
||||
- Execution handling
|
||||
- Result processing
|
||||
- Tool discovery and capability checking
|
||||
- Parameter validation and processing
|
||||
- Execution handling with timeout support
|
||||
- Result processing with error handling
|
||||
|
||||
## Error Handling
|
||||
|
||||
The SDK provides comprehensive error handling through the McpError class:
|
||||
|
||||
- Transport-level errors
|
||||
- Protocol violations
|
||||
- Protocol version incompatibility
|
||||
- Transport-level communication errors
|
||||
- JSON-RPC protocol violations
|
||||
- Tool execution failures
|
||||
- Resource access errors
|
||||
- Subscription handling errors
|
||||
- Resource access and subscription errors
|
||||
- Prompt management errors
|
||||
- Timeout handling
|
||||
- Request timeout handling
|
||||
- Server capability mismatches
|
||||
- SSE connection failures and retry handling
|
||||
- HTTP request/response errors
|
||||
|
||||
## Contributing
|
||||
|
||||
|
|
|
|||
144
spring-ai-mcp-core/docs/CLIENT.SSE.TRANSPORT.md
Normal file
144
spring-ai-mcp-core/docs/CLIENT.SSE.TRANSPORT.md
Normal file
|
|
@ -0,0 +1,144 @@
|
|||
# SSE Server Transport Implementation for MCP
|
||||
|
||||
## Overview
|
||||
|
||||
The `SseServerTransport` class implements the Model Context Protocol (MCP) HTTP with SSE transport specification, providing a bidirectional communication channel between clients and servers. This implementation is part of the Spring AI MCP Core library and follows the [MCP HTTP with SSE Transport Specification](https://spec.modelcontextprotocol.io/specification/basic/transports/#http-with-sse).
|
||||
|
||||
## Key Features
|
||||
|
||||
- **Bidirectional Communication**:
|
||||
- Inbound: Receives messages through SSE connection
|
||||
- Outbound: Sends messages via HTTP POST requests
|
||||
|
||||
- **Error Handling**:
|
||||
- Automatic reconnection for transient failures
|
||||
- Graceful shutdown capabilities
|
||||
- Comprehensive error reporting
|
||||
|
||||
- **Message Flow**:
|
||||
1. Client establishes SSE connection to `/sse` endpoint
|
||||
2. Server sends 'endpoint' event with message URI
|
||||
3. Client uses provided URI for outbound messages
|
||||
|
||||
## Technical Details
|
||||
|
||||
### Transport Architecture
|
||||
|
||||
The transport consists of two main processing pipelines:
|
||||
|
||||
1. **Inbound Processing**:
|
||||
- Establishes SSE connection to `/sse` endpoint
|
||||
- Handles two types of events:
|
||||
- `message`: JSON-RPC protocol messages
|
||||
- `endpoint`: Server-provided URI for outbound messages
|
||||
- Implements automatic retry logic for connection failures
|
||||
|
||||
2. **Outbound Processing**:
|
||||
- Sends messages via HTTP POST to server-provided endpoint
|
||||
- Uses dedicated scheduler for sequential message processing
|
||||
- Handles JSON serialization and HTTP communication
|
||||
|
||||
### Key Components
|
||||
|
||||
- **WebClient**: Handles HTTP communications for both SSE and POST requests
|
||||
- **ObjectMapper**: Manages JSON serialization/deserialization
|
||||
- **Scheduler**: Ensures sequential outbound message processing
|
||||
- **Message Endpoint Sink**: Manages server-provided endpoint URI
|
||||
|
||||
## Usage Example
|
||||
|
||||
```java
|
||||
// Create WebClient builder with base URL
|
||||
WebClient.Builder webClientBuilder = WebClient.builder()
|
||||
.baseUrl("http://localhost:3001");
|
||||
|
||||
// Initialize transport
|
||||
SseServerTransport transport = new SseServerTransport(webClientBuilder);
|
||||
|
||||
// Create a JSON-RPC request
|
||||
JSONRPCRequest request = new JSONRPCRequest(
|
||||
McpSchema.JSONRPC_VERSION,
|
||||
"method-name",
|
||||
"request-id",
|
||||
Map.of("key", "value")
|
||||
);
|
||||
|
||||
// Send message
|
||||
transport.sendMessage(request)
|
||||
.subscribe(
|
||||
success -> System.out.println("Message sent successfully"),
|
||||
error -> System.err.println("Error sending message: " + error)
|
||||
);
|
||||
|
||||
// Graceful shutdown when done
|
||||
transport.closeGracefully()
|
||||
.subscribe();
|
||||
```
|
||||
|
||||
## Implementation Details
|
||||
|
||||
### Message Processing
|
||||
|
||||
1. **Inbound Messages**:
|
||||
- Received through SSE connection
|
||||
- Automatically deserialized from JSON
|
||||
- Routed to appropriate handlers based on event type
|
||||
|
||||
2. **Outbound Messages**:
|
||||
- Serialized to JSON
|
||||
- Sent via HTTP POST
|
||||
- Processed sequentially on dedicated thread
|
||||
|
||||
### Error Handling
|
||||
|
||||
- **Connection Failures**:
|
||||
- Automatic retry for transient failures
|
||||
- Configurable retry policies
|
||||
- Graceful handling of permanent failures
|
||||
|
||||
- **Message Processing Errors**:
|
||||
- Comprehensive error reporting
|
||||
- Clean failure handling
|
||||
- Maintains system stability
|
||||
|
||||
### Shutdown Process
|
||||
|
||||
The transport implements a graceful shutdown mechanism that:
|
||||
1. Marks transport as closing
|
||||
2. Disposes of all subscriptions
|
||||
3. Cleans up schedulers
|
||||
4. Completes message endpoint sink
|
||||
|
||||
## Technical Specifications
|
||||
|
||||
- **Protocol**: HTTP with Server-Sent Events (SSE)
|
||||
- **Message Format**: JSON-RPC 2.0
|
||||
- **Content Type**: application/json for outbound, text/event-stream for inbound
|
||||
- **Endpoints**:
|
||||
- SSE Connection: `/sse`
|
||||
- Message Endpoint: Dynamically provided by server
|
||||
|
||||
## Dependencies
|
||||
|
||||
- Spring WebFlux for reactive HTTP client
|
||||
- Jackson for JSON processing
|
||||
- Project Reactor for reactive streams
|
||||
- SLF4J for logging
|
||||
|
||||
## Thread Safety
|
||||
|
||||
The implementation is thread-safe and designed for concurrent use:
|
||||
- Inbound processing uses SSE event loop
|
||||
- Outbound processing uses dedicated single-threaded scheduler
|
||||
- State management uses thread-safe constructs
|
||||
|
||||
## Testing
|
||||
|
||||
The implementation includes comprehensive tests covering:
|
||||
- Message processing
|
||||
- Error handling
|
||||
- Connection management
|
||||
- Shutdown procedures
|
||||
- Edge cases and failure scenarios
|
||||
|
||||
For detailed test examples, see `SseServerTransportTests.java`.
|
||||
123
spring-ai-mcp-core/docs/CLIENT.STDIO.TRANSPORT.md
Normal file
123
spring-ai-mcp-core/docs/CLIENT.STDIO.TRANSPORT.md
Normal file
|
|
@ -0,0 +1,123 @@
|
|||
# StdioServerTransport
|
||||
|
||||
The `StdioServerTransport` class implements the [MCP Stdio Transport](https://spec.modelcontextprotocol.io/specification/basic/transports/#stdio) specification, providing communication with MCP servers over standard input/output streams.
|
||||
|
||||
## Overview
|
||||
|
||||
The Stdio transport uses standard input/output streams for bidirectional communication between the client and server:
|
||||
- Messages are sent to the server via stdout
|
||||
- Messages are received from the server via stdin
|
||||
- Error and debug information is sent via stderr
|
||||
|
||||
## Message Format
|
||||
|
||||
Messages are exchanged as newline-delimited JSON-RPC messages:
|
||||
- Each message is a single line of JSON text
|
||||
- Messages are terminated with a newline character (`\n`)
|
||||
- Messages follow the JSON-RPC 2.0 specification
|
||||
- UTF-8 encoding is used for all messages
|
||||
|
||||
### Message Types
|
||||
|
||||
The transport handles three types of JSON-RPC messages:
|
||||
|
||||
1. **Requests** - Messages with both `method` and `id` fields:
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": 1,
|
||||
"method": "example.method",
|
||||
"params": {}
|
||||
}
|
||||
```
|
||||
|
||||
2. **Notifications** - Messages with `method` but no `id` field:
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"method": "example.notification",
|
||||
"params": {}
|
||||
}
|
||||
```
|
||||
|
||||
3. **Responses** - Messages with either `result` or `error` fields:
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": 1,
|
||||
"result": {}
|
||||
}
|
||||
```
|
||||
|
||||
## Implementation Details
|
||||
|
||||
### Process Management
|
||||
|
||||
The transport manages a server process with:
|
||||
- Command and arguments specified via `ServerParameters`
|
||||
- Environment variables for configuration
|
||||
- Proper process cleanup on shutdown
|
||||
|
||||
### Threading Model
|
||||
|
||||
The implementation uses three dedicated threads:
|
||||
1. **Inbound** - Reads and processes messages from stdin
|
||||
2. **Outbound** - Writes messages to stdout
|
||||
3. **Error** - Handles stderr messages
|
||||
|
||||
### Error Handling
|
||||
|
||||
Errors are handled at multiple levels:
|
||||
- Process startup failures
|
||||
- Stream I/O errors
|
||||
- Message parsing errors
|
||||
- Process termination handling
|
||||
|
||||
## Usage Example
|
||||
|
||||
```java
|
||||
// Create server parameters
|
||||
ServerParameters params = ServerParameters.builder()
|
||||
.command("path/to/server")
|
||||
.args(List.of("--arg1", "value1"))
|
||||
.env(Map.of("ENV_VAR", "value"))
|
||||
.build();
|
||||
|
||||
// Create transport
|
||||
StdioServerTransport transport = new StdioServerTransport(params);
|
||||
|
||||
// Start transport
|
||||
transport.start();
|
||||
|
||||
// Send message
|
||||
JSONRPCRequest request = new JSONRPCRequest();
|
||||
request.setId(1);
|
||||
request.setMethod("example.method");
|
||||
request.setParams(new HashMap<>());
|
||||
transport.send(request);
|
||||
|
||||
// Close transport
|
||||
transport.closeGracefully().block();
|
||||
```
|
||||
|
||||
## Error Messages
|
||||
|
||||
Error messages from the server process are:
|
||||
- Written to stderr
|
||||
- Logged at ERROR level
|
||||
- Emitted through the error sink for handling
|
||||
- UTF-8 encoded
|
||||
|
||||
## Debugging
|
||||
|
||||
For debugging purposes:
|
||||
1. Enable DEBUG level logging to see detailed message flow
|
||||
2. Monitor stderr for server process errors
|
||||
3. Check process exit codes for abnormal termination
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
- The transport ensures thread safety through dedicated schedulers
|
||||
- Messages are processed asynchronously using Project Reactor
|
||||
- Resources are properly cleaned up on shutdown
|
||||
- The implementation is resilient to various failure modes
|
||||
|
|
@ -57,6 +57,10 @@ class StdioServerTransport {
|
|||
-ServerParameters parameters
|
||||
}
|
||||
|
||||
class SseServerTransport {
|
||||
-WebClient webClient
|
||||
}
|
||||
|
||||
class McpSchema {
|
||||
+class ErrorCodes
|
||||
+interface Request
|
||||
|
|
@ -72,6 +76,7 @@ class McpError {
|
|||
|
||||
McpTransport <|.. AbstractMcpTransport
|
||||
AbstractMcpTransport <|-- StdioServerTransport
|
||||
AbstractMcpTransport <|-- SseServerTransport
|
||||
McpSession <|-- DefaultMcpSession
|
||||
DefaultMcpSession <|-- McpAsyncClient
|
||||
McpClient ..> McpAsyncClient : creates
|
||||
|
|
|
|||
File diff suppressed because one or more lines are too long
|
Before Width: | Height: | Size: 28 KiB After Width: | Height: | Size: 30 KiB |
|
|
@ -21,16 +21,35 @@
|
|||
</scm>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-api</artifactId>
|
||||
<version>${slf4j-api.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>ch.qos.logback</groupId>
|
||||
<artifactId>logback-classic</artifactId>
|
||||
<version>${logback.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-databind</artifactId>
|
||||
<version>2.18.2</version>
|
||||
<version>${jackson.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>io.projectreactor</groupId>
|
||||
<artifactId>reactor-core</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework</groupId>
|
||||
<artifactId>spring-webflux</artifactId>
|
||||
<version>${springframework.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.assertj</groupId>
|
||||
<artifactId>assertj-core</artifactId>
|
||||
|
|
@ -54,8 +73,14 @@
|
|||
<artifactId>reactor-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>junit-jupiter</artifactId>
|
||||
<version>${testcontainers.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
|
||||
</project>
|
||||
</project>
|
||||
|
|
|
|||
|
|
@ -36,7 +36,7 @@ import org.springframework.ai.mcp.spec.McpTransport;
|
|||
*/
|
||||
public class McpAsyncClient extends DefaultMcpSession {
|
||||
|
||||
private static TypeReference<Void> VOID_TYPE_REFERENCE = new TypeReference<Void>() {
|
||||
private static TypeReference<Void> VOID_TYPE_REFERENCE = new TypeReference<>() {
|
||||
};
|
||||
|
||||
public McpAsyncClient(McpTransport transport) {
|
||||
|
|
@ -113,10 +113,10 @@ public class McpAsyncClient extends DefaultMcpSession {
|
|||
// --------------------------
|
||||
// Tools
|
||||
// --------------------------
|
||||
private static TypeReference<McpSchema.CallToolResult> CALL_TOOL_RESULT_TYPE_REF = new TypeReference<McpSchema.CallToolResult>() {
|
||||
private static TypeReference<McpSchema.CallToolResult> CALL_TOOL_RESULT_TYPE_REF = new TypeReference<>() {
|
||||
};
|
||||
|
||||
private static TypeReference<McpSchema.ListToolsResult> LIST_TOOLS_RESULT_TYPE_REF = new TypeReference<McpSchema.ListToolsResult>() {
|
||||
private static TypeReference<McpSchema.ListToolsResult> LIST_TOOLS_RESULT_TYPE_REF = new TypeReference<>() {
|
||||
};
|
||||
|
||||
/**
|
||||
|
|
@ -149,13 +149,13 @@ public class McpAsyncClient extends DefaultMcpSession {
|
|||
// Resources
|
||||
// --------------------------
|
||||
|
||||
private static TypeReference<McpSchema.ListResourcesResult> LIST_RESOURCES_RESULT_TYPE_REF = new TypeReference<McpSchema.ListResourcesResult>() {
|
||||
private static TypeReference<McpSchema.ListResourcesResult> LIST_RESOURCES_RESULT_TYPE_REF = new TypeReference<>() {
|
||||
};
|
||||
|
||||
private static TypeReference<McpSchema.ReadResourceResult> READ_RESOURCE_RESULT_TYPE_REF = new TypeReference<McpSchema.ReadResourceResult>() {
|
||||
private static TypeReference<McpSchema.ReadResourceResult> READ_RESOURCE_RESULT_TYPE_REF = new TypeReference<>() {
|
||||
};
|
||||
|
||||
private static TypeReference<McpSchema.ListResourceTemplatesResult> LIST_RESOURCE_TEMPLATES_RESULT_TYPE_REF = new TypeReference<McpSchema.ListResourceTemplatesResult>() {
|
||||
private static TypeReference<McpSchema.ListResourceTemplatesResult> LIST_RESOURCE_TEMPLATES_RESULT_TYPE_REF = new TypeReference<>() {
|
||||
};
|
||||
|
||||
/**
|
||||
|
|
@ -251,10 +251,10 @@ public class McpAsyncClient extends DefaultMcpSession {
|
|||
// --------------------------
|
||||
// Prompts
|
||||
// --------------------------
|
||||
private static TypeReference<McpSchema.ListPromptsResult> LIST_PROMPTS_RESULT_TYPE_REF = new TypeReference<McpSchema.ListPromptsResult>() {
|
||||
private static TypeReference<McpSchema.ListPromptsResult> LIST_PROMPTS_RESULT_TYPE_REF = new TypeReference<>() {
|
||||
};
|
||||
|
||||
private static TypeReference<McpSchema.GetPromptResult> GET_PROMPT_RESULT_TYPE_REF = new TypeReference<McpSchema.GetPromptResult>() {
|
||||
private static TypeReference<McpSchema.GetPromptResult> GET_PROMPT_RESULT_TYPE_REF = new TypeReference<>() {
|
||||
};
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -0,0 +1,353 @@
|
|||
/*
|
||||
* Copyright 2024 - 2024 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.ai.mcp.client.sse;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.function.BiConsumer;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import reactor.core.Disposable;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.core.publisher.Sinks;
|
||||
import reactor.core.publisher.SynchronousSink;
|
||||
import reactor.core.scheduler.Scheduler;
|
||||
import reactor.core.scheduler.Schedulers;
|
||||
import reactor.util.retry.Retry;
|
||||
import reactor.util.retry.Retry.RetrySignal;
|
||||
|
||||
import org.springframework.ai.mcp.client.util.Assert;
|
||||
import org.springframework.ai.mcp.spec.AbstractMcpTransport;
|
||||
import org.springframework.ai.mcp.spec.McpSchema;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.JSONRPCMessage;
|
||||
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 SseServerTransport extends AbstractMcpTransport {
|
||||
|
||||
private final static Logger logger = LoggerFactory.getLogger(SseServerTransport.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.
|
||||
*/
|
||||
private ObjectMapper objectMapper;
|
||||
|
||||
/**
|
||||
* Dedicated scheduler for processing outbound messages. Ensures sequential processing
|
||||
* of outbound messages on a single thread.
|
||||
*/
|
||||
private Scheduler outboundScheduler;
|
||||
|
||||
/**
|
||||
* Subscription for the SSE connection handling inbound messages. Used for cleanup
|
||||
* during transport shutdown.
|
||||
*/
|
||||
private Disposable inboundSubscription;
|
||||
|
||||
/**
|
||||
* Subscription for the outbound message processing pipeline. Used for cleanup during
|
||||
* transport shutdown.
|
||||
*/
|
||||
private Disposable outboundSubscription;
|
||||
|
||||
/**
|
||||
* 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.Many<String> messageEndpointSink = Sinks.many().replay().latest();
|
||||
|
||||
/**
|
||||
* Constructs a new SseServerTransport 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 SseServerTransport(WebClient.Builder webClientBuilder) {
|
||||
this(webClientBuilder, new ObjectMapper());
|
||||
}
|
||||
|
||||
/**
|
||||
* Constructs a new SseServerTransport 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 SseServerTransport(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();
|
||||
|
||||
this.outboundScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(), "outbound");
|
||||
|
||||
startInboundProcessing();
|
||||
startOutboundProcessing();
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
private void startInboundProcessing() {// @formatter:off
|
||||
this.inboundSubscription = this.webClient
|
||||
.get()
|
||||
.uri(SSE_ENDPOINT)
|
||||
.accept(MediaType.TEXT_EVENT_STREAM)
|
||||
.retrieve()
|
||||
.bodyToFlux(SSE_TYPE)
|
||||
.retryWhen(Retry.from(retrySignal -> retrySignal.handle(inboundRetryHandler)))
|
||||
.subscribe(
|
||||
onInboudEvent,
|
||||
onInboundError,
|
||||
onInboundComplete);
|
||||
} // @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());
|
||||
};
|
||||
|
||||
/**
|
||||
* Processes incoming SSE events, handling both message and endpoint events. For
|
||||
* message events, deserializes and emits JSON-RPC messages to the inbound sink. For
|
||||
* endpoint events, updates the message endpoint URI for outbound messages.
|
||||
*/
|
||||
private Consumer<ServerSentEvent<String>> onInboudEvent = event -> {
|
||||
if (!isClosing) {
|
||||
try {
|
||||
logger.debug("Received SSE event: {}", event);
|
||||
|
||||
// When a client connects, the server MUST send an endpoint event
|
||||
// containing a URI for the client to use for sending messages.
|
||||
// All subsequent client messages MUST be sent as HTTP POST requests
|
||||
// to this endpoint.
|
||||
if (ENDPOINT_EVENT_TYPE.equals(event.event())) {
|
||||
|
||||
String messageEndpointUri = event.data();
|
||||
messageEndpointSink.tryEmitNext(messageEndpointUri);
|
||||
|
||||
}
|
||||
else if (MESSAGE_EVENT_TYPE.equals(event.event())) {
|
||||
JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(this.objectMapper, event.data());
|
||||
|
||||
if (!this.getInboundSink().tryEmitNext(message).isSuccess()) {
|
||||
throw new RuntimeException("Failed to enqueue message");
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
if (!isClosing) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Error handler for the inbound SSE stream.
|
||||
*/
|
||||
private Consumer<Throwable> onInboundError = error -> {
|
||||
if (!isClosing) {
|
||||
logger.error("Error receiving SSE: {}", error);
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Completion handler for the inbound SSE stream.
|
||||
*/
|
||||
private Runnable onInboundComplete = () -> {
|
||||
if (!isClosing) {
|
||||
logger.info("SSE stream completed!");
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Initializes and starts the outbound message processing. Sets up the pipeline for
|
||||
* sending messages to the server-provided endpoint. Messages are sent as HTTP POST
|
||||
* requests with JSON content.
|
||||
*
|
||||
* <p>
|
||||
*
|
||||
* This pipeline depends on the inbound pipeline to provide the message endpoint URI
|
||||
* for the client to use for sending messages. The messageEndpointSink is used to
|
||||
* store the most recent message endpoint URI provided by the server.
|
||||
*/
|
||||
private void startOutboundProcessing() {// @formatter:off
|
||||
this.outboundSubscription = messageEndpointSink.asFlux()
|
||||
// The messageEndpoint is the endpoint URI to send the messages
|
||||
// It is provided by the server as part of the endpoint event
|
||||
.flatMap(messageEndpointUri -> this.getOutboundSink()
|
||||
.asFlux()
|
||||
.publishOn(outboundScheduler)
|
||||
.flatMap(jsonRpcMessage -> {
|
||||
if (isClosing) {
|
||||
return Mono.empty();
|
||||
}
|
||||
|
||||
try {
|
||||
String jsonText = this.objectMapper.writeValueAsString(jsonRpcMessage);
|
||||
|
||||
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());
|
||||
}
|
||||
})
|
||||
.thenReturn(jsonRpcMessage);
|
||||
}
|
||||
catch (IOException e) {
|
||||
if (!isClosing) {
|
||||
return Mono.error(new RuntimeException("Failed to serialize message", e));
|
||||
}
|
||||
return Mono.empty();
|
||||
}
|
||||
}))
|
||||
.subscribe();
|
||||
} // @formatter:on
|
||||
|
||||
/**
|
||||
* 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();
|
||||
}
|
||||
|
||||
if (outboundSubscription != null) {
|
||||
outboundSubscription.dispose();
|
||||
}
|
||||
|
||||
// Dispose of scheduler
|
||||
if (outboundScheduler != null) {
|
||||
outboundScheduler.dispose();
|
||||
}
|
||||
|
||||
// Complete the endpoint sink
|
||||
// messageEndpointSink.tryEmitComplete();
|
||||
})
|
||||
.then()
|
||||
.subscribeOn(Schedulers.boundedElastic());
|
||||
} // @formatter:on
|
||||
|
||||
}
|
||||
|
|
@ -21,51 +21,64 @@ import java.io.IOException;
|
|||
import java.io.InputStreamReader;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.Executors;
|
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.core.scheduler.Scheduler;
|
||||
import reactor.core.scheduler.Schedulers;
|
||||
|
||||
import org.springframework.ai.mcp.client.util.Assert;
|
||||
import org.springframework.ai.mcp.spec.AbstractMcpTransport;
|
||||
import org.springframework.ai.mcp.spec.McpSchema;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.JSONRPCMessage;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.JSONRPCNotification;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.JSONRPCRequest;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.JSONRPCResponse;
|
||||
|
||||
/**
|
||||
* Stdio client for communicating with a server process.
|
||||
* 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 StdioServerTransport extends AbstractMcpTransport {
|
||||
|
||||
private final static TypeReference<HashMap<String, Object>> MAP_TYPE_REF = new TypeReference<HashMap<String, Object>>() {
|
||||
};
|
||||
private static final Logger logger = LoggerFactory.getLogger(StdioServerTransport.class);
|
||||
|
||||
/** The server process being communicated with */
|
||||
private Process process;
|
||||
|
||||
/** 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;
|
||||
|
||||
/**
|
||||
* Creates a new StdioServerTransport with the specified parameters and default
|
||||
* ObjectMapper.
|
||||
* @param params The parameters for configuring the server process
|
||||
*/
|
||||
public StdioServerTransport(ServerParameters params) {
|
||||
this(params, new ObjectMapper());
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new StdioServerTransport 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 StdioServerTransport(ServerParameters params, ObjectMapper objectMapper) {
|
||||
|
||||
super(objectMapper);
|
||||
|
||||
Assert.notNull(params, "The params can not be null");
|
||||
|
|
@ -79,6 +92,13 @@ public class StdioServerTransport extends AbstractMcpTransport {
|
|||
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 void start() {
|
||||
// Let's kick off the abstraction layer that will consume the logical messages
|
||||
|
|
@ -115,10 +135,19 @@ public class StdioServerTransport extends AbstractMcpTransport {
|
|||
startErrorProcessing();
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates and returns a new ProcessBuilder instance. Protected to allow overriding in
|
||||
* tests.
|
||||
* @return A new ProcessBuilder instance
|
||||
*/
|
||||
protected ProcessBuilder getProcessBuilder() {
|
||||
return new ProcessBuilder();
|
||||
}
|
||||
|
||||
/**
|
||||
* Waits for the server process to exit.
|
||||
* @throws RuntimeException if the process is interrupted while waiting
|
||||
*/
|
||||
public void awaitForExit() {
|
||||
try {
|
||||
this.process.waitFor();
|
||||
|
|
@ -128,6 +157,10 @@ public class StdioServerTransport extends AbstractMcpTransport {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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(
|
||||
|
|
@ -135,7 +168,7 @@ public class StdioServerTransport extends AbstractMcpTransport {
|
|||
String line;
|
||||
while ((line = processErrorReader.readLine()) != null) {
|
||||
try {
|
||||
System.out.println("Received error line: " + line);
|
||||
logger.error("Received error line: {}", line);
|
||||
// TODO: handle errors, etc.
|
||||
this.getErrorSink().tryEmitNext(line);
|
||||
}
|
||||
|
|
@ -150,13 +183,17 @@ public class StdioServerTransport extends AbstractMcpTransport {
|
|||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 = deserializeJsonRpcMessage(line);
|
||||
JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(this.objectMapper, line);
|
||||
if (!this.getInboundSink().tryEmitNext(message).isSuccess()) {
|
||||
// TODO: Back off, reschedule, give up?
|
||||
throw new RuntimeException("Failed to enqueue message");
|
||||
|
|
@ -173,24 +210,11 @@ public class StdioServerTransport extends AbstractMcpTransport {
|
|||
});
|
||||
}
|
||||
|
||||
private JSONRPCMessage deserializeJsonRpcMessage(String jsonText) throws IOException {
|
||||
|
||||
var map = this.objectMapper.readValue(jsonText, MAP_TYPE_REF);
|
||||
|
||||
// Determine message type based on specific JSON structure
|
||||
if (map.containsKey("method") && map.containsKey("id")) {
|
||||
return this.objectMapper.convertValue(map, JSONRPCRequest.class);
|
||||
}
|
||||
else if (map.containsKey("method") && !map.containsKey("id")) {
|
||||
return this.objectMapper.convertValue(map, JSONRPCNotification.class);
|
||||
}
|
||||
else if (map.containsKey("result") || map.containsKey("error")) {
|
||||
return this.objectMapper.convertValue(map, JSONRPCResponse.class);
|
||||
}
|
||||
|
||||
throw new IllegalArgumentException("Cannot deserialize JSONRPCMessage: " + jsonText);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.getOutboundSink()
|
||||
.asFlux()
|
||||
|
|
@ -214,11 +238,16 @@ public class StdioServerTransport extends AbstractMcpTransport {
|
|||
.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(() -> {
|
||||
System.out.println("Sending TERM to process");
|
||||
logger.info("Sending TERM to process");
|
||||
if (this.process != null) {
|
||||
this.process.destroy();
|
||||
return process.onExit();
|
||||
|
|
@ -228,7 +257,7 @@ public class StdioServerTransport extends AbstractMcpTransport {
|
|||
}
|
||||
}).doOnNext(process -> {
|
||||
if (process.exitValue() != 0) {
|
||||
System.out.println("Process terminated with code " + process.exitValue());
|
||||
logger.warn("Process terminated with code " + process.exitValue());
|
||||
}
|
||||
}).then(Mono.fromRunnable(() -> {
|
||||
// The Threads are blocked on readLine so disposeGracefully would not
|
||||
|
|
@ -239,4 +268,4 @@ public class StdioServerTransport extends AbstractMcpTransport {
|
|||
})).then().subscribeOn(Schedulers.boundedElastic());
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,6 +19,8 @@ package org.springframework.ai.mcp.spec;
|
|||
import java.util.function.Consumer;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.core.publisher.Sinks;
|
||||
|
||||
|
|
@ -34,6 +36,8 @@ import org.springframework.ai.mcp.spec.McpSchema.JSONRPCMessage;
|
|||
// proper encapsulation and state management.
|
||||
public abstract class AbstractMcpTransport implements McpTransport {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(AbstractMcpTransport.class);
|
||||
|
||||
protected final ObjectMapper objectMapper;
|
||||
|
||||
private final Sinks.Many<String> errorSink;
|
||||
|
|
@ -42,10 +46,9 @@ public abstract class AbstractMcpTransport implements McpTransport {
|
|||
|
||||
private final Sinks.Many<JSONRPCMessage> outboundSink;
|
||||
|
||||
private Consumer<JSONRPCMessage> inboundMessageHandler = message -> System.out
|
||||
.println("Received message: " + message);
|
||||
private Consumer<JSONRPCMessage> inboundMessageHandler = message -> logger.debug("Received message: {}", message);
|
||||
|
||||
private Consumer<String> errorHandler = error -> System.err.println("Received error: " + error);
|
||||
private Consumer<String> errorHandler = error -> logger.error("Received error: {}", error);
|
||||
|
||||
public AbstractMcpTransport() {
|
||||
this(new ObjectMapper());
|
||||
|
|
|
|||
|
|
@ -23,6 +23,8 @@ import java.util.concurrent.ConcurrentHashMap;
|
|||
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.core.publisher.MonoSink;
|
||||
|
||||
|
|
@ -36,6 +38,8 @@ import org.springframework.ai.mcp.client.util.Assert;
|
|||
*/
|
||||
public class DefaultMcpSession implements McpSession {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(DefaultMcpSession.class);
|
||||
|
||||
private final ConcurrentHashMap<Object, MonoSink<McpSchema.JSONRPCResponse>> pendingResponses = new ConcurrentHashMap<>();
|
||||
|
||||
private final Duration requestTimeout;
|
||||
|
|
@ -59,21 +63,21 @@ public class DefaultMcpSession implements McpSession {
|
|||
if (message instanceof McpSchema.JSONRPCResponse response) {
|
||||
var sink = pendingResponses.remove(response.id());
|
||||
if (sink == null) {
|
||||
System.out.println("Unexpected response for unkown id " + response.id());
|
||||
logger.warn("Unexpected response for unkown id {}", response.id());
|
||||
}
|
||||
else {
|
||||
sink.success(response);
|
||||
}
|
||||
}
|
||||
else if (message instanceof McpSchema.JSONRPCRequest request) {
|
||||
System.out.println("Client does not yet support server requests");
|
||||
logger.info("Client does not yet support server requests");
|
||||
}
|
||||
else if (message instanceof McpSchema.JSONRPCNotification notification) {
|
||||
System.out.println("Notifications not yet supported");
|
||||
logger.info("Notifications not yet supported");
|
||||
}
|
||||
});
|
||||
|
||||
this.transport.setInboundErrorHandler(error -> System.out.println("Error received " + error));
|
||||
this.transport.setInboundErrorHandler(error -> logger.error("Error received: {}", error));
|
||||
|
||||
this.transport.start();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,6 +16,8 @@
|
|||
|
||||
package org.springframework.ai.mcp.spec;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
|
|
@ -25,6 +27,8 @@ import com.fasterxml.jackson.annotation.JsonProperty;
|
|||
import com.fasterxml.jackson.annotation.JsonSubTypes;
|
||||
import com.fasterxml.jackson.annotation.JsonTypeInfo;
|
||||
import com.fasterxml.jackson.annotation.JsonTypeInfo.As;
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
||||
/**
|
||||
* Based on the <a href="http://www.jsonrpc.org/specification">JSON-RPC 2.0
|
||||
|
|
@ -33,7 +37,6 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo.As;
|
|||
* Context Protocol Schema</a>.
|
||||
*
|
||||
* @author Christian Tzolov
|
||||
* @since 1.0.0
|
||||
*/
|
||||
public class McpSchema {
|
||||
|
||||
|
|
@ -63,6 +66,38 @@ public class McpSchema {
|
|||
|
||||
}
|
||||
|
||||
private final static TypeReference<HashMap<String, Object>> MAP_TYPE_REF = new TypeReference<>() {
|
||||
};
|
||||
|
||||
/**
|
||||
* Deserializes a JSON string into a JSONRPCMessage object.
|
||||
* @param objectMapper The ObjectMapper instance to use for deserialization
|
||||
* @param jsonText The JSON string to deserialize
|
||||
* @return A JSONRPCMessage instance using either the {@link JSONRPCRequest},
|
||||
* {@link JSONRPCNotification}, or {@link JSONRPCResponse} classes.
|
||||
* @throws IOException If there's an error during deserialization
|
||||
* @throws IllegalArgumentException If the JSON structure doesn't match any known
|
||||
* message type
|
||||
*/
|
||||
public static JSONRPCMessage deserializeJsonRpcMessage(ObjectMapper objectMapper, String jsonText)
|
||||
throws IOException {
|
||||
|
||||
var map = objectMapper.readValue(jsonText, MAP_TYPE_REF);
|
||||
|
||||
// Determine message type based on specific JSON structure
|
||||
if (map.containsKey("method") && map.containsKey("id")) {
|
||||
return objectMapper.convertValue(map, JSONRPCRequest.class);
|
||||
}
|
||||
else if (map.containsKey("method") && !map.containsKey("id")) {
|
||||
return objectMapper.convertValue(map, JSONRPCNotification.class);
|
||||
}
|
||||
else if (map.containsKey("result") || map.containsKey("error")) {
|
||||
return objectMapper.convertValue(map, JSONRPCResponse.class);
|
||||
}
|
||||
|
||||
throw new IllegalArgumentException("Cannot deserialize JSONRPCMessage: " + jsonText);
|
||||
}
|
||||
|
||||
// ---------------------------
|
||||
// JSON-RPC Message Types
|
||||
// ---------------------------
|
||||
|
|
@ -311,6 +346,7 @@ public class McpSchema {
|
|||
* actually be represented as text (not binary data).
|
||||
*/
|
||||
@JsonInclude(JsonInclude.Include.NON_ABSENT)
|
||||
@JsonIgnoreProperties(ignoreUnknown = true)
|
||||
public record TextResourceContents( // @formatter:off
|
||||
@JsonProperty("uri") String uri,
|
||||
@JsonProperty("mimeType") String mimeType,
|
||||
|
|
@ -327,6 +363,7 @@ public class McpSchema {
|
|||
* (not text).
|
||||
*/
|
||||
@JsonInclude(JsonInclude.Include.NON_ABSENT)
|
||||
@JsonIgnoreProperties(ignoreUnknown = true)
|
||||
public record BlobResourceContents( // @formatter:off
|
||||
@JsonProperty("uri") String uri,
|
||||
@JsonProperty("mimeType") String mimeType,
|
||||
|
|
@ -452,11 +489,11 @@ public class McpSchema {
|
|||
// Sampling and Message Creation
|
||||
@JsonInclude(JsonInclude.Include.NON_ABSENT)
|
||||
public record CreateMessageRequest(// @formatter:off
|
||||
@JsonProperty("messages") List<SamplingMessage> messages,
|
||||
@JsonProperty("messages") List<SamplingMessage> messages,
|
||||
@JsonProperty("modelPreferences") ModelPreferences modelPreferences,
|
||||
@JsonProperty("systemPrompt") String systemPrompt,
|
||||
@JsonProperty("includeContext") ContextInclusionStrategy includeContext,
|
||||
@JsonProperty("temperature") Double temperature,
|
||||
@JsonProperty("systemPrompt") String systemPrompt,
|
||||
@JsonProperty("includeContext") ContextInclusionStrategy includeContext,
|
||||
@JsonProperty("temperature") Double temperature,
|
||||
@JsonProperty("maxTokens") int maxTokens,
|
||||
@JsonProperty("stopSequences") List<String> stopSequences,
|
||||
@JsonProperty("metadata") Map<String, Object> metadata) implements Request {
|
||||
|
|
@ -506,14 +543,14 @@ public class McpSchema {
|
|||
// Progress and Logging
|
||||
// ---------------------------
|
||||
public record ProgressNotification(// @formatter:off
|
||||
@JsonProperty("progressToken") String progressToken,
|
||||
@JsonProperty("progress") double progress,
|
||||
@JsonProperty("progressToken") String progressToken,
|
||||
@JsonProperty("progress") double progress,
|
||||
@JsonProperty("total") Double total) {
|
||||
}// @formatter:on
|
||||
|
||||
public record LoggingMessageNotification(// @formatter:off
|
||||
@JsonProperty("level") LoggingLevel level,
|
||||
@JsonProperty("logger") String logger,
|
||||
@JsonProperty("level") LoggingLevel level,
|
||||
@JsonProperty("logger") String logger,
|
||||
@JsonProperty("data") Object data) {
|
||||
}// @formatter:on
|
||||
|
||||
|
|
@ -528,25 +565,25 @@ public class McpSchema {
|
|||
}
|
||||
|
||||
public record PromptReference(// @formatter:off
|
||||
@JsonProperty("type") String type,
|
||||
@JsonProperty("type") String type,
|
||||
@JsonProperty("name") String name) implements PromptOrResourceReference {
|
||||
}// @formatter:on
|
||||
|
||||
public record ResourceReference(// @formatter:off
|
||||
@JsonProperty("type") String type,
|
||||
@JsonProperty("type") String type,
|
||||
@JsonProperty("uri") String uri) implements PromptOrResourceReference {
|
||||
}// @formatter:on
|
||||
|
||||
public record CompleteArgument(// @formatter:off
|
||||
@JsonProperty("name") String name,
|
||||
@JsonProperty("name") String name,
|
||||
@JsonProperty("value") String value) {
|
||||
}// @formatter:on
|
||||
}
|
||||
|
||||
public record CompleteResult(CompleteCompletion completion) {
|
||||
public record CompleteCompletion(// @formatter:off
|
||||
@JsonProperty("values") List<String> values,
|
||||
@JsonProperty("total") Integer total,
|
||||
@JsonProperty("values") List<String> values,
|
||||
@JsonProperty("total") Integer total,
|
||||
@JsonProperty("hasMore") Boolean hasMore) {
|
||||
}// @formatter:on
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,7 +16,6 @@
|
|||
|
||||
package org.springframework.ai.mcp.spec;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
|
|
|
|||
31
spring-ai-mcp-core/src/main/resources/logback.xml
Normal file
31
spring-ai-mcp-core/src/main/resources/logback.xml
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!DOCTYPE configuration>
|
||||
|
||||
<configuration>
|
||||
<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
|
||||
<encoder>
|
||||
<pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n</pattern>
|
||||
</encoder>
|
||||
</appender>
|
||||
|
||||
<!-- Main MCP package -->
|
||||
<logger name="org.springframework.ai.mcp" level="INFO"/>
|
||||
|
||||
<!-- Client packages -->
|
||||
<logger name="org.springframework.ai.mcp.client" level="INFO"/>
|
||||
<logger name="org.springframework.ai.mcp.client.sse" level="INFO"/>
|
||||
<logger name="org.springframework.ai.mcp.client.stdio" level="INFO"/>
|
||||
|
||||
<!-- Spec package -->
|
||||
<logger name="org.springframework.ai.mcp.spec" level="INFO"/>
|
||||
|
||||
<!-- Test packages -->
|
||||
<logger name="org.springframework.ai.mcp.client.sse.SseServerTransportTests" level="DEBUG"/>
|
||||
<logger name="org.springframework.ai.mcp.client.stdio.McpAsyncClientTests" level="DEBUG"/>
|
||||
<logger name="org.springframework.ai.mcp.client.stdio.McpSyncClientTests" level="DEBUG"/>
|
||||
|
||||
<!-- Root logger -->
|
||||
<root level="INFO">
|
||||
<appender-ref ref="CONSOLE"/>
|
||||
</root>
|
||||
</configuration>
|
||||
|
|
@ -0,0 +1,82 @@
|
|||
/*
|
||||
* Copyright 2024 - 2024 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.ai.mcp.attic;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
||||
import org.springframework.ai.mcp.client.McpClient;
|
||||
import org.springframework.ai.mcp.client.McpSyncClient;
|
||||
import org.springframework.ai.mcp.client.sse.SseServerTransport;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.CallToolRequest;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.CallToolResult;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.ListPromptsResult;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.ListResourcesResult;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.ListToolsResult;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.Resource;
|
||||
import org.springframework.web.reactive.function.client.WebClient;
|
||||
|
||||
/**
|
||||
* @author Christian Tzolov
|
||||
* @since 1.0.0
|
||||
*/
|
||||
|
||||
public class MainSSE {
|
||||
|
||||
public static void main(String[] args) {
|
||||
|
||||
try (McpSyncClient mcpClient = McpClient.sync(
|
||||
new SseServerTransport(WebClient.builder().baseUrl("http://localhost:3001")), Duration.ofSeconds(1000),
|
||||
new ObjectMapper())) {
|
||||
|
||||
mcpClient.initialize();
|
||||
|
||||
ListToolsResult tools = mcpClient.listTools();
|
||||
System.out.println("Tools: " + tools);
|
||||
|
||||
mcpClient.ping();
|
||||
|
||||
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", "Hello MCP Spring AI!"));
|
||||
CallToolResult callToolResult = mcpClient.callTool(callToolRequest);
|
||||
System.out.println("Call Tool Result: " + callToolResult);
|
||||
|
||||
// mcpClient.sendRootsListChanged();
|
||||
|
||||
// Resources
|
||||
ListResourcesResult resources = mcpClient.listResources();
|
||||
System.out.println("Resources Size: " + resources.resources().size());
|
||||
System.out.println("Resources: " + resources);
|
||||
for (Resource resource : resources.resources()) {
|
||||
System.out.println(mcpClient.readResource(resource));
|
||||
}
|
||||
|
||||
var resourceTemplate = mcpClient.listResourceTemplates();
|
||||
System.out.println("Resource Templates: " + resourceTemplate);
|
||||
|
||||
ListPromptsResult prompts = mcpClient.listPrompts();
|
||||
for (var prompt : prompts.prompts()) {
|
||||
System.out.println("Prompt: " + prompt);
|
||||
}
|
||||
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,176 @@
|
|||
/*
|
||||
* Copyright 2024-2024 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.ai.mcp.client;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import reactor.test.StepVerifier;
|
||||
|
||||
import org.springframework.ai.mcp.spec.McpSchema.CallToolRequest;
|
||||
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.Tool;
|
||||
import org.springframework.ai.mcp.spec.McpTransport;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatCode;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
/**
|
||||
* Test suite for the {@link McpAsyncClient} that can be used with different
|
||||
* {@link McpTransport} implementations.
|
||||
*
|
||||
* @author Christian Tzolov
|
||||
* @author Dariusz Jędrzejczyk
|
||||
*/
|
||||
public abstract class AbstractMcpAsyncClientTests {
|
||||
|
||||
private McpAsyncClient mcpAsyncClient;
|
||||
|
||||
protected McpTransport mcpTransport;
|
||||
|
||||
private static final Duration TIMEOUT = Duration.ofSeconds(20);
|
||||
|
||||
private static final String ECHO_TEST_MESSAGE = "Hello MCP Spring AI!";
|
||||
|
||||
abstract protected void createMcpTransport();
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
|
||||
createMcpTransport();
|
||||
|
||||
assertThatCode(() -> {
|
||||
mcpAsyncClient = McpClient.async(mcpTransport, TIMEOUT, new ObjectMapper());
|
||||
mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
|
||||
}).doesNotThrowAnyException();
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void tearDown() {
|
||||
if (mcpAsyncClient != null) {
|
||||
assertThatCode(() -> mcpAsyncClient.closeGracefully().block(Duration.ofSeconds(10)))
|
||||
.doesNotThrowAnyException();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void testConstructorWithInvalidArguments() {
|
||||
assertThatThrownBy(() -> McpClient.sync(null, TIMEOUT, new ObjectMapper()))
|
||||
.isInstanceOf(IllegalArgumentException.class);
|
||||
|
||||
assertThatThrownBy(() -> McpClient.sync(mcpTransport, null, new ObjectMapper()))
|
||||
.isInstanceOf(IllegalArgumentException.class);
|
||||
|
||||
assertThatThrownBy(() -> McpClient.sync(mcpTransport, TIMEOUT, null))
|
||||
.isInstanceOf(IllegalArgumentException.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testListTools() {
|
||||
StepVerifier.create(mcpAsyncClient.listTools(null)).consumeNextWith(result -> {
|
||||
assertThat(result.tools()).isNotNull().isNotEmpty();
|
||||
|
||||
Tool firstTool = result.tools().get(0);
|
||||
assertThat(firstTool.name()).isNotNull();
|
||||
assertThat(firstTool.description()).isNotNull();
|
||||
}).verifyComplete();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPing() {
|
||||
assertThatCode(() -> mcpAsyncClient.ping().block()).doesNotThrowAnyException();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCallTool() {
|
||||
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE));
|
||||
|
||||
StepVerifier.create(mcpAsyncClient.callTool(callToolRequest)).consumeNextWith(callToolResult -> {
|
||||
assertThat(callToolResult).isNotNull().satisfies(result -> {
|
||||
assertThat(result.content()).isNotNull();
|
||||
assertThat(result.isError()).isNull();
|
||||
});
|
||||
}).verifyComplete();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCallToolWithInvalidTool() {
|
||||
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", ECHO_TEST_MESSAGE));
|
||||
|
||||
assertThatThrownBy(() -> mcpAsyncClient.callTool(invalidRequest).block()).isInstanceOf(Exception.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testRootsListChanged() {
|
||||
assertThatCode(() -> mcpAsyncClient.sendRootsListChanged().block()).doesNotThrowAnyException();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testListResources() {
|
||||
StepVerifier.create(mcpAsyncClient.listResources(null)).consumeNextWith(resources -> {
|
||||
assertThat(resources).isNotNull().satisfies(result -> {
|
||||
assertThat(result.resources()).isNotNull();
|
||||
|
||||
if (!result.resources().isEmpty()) {
|
||||
Resource firstResource = result.resources().get(0);
|
||||
assertThat(firstResource.uri()).isNotNull();
|
||||
assertThat(firstResource.name()).isNotNull();
|
||||
}
|
||||
});
|
||||
}).verifyComplete();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testMcpAsyncClientState() {
|
||||
assertThat(mcpAsyncClient).isNotNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testListPrompts() {
|
||||
StepVerifier.create(mcpAsyncClient.listPrompts(null)).consumeNextWith(prompts -> {
|
||||
assertThat(prompts).isNotNull().satisfies(result -> {
|
||||
assertThat(result.prompts()).isNotNull();
|
||||
|
||||
if (!result.prompts().isEmpty()) {
|
||||
Prompt firstPrompt = result.prompts().get(0);
|
||||
assertThat(firstPrompt.name()).isNotNull();
|
||||
assertThat(firstPrompt.description()).isNotNull();
|
||||
}
|
||||
});
|
||||
}).verifyComplete();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testGetPrompt() {
|
||||
StepVerifier.create(mcpAsyncClient.getPrompt(new GetPromptRequest("simple_prompt", Map.of())))
|
||||
.consumeNextWith(prompt -> {
|
||||
assertThat(prompt).isNotNull().satisfies(result -> {
|
||||
assertThat(result.messages()).isNotEmpty();
|
||||
assertThat(result.messages()).hasSize(1);
|
||||
});
|
||||
})
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,165 @@
|
|||
/*
|
||||
* Copyright 2024-2024 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.ai.mcp.client;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.ai.mcp.spec.McpSchema.CallToolRequest;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.CallToolResult;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.ListResourcesResult;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.ListToolsResult;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.Resource;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.TextContent;
|
||||
import org.springframework.ai.mcp.spec.McpSchema.Tool;
|
||||
import org.springframework.ai.mcp.spec.McpTransport;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatCode;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
/**
|
||||
* Unit tests for MCP Client Session functionality.
|
||||
*
|
||||
* @author Christian Tzolov
|
||||
* @author Dariusz Jędrzejczyk
|
||||
*/
|
||||
public abstract class AbstractMcpSyncClientTests {
|
||||
|
||||
private McpSyncClient mcpSyncClient;
|
||||
|
||||
private static final Duration TIMEOUT = Duration.ofSeconds(20);
|
||||
|
||||
private static final String TEST_MESSAGE = "Hello MCP Spring AI!";
|
||||
|
||||
protected McpTransport mcpTransport;
|
||||
|
||||
abstract protected void createMcpTransport();
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
createMcpTransport();
|
||||
|
||||
assertThatCode(() -> {
|
||||
mcpSyncClient = McpClient.sync(mcpTransport, TIMEOUT, new ObjectMapper());
|
||||
mcpSyncClient.initialize();
|
||||
}).doesNotThrowAnyException();
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void tearDown() {
|
||||
if (mcpSyncClient != null) {
|
||||
assertThatCode(() -> mcpSyncClient.close()).doesNotThrowAnyException();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void testConstructorWithInvalidArguments() {
|
||||
assertThatThrownBy(() -> McpClient.sync(null, TIMEOUT, new ObjectMapper()))
|
||||
.isInstanceOf(IllegalArgumentException.class);
|
||||
|
||||
assertThatThrownBy(() -> McpClient.sync(mcpTransport, null, new ObjectMapper()))
|
||||
.isInstanceOf(IllegalArgumentException.class);
|
||||
|
||||
assertThatThrownBy(() -> McpClient.sync(mcpTransport, TIMEOUT, null))
|
||||
.isInstanceOf(IllegalArgumentException.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testListTools() {
|
||||
ListToolsResult tools = mcpSyncClient.listTools(null);
|
||||
|
||||
assertThat(tools).isNotNull().satisfies(result -> {
|
||||
assertThat(result.tools()).isNotNull().isNotEmpty();
|
||||
|
||||
Tool firstTool = result.tools().get(0);
|
||||
assertThat(firstTool.name()).isNotNull();
|
||||
assertThat(firstTool.description()).isNotNull();
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCallTools() {
|
||||
CallToolResult toolResult = mcpSyncClient.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4)));
|
||||
|
||||
assertThat(toolResult).isNotNull().satisfies(result -> {
|
||||
|
||||
assertThat(result.content()).hasSize(1);
|
||||
|
||||
TextContent content = (TextContent) result.content().get(0);
|
||||
|
||||
assertThat(content).isNotNull();
|
||||
assertThat(content.text()).isNotNull();
|
||||
assertThat(content.text()).contains("7");
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPing() {
|
||||
assertThatCode(() -> mcpSyncClient.ping()).doesNotThrowAnyException();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCallTool() {
|
||||
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE));
|
||||
|
||||
CallToolResult callToolResult = mcpSyncClient.callTool(callToolRequest);
|
||||
|
||||
assertThat(callToolResult).isNotNull().satisfies(result -> {
|
||||
assertThat(result.content()).isNotNull();
|
||||
assertThat(result.isError()).isNull();
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCallToolWithInvalidTool() {
|
||||
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", TEST_MESSAGE));
|
||||
|
||||
assertThatThrownBy(() -> mcpSyncClient.callTool(invalidRequest)).isInstanceOf(Exception.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testRootsListChanged() {
|
||||
assertThatCode(() -> mcpSyncClient.sendRootsListChanged()).doesNotThrowAnyException();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testListResources() {
|
||||
ListResourcesResult resources = mcpSyncClient.listResources(null);
|
||||
|
||||
assertThat(resources).isNotNull().satisfies(result -> {
|
||||
assertThat(result.resources()).isNotNull();
|
||||
|
||||
if (!result.resources().isEmpty()) {
|
||||
Resource firstResource = result.resources().get(0);
|
||||
assertThat(firstResource.uri()).isNotNull();
|
||||
assertThat(firstResource.name()).isNotNull();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void testClientSessionState() {
|
||||
assertThat(mcpSyncClient).isNotNull();
|
||||
}
|
||||
|
||||
}
|
||||
Some files were not shown because too many files have changed in this diff Show more
Loading…
Reference in a new issue