feat(mcp): refactor logging to use exchange for targeted client notifications

mcp-test/src/main/java/io/modelcontextprotocol/client/AbstractMcpAsyncClientTests.java

@@ -453,15 +453,15 @@ void testLoggingLevelsWithoutInitialization() {
 	@Test
 	void testLoggingLevels() {
 		withClient(createMcpTransport(), mcpAsyncClient -> {
-			Mono<Void> testAllLevels = mcpAsyncClient.initialize().then(Mono.defer(() -> {
-				Mono<Void> chain = Mono.empty();
+			Mono<Object> testAllLevels = mcpAsyncClient.initialize().then(Mono.defer(() -> {
+				Mono<Object> chain = Mono.empty();
 				for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
 					chain = chain.then(mcpAsyncClient.setLoggingLevel(level));
 				}
 				return chain;
 			}));
 
-			StepVerifier.create(testAllLevels).verifyComplete();
+			StepVerifier.create(testAllLevels).expectNextCount(1).verifyComplete();
 		});
 	}
 

mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java

@@ -416,53 +416,4 @@ void testRootsChangeHandlers() {
 			.doesNotThrowAnyException();
 	}
 
-	// ---------------------------------------
-	// Logging Tests
-	// ---------------------------------------
-
-	@Test
-	void testLoggingLevels() {
-		var mcpAsyncServer = McpServer.async(createMcpTransportProvider())
-			.serverInfo("test-server", "1.0.0")
-			.capabilities(ServerCapabilities.builder().logging().build())
-			.build();
-
-		// Test all logging levels
-		for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
-			var notification = McpSchema.LoggingMessageNotification.builder()
-				.level(level)
-				.logger("test-logger")
-				.data("Test message with level " + level)
-				.build();
-
-			StepVerifier.create(mcpAsyncServer.loggingNotification(notification)).verifyComplete();
-		}
-	}
-
-	@Test
-	void testLoggingWithoutCapability() {
-		var mcpAsyncServer = McpServer.async(createMcpTransportProvider())
-			.serverInfo("test-server", "1.0.0")
-			.capabilities(ServerCapabilities.builder().build()) // No logging capability
-			.build();
-
-		var notification = McpSchema.LoggingMessageNotification.builder()
-			.level(McpSchema.LoggingLevel.INFO)
-			.logger("test-logger")
-			.data("Test log message")
-			.build();
-
-		StepVerifier.create(mcpAsyncServer.loggingNotification(notification)).verifyComplete();
-	}
-
-	@Test
-	void testLoggingWithNullNotification() {
-		var mcpAsyncServer = McpServer.async(createMcpTransportProvider())
-			.serverInfo("test-server", "1.0.0")
-			.capabilities(ServerCapabilities.builder().logging().build())
-			.build();
-
-		StepVerifier.create(mcpAsyncServer.loggingNotification(null)).verifyError(McpError.class);
-	}
-
 }

mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpSyncServerTests.java

@@ -388,53 +388,4 @@ void testRootsChangeHandlers() {
 		assertThatCode(() -> noConsumersServer.closeGracefully()).doesNotThrowAnyException();
 	}
 
-	// ---------------------------------------
-	// Logging Tests
-	// ---------------------------------------
-
-	@Test
-	void testLoggingLevels() {
-		var mcpSyncServer = McpServer.sync(createMcpTransportProvider())
-			.serverInfo("test-server", "1.0.0")
-			.capabilities(ServerCapabilities.builder().logging().build())
-			.build();
-
-		// Test all logging levels
-		for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
-			var notification = McpSchema.LoggingMessageNotification.builder()
-				.level(level)
-				.logger("test-logger")
-				.data("Test message with level " + level)
-				.build();
-
-			assertThatCode(() -> mcpSyncServer.loggingNotification(notification)).doesNotThrowAnyException();
-		}
-	}
-
-	@Test
-	void testLoggingWithoutCapability() {
-		var mcpSyncServer = McpServer.sync(createMcpTransportProvider())
-			.serverInfo("test-server", "1.0.0")
-			.capabilities(ServerCapabilities.builder().build()) // No logging capability
-			.build();
-
-		var notification = McpSchema.LoggingMessageNotification.builder()
-			.level(McpSchema.LoggingLevel.INFO)
-			.logger("test-logger")
-			.data("Test log message")
-			.build();
-
-		assertThatCode(() -> mcpSyncServer.loggingNotification(notification)).doesNotThrowAnyException();
-	}
-
-	@Test
-	void testLoggingWithNullNotification() {
-		var mcpSyncServer = McpServer.sync(createMcpTransportProvider())
-			.serverInfo("test-server", "1.0.0")
-			.capabilities(ServerCapabilities.builder().logging().build())
-			.build();
-
-		assertThatThrownBy(() -> mcpSyncServer.loggingNotification(null)).isInstanceOf(McpError.class);
-	}
-
 }

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

