refactor: reduce log verbosity by adjusting log levels
- Change most INFO logs to DEBUG level for operational messages - Keep stderr messages at INFO level - Rename errorHandler to stdErrorHandler in StdioClientTransport - Add debug logging for JSON message deserialization Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
This commit is contained in:
parent
5d27a13a4d
commit
871add4e45
9 changed files with 35 additions and 28 deletions
|
|
@ -251,7 +251,7 @@ public class WebFluxSseServerTransport implements ServerMcpTransport {
|
||||||
.then(Mono.fromRunnable(() -> sessions.remove(sessionId)));
|
.then(Mono.fromRunnable(() -> sessions.remove(sessionId)));
|
||||||
}).toList()))
|
}).toList()))
|
||||||
.timeout(Duration.ofSeconds(5))
|
.timeout(Duration.ofSeconds(5))
|
||||||
.doOnSuccess(v -> logger.info("Graceful shutdown completed"))
|
.doOnSuccess(v -> logger.debug("Graceful shutdown completed"))
|
||||||
.doOnError(e -> logger.error("Error during graceful shutdown: {}", e.getMessage()));
|
.doOnError(e -> logger.error("Error during graceful shutdown: {}", e.getMessage()));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -353,7 +353,7 @@ public class WebMvcSseServerTransport implements ServerMcpTransport {
|
||||||
sessions.remove(sessionId);
|
sessions.remove(sessionId);
|
||||||
});
|
});
|
||||||
|
|
||||||
logger.info("Graceful shutdown completed");
|
logger.debug("Graceful shutdown completed");
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -173,7 +173,7 @@ public class McpAsyncClient {
|
||||||
// Tools Change Notification
|
// Tools Change Notification
|
||||||
List<Function<List<McpSchema.Tool>, Mono<Void>>> toolsChangeConsumersFinal = new ArrayList<>();
|
List<Function<List<McpSchema.Tool>, Mono<Void>>> toolsChangeConsumersFinal = new ArrayList<>();
|
||||||
toolsChangeConsumersFinal
|
toolsChangeConsumersFinal
|
||||||
.add((notification) -> Mono.fromRunnable(() -> logger.info("Tools changed: {}", notification)));
|
.add((notification) -> Mono.fromRunnable(() -> logger.debug("Tools changed: {}", notification)));
|
||||||
|
|
||||||
if (!Utils.isEmpty(features.toolsChangeConsumers())) {
|
if (!Utils.isEmpty(features.toolsChangeConsumers())) {
|
||||||
toolsChangeConsumersFinal.addAll(features.toolsChangeConsumers());
|
toolsChangeConsumersFinal.addAll(features.toolsChangeConsumers());
|
||||||
|
|
@ -184,7 +184,7 @@ public class McpAsyncClient {
|
||||||
// Resources Change Notification
|
// Resources Change Notification
|
||||||
List<Function<List<McpSchema.Resource>, Mono<Void>>> resourcesChangeConsumersFinal = new ArrayList<>();
|
List<Function<List<McpSchema.Resource>, Mono<Void>>> resourcesChangeConsumersFinal = new ArrayList<>();
|
||||||
resourcesChangeConsumersFinal
|
resourcesChangeConsumersFinal
|
||||||
.add((notification) -> Mono.fromRunnable(() -> logger.info("Resources changed: {}", notification)));
|
.add((notification) -> Mono.fromRunnable(() -> logger.debug("Resources changed: {}", notification)));
|
||||||
|
|
||||||
if (!Utils.isEmpty(features.resourcesChangeConsumers())) {
|
if (!Utils.isEmpty(features.resourcesChangeConsumers())) {
|
||||||
resourcesChangeConsumersFinal.addAll(features.resourcesChangeConsumers());
|
resourcesChangeConsumersFinal.addAll(features.resourcesChangeConsumers());
|
||||||
|
|
@ -196,7 +196,7 @@ public class McpAsyncClient {
|
||||||
// Prompts Change Notification
|
// Prompts Change Notification
|
||||||
List<Function<List<McpSchema.Prompt>, Mono<Void>>> promptsChangeConsumersFinal = new ArrayList<>();
|
List<Function<List<McpSchema.Prompt>, Mono<Void>>> promptsChangeConsumersFinal = new ArrayList<>();
|
||||||
promptsChangeConsumersFinal
|
promptsChangeConsumersFinal
|
||||||
.add((notification) -> Mono.fromRunnable(() -> logger.info("Prompts changed: {}", notification)));
|
.add((notification) -> Mono.fromRunnable(() -> logger.debug("Prompts changed: {}", notification)));
|
||||||
if (!Utils.isEmpty(features.promptsChangeConsumers())) {
|
if (!Utils.isEmpty(features.promptsChangeConsumers())) {
|
||||||
promptsChangeConsumersFinal.addAll(features.promptsChangeConsumers());
|
promptsChangeConsumersFinal.addAll(features.promptsChangeConsumers());
|
||||||
}
|
}
|
||||||
|
|
@ -205,7 +205,7 @@ public class McpAsyncClient {
|
||||||
|
|
||||||
// Utility Logging Notification
|
// Utility Logging Notification
|
||||||
List<Function<LoggingMessageNotification, Mono<Void>>> loggingConsumersFinal = new ArrayList<>();
|
List<Function<LoggingMessageNotification, Mono<Void>>> loggingConsumersFinal = new ArrayList<>();
|
||||||
loggingConsumersFinal.add((notification) -> Mono.fromRunnable(() -> logger.info("Logging: {}", notification)));
|
loggingConsumersFinal.add((notification) -> Mono.fromRunnable(() -> logger.debug("Logging: {}", notification)));
|
||||||
if (!Utils.isEmpty(features.loggingConsumers())) {
|
if (!Utils.isEmpty(features.loggingConsumers())) {
|
||||||
loggingConsumersFinal.addAll(features.loggingConsumers());
|
loggingConsumersFinal.addAll(features.loggingConsumers());
|
||||||
}
|
}
|
||||||
|
|
@ -364,7 +364,7 @@ public class McpAsyncClient {
|
||||||
|
|
||||||
this.roots.put(root.uri(), root);
|
this.roots.put(root.uri(), root);
|
||||||
|
|
||||||
logger.info("Added root: {}", root);
|
logger.debug("Added root: {}", root);
|
||||||
|
|
||||||
if (this.clientCapabilities.roots().listChanged()) {
|
if (this.clientCapabilities.roots().listChanged()) {
|
||||||
return this.rootsListChangedNotification();
|
return this.rootsListChangedNotification();
|
||||||
|
|
@ -390,7 +390,7 @@ public class McpAsyncClient {
|
||||||
Root removed = this.roots.remove(rootUri);
|
Root removed = this.roots.remove(rootUri);
|
||||||
|
|
||||||
if (removed != null) {
|
if (removed != null) {
|
||||||
logger.info("Removed Root: {}", rootUri);
|
logger.debug("Removed Root: {}", rootUri);
|
||||||
if (this.clientCapabilities.roots().listChanged()) {
|
if (this.clientCapabilities.roots().listChanged()) {
|
||||||
return this.rootsListChangedNotification();
|
return this.rootsListChangedNotification();
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -68,7 +68,7 @@ public class StdioClientTransport implements ClientMcpTransport {
|
||||||
private volatile boolean isClosing = false;
|
private volatile boolean isClosing = false;
|
||||||
|
|
||||||
// visible for tests
|
// visible for tests
|
||||||
private Consumer<String> errorHandler = error -> logger.error("Error received: {}", error);
|
private Consumer<String> stdErrorHandler = error -> logger.info("STDERR Message received: {}", error);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Creates a new StdioClientTransport with the specified parameters and default
|
* Creates a new StdioClientTransport with the specified parameters and default
|
||||||
|
|
@ -164,8 +164,8 @@ public class StdioClientTransport implements ClientMcpTransport {
|
||||||
* </p>
|
* </p>
|
||||||
* @param errorHandler a consumer that processes error messages
|
* @param errorHandler a consumer that processes error messages
|
||||||
*/
|
*/
|
||||||
public void setErrorHandler(Consumer<String> errorHandler) {
|
public void setStdErrorHandler(Consumer<String> errorHandler) {
|
||||||
this.errorHandler = errorHandler;
|
this.stdErrorHandler = errorHandler;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -192,7 +192,6 @@ public class StdioClientTransport implements ClientMcpTransport {
|
||||||
String line;
|
String line;
|
||||||
while (!isClosing && (line = processErrorReader.readLine()) != null) {
|
while (!isClosing && (line = processErrorReader.readLine()) != null) {
|
||||||
try {
|
try {
|
||||||
logger.error("Received error line: {}", line);
|
|
||||||
if (!this.errorSink.tryEmitNext(line).isSuccess()) {
|
if (!this.errorSink.tryEmitNext(line).isSuccess()) {
|
||||||
if (!isClosing) {
|
if (!isClosing) {
|
||||||
logger.error("Failed to emit error message");
|
logger.error("Failed to emit error message");
|
||||||
|
|
@ -230,7 +229,7 @@ public class StdioClientTransport implements ClientMcpTransport {
|
||||||
|
|
||||||
private void handleIncomingErrors() {
|
private void handleIncomingErrors() {
|
||||||
this.errorSink.asFlux().subscribe(e -> {
|
this.errorSink.asFlux().subscribe(e -> {
|
||||||
this.errorHandler.accept(e);
|
this.stdErrorHandler.accept(e);
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -355,7 +354,7 @@ public class StdioClientTransport implements ClientMcpTransport {
|
||||||
// Give a short time for any pending messages to be processed
|
// Give a short time for any pending messages to be processed
|
||||||
return Mono.delay(Duration.ofMillis(100));
|
return Mono.delay(Duration.ofMillis(100));
|
||||||
})).then(Mono.fromFuture(() -> {
|
})).then(Mono.fromFuture(() -> {
|
||||||
logger.info("Sending TERM to process");
|
logger.debug("Sending TERM to process");
|
||||||
if (this.process != null) {
|
if (this.process != null) {
|
||||||
this.process.destroy();
|
this.process.destroy();
|
||||||
return process.onExit();
|
return process.onExit();
|
||||||
|
|
@ -375,7 +374,7 @@ public class StdioClientTransport implements ClientMcpTransport {
|
||||||
errorScheduler.dispose();
|
errorScheduler.dispose();
|
||||||
outboundScheduler.dispose();
|
outboundScheduler.dispose();
|
||||||
|
|
||||||
logger.info("Graceful shutdown completed");
|
logger.debug("Graceful shutdown completed");
|
||||||
}
|
}
|
||||||
catch (Exception e) {
|
catch (Exception e) {
|
||||||
logger.error("Error during graceful shutdown", e);
|
logger.error("Error during graceful shutdown", e);
|
||||||
|
|
|
||||||
|
|
@ -318,7 +318,7 @@ public class McpAsyncServer {
|
||||||
}
|
}
|
||||||
|
|
||||||
this.tools.add(toolRegistration);
|
this.tools.add(toolRegistration);
|
||||||
logger.info("Added tool handler: {}", toolRegistration.tool().name());
|
logger.debug("Added tool handler: {}", toolRegistration.tool().name());
|
||||||
|
|
||||||
if (this.serverCapabilities.tools().listChanged()) {
|
if (this.serverCapabilities.tools().listChanged()) {
|
||||||
return notifyToolsListChanged();
|
return notifyToolsListChanged();
|
||||||
|
|
@ -343,7 +343,7 @@ public class McpAsyncServer {
|
||||||
return Mono.defer(() -> {
|
return Mono.defer(() -> {
|
||||||
boolean removed = this.tools.removeIf(toolRegistration -> toolRegistration.tool().name().equals(toolName));
|
boolean removed = this.tools.removeIf(toolRegistration -> toolRegistration.tool().name().equals(toolName));
|
||||||
if (removed) {
|
if (removed) {
|
||||||
logger.info("Removed tool handler: {}", toolName);
|
logger.debug("Removed tool handler: {}", toolName);
|
||||||
if (this.serverCapabilities.tools().listChanged()) {
|
if (this.serverCapabilities.tools().listChanged()) {
|
||||||
return notifyToolsListChanged();
|
return notifyToolsListChanged();
|
||||||
}
|
}
|
||||||
|
|
@ -411,7 +411,7 @@ public class McpAsyncServer {
|
||||||
return Mono
|
return Mono
|
||||||
.error(new McpError("Resource with URI '" + resourceHandler.resource().uri() + "' already exists"));
|
.error(new McpError("Resource with URI '" + resourceHandler.resource().uri() + "' already exists"));
|
||||||
}
|
}
|
||||||
logger.info("Added resource handler: {}", resourceHandler.resource().uri());
|
logger.debug("Added resource handler: {}", resourceHandler.resource().uri());
|
||||||
if (this.serverCapabilities.resources().listChanged()) {
|
if (this.serverCapabilities.resources().listChanged()) {
|
||||||
return notifyResourcesListChanged();
|
return notifyResourcesListChanged();
|
||||||
}
|
}
|
||||||
|
|
@ -435,7 +435,7 @@ public class McpAsyncServer {
|
||||||
return Mono.defer(() -> {
|
return Mono.defer(() -> {
|
||||||
McpServerFeatures.AsyncResourceRegistration removed = this.resources.remove(resourceUri);
|
McpServerFeatures.AsyncResourceRegistration removed = this.resources.remove(resourceUri);
|
||||||
if (removed != null) {
|
if (removed != null) {
|
||||||
logger.info("Removed resource handler: {}", resourceUri);
|
logger.debug("Removed resource handler: {}", resourceUri);
|
||||||
if (this.serverCapabilities.resources().listChanged()) {
|
if (this.serverCapabilities.resources().listChanged()) {
|
||||||
return notifyResourcesListChanged();
|
return notifyResourcesListChanged();
|
||||||
}
|
}
|
||||||
|
|
@ -507,7 +507,7 @@ public class McpAsyncServer {
|
||||||
new McpError("Prompt with name '" + promptRegistration.prompt().name() + "' already exists"));
|
new McpError("Prompt with name '" + promptRegistration.prompt().name() + "' already exists"));
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.info("Added prompt handler: {}", promptRegistration.prompt().name());
|
logger.debug("Added prompt handler: {}", promptRegistration.prompt().name());
|
||||||
|
|
||||||
// Servers that declared the listChanged capability SHOULD send a
|
// Servers that declared the listChanged capability SHOULD send a
|
||||||
// notification,
|
// notification,
|
||||||
|
|
@ -536,7 +536,7 @@ public class McpAsyncServer {
|
||||||
McpServerFeatures.AsyncPromptRegistration removed = this.prompts.remove(promptName);
|
McpServerFeatures.AsyncPromptRegistration removed = this.prompts.remove(promptName);
|
||||||
|
|
||||||
if (removed != null) {
|
if (removed != null) {
|
||||||
logger.info("Removed prompt handler: {}", promptName);
|
logger.debug("Removed prompt handler: {}", promptName);
|
||||||
// Servers that declared the listChanged capability SHOULD send a
|
// Servers that declared the listChanged capability SHOULD send a
|
||||||
// notification, when the list of available prompts changes
|
// notification, when the list of available prompts changes
|
||||||
if (this.serverCapabilities.prompts().listChanged()) {
|
if (this.serverCapabilities.prompts().listChanged()) {
|
||||||
|
|
|
||||||
|
|
@ -141,6 +141,8 @@ public class StdioServerTransport implements ServerMcpTransport {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
logger.debug("Received JSON message: {}", line);
|
||||||
|
|
||||||
try {
|
try {
|
||||||
JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(this.objectMapper, line);
|
JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(this.objectMapper, line);
|
||||||
if (!this.inboundSink.tryEmitNext(message).isSuccess()) {
|
if (!this.inboundSink.tryEmitNext(message).isSuccess()) {
|
||||||
|
|
@ -229,7 +231,7 @@ public class StdioServerTransport implements ServerMcpTransport {
|
||||||
// Completing the inbound causes the outbound to be completed as well, so
|
// Completing the inbound causes the outbound to be completed as well, so
|
||||||
// we only close the inbound.
|
// we only close the inbound.
|
||||||
inboundSink.tryEmitComplete();
|
inboundSink.tryEmitComplete();
|
||||||
logger.info("Graceful shutdown complete");
|
logger.debug("Graceful shutdown complete");
|
||||||
return Mono.empty();
|
return Mono.empty();
|
||||||
}).subscribeOn(Schedulers.boundedElastic());
|
}).subscribeOn(Schedulers.boundedElastic());
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -124,7 +124,7 @@ public class DefaultMcpSession implements McpSession {
|
||||||
// consumer
|
// consumer
|
||||||
this.connection = this.transport.connect(mono -> mono.doOnNext(message -> {
|
this.connection = this.transport.connect(mono -> mono.doOnNext(message -> {
|
||||||
if (message instanceof McpSchema.JSONRPCResponse response) {
|
if (message instanceof McpSchema.JSONRPCResponse response) {
|
||||||
logger.info("Received Response: {}", response);
|
logger.debug("Received Response: {}", response);
|
||||||
var sink = pendingResponses.remove(response.id());
|
var sink = pendingResponses.remove(response.id());
|
||||||
if (sink == null) {
|
if (sink == null) {
|
||||||
logger.warn("Unexpected response for unkown id {}", response.id());
|
logger.warn("Unexpected response for unkown id {}", response.id());
|
||||||
|
|
@ -134,7 +134,7 @@ public class DefaultMcpSession implements McpSession {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
else if (message instanceof McpSchema.JSONRPCRequest request) {
|
else if (message instanceof McpSchema.JSONRPCRequest request) {
|
||||||
logger.info("Received request: {}", request);
|
logger.debug("Received request: {}", request);
|
||||||
handleIncomingRequest(request).subscribe(response -> transport.sendMessage(response).subscribe(),
|
handleIncomingRequest(request).subscribe(response -> transport.sendMessage(response).subscribe(),
|
||||||
error -> {
|
error -> {
|
||||||
var errorResponse = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
|
var errorResponse = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
|
||||||
|
|
@ -144,7 +144,7 @@ public class DefaultMcpSession implements McpSession {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
else if (message instanceof McpSchema.JSONRPCNotification notification) {
|
else if (message instanceof McpSchema.JSONRPCNotification notification) {
|
||||||
logger.info("Received notification: {}", notification);
|
logger.debug("Received notification: {}", notification);
|
||||||
handleIncomingNotification(notification).subscribe(null,
|
handleIncomingNotification(notification).subscribe(null,
|
||||||
error -> logger.error("Error handling notification: {}", error.getMessage()));
|
error -> logger.error("Error handling notification: {}", error.getMessage()));
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -17,6 +17,8 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo;
|
||||||
import com.fasterxml.jackson.annotation.JsonTypeInfo.As;
|
import com.fasterxml.jackson.annotation.JsonTypeInfo.As;
|
||||||
import com.fasterxml.jackson.core.type.TypeReference;
|
import com.fasterxml.jackson.core.type.TypeReference;
|
||||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Based on the <a href="http://www.jsonrpc.org/specification">JSON-RPC 2.0
|
* Based on the <a href="http://www.jsonrpc.org/specification">JSON-RPC 2.0
|
||||||
|
|
@ -28,6 +30,8 @@ import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
*/
|
*/
|
||||||
public final class McpSchema {
|
public final class McpSchema {
|
||||||
|
|
||||||
|
private static final Logger logger = LoggerFactory.getLogger(McpSchema.class);
|
||||||
|
|
||||||
private McpSchema() {
|
private McpSchema() {
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -144,6 +148,8 @@ public final class McpSchema {
|
||||||
public static JSONRPCMessage deserializeJsonRpcMessage(ObjectMapper objectMapper, String jsonText)
|
public static JSONRPCMessage deserializeJsonRpcMessage(ObjectMapper objectMapper, String jsonText)
|
||||||
throws IOException {
|
throws IOException {
|
||||||
|
|
||||||
|
logger.debug("Received JSON message: {}", jsonText);
|
||||||
|
|
||||||
var map = objectMapper.readValue(jsonText, MAP_TYPE_REF);
|
var map = objectMapper.readValue(jsonText, MAP_TYPE_REF);
|
||||||
|
|
||||||
// Determine message type based on specific JSON structure
|
// Determine message type based on specific JSON structure
|
||||||
|
|
|
||||||
|
|
@ -36,7 +36,7 @@ class StdioMcpSyncClientTests extends AbstractMcpSyncClientTests {
|
||||||
void customErrorHandlerShouldReceiveErrors() {
|
void customErrorHandlerShouldReceiveErrors() {
|
||||||
AtomicReference<String> receivedError = new AtomicReference<>();
|
AtomicReference<String> receivedError = new AtomicReference<>();
|
||||||
|
|
||||||
((StdioClientTransport) mcpTransport).setErrorHandler(error -> receivedError.set(error));
|
((StdioClientTransport) mcpTransport).setStdErrorHandler(error -> receivedError.set(error));
|
||||||
|
|
||||||
String errorMessage = "Test error";
|
String errorMessage = "Test error";
|
||||||
((StdioClientTransport) mcpTransport).getErrorSink().tryEmitNext(errorMessage);
|
((StdioClientTransport) mcpTransport).getErrorSink().tryEmitNext(errorMessage);
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue