Implement for client

BREAKING : Client Specification record is braking

Author: 134130

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

@@ -234,8 +234,17 @@ public class McpAsyncClient {
 		notificationHandlers.put(McpSchema.METHOD_NOTIFICATION_MESSAGE,
 				asyncLoggingNotificationHandler(loggingConsumersFinal));
 
-		this.mcpSession = new McpClientSession(requestTimeout, transport, requestHandlers, notificationHandlers);
+		// Utility Progress Notification
+		List<Function<McpSchema.ProgressNotification, Mono<Void>>> progressConsumersFinal = new ArrayList<>();
+		progressConsumersFinal
+			.add((notification) -> Mono.fromRunnable(() -> logger.debug("Progress: {}", notification)));
+		if (!Utils.isEmpty(features.progressConsumers())) {
+			progressConsumersFinal.addAll(features.progressConsumers());
+		}
+		notificationHandlers.put(McpSchema.METHOD_NOTIFICATION_PROGRESS,
+				asyncProgressNotificationHandler(progressConsumersFinal));
 
+		this.mcpSession = new McpClientSession(requestTimeout, transport, requestHandlers, notificationHandlers);
 	}
 
 	/**
@@ -789,6 +798,20 @@ private NotificationHandler asyncLoggingNotificationHandler(
 		};
 	}
 
+	private NotificationHandler asyncProgressNotificationHandler(
+			List<Function<McpSchema.ProgressNotification, Mono<Void>>> progressConsumers) {
+
+		return params -> {
+			McpSchema.ProgressNotification progressNotification = transport.unmarshalFrom(params,
+					new TypeReference<McpSchema.ProgressNotification>() {
+					});
+
+			return Flux.fromIterable(progressConsumers)
+				.flatMap(consumer -> consumer.apply(progressNotification))
+				.then();
+		};
+	}
+
 	/**
 	 * Sets the minimum logging level for messages received from the server. The client
 	 * will only receive log messages at or above the specified severity level.

mcp/src/main/java/io/modelcontextprotocol/client/McpClient.java

@@ -173,6 +173,8 @@ class SyncSpec {
 
 		private final List<Consumer<McpSchema.LoggingMessageNotification>> loggingConsumers = new ArrayList<>();
 
+		private final List<Consumer<McpSchema.ProgressNotification>> progressConsumers = new ArrayList<>();
+
 		private Function<CreateMessageRequest, CreateMessageResult> samplingHandler;
 
 		private SyncSpec(McpClientTransport transport) {
@@ -356,6 +358,36 @@ public SyncSpec loggingConsumers(List<Consumer<McpSchema.LoggingMessageNotificat
 			return this;
 		}
 
+		/**
+		 * Adds a consumer to be notified of progress notifications from the server. This
+		 * allows the client to track long-running operations and provide feedback to
+		 * users.
+		 * @param progressConsumer A consumer that receives progress notifications. Must
+		 * not be null.
+		 * @return This builder instance for method chaining
+		 * @throws IllegalArgumentException if progressConsumer is null
+		 */
+		public SyncSpec progressConsumer(Consumer<McpSchema.ProgressNotification> progressConsumer) {
+			Assert.notNull(progressConsumer, "Progress consumer must not be null");
+			this.progressConsumers.add(progressConsumer);
+			return this;
+		}
+
+		/**
+		 * Adds a multiple consumers to be notified of progress notifications from the
+		 * server. This allows the client to track long-running operations and provide
+		 * feedback to users.
+		 * @param progressConsumers A list of consumers that receives progress
+		 * notifications. Must not be null.
+		 * @return This builder instance for method chaining
+		 * @throws IllegalArgumentException if progressConsumer is null
+		 */
+		public SyncSpec progressConsumers(List<Consumer<McpSchema.ProgressNotification>> progressConsumers) {
+			Assert.notNull(progressConsumers, "Progress consumers must not be null");
+			this.progressConsumers.addAll(progressConsumers);
+			return this;
+		}
+
 		/**
 		 * Create an instance of {@link McpSyncClient} with the provided configurations or
 		 * sensible defaults.
@@ -364,7 +396,7 @@ public SyncSpec loggingConsumers(List<Consumer<McpSchema.LoggingMessageNotificat
 		public McpSyncClient build() {
 			McpClientFeatures.Sync syncFeatures = new McpClientFeatures.Sync(this.clientInfo, this.capabilities,
 					this.roots, this.toolsChangeConsumers, this.resourcesChangeConsumers, this.promptsChangeConsumers,
-					this.loggingConsumers, this.samplingHandler);
+					this.loggingConsumers, this.progressConsumers, this.samplingHandler);
 
 			McpClientFeatures.Async asyncFeatures = McpClientFeatures.Async.fromSync(syncFeatures);
 
@@ -412,6 +444,8 @@ class AsyncSpec {
 
 		private final List<Function<McpSchema.LoggingMessageNotification, Mono<Void>>> loggingConsumers = new ArrayList<>();
 
+		private final List<Function<McpSchema.ProgressNotification, Mono<Void>>> progressConsumers = new ArrayList<>();
+
 		private Function<CreateMessageRequest, Mono<CreateMessageResult>> samplingHandler;
 
 		private AsyncSpec(McpClientTransport transport) {
@@ -606,7 +640,7 @@ public McpAsyncClient build() {
 			return new McpAsyncClient(this.transport, this.requestTimeout, this.initializationTimeout,
 					new McpClientFeatures.Async(this.clientInfo, this.capabilities, this.roots,
 							this.toolsChangeConsumers, this.resourcesChangeConsumers, this.promptsChangeConsumers,
-							this.loggingConsumers, this.samplingHandler));
+							this.loggingConsumers, this.progressConsumers, this.samplingHandler));
 		}
 
 	}

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

@@ -59,13 +59,15 @@ class McpClientFeatures {
 	 * @param resourcesChangeConsumers the resources change consumers.
 	 * @param promptsChangeConsumers the prompts change consumers.
 	 * @param loggingConsumers the logging consumers.
+	 * @param progressConsumers the progress consumers.
 	 * @param samplingHandler the sampling handler.
 	 */
 	record Async(McpSchema.Implementation clientInfo, McpSchema.ClientCapabilities clientCapabilities,
 			Map<String, McpSchema.Root> roots, List<Function<List<McpSchema.Tool>, Mono<Void>>> toolsChangeConsumers,
 			List<Function<List<McpSchema.Resource>, Mono<Void>>> resourcesChangeConsumers,
 			List<Function<List<McpSchema.Prompt>, Mono<Void>>> promptsChangeConsumers,
 			List<Function<McpSchema.LoggingMessageNotification, Mono<Void>>> loggingConsumers,
+			List<Function<McpSchema.ProgressNotification, Mono<Void>>> progressConsumers,
 			Function<McpSchema.CreateMessageRequest, Mono<McpSchema.CreateMessageResult>> samplingHandler) {
 
 		/**
@@ -76,6 +78,7 @@ record Async(McpSchema.Implementation clientInfo, McpSchema.ClientCapabilities c
 		 * @param resourcesChangeConsumers the resources change consumers.
 		 * @param promptsChangeConsumers the prompts change consumers.
 		 * @param loggingConsumers the logging consumers.
+		 * @param progressConsumers the progressconsumers.
 		 * @param samplingHandler the sampling handler.
 		 */
 		public Async(McpSchema.Implementation clientInfo, McpSchema.ClientCapabilities clientCapabilities,
@@ -84,6 +87,7 @@ public Async(McpSchema.Implementation clientInfo, McpSchema.ClientCapabilities c
 				List<Function<List<McpSchema.Resource>, Mono<Void>>> resourcesChangeConsumers,
 				List<Function<List<McpSchema.Prompt>, Mono<Void>>> promptsChangeConsumers,
 				List<Function<McpSchema.LoggingMessageNotification, Mono<Void>>> loggingConsumers,
+				List<Function<McpSchema.ProgressNotification, Mono<Void>>> progressConsumers,
 				Function<McpSchema.CreateMessageRequest, Mono<McpSchema.CreateMessageResult>> samplingHandler) {
 
 			Assert.notNull(clientInfo, "Client info must not be null");
@@ -98,6 +102,7 @@ public Async(McpSchema.Implementation clientInfo, McpSchema.ClientCapabilities c
 			this.resourcesChangeConsumers = resourcesChangeConsumers != null ? resourcesChangeConsumers : List.of();
 			this.promptsChangeConsumers = promptsChangeConsumers != null ? promptsChangeConsumers : List.of();
 			this.loggingConsumers = loggingConsumers != null ? loggingConsumers : List.of();
+			this.progressConsumers = progressConsumers != null ? progressConsumers : List.of();
 			this.samplingHandler = samplingHandler;
 		}
 
@@ -135,12 +140,18 @@ public static Async fromSync(Sync syncSpec) {
 					.subscribeOn(Schedulers.boundedElastic()));
 			}
 
+			List<Function<McpSchema.ProgressNotification, Mono<Void>>> progressConsumers = new ArrayList<>();
+			for (Consumer<McpSchema.ProgressNotification> consumer : syncSpec.progressConsumers()) {
+				progressConsumers.add(p -> Mono.<Void>fromRunnable(() -> consumer.accept(p))
+					.subscribeOn(Schedulers.boundedElastic()));
+			}
+
 			Function<McpSchema.CreateMessageRequest, Mono<McpSchema.CreateMessageResult>> samplingHandler = r -> Mono
 				.fromCallable(() -> syncSpec.samplingHandler().apply(r))
 				.subscribeOn(Schedulers.boundedElastic());
 			return new Async(syncSpec.clientInfo(), syncSpec.clientCapabilities(), syncSpec.roots(),
 					toolsChangeConsumers, resourcesChangeConsumers, promptsChangeConsumers, loggingConsumers,
-					samplingHandler);
+					progressConsumers, samplingHandler);
 		}
 	}
 
