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);
 
@@ -144,7 +144,7 @@ public void close() {
 	 * @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) {
 		return this.delegate.addTool(toolSpecification);
 	}
 
@@ -173,7 +173,7 @@ public Mono<Void> notifyToolsListChanged() {
 	 * @param resourceHandler The resource handler to add
 	 * @return Mono that completes when clients have been notified of the change
 	 */
-	public Mono<Void> addResource(McpServerFeatures.AsyncResourceSpecification resourceHandler) {
+	public Mono<Void> addResource(McpServer.AsyncResourceSpecification resourceHandler) {
 		return this.delegate.addResource(resourceHandler);
 	}
 
@@ -202,7 +202,7 @@ public Mono<Void> notifyResourcesListChanged() {
 	 * @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) {
 		return this.delegate.addPrompt(promptSpecification);
 	}
 
@@ -251,7 +251,7 @@ public Mono<Void> loggingNotification(LoggingMessageNotification loggingMessageN
 	 * code.
 	 * @param protocolVersions the Client supported protocol versions.
 	 */
-	void setProtocolVersions(List<String> protocolVersions) {
+	public void setProtocolVersions(List<String> protocolVersions) {
 		this.delegate.setProtocolVersions(protocolVersions);
 	}
 
@@ -267,13 +267,13 @@ private static class AsyncServerImpl extends 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.
@@ -283,7 +283,7 @@ private static class AsyncServerImpl extends McpAsyncServer {
 
 		private List<String> protocolVersions = List.of(McpSchema.LATEST_PROTOCOL_VERSION);
 
-		private McpUriTemplateManagerFactory uriTemplateManagerFactory = new DeafaultMcpUriTemplateManagerFactory();
+		private final McpUriTemplateManagerFactory uriTemplateManagerFactory;
 
 		AsyncServerImpl(McpServerTransportProvider mcpTransportProvider, ObjectMapper objectMapper,
 				Duration requestTimeout, McpServerFeatures.Async features,
@@ -432,7 +432,7 @@ private McpServerSession.NotificationHandler asyncRootsListChangedNotificationHa
 		// ---------------------------------------
 
 		@Override
-		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"));
 			}
@@ -494,7 +494,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));
 			};
@@ -505,15 +505,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())));
 			};
 		}
@@ -523,7 +524,7 @@ private McpServerSession.RequestHandler<CallToolResult> toolsCallRequestHandler(
 		// ---------------------------------------
 
 		@Override
-		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"));
 			}
@@ -555,7 +556,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()) {
@@ -577,7 +578,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));
 			};
@@ -613,15 +614,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));
 			};
 		}
 
@@ -630,7 +631,7 @@ private McpServerSession.RequestHandler<McpSchema.ReadResourceResult> resourcesR
 		// ---------------------------------------
 
 		@Override
-		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"));
 			}
@@ -639,7 +640,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(new McpError(
@@ -668,7 +669,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);
@@ -698,7 +699,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));
@@ -711,12 +712,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));
 			};
 		}
 
@@ -775,25 +776,19 @@ 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());
+					McpServer.AsyncResourceSpecification resourceSpec = this.resources.get(resourceReference.uri());
 					if (resourceSpec == null) {
 						return Mono.error(new McpError("Resource not found: " + resourceReference.uri()));
 					}
@@ -855,7 +850,7 @@ private McpSchema.CompleteRequest parseCompletionParams(Object object) {
 		// ---------------------------------------
 
 		@Override
-		void setProtocolVersions(List<String> 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;
 	}