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

Refactors the MCP logging system to use the exchange mechanism for sending
logging notifications only to specific client sessions rather than broadcasting to all clients.

- Move logging notification delivery from server-wide broadcast to per-session exchange
- Implement per-session minimum logging level tracking and filtering
- Add proper logging level filtering at the exchange level
- Change setLoggingLevel from notification to request/response pattern (breaking change)
- Deprecate global server.loggingNotification in favor of exchange.loggingNotification
- Add SetLevelRequest record to McpSchema
- Add integration test demonstrating filtered logging notifications

Resolves #131

Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
Co-authored-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>

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

@@ -786,10 +786,9 @@ public class McpAsyncClient {
 		}
 
 		return this.withInitializationCheck("setting logging level", initializedResult -> {
-			String levelName = this.transport.unmarshalFrom(loggingLevel, new TypeReference<String>() {
-			});
-			Map<String, Object> params = Map.of("level", levelName);
-			return this.mcpSession.sendNotification(McpSchema.METHOD_LOGGING_SET_LEVEL, params);
+			var params = new McpSchema.SetLevelRequest(loggingLevel);
+			return this.mcpSession.sendRequest(McpSchema.METHOD_LOGGING_SET_LEVEL, params, new TypeReference<Object>() {
+			}).then();
 		});
 	}
 

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

@@ -6,7 +6,6 @@ package io.modelcontextprotocol.client;
 
 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;

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

@@ -21,6 +21,7 @@ import io.modelcontextprotocol.spec.McpSchema;
 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;
@@ -216,11 +217,17 @@ public class McpAsyncServer {
 	// ---------------------------------------
 
 	/**
-	 * Send a logging message notification to all connected clients. Messages below the
-	 * current minimum logging level will be filtered out.
+	 * 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.
 	 * @param loggingMessageNotification The logging message to send
 	 * @return A Mono that completes when the notification has been sent
+	 * @deprecated Use
+	 * {@link McpAsyncServerExchange#loggingNotification(LoggingMessageNotification)}
+	 * instead.
 	 */
+	@Deprecated
 	public Mono<Void> loggingNotification(LoggingMessageNotification loggingMessageNotification) {
 		return this.delegate.loggingNotification(loggingMessageNotification);
 	}
@@ -257,6 +264,8 @@ public class McpAsyncServer {
 
 		private final ConcurrentHashMap<String, McpServerFeatures.AsyncPromptSpecification> prompts = new ConcurrentHashMap<>();
 
+		// FIXME: this field is deprecated and should be remvoed together with the
+		// broadcasting loggingNotification.
 		private LoggingLevel minLoggingLevel = LoggingLevel.DEBUG;
 
 		private List<String> protocolVersions = List.of(McpSchema.LATEST_PROTOCOL_VERSION);
@@ -677,12 +686,22 @@ public class McpAsyncServer {
 					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());
+
+					// FIXME: this field is deprecated and should be removed together
+					// with the broadcasting loggingNotification.
+					this.minLoggingLevel = newMinLoggingLevel.level();
+
+					return Mono.just(Map.of());
+				});
 			};
 		}
 

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

@@ -1,9 +1,16 @@
+/*
+ * 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;
 
 /**
@@ -11,6 +18,7 @@ import reactor.core.publisher.Mono;
  * 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,38 @@ public class McpAsyncServerExchange {
 				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;
+	}
+
+	private boolean isNotificationForLevelAllowed(LoggingLevel loggingLevel) {
+		return loggingLevel.level() >= this.minLoggingLevel.level();
+	}
+
 }

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

@@ -4,9 +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;
 
@@ -151,9 +149,16 @@ public class McpSyncServer {
 	}
 
 	/**
-	 * Send a logging message notification to all clients.
-	 * @param loggingMessageNotification The logging message notification to send
+	 * 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.
+	 * @param loggingMessageNotification The logging message to send
+	 * @deprecated Use
+	 * {@link McpSyncServerExchange#loggingNotification(LoggingMessageNotification)}
+	 * instead.
 	 */
+	@Deprecated
 	public void loggingNotification(LoggingMessageNotification loggingMessageNotification) {
 		this.asyncServer.loggingNotification(loggingMessageNotification).block();
 	}

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

@@ -1,13 +1,19 @@
+/*
+ * 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;
 
 /**
  * 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 +81,13 @@ public class McpSyncServerExchange {
 		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
+	 */
+	public void loggingNotification(LoggingMessageNotification loggingMessageNotification) {
+		this.exchange.loggingNotification(loggingMessageNotification).block();
+	}
+
 }

mcp/src/main/java/io/modelcontextprotocol/spec/McpSchema.java

@@ -1165,6 +1165,11 @@ public final class McpSchema {
 
 	} // @formatter:on
 
+	@JsonInclude(JsonInclude.Include.NON_ABSENT)
+	@JsonIgnoreProperties(ignoreUnknown = true)
+	public record SetLevelRequest(@JsonProperty("level") LoggingLevel level) {
+	}
+
 	// ---------------------------
 	// Autocomplete
 	// ---------------------------