feat: extracts McpClient/McpServer abstraction into mcp-spi

Author: Aliaksie

mcp-reactor/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java

@@ -72,19 +72,19 @@
  * @author Dariusz Jędrzejczyk
  * @author Christian Tzolov
  * @author Jihoon Kim
- * @see McpClient
+ * @see McpClientFactory
  * @see McpSchema
  * @see McpClientSession
  */
-public class McpAsyncClient {
+public class McpAsyncClient implements McpClient {
 
 	private static final Logger logger = LoggerFactory.getLogger(McpAsyncClient.class);
 
 	private static final McpType<Void> VOID_TYPE_REFERENCE = McpType.of(Void.class);
 
 	protected final Sinks.One<McpSchema.InitializeResult> initializedSink = Sinks.one();
 
-	private AtomicBoolean initialized = new AtomicBoolean(false);
+	private final AtomicBoolean initialized = new AtomicBoolean(false);
 
 	/**
 	 * The max timeout to await for the client-server connection to be initialized.
@@ -808,7 +808,7 @@ public Mono<Void> setLoggingLevel(LoggingLevel loggingLevel) {
 	 * code.
 	 * @param protocolVersions the Client supported protocol versions.
 	 */
-	void setProtocolVersions(List<String> protocolVersions) {
+	public void setProtocolVersions(List<String> protocolVersions) {
 		this.protocolVersions = protocolVersions;
 
 	}

mcp-reactor/src/main/java/io/modelcontextprotocol/client/McpClient.java renamed to mcp-reactor/src/main/java/io/modelcontextprotocol/client/McpClientFactory.java

@@ -99,7 +99,7 @@
  * @see McpSyncClient
  * @see McpTransport
  */
-public interface McpClient {
+public interface McpClientFactory {
 
 	/**
 	 * Start building a synchronous MCP client with the specified transport layer. The

mcp-reactor/src/main/java/io/modelcontextprotocol/client/McpClientFeatures.java

@@ -42,7 +42,7 @@
  * operations in non-blocking contexts by scheduling them on a bounded elastic scheduler.
  *
  * @author Dariusz Jędrzejczyk
- * @see McpClient
+ * @see McpClientFactory
  * @see McpSchema.Implementation
  * @see McpSchema.ClientCapabilities
  */

mcp-reactor/src/main/java/io/modelcontextprotocol/client/McpSyncClient.java

@@ -47,7 +47,7 @@
  * @author Dariusz Jędrzejczyk
  * @author Christian Tzolov
  * @author Jihoon Kim
- * @see McpClient
+ * @see McpClientFactory
  * @see McpAsyncClient
  * @see McpSchema
  */

mcp-reactor/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.java

@@ -77,11 +77,11 @@
  * @author Christian Tzolov
  * @author Dariusz Jędrzejczyk
  * @author Jihoon Kim
- * @see McpServer
+ * @see McpServerFactory
  * @see McpSchema
  * @see McpClientSession
  */
-public class McpAsyncServer {
+public class McpAsyncServer implements McpServer {
 
 	private static final Logger logger = LoggerFactory.getLogger(McpAsyncServer.class);
 
@@ -95,13 +95,13 @@ public class McpAsyncServer {
 
 	private final String instructions;
 
-	private final CopyOnWriteArrayList<McpServerFeatures.AsyncToolSpecification> tools = new CopyOnWriteArrayList<>();
+	private final CopyOnWriteArrayList<McpServer.AsyncToolSpecification> tools = new CopyOnWriteArrayList<>();
 
 	private final CopyOnWriteArrayList<McpSchema.ResourceTemplate> resourceTemplates = new CopyOnWriteArrayList<>();
 
-	private final ConcurrentHashMap<String, McpServerFeatures.AsyncResourceSpecification> resources = new ConcurrentHashMap<>();
+	private final ConcurrentHashMap<String, McpServer.AsyncResourceSpecification> resources = new ConcurrentHashMap<>();
 
-	private final ConcurrentHashMap<String, McpServerFeatures.AsyncPromptSpecification> prompts = new ConcurrentHashMap<>();
+	private final ConcurrentHashMap<String, McpServer.AsyncPromptSpecification> prompts = new ConcurrentHashMap<>();
 
 	// FIXME: this field is deprecated and should be remvoed together with the
 	// broadcasting loggingNotification.
@@ -111,7 +111,7 @@ public class McpAsyncServer {
 
 	private List<String> protocolVersions = List.of(McpSchema.LATEST_PROTOCOL_VERSION);
 
-	private McpUriTemplateManagerFactory uriTemplateManagerFactory = new DeafaultMcpUriTemplateManagerFactory();
+	private final McpUriTemplateManagerFactory uriTemplateManagerFactory;
 
 	/**
 	 * Create a new McpAsyncServer with the given transport provider and capabilities.
@@ -282,7 +282,7 @@ private McpServerSession.NotificationHandler asyncRootsListChangedNotificationHa
 	 * @param toolSpecification The tool specification to add
 	 * @return Mono that completes when clients have been notified of the change
 	 */
-	public Mono<Void> addTool(McpServerFeatures.AsyncToolSpecification toolSpecification) {
+	public Mono<Void> addTool(McpServer.AsyncToolSpecification toolSpecification) {
 		if (toolSpecification == null) {
 			return Mono.error(new McpError("Tool specification must not be null"));
 		}
@@ -351,7 +351,7 @@ public Mono<Void> notifyToolsListChanged() {
 
 	private McpServerSession.RequestHandler<McpSchema.ListToolsResult> toolsListRequestHandler() {
 		return (exchange, params) -> {
-			List<Tool> tools = this.tools.stream().map(McpServerFeatures.AsyncToolSpecification::tool).toList();
+			List<Tool> tools = this.tools.stream().map(McpServer.AsyncToolSpecification::tool).toList();
 
 			return Mono.just(new McpSchema.ListToolsResult(tools, null));
 		};
@@ -362,15 +362,16 @@ private McpServerSession.RequestHandler<CallToolResult> toolsCallRequestHandler(
 			McpSchema.CallToolRequest callToolRequest = schemaCodec.decodeResult(params,
 					McpType.of(McpSchema.CallToolRequest.class));
 
-			Optional<McpServerFeatures.AsyncToolSpecification> toolSpecification = this.tools.stream()
+			Optional<McpServer.AsyncToolSpecification> toolSpecification = this.tools.stream()
 				.filter(tr -> callToolRequest.name().equals(tr.tool().name()))
 				.findAny();
 
 			if (toolSpecification.isEmpty()) {
 				return Mono.error(new McpError("Tool not found: " + callToolRequest.name()));
 			}
 
-			return toolSpecification.map(tool -> tool.call().apply(exchange, callToolRequest.arguments()))
+			return toolSpecification
+					.map(tool -> Mono.from(tool.call().apply(exchange, callToolRequest.arguments())))
 				.orElse(Mono.error(new McpError("Tool not found: " + callToolRequest.name())));
 		};
 	}
@@ -384,7 +385,7 @@ private McpServerSession.RequestHandler<CallToolResult> toolsCallRequestHandler(
 	 * @param resourceSpecification The resource handler to add
 	 * @return Mono that completes when clients have been notified of the change
 	 */
-	public Mono<Void> addResource(McpServerFeatures.AsyncResourceSpecification resourceSpecification) {
+	public Mono<Void> addResource(McpServer.AsyncResourceSpecification resourceSpecification) {
 		if (resourceSpecification == null || resourceSpecification.resource() == null) {
 			return Mono.error(new McpError("Resource must not be null"));
 		}
@@ -420,7 +421,7 @@ public Mono<Void> removeResource(String resourceUri) {
 		}
 
 		return Mono.defer(() -> {
-			McpServerFeatures.AsyncResourceSpecification removed = this.resources.remove(resourceUri);
+			McpServer.AsyncResourceSpecification removed = this.resources.remove(resourceUri);
 			if (removed != null) {
 				logger.debug("Removed resource handler: {}", resourceUri);
 				if (this.serverCapabilities.resources().listChanged()) {
@@ -445,7 +446,7 @@ private McpServerSession.RequestHandler<McpSchema.ListResourcesResult> resources
 		return (exchange, params) -> {
 			var resourceList = this.resources.values()
 				.stream()
-				.map(McpServerFeatures.AsyncResourceSpecification::resource)
+				.map(McpServer.AsyncResourceSpecification::resource)
 				.toList();
 			return Mono.just(new McpSchema.ListResourcesResult(resourceList, null));
 		};
@@ -481,15 +482,15 @@ private McpServerSession.RequestHandler<McpSchema.ReadResourceResult> resourcesR
 					McpType.of(McpSchema.ReadResourceRequest.class));
 			var resourceUri = resourceRequest.uri();
 
-			McpServerFeatures.AsyncResourceSpecification specification = this.resources.values()
+			McpServer.AsyncResourceSpecification specification = this.resources.values()
 				.stream()
 				.filter(resourceSpecification -> this.uriTemplateManagerFactory
 					.create(resourceSpecification.resource().uri())
 					.matches(resourceUri))
 				.findFirst()
 				.orElseThrow(() -> new McpError("Resource not found: " + resourceUri));
 
-			return specification.readHandler().apply(exchange, resourceRequest);
+			return Mono.from(specification.readHandler().apply(exchange, resourceRequest));
 		};
 	}
 
@@ -502,7 +503,7 @@ private McpServerSession.RequestHandler<McpSchema.ReadResourceResult> resourcesR
 	 * @param promptSpecification The prompt handler to add
 	 * @return Mono that completes when clients have been notified of the change
 	 */
-	public Mono<Void> addPrompt(McpServerFeatures.AsyncPromptSpecification promptSpecification) {
+	public Mono<Void> addPrompt(McpServer.AsyncPromptSpecification promptSpecification) {
 		if (promptSpecification == null) {
 			return Mono.error(new McpError("Prompt specification must not be null"));
 		}
@@ -511,7 +512,7 @@ public Mono<Void> addPrompt(McpServerFeatures.AsyncPromptSpecification promptSpe
 		}
 
 		return Mono.defer(() -> {
-			McpServerFeatures.AsyncPromptSpecification specification = this.prompts
+			McpServer.AsyncPromptSpecification specification = this.prompts
 				.putIfAbsent(promptSpecification.prompt().name(), promptSpecification);
 			if (specification != null) {
 				return Mono.error(
@@ -544,7 +545,7 @@ public Mono<Void> removePrompt(String promptName) {
 		}
 
 		return Mono.defer(() -> {
-			McpServerFeatures.AsyncPromptSpecification removed = this.prompts.remove(promptName);
+			McpServer.AsyncPromptSpecification removed = this.prompts.remove(promptName);
 
 			if (removed != null) {
 				logger.debug("Removed prompt handler: {}", promptName);
@@ -577,7 +578,7 @@ private McpServerSession.RequestHandler<McpSchema.ListPromptsResult> promptsList
 
 			var promptList = this.prompts.values()
 				.stream()
-				.map(McpServerFeatures.AsyncPromptSpecification::prompt)
+				.map(McpServer.AsyncPromptSpecification::prompt)
 				.toList();
 
 			return Mono.just(new McpSchema.ListPromptsResult(promptList, null));
@@ -590,12 +591,12 @@ private McpServerSession.RequestHandler<McpSchema.GetPromptResult> promptsGetReq
 					McpType.of(McpSchema.GetPromptRequest.class));
 
 			// Implement prompt retrieval logic here
-			McpServerFeatures.AsyncPromptSpecification specification = this.prompts.get(promptRequest.name());
+			McpServer.AsyncPromptSpecification specification = this.prompts.get(promptRequest.name());
 			if (specification == null) {
 				return Mono.error(new McpError("Prompt not found: " + promptRequest.name()));
 			}
 
-			return specification.promptHandler().apply(exchange, promptRequest);
+			return Mono.from(specification.promptHandler().apply(exchange, promptRequest));
 		};
 	}
 
@@ -665,31 +666,26 @@ private McpServerSession.RequestHandler<McpSchema.CompleteResult> completionComp
 
 			// check if the referenced resource exists
 			if (type.equals("ref/prompt") && request.ref() instanceof McpSchema.PromptReference promptReference) {
-				McpServerFeatures.AsyncPromptSpecification promptSpec = this.prompts.get(promptReference.name());
+				McpServer.AsyncPromptSpecification promptSpec = this.prompts.get(promptReference.name());
 				if (promptSpec == null) {
 					return Mono.error(new McpError("Prompt not found: " + promptReference.name()));
 				}
-				if (!promptSpec.prompt()
-					.arguments()
-					.stream()
-					.filter(arg -> arg.name().equals(argumentName))
-					.findFirst()
-					.isPresent()) {
+				if (promptSpec.prompt().arguments().stream().noneMatch(arg -> arg.name().equals(argumentName))) {
 
 					return Mono.error(new McpError("Argument not found: " + argumentName));
 				}
 			}
 
 			if (type.equals("ref/resource") && request.ref() instanceof McpSchema.ResourceReference resourceReference) {
-				McpServerFeatures.AsyncResourceSpecification resourceSpec = this.resources.get(resourceReference.uri());
-				if (resourceSpec == null) {
-					return Mono.error(new McpError("Resource not found: " + resourceReference.uri()));
-				}
-				if (!uriTemplateManagerFactory.create(resourceSpec.resource().uri())
-					.getVariableNames()
-					.contains(argumentName)) {
-					return Mono.error(new McpError("Argument not found: " + argumentName));
-				}
+				McpServer.AsyncResourceSpecification resourceSpec = this.resources.get(resourceReference.uri());
+					if (resourceSpec == null) {
+						return Mono.error(new McpError("Resource not found: " + resourceReference.uri()));
+					}
+					if (!uriTemplateManagerFactory.create(resourceSpec.resource().uri())
+						.getVariableNames()
+						.contains(argumentName)) {
+						return Mono.error(new McpError("Argument not found: " + argumentName));
+					}
 
 			}
 
@@ -743,8 +739,8 @@ private McpSchema.CompleteRequest parseCompletionParams(Object object) {
 	 * code.
 	 * @param protocolVersions the Client supported protocol versions.
 	 */
-	void setProtocolVersions(List<String> protocolVersions) {
-		this.protocolVersions = protocolVersions;
-	}
+	public void setProtocolVersions(List<String> protocolVersions) {
+			this.protocolVersions = protocolVersions;
+		}
 
 }

mcp-reactor/src/main/java/io/modelcontextprotocol/server/McpAsyncServerExchange.java

@@ -20,7 +20,7 @@
  * @author Dariusz Jędrzejczyk
  * @author Christian Tzolov
  */
-public class McpAsyncServerExchange {
+public class McpAsyncServerExchange implements McpServerExchange {
 
 	private final McpServerSession session;
 
@@ -136,7 +136,7 @@ public Mono<Void> loggingNotification(LoggingMessageNotification loggingMessageN
 	 * filtered out.
 	 * @param minLoggingLevel The minimum logging level
 	 */
-	void setMinLoggingLevel(LoggingLevel minLoggingLevel) {
+	public void setMinLoggingLevel(LoggingLevel minLoggingLevel) {
 		Assert.notNull(minLoggingLevel, "minLoggingLevel must not be null");
 		this.minLoggingLevel = minLoggingLevel;
 	}