diff --git a/mcp/src/main/java/org/springframework/ai/mcp/client/McpAsyncClient.java b/mcp/src/main/java/org/springframework/ai/mcp/client/McpAsyncClient.java index 22d0804..5d82881 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/client/McpAsyncClient.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/client/McpAsyncClient.java @@ -517,7 +517,7 @@ public class McpAsyncClient { return this.mcpSession.sendNotification(McpSchema.METHOD_NOTIFICATION_ROOTS_LIST_CHANGED); } - private RequestHandler rootsListRequestHandler() { + private RequestHandler rootsListRequestHandler() { return params -> { McpSchema.PaginatedRequest request = transport.unmarshalFrom(params, new TypeReference() { @@ -545,10 +545,10 @@ public class McpAsyncClient { // -------------------------- // Tools // -------------------------- - private static TypeReference CALL_TOOL_RESULT_TYPE_REF = new TypeReference<>() { + private static final TypeReference CALL_TOOL_RESULT_TYPE_REF = new TypeReference<>() { }; - private static TypeReference LIST_TOOLS_RESULT_TYPE_REF = new TypeReference<>() { + private static final TypeReference LIST_TOOLS_RESULT_TYPE_REF = new TypeReference<>() { }; /** @@ -601,6 +601,7 @@ public class McpAsyncClient { * list to all registered consumers 3. Handling any errors that occur during this * process */ + @Deprecated private NotificationHandler toolsChangeNotificationHandler( List>> toolsChangeConsumers) { @@ -631,6 +632,10 @@ public class McpAsyncClient { // TODO: params are not used yet return params -> listTools().flatMap(listToolsResult -> Flux.fromIterable(toolsChangeConsumers) .flatMap(consumer -> consumer.apply(listToolsResult.tools())) + .onErrorResume(error -> { + logger.error("Error handling tools list change notification", error); + return Mono.empty(); + }) .then()); } @@ -638,13 +643,13 @@ public class McpAsyncClient { // Resources // -------------------------- - private static TypeReference LIST_RESOURCES_RESULT_TYPE_REF = new TypeReference<>() { + private static final TypeReference LIST_RESOURCES_RESULT_TYPE_REF = new TypeReference<>() { }; - private static TypeReference READ_RESOURCE_RESULT_TYPE_REF = new TypeReference<>() { + private static final TypeReference READ_RESOURCE_RESULT_TYPE_REF = new TypeReference<>() { }; - private static TypeReference LIST_RESOURCE_TEMPLATES_RESULT_TYPE_REF = new TypeReference<>() { + private static final TypeReference LIST_RESOURCE_TEMPLATES_RESULT_TYPE_REF = new TypeReference<>() { }; /** @@ -742,6 +747,7 @@ public class McpAsyncClient { VOID_TYPE_REFERENCE); } + @Deprecated private NotificationHandler resourcesChangeNotificationHandler( List>> resourcesChangeConsumers) { @@ -759,16 +765,20 @@ public class McpAsyncClient { List, Mono>> resourcesChangeConsumers) { return params -> listResources().flatMap(listResourcesResult -> Flux.fromIterable(resourcesChangeConsumers) .flatMap(consumer -> consumer.apply(listResourcesResult.resources())) + .onErrorResume(error -> { + logger.error("Error handling resources list change notification", error); + return Mono.empty(); + }) .then()); } // -------------------------- // Prompts // -------------------------- - private static TypeReference LIST_PROMPTS_RESULT_TYPE_REF = new TypeReference<>() { + private static final TypeReference LIST_PROMPTS_RESULT_TYPE_REF = new TypeReference<>() { }; - private static TypeReference GET_PROMPT_RESULT_TYPE_REF = new TypeReference<>() { + private static final TypeReference GET_PROMPT_RESULT_TYPE_REF = new TypeReference<>() { }; /** @@ -808,6 +818,7 @@ public class McpAsyncClient { return this.mcpSession.sendNotification(McpSchema.METHOD_NOTIFICATION_PROMPTS_LIST_CHANGED); } + @Deprecated private NotificationHandler promptsChangeNotificationHandler( List>> promptsChangeConsumers) { @@ -827,6 +838,10 @@ public class McpAsyncClient { List, Mono>> promptsChangeConsumers) { return params -> listPrompts().flatMap(listPromptsResult -> Flux.fromIterable(promptsChangeConsumers) .flatMap(consumer -> consumer.apply(listPromptsResult.prompts())) + .onErrorResume(error -> { + logger.error("Error handling prompts list change notification", error); + return Mono.empty(); + }) .then()); } diff --git a/mcp/src/main/java/org/springframework/ai/mcp/client/McpClient.java b/mcp/src/main/java/org/springframework/ai/mcp/client/McpClient.java index 69a811e..e11a9a7 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/client/McpClient.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/client/McpClient.java @@ -65,7 +65,7 @@ import org.springframework.ai.mcp.util.Assert; * .requestTimeout(Duration.ofSeconds(5)) * .build(); * } - * + * * Example of creating a basic asynchronous client:
{@code
  * McpClient.async(transport)
  *     .requestTimeout(Duration.ofSeconds(5))
@@ -114,6 +114,40 @@ import org.springframework.ai.mcp.util.Assert;
  */
 public interface McpClient {
 
+	/**
+	 * Start building a synchronous MCP client with the specified transport layer. The
+	 * synchronous MCP client provides blocking operations. Synchronous clients wait for
+	 * each operation to complete before returning, making them simpler to use but
+	 * potentially less performant for concurrent operations. The transport layer handles
+	 * the low-level communication between client and server using protocols like stdio or
+	 * Server-Sent Events (SSE).
+	 * @param transport The transport layer implementation for MCP communication. Common
+	 * implementations include {@code StdioClientTransport} for stdio-based communication
+	 * and {@code SseClientTransport} for SSE-based communication.
+	 * @return A new builder instance for configuring the client
+	 * @throws IllegalArgumentException if transport is null
+	 */
+	static SyncSpec sync(ClientMcpTransport transport) {
+		return new SyncSpec(transport);
+	}
+
+	/**
+	 * Start building an asynchronous MCP client with the specified transport layer. The
+	 * asynchronous MCP client provides non-blocking operations. Asynchronous clients
+	 * return reactive primitives (Mono/Flux) immediately, allowing for concurrent
+	 * operations and reactive programming patterns. The transport layer handles the
+	 * low-level communication between client and server using protocols like stdio or
+	 * Server-Sent Events (SSE).
+	 * @param transport The transport layer implementation for MCP communication. Common
+	 * implementations include {@code StdioClientTransport} for stdio-based communication
+	 * and {@code SseClientTransport} for SSE-based communication.
+	 * @return A new builder instance for configuring the client
+	 * @throws IllegalArgumentException if transport is null
+	 */
+	static AsyncSpec async(ClientMcpTransport transport) {
+		return new AsyncSpec(transport);
+	}
+
 	/**
 	 * Start building an MCP client with the specified transport layer. The transport
 	 * layer handles the low-level communication between client and server using protocols
@@ -371,7 +405,20 @@ public interface McpClient {
 	}
 
 	/**
-	 * Synchronous client specification.
+	 * Synchronous client specification. This class follows the builder pattern to provide
+	 * a fluent API for setting up clients with custom configurations.
+	 *
+	 * 

+ * The builder supports configuration of: + *

    + *
  • Transport layer for client-server communication + *
  • Request timeouts for operation boundaries + *
  • Client capabilities for feature negotiation + *
  • Client implementation details for version tracking + *
  • Root URIs for resource access + *
  • Change notification handlers for tools, resources, and prompts + *
  • Custom message sampling logic + *
*/ class SyncSpec { @@ -381,21 +428,22 @@ public interface McpClient { private ClientCapabilities capabilities; - private Implementation clientInfo = new Implementation("Spring AI MCP Client", "0.3.1"); + private Implementation clientInfo = new Implementation("Spring AI MCP Client", "1.0.0"); - private Map roots = new HashMap<>(); + private final Map roots = new HashMap<>(); - private List>> toolsChangeConsumers = new ArrayList<>(); + private final List>> toolsChangeConsumers = new ArrayList<>(); - private List>> resourcesChangeConsumers = new ArrayList<>(); + private final List>> resourcesChangeConsumers = new ArrayList<>(); - private List>> promptsChangeConsumers = new ArrayList<>(); + private final List>> promptsChangeConsumers = new ArrayList<>(); - private List> loggingConsumers = new ArrayList<>(); + private final List> loggingConsumers = new ArrayList<>(); private Function samplingHandler; private SyncSpec(ClientMcpTransport transport) { + Assert.notNull(transport, "Transport must not be null"); this.transport = transport; } @@ -581,7 +629,20 @@ public interface McpClient { } /** - * Asynchronous client specification. + * Asynchronous client specification. This class follows the builder pattern to + * provide a fluent API for setting up clients with custom configurations. + * + *

+ * The builder supports configuration of: + *

    + *
  • Transport layer for client-server communication + *
  • Request timeouts for operation boundaries + *
  • Client capabilities for feature negotiation + *
  • Client implementation details for version tracking + *
  • Root URIs for resource access + *
  • Change notification handlers for tools, resources, and prompts + *
  • Custom message sampling logic + *
*/ class AsyncSpec { @@ -593,19 +654,20 @@ public interface McpClient { private Implementation clientInfo = new Implementation("Spring AI MCP Client", "0.3.1"); - private Map roots = new HashMap<>(); + private final Map roots = new HashMap<>(); - private List, Mono>> toolsChangeConsumers = new ArrayList<>(); + private final List, Mono>> toolsChangeConsumers = new ArrayList<>(); - private List, Mono>> resourcesChangeConsumers = new ArrayList<>(); + private final List, Mono>> resourcesChangeConsumers = new ArrayList<>(); - private List, Mono>> promptsChangeConsumers = new ArrayList<>(); + private final List, Mono>> promptsChangeConsumers = new ArrayList<>(); - private List>> loggingConsumers = new ArrayList<>(); + private final List>> loggingConsumers = new ArrayList<>(); private Function> samplingHandler; private AsyncSpec(ClientMcpTransport transport) { + Assert.notNull(transport, "Transport must not be null"); this.transport = transport; } @@ -789,12 +851,4 @@ public interface McpClient { } - static SyncSpec sync(ClientMcpTransport transport) { - return new SyncSpec(transport); - } - - static AsyncSpec async(ClientMcpTransport transport) { - return new AsyncSpec(transport); - } - } diff --git a/mcp/src/main/java/org/springframework/ai/mcp/client/McpClientFeatures.java b/mcp/src/main/java/org/springframework/ai/mcp/client/McpClientFeatures.java index 9277398..7d8bb17 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/client/McpClientFeatures.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/client/McpClientFeatures.java @@ -16,8 +16,8 @@ import org.springframework.ai.mcp.util.Assert; import org.springframework.ai.mcp.util.Utils; /** - * Internal representation of features and capabilities for Model Context Protocol (MCP) - * clients. This class provides two record types for managing client features: + * Representation of features and capabilities for Model Context Protocol (MCP) clients. + * This class provides two record types for managing client features: *
    *
  • {@link Async} for non-blocking operations with Project Reactor's Mono responses *
  • {@link Sync} for blocking operations with direct responses @@ -38,6 +38,7 @@ import org.springframework.ai.mcp.util.Utils; * through the {@link Async#fromSync} method, which ensures proper handling of blocking * operations in non-blocking contexts by scheduling them on a bounded elastic scheduler. * + * @author Dariusz Jędrzejczyk * @see McpClient * @see McpSchema.Implementation * @see McpSchema.ClientCapabilities @@ -92,10 +93,10 @@ class McpClientFeatures { samplingHandler != null ? new McpSchema.ClientCapabilities.Sampling() : null); this.roots = roots != null ? new ConcurrentHashMap<>(roots) : new ConcurrentHashMap<>(); - this.toolsChangeConsumers = toolsChangeConsumers; - this.resourcesChangeConsumers = resourcesChangeConsumers; - this.promptsChangeConsumers = promptsChangeConsumers; - this.loggingConsumers = loggingConsumers; + this.toolsChangeConsumers = toolsChangeConsumers != null ? toolsChangeConsumers : List.of(); + this.resourcesChangeConsumers = resourcesChangeConsumers != null ? resourcesChangeConsumers : List.of(); + this.promptsChangeConsumers = promptsChangeConsumers != null ? promptsChangeConsumers : List.of(); + this.loggingConsumers = loggingConsumers != null ? loggingConsumers : List.of(); this.samplingHandler = samplingHandler; } @@ -189,10 +190,10 @@ class McpClientFeatures { samplingHandler != null ? new McpSchema.ClientCapabilities.Sampling() : null); this.roots = roots != null ? new HashMap<>(roots) : new HashMap<>(); - this.toolsChangeConsumers = toolsChangeConsumers; - this.resourcesChangeConsumers = resourcesChangeConsumers; - this.promptsChangeConsumers = promptsChangeConsumers; - this.loggingConsumers = loggingConsumers; + this.toolsChangeConsumers = toolsChangeConsumers != null ? toolsChangeConsumers : List.of(); + this.resourcesChangeConsumers = resourcesChangeConsumers != null ? resourcesChangeConsumers : List.of(); + this.promptsChangeConsumers = promptsChangeConsumers != null ? promptsChangeConsumers : List.of(); + this.loggingConsumers = loggingConsumers != null ? loggingConsumers : List.of(); this.samplingHandler = samplingHandler; } } diff --git a/mcp/src/main/java/org/springframework/ai/mcp/client/McpSyncClient.java b/mcp/src/main/java/org/springframework/ai/mcp/client/McpSyncClient.java index 954f7aa..1d008ab 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/client/McpSyncClient.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/client/McpSyncClient.java @@ -21,6 +21,7 @@ import java.time.Duration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.ai.mcp.spec.ClientMcpTransport; import org.springframework.ai.mcp.spec.McpSchema; import org.springframework.ai.mcp.spec.McpSchema.ClientCapabilities; import org.springframework.ai.mcp.spec.McpSchema.GetPromptRequest; @@ -74,6 +75,14 @@ public class McpSyncClient implements AutoCloseable { private final McpAsyncClient delegate; + /** + * Create a new McpSyncClient with the given delegate. + * @param delegate the asynchronous kernel on top of which this synchronous client + * provides a blocking API. + * @deprecated Use {@link McpClient#sync(ClientMcpTransport)} to obtain an instance. + */ + @Deprecated + // TODO make the constructor package private post-deprecation public McpSyncClient(McpAsyncClient delegate) { Assert.notNull(delegate, "The delegate can not be null"); this.delegate = delegate; diff --git a/mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java b/mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java index d5e3043..8d406a5 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java @@ -84,6 +84,7 @@ import org.springframework.ai.mcp.util.Utils; * connected clients of changes when configured to do so. * * @author Christian Tzolov + * @author Dariusz Jędrzejczyk * @see McpServer * @see McpSchema * @see DefaultMcpSession @@ -137,7 +138,7 @@ public class McpAsyncServer { Map> requestHandlers = new HashMap<>(); // Initialize request handlers for standard MCP methods - requestHandlers.put(McpSchema.METHOD_INITIALIZE, initializeRequestHandler()); + requestHandlers.put(McpSchema.METHOD_INITIALIZE, asyncInitializeRequestHandler()); // Ping MUST respond with an empty data, but not NULL response. requestHandlers.put(McpSchema.METHOD_PING, (params) -> Mono.just("")); @@ -287,6 +288,28 @@ public class McpAsyncServer { // --------------------------------------- // Lifecycle Management // --------------------------------------- + private DefaultMcpSession.RequestHandler asyncInitializeRequestHandler() { + return params -> { + McpSchema.InitializeRequest initializeRequest = transport.unmarshalFrom(params, + new TypeReference() { + }); + this.clientCapabilities = initializeRequest.capabilities(); + this.clientInfo = initializeRequest.clientInfo(); + logger.info("Client initialize request - Protocol: {}, Capabilities: {}, Info: {}", + initializeRequest.protocolVersion(), initializeRequest.capabilities(), + initializeRequest.clientInfo()); + + if (!McpSchema.LATEST_PROTOCOL_VERSION.equals(initializeRequest.protocolVersion())) { + return Mono.error(new McpError( + "Unsupported protocol version from client: " + initializeRequest.protocolVersion())); + } + + return Mono.just(new McpSchema.InitializeResult(McpSchema.LATEST_PROTOCOL_VERSION, this.serverCapabilities, + this.serverInfo, null)); + }; + } + + @Deprecated private DefaultMcpSession.RequestHandler initializeRequestHandler() { return params -> { McpSchema.InitializeRequest initializeRequest = transport.unmarshalFrom(params, @@ -360,7 +383,7 @@ public class McpAsyncServer { this.mcpSession.close(); } - private static TypeReference LIST_ROOTS_RESULT_TYPE_REF = new TypeReference<>() { + private static final TypeReference LIST_ROOTS_RESULT_TYPE_REF = new TypeReference<>() { }; /** @@ -399,6 +422,10 @@ public class McpAsyncServer { List, Mono>> rootsChangeConsumers) { return params -> listRoots().flatMap(listRootsResult -> Flux.fromIterable(rootsChangeConsumers) .flatMap(consumer -> consumer.apply(listRootsResult.roots())) + .onErrorResume(error -> { + logger.error("Error handling roots list change notification", error); + return Mono.empty(); + }) .then()); } @@ -654,7 +681,7 @@ public class McpAsyncServer { var resourceUri = resourceRequest.uri(); McpServerFeatures.AsyncResourceRegistration registration = this.resources.get(resourceUri); if (registration != null) { - return this.resources.get(resourceUri).readHandler().apply(resourceRequest); + return registration.readHandler().apply(resourceRequest); } return Mono.error(new McpError("Resource not found: " + resourceUri)); }; diff --git a/mcp/src/main/java/org/springframework/ai/mcp/server/McpServer.java b/mcp/src/main/java/org/springframework/ai/mcp/server/McpServer.java index 1fbf47b..615fd51 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/server/McpServer.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/server/McpServer.java @@ -120,12 +120,37 @@ import org.springframework.ai.mcp.util.Assert; * }
* * @author Christian Tzolov + * @author Dariusz Jędrzejczyk * @see McpAsyncServer * @see McpSyncServer * @see McpTransport */ public interface McpServer { + /** + * Starts building a synchronous MCP server that provides blocking operations. + * Synchronous servers process each request to completion before handling the next + * one, making them simpler to implement but potentially less performant for + * concurrent operations. + * @param transport The transport layer implementation for MCP communication + * @return A new instance of {@link SyncSpec} for configuring the server. + */ + static SyncSpec sync(ServerMcpTransport transport) { + return new SyncSpec(transport); + } + + /** + * Starts building an asynchronous MCP server that provides blocking operations. + * Asynchronous servers can handle multiple requests concurrently using a functional + * paradigm with non-blocking server transports, making them more efficient for + * high-concurrency scenarios but more complex to implement. + * @param transport The transport layer implementation for MCP communication + * @return A new instance of {@link SyncSpec} for configuring the server. + */ + static AsyncSpec async(ServerMcpTransport transport) { + return new AsyncSpec(transport); + } + /** * Start building an MCP server with the specified transport. * @param transport The transport layer implementation for MCP communication @@ -170,7 +195,7 @@ public interface McpServer { */ private final Map resources = new HashMap<>(); - private List resourceTemplates = new ArrayList<>(); + private final List resourceTemplates = new ArrayList<>(); /** * The Model Context Protocol (MCP) provides a standardized way for servers to @@ -547,7 +572,7 @@ public interface McpServer { */ private final Map resources = new HashMap<>(); - private List resourceTemplates = new ArrayList<>(); + private final List resourceTemplates = new ArrayList<>(); /** * The Model Context Protocol (MCP) provides a standardized way for servers to @@ -556,9 +581,9 @@ public interface McpServer { * discover available prompts, retrieve their contents, and provide arguments to * customize them. */ - private Map prompts = new HashMap<>(); + private final Map prompts = new HashMap<>(); - private List>> rootsChangeConsumers = new ArrayList<>(); + private final List>> rootsChangeConsumers = new ArrayList<>(); private SyncSpec(ServerMcpTransport transport) { Assert.notNull(transport, "Transport must not be null"); @@ -895,28 +920,6 @@ public interface McpServer { } - /** - * Builds a synchronous MCP server that provides blocking operations. Synchronous - * servers process each request to completion before handling the next one, making - * them simpler to implement but potentially less performant for concurrent - * operations. - * @return A new instance of {@link SyncSpec} for configuring the server. - */ - static SyncSpec sync(ServerMcpTransport transport) { - return new SyncSpec(transport); - } - - /** - * Builds an asynchronous MCP server that provides non-blocking operations. - * Asynchronous servers can handle multiple requests concurrently using a functional - * paradigm with non-blocking server transports, making them more efficient for - * high-concurrency scenarios but more complex to implement. - * @return A new instance of {@link SyncSpec} for configuring the server. - */ - static AsyncSpec async(ServerMcpTransport transport) { - return new AsyncSpec(transport); - } - /** * Builder class for creating MCP servers with custom configuration. * diff --git a/mcp/src/main/java/org/springframework/ai/mcp/server/McpServerFeatures.java b/mcp/src/main/java/org/springframework/ai/mcp/server/McpServerFeatures.java index 0faba95..2dd4e66 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/server/McpServerFeatures.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/server/McpServerFeatures.java @@ -16,6 +16,8 @@ import org.springframework.ai.mcp.util.Utils; /** * MCP server features specification that a particular server can choose to support. + * + * @author Dariusz Jędrzejczyk */ public class McpServerFeatures { diff --git a/mcp/src/main/java/org/springframework/ai/mcp/server/McpSyncServer.java b/mcp/src/main/java/org/springframework/ai/mcp/server/McpSyncServer.java index 97c949f..004f7a3 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/server/McpSyncServer.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/server/McpSyncServer.java @@ -60,6 +60,7 @@ import org.springframework.ai.mcp.util.Assert; * connected clients of changes when configured to do so. * * @author Christian Tzolov + * @author Dariusz Jędrzejczyk * @see McpAsyncServer * @see McpSchema */