@@ -780,16 +780,15 @@ private NotificationHandler asyncLoggingNotificationHandler(
 	 * @return A Mono that completes when the logging level is set.
 	 * @see McpSchema.LoggingLevel
 	 */
-	public Mono<Void> setLoggingLevel(LoggingLevel loggingLevel) {
+	public Mono<Object> setLoggingLevel(LoggingLevel loggingLevel) {
 		if (loggingLevel == null) {
 			return Mono.error(new McpError("Logging level must not be null"));
 		}
 
 		return this.withInitializationCheck("setting logging level", initializedResult -> {
-			String levelName = this.transport.unmarshalFrom(loggingLevel, new TypeReference<String>() {
+			var params = new McpSchema.SetLevelRequest(loggingLevel);
+			return this.mcpSession.sendRequest(McpSchema.METHOD_LOGGING_SET_LEVEL, params, new TypeReference<Object>() {
 			});
-			Map<String, Object> params = Map.of("level", levelName);
-			return this.mcpSession.sendNotification(McpSchema.METHOD_LOGGING_SET_LEVEL, params);
 		});
 	}
 

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

@@ -6,7 +6,6 @@
 
 import java.time.Duration;
 
-import io.modelcontextprotocol.spec.McpClientTransport;
 import io.modelcontextprotocol.spec.McpSchema;
 import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
 import io.modelcontextprotocol.spec.McpSchema.GetPromptRequest;
@@ -321,9 +320,10 @@ public GetPromptResult getPrompt(GetPromptRequest getPromptRequest) {
 	/**
 	 * Client can set the minimum logging level it wants to receive from the server.
 	 * @param loggingLevel the min logging level
+	 * @return empty response
 	 */
-	public void setLoggingLevel(McpSchema.LoggingLevel loggingLevel) {
-		this.delegate.setLoggingLevel(loggingLevel).block();
+	public Object setLoggingLevel(McpSchema.LoggingLevel loggingLevel) {
+		return this.delegate.setLoggingLevel(loggingLevel).block();
 	}
 
 }

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

@@ -21,6 +21,7 @@
 import io.modelcontextprotocol.spec.McpSchema.CallToolResult;
 import io.modelcontextprotocol.spec.McpSchema.LoggingLevel;
 import io.modelcontextprotocol.spec.McpSchema.LoggingMessageNotification;
+import io.modelcontextprotocol.spec.McpSchema.SetLevelRequest;
 import io.modelcontextprotocol.spec.McpSchema.Tool;
 import io.modelcontextprotocol.spec.McpServerSession;
 import io.modelcontextprotocol.spec.McpServerTransportProvider;
@@ -257,6 +258,9 @@ private static class AsyncServerImpl extends McpAsyncServer {
 
 		private final ConcurrentHashMap<String, McpServerFeatures.AsyncPromptSpecification> prompts = new ConcurrentHashMap<>();
 
+		// TODO: this field is deprecated and should be remvoed together with the
+		// broadcasting loggingNotification.
+		@Deprecated
 		private LoggingLevel minLoggingLevel = LoggingLevel.DEBUG;
 
 		private List<String> protocolVersions = List.of(McpSchema.LATEST_PROTOCOL_VERSION);
@@ -662,7 +666,17 @@ private McpServerSession.RequestHandler<McpSchema.GetPromptResult> promptsGetReq
 		// Logging Management
 		// ---------------------------------------
 
+		/**
+		 * This implementation would, incorrectly, broadcast the logging message to all
+		 * connected clients, using a single minLoggingLevel for all of them. Similar to
+		 * the sampling and roots, the logging level should be set per client session and
+		 * use the ServerExchange to send the logging message to the right client.
+		 * @deprecated Use
+		 * {@link McpAsyncServerExchange#loggingNotification(LoggingMessageNotification)}
+		 * instead.
+		 */
 		@Override
+		@Deprecated
 		public Mono<Void> loggingNotification(LoggingMessageNotification loggingMessageNotification) {
 
 			if (loggingMessageNotification == null) {
@@ -677,12 +691,23 @@ public Mono<Void> loggingNotification(LoggingMessageNotification loggingMessageN
 					loggingMessageNotification);
 		}
 
-		private McpServerSession.RequestHandler<Void> setLoggerRequestHandler() {
+		private McpServerSession.RequestHandler<Object> setLoggerRequestHandler() {
 			return (exchange, params) -> {
-				this.minLoggingLevel = objectMapper.convertValue(params, new TypeReference<LoggingLevel>() {
-				});
+				return Mono.defer(() -> {
 
-				return Mono.empty();
+					SetLevelRequest newMinLoggingLevel = objectMapper.convertValue(params,
+							new TypeReference<SetLevelRequest>() {
+							});
+
+					exchange.setMinLoggingLevel(newMinLoggingLevel.level());
+
+					// TODO: this field is deprecated and should be remvoed together with
+					// the
+					// broadcasting loggingNotification.
+					this.minLoggingLevel = newMinLoggingLevel.level();
+
+					return Mono.just(Map.of());
+				});
 			};
 		}
 

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

@@ -1,16 +1,24 @@
+/*
+ * Copyright 2024-2024 the original author or authors.
+ */
+
 package io.modelcontextprotocol.server;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import io.modelcontextprotocol.spec.McpError;
 import io.modelcontextprotocol.spec.McpSchema;
+import io.modelcontextprotocol.spec.McpSchema.LoggingLevel;
+import io.modelcontextprotocol.spec.McpSchema.LoggingMessageNotification;
 import io.modelcontextprotocol.spec.McpServerSession;
+import io.modelcontextprotocol.util.Assert;
 import reactor.core.publisher.Mono;
 
 /**
  * Represents an asynchronous exchange with a Model Context Protocol (MCP) client. The
  * exchange provides methods to interact with the client and query its capabilities.
  *
  * @author Dariusz Jędrzejczyk
+ * @author Christian Tzolov
  */
 public class McpAsyncServerExchange {
 
@@ -20,6 +28,8 @@ public class McpAsyncServerExchange {
 
 	private final McpSchema.Implementation clientInfo;
 
+	private volatile LoggingLevel minLoggingLevel = LoggingLevel.INFO;
+
 	private static final TypeReference<McpSchema.CreateMessageResult> CREATE_MESSAGE_RESULT_TYPE_REF = new TypeReference<>() {
 	};
 
@@ -101,4 +111,42 @@ public Mono<McpSchema.ListRootsResult> listRoots(String cursor) {
 				LIST_ROOTS_RESULT_TYPE_REF);
 	}
 
+	/**
+	 * Send a logging message notification to all connected clients. Messages below the
+	 * current minimum logging level will be filtered out.
+	 * @param loggingMessageNotification The logging message to send
+	 * @return A Mono that completes when the notification has been sent
+	 */
+	public Mono<Void> loggingNotification(LoggingMessageNotification loggingMessageNotification) {
+
+		if (loggingMessageNotification == null) {
+			return Mono.error(new McpError("Logging message must not be null"));
+		}
+
+		return Mono.defer(() -> {
+			if (this.isNotificationForLevelAllowed(loggingMessageNotification.level())) {
+				return this.session.sendNotification(McpSchema.METHOD_NOTIFICATION_MESSAGE, loggingMessageNotification);
+			}
+			return Mono.empty();
+		});
+	}
+
+	/**
+	 * Set the minimum logging level for the client. Messages below this level will be
+	 * filtered out.
+	 * @param minLoggingLevel The minimum logging level
+	 */
+	void setMinLoggingLevel(LoggingLevel minLoggingLevel) {
+		Assert.notNull(minLoggingLevel, "minLoggingLevel must not be null");
+		this.minLoggingLevel = minLoggingLevel;
+	}
+
+	/**
+	 * Checks if the logging level bigger or equal to the minimum set logging level.
+	 * @return true if the logging level is enabled, false otherwise
+	 */
+	private boolean isNotificationForLevelAllowed(LoggingLevel loggingLevel) {
+		return loggingLevel.level() >= this.minLoggingLevel.level();
+	}
+
 }

mcp/src/main/java/io/modelcontextprotocol/server/McpSyncServer.java

@@ -4,10 +4,7 @@
 
 package io.modelcontextprotocol.server;
 
-import io.modelcontextprotocol.spec.McpError;
 import io.modelcontextprotocol.spec.McpSchema;
-import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
-import io.modelcontextprotocol.spec.McpSchema.LoggingMessageNotification;
 import io.modelcontextprotocol.util.Assert;
 
 /**
@@ -49,7 +46,7 @@
  * @see McpAsyncServer
  * @see McpSchema
  */
-public class McpSyncServer {
+public class McpSyncServer implements AutoCloseable {
 
 	/**
 	 * The async server to wrap.
@@ -150,14 +147,6 @@ public void notifyPromptsListChanged() {
 		this.asyncServer.notifyPromptsListChanged().block();
 	}
 
-	/**
-	 * Send a logging message notification to all clients.
-	 * @param loggingMessageNotification The logging message notification to send
-	 */
-	public void loggingNotification(LoggingMessageNotification loggingMessageNotification) {
-		this.asyncServer.loggingNotification(loggingMessageNotification).block();
-	}
-
 	/**
 	 * Close the server gracefully.
 	 */
@@ -168,6 +157,7 @@ public void closeGracefully() {
 	/**
 	 * Close the server immediately.
 	 */
+	@Override
 	public void close() {
 		this.asyncServer.close();
 	}

mcp/src/main/java/io/modelcontextprotocol/server/McpSyncServerExchange.java

@@ -1,13 +1,20 @@
+/*
+ * Copyright 2024-2024 the original author or authors.
+ */
+
 package io.modelcontextprotocol.server;
 
-import com.fasterxml.jackson.core.type.TypeReference;
 import io.modelcontextprotocol.spec.McpSchema;
+import io.modelcontextprotocol.spec.McpSchema.LoggingLevel;
+import io.modelcontextprotocol.spec.McpSchema.LoggingMessageNotification;
+import reactor.core.publisher.Mono;
 
 /**
  * Represents a synchronous exchange with a Model Context Protocol (MCP) client. The
  * exchange provides methods to interact with the client and query its capabilities.
  *
  * @author Dariusz Jędrzejczyk
+ * @author Christian Tzolov
  */
 public class McpSyncServerExchange {
 
@@ -75,4 +82,23 @@ public McpSchema.ListRootsResult listRoots(String cursor) {
 		return this.exchange.listRoots(cursor).block();
 	}
 
+	/**
+	 * Send a logging message notification to all connected clients. Messages below the
+	 * current minimum logging level will be filtered out.
+	 * @param loggingMessageNotification The logging message to send
+	 * @return A Mono that completes when the notification has been sent
+	 */
+	public Mono<Void> loggingNotification(LoggingMessageNotification loggingMessageNotification) {
+		return this.exchange.loggingNotification(loggingMessageNotification);
+	}
+
+	/**
+	 * Set the minimum logging level for the client. Messages below this level will be
+	 * filtered out.
+	 * @param minLoggingLevel The minimum logging level
+	 */
+	void setMinLoggingLevel(LoggingLevel minLoggingLevel) {
+		this.exchange.setMinLoggingLevel(minLoggingLevel);
+	}
+
 }