refactor: wrap handler invocations with Mono.defer for lazy evaluation

- Wrap consumer, tool, resource, prompt, and completion handler calls with Mono.defer()
- Ensures proper lazy evaluation and error handling in reactive streams

Signed-off-by: Christian Tzolov <[email protected]>

Author: tzolov

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

@@ -296,7 +296,7 @@ private McpNotificationHandler asyncRootsListChangedNotificationHandler(
 			List<BiFunction<McpAsyncServerExchange, List<McpSchema.Root>, Mono<Void>>> rootsChangeConsumers) {
 		return (exchange, params) -> exchange.listRoots()
 			.flatMap(listRootsResult -> Flux.fromIterable(rootsChangeConsumers)
-				.flatMap(consumer -> consumer.apply(exchange, listRootsResult.roots()))
+				.flatMap(consumer -> Mono.defer(() -> consumer.apply(exchange, listRootsResult.roots())))
 				.onErrorResume(error -> {
 					logger.error("Error handling roots list change notification", error);
 					return Mono.empty();
@@ -506,7 +506,7 @@ private McpRequestHandler<CallToolResult> toolsCallRequestHandler() {
 				return Mono.error(new McpError("Tool not found: " + callToolRequest.name()));
 			}
 
-			return toolSpecification.map(tool -> tool.callHandler().apply(exchange, callToolRequest))
+			return toolSpecification.map(tool -> Mono.defer(() -> tool.callHandler().apply(exchange, callToolRequest)))
 				.orElse(Mono.error(new McpError("Tool not found: " + callToolRequest.name())));
 		};
 	}
@@ -634,7 +634,7 @@ private McpRequestHandler<McpSchema.ReadResourceResult> resourcesReadRequestHand
 				.findFirst()
 				.orElseThrow(() -> new McpError("Resource not found: " + resourceUri));
 
-			return specification.readHandler().apply(exchange, resourceRequest);
+			return Mono.defer(() -> specification.readHandler().apply(exchange, resourceRequest));
 		};
 	}
 
@@ -740,7 +740,7 @@ private McpRequestHandler<McpSchema.GetPromptResult> promptsGetRequestHandler()
 				return Mono.error(new McpError("Prompt not found: " + promptRequest.name()));
 			}
 
-			return specification.promptHandler().apply(exchange, promptRequest);
+			return Mono.defer(() -> specification.promptHandler().apply(exchange, promptRequest));
 		};
 	}
 
@@ -845,7 +845,7 @@ private McpRequestHandler<McpSchema.CompleteResult> completionCompleteRequestHan
 				return Mono.error(new McpError("AsyncCompletionSpecification not found: " + request.ref()));
 			}
 
-			return specification.completionHandler().apply(exchange, request);
+			return Mono.defer(() -> specification.completionHandler().apply(exchange, request));
 		};
 	}
 

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

@@ -179,10 +179,6 @@ public Mono<Void> responseStream(McpSchema.JSONRPCRequest jsonrpcRequest, McpStr
 				})
 				.flatMap(transport::sendMessage)
 				.then(transport.closeGracefully());
-		}).onErrorResume(error -> {
-			return transport.sendMessage(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, jsonrpcRequest.id(),
-					null, new McpSchema.JSONRPCResponse.JSONRPCError(McpSchema.ErrorCodes.INTERNAL_ERROR,
-							error.getMessage(), null)));
 		});
 	}