@@ -155,13 +166,15 @@ public static Async fromSync(Sync syncSpec) {
 	 * @param resourcesChangeConsumers the resources change consumers.
 	 * @param promptsChangeConsumers the prompts change consumers.
 	 * @param loggingConsumers the logging consumers.
+	 * @param progressConsumers the progress consumers.
 	 * @param samplingHandler the sampling handler.
 	 */
 	public record Sync(McpSchema.Implementation clientInfo, McpSchema.ClientCapabilities clientCapabilities,
 			Map<String, McpSchema.Root> roots, List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumers,
 			List<Consumer<List<McpSchema.Resource>>> resourcesChangeConsumers,
 			List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumers,
 			List<Consumer<McpSchema.LoggingMessageNotification>> loggingConsumers,
+			List<Consumer<McpSchema.ProgressNotification>> progressConsumers,
 			Function<McpSchema.CreateMessageRequest, McpSchema.CreateMessageResult> samplingHandler) {
 
 		/**
@@ -173,13 +186,15 @@ public record Sync(McpSchema.Implementation clientInfo, McpSchema.ClientCapabili
 		 * @param resourcesChangeConsumers the resources change consumers.
 		 * @param promptsChangeConsumers the prompts change consumers.
 		 * @param loggingConsumers the logging consumers.
+		 * @param progressConsumers the progress consumers.
 		 * @param samplingHandler the sampling handler.
 		 */
 		public Sync(McpSchema.Implementation clientInfo, McpSchema.ClientCapabilities clientCapabilities,
 				Map<String, McpSchema.Root> roots, List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumers,
 				List<Consumer<List<McpSchema.Resource>>> resourcesChangeConsumers,
 				List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumers,
 				List<Consumer<McpSchema.LoggingMessageNotification>> loggingConsumers,
+				List<Consumer<McpSchema.ProgressNotification>> progressConsumers,
 				Function<McpSchema.CreateMessageRequest, McpSchema.CreateMessageResult> samplingHandler) {
 
 			Assert.notNull(clientInfo, "Client info must not be null");
@@ -194,6 +209,7 @@ public Sync(McpSchema.Implementation clientInfo, McpSchema.ClientCapabilities cl
 			this.resourcesChangeConsumers = resourcesChangeConsumers != null ? resourcesChangeConsumers : List.of();
 			this.promptsChangeConsumers = promptsChangeConsumers != null ? promptsChangeConsumers : List.of();
 			this.loggingConsumers = loggingConsumers != null ? loggingConsumers : List.of();
+			this.progressConsumers = progressConsumers != null ? progressConsumers : List.of();
 			this.samplingHandler = samplingHandler;
 		}
 	}