Add option for immediate execution in McpSyncServer

- The McpSyncServer wraps an async server. By default, reactive
  operations are scheduled on a bounded-elastic scheduler, to offload
  blocking work and prevent accidental blocking of non-blocking
  operations.
- With the default behavior, there will be thead ops, even in a blocking
  context, which means thread-locals from the request thread will be
  lost. This is inconenvient for frameworks that store state in
  thread-locals.
- This commit adds the ability to avoid offloading, when the user is
  sure they are executing code in a blocking environment. Work happens
  in the calling thread, and thread-locals are available throughout the
  execution.

Author: Kehrlann

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

@@ -695,6 +695,8 @@ class SyncSpecification {
 
 		private Duration requestTimeout = Duration.ofSeconds(10); // Default timeout
 
+		private boolean immediate = false;
+
 		private SyncSpecification(McpServerTransportProvider transportProvider) {
 			Assert.notNull(transportProvider, "Transport provider must not be null");
 			this.transportProvider = transportProvider;
@@ -1116,6 +1118,19 @@ public SyncSpecification objectMapper(ObjectMapper objectMapper) {
 			return this;
 		}
 
+		/**
+		 * Sets a flag to turn on "immediate execution" of the operations when creating
+		 * the underlying {@link McpAsyncServer}. Defaults to false, which does blocking
+		 * code offloading to prevent accidental blocking of the non-blocking transport.
+		 * @param immediateExecution When true, do not offload work asynchronously.
+		 * @return This builder instance for method chaining.
+		 *
+		 */
+		public SyncSpecification immediateExecution(boolean immediateExecution) {
+			this.immediate = immediateExecution;
+			return this;
+		}
+
 		/**
 		 * Builds a synchronous MCP server that provides blocking operations.
 		 * @return A new instance of {@link McpSyncServer} configured with this builder's
@@ -1125,12 +1140,12 @@ public McpSyncServer build() {
 			McpServerFeatures.Sync syncFeatures = new McpServerFeatures.Sync(this.serverInfo, this.serverCapabilities,
 					this.tools, this.resources, this.resourceTemplates, this.prompts, this.completions,
 					this.rootsChangeHandlers, this.instructions);
-			McpServerFeatures.Async asyncFeatures = McpServerFeatures.Async.fromSync(syncFeatures);
+			McpServerFeatures.Async asyncFeatures = McpServerFeatures.Async.fromSync(syncFeatures, this.immediate);
 			var mapper = this.objectMapper != null ? this.objectMapper : new ObjectMapper();
 			var asyncServer = new McpAsyncServer(this.transportProvider, mapper, asyncFeatures, this.requestTimeout,
 					this.uriTemplateManagerFactory);
 
-			return new McpSyncServer(asyncServer);
+			return new McpSyncServer(asyncServer, this.immediate);
 		}
 
 	}

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

@@ -95,28 +95,30 @@ record Async(McpSchema.Implementation serverInfo, McpSchema.ServerCapabilities s
 		 * blocking code offloading to prevent accidental blocking of the non-blocking
 		 * transport.
 		 * @param syncSpec a potentially blocking, synchronous specification.
+		 * @param immediateExecution when true, do not offload. Only use if you are sure
+		 * you are in a blocking context.
 		 * @return a specification which is protected from blocking calls specified by the
 		 * user.
 		 */
-		static Async fromSync(Sync syncSpec) {
+		static Async fromSync(Sync syncSpec, boolean immediateExecution) {
 			List<McpServerFeatures.AsyncToolSpecification> tools = new ArrayList<>();
 			for (var tool : syncSpec.tools()) {
-				tools.add(AsyncToolSpecification.fromSync(tool));
+				tools.add(AsyncToolSpecification.fromSync(tool, immediateExecution));
 			}
 
 			Map<String, AsyncResourceSpecification> resources = new HashMap<>();
 			syncSpec.resources().forEach((key, resource) -> {
-				resources.put(key, AsyncResourceSpecification.fromSync(resource));
+				resources.put(key, AsyncResourceSpecification.fromSync(resource, immediateExecution));
 			});
 
 			Map<String, AsyncPromptSpecification> prompts = new HashMap<>();
 			syncSpec.prompts().forEach((key, prompt) -> {
-				prompts.put(key, AsyncPromptSpecification.fromSync(prompt));
+				prompts.put(key, AsyncPromptSpecification.fromSync(prompt, immediateExecution));
 			});
 
 			Map<McpSchema.CompleteReference, McpServerFeatures.AsyncCompletionSpecification> completions = new HashMap<>();
 			syncSpec.completions().forEach((key, completion) -> {
-				completions.put(key, AsyncCompletionSpecification.fromSync(completion));
+				completions.put(key, AsyncCompletionSpecification.fromSync(completion, immediateExecution));
 			});
 
 			List<BiFunction<McpAsyncServerExchange, List<McpSchema.Root>, Mono<Void>>> rootChangeConsumers = new ArrayList<>();
@@ -239,15 +241,15 @@ record Sync(McpSchema.Implementation serverInfo, McpSchema.ServerCapabilities se
 	public record AsyncToolSpecification(McpSchema.Tool tool,
 			BiFunction<McpAsyncServerExchange, Map<String, Object>, Mono<McpSchema.CallToolResult>> call) {
 
-		static AsyncToolSpecification fromSync(SyncToolSpecification tool) {
+		static AsyncToolSpecification fromSync(SyncToolSpecification tool, boolean immediate) {
 			// FIXME: This is temporary, proper validation should be implemented
 			if (tool == null) {
 				return null;
 			}
-			return new AsyncToolSpecification(tool.tool(),
-					(exchange, map) -> Mono
-						.fromCallable(() -> tool.call().apply(new McpSyncServerExchange(exchange), map))
-						.subscribeOn(Schedulers.boundedElastic()));
+			return new AsyncToolSpecification(tool.tool(), (exchange, map) -> {
+				var toolResult = Mono.fromCallable(() -> tool.call().apply(new McpSyncServerExchange(exchange), map));
+				return immediate ? toolResult : toolResult.subscribeOn(Schedulers.boundedElastic());
+			});
 		}
 	}
 
@@ -281,15 +283,16 @@ static AsyncToolSpecification fromSync(SyncToolSpecification tool) {
 	public record AsyncResourceSpecification(McpSchema.Resource resource,
 			BiFunction<McpAsyncServerExchange, McpSchema.ReadResourceRequest, Mono<McpSchema.ReadResourceResult>> readHandler) {
 
-		static AsyncResourceSpecification fromSync(SyncResourceSpecification resource) {
+		static AsyncResourceSpecification fromSync(SyncResourceSpecification resource, boolean immediateExecution) {
 			// FIXME: This is temporary, proper validation should be implemented
 			if (resource == null) {
 				return null;
 			}
-			return new AsyncResourceSpecification(resource.resource(),
-					(exchange, req) -> Mono
-						.fromCallable(() -> resource.readHandler().apply(new McpSyncServerExchange(exchange), req))
-						.subscribeOn(Schedulers.boundedElastic()));
+			return new AsyncResourceSpecification(resource.resource(), (exchange, req) -> {
+				var resourceResult = Mono
+					.fromCallable(() -> resource.readHandler().apply(new McpSyncServerExchange(exchange), req));
+				return immediateExecution ? resourceResult : resourceResult.subscribeOn(Schedulers.boundedElastic());
+			});
 		}
 	}
 
@@ -327,15 +330,16 @@ static AsyncResourceSpecification fromSync(SyncResourceSpecification resource) {
 	public record AsyncPromptSpecification(McpSchema.Prompt prompt,
 			BiFunction<McpAsyncServerExchange, McpSchema.GetPromptRequest, Mono<McpSchema.GetPromptResult>> promptHandler) {
 
-		static AsyncPromptSpecification fromSync(SyncPromptSpecification prompt) {
+		static AsyncPromptSpecification fromSync(SyncPromptSpecification prompt, boolean immediateExecution) {
 			// FIXME: This is temporary, proper validation should be implemented
 			if (prompt == null) {
 				return null;
 			}
-			return new AsyncPromptSpecification(prompt.prompt(),
-					(exchange, req) -> Mono
-						.fromCallable(() -> prompt.promptHandler().apply(new McpSyncServerExchange(exchange), req))
-						.subscribeOn(Schedulers.boundedElastic()));
+			return new AsyncPromptSpecification(prompt.prompt(), (exchange, req) -> {
+				var promptResult = Mono
+					.fromCallable(() -> prompt.promptHandler().apply(new McpSyncServerExchange(exchange), req));
+				return immediateExecution ? promptResult : promptResult.subscribeOn(Schedulers.boundedElastic());
+			});
 		}
 	}
 
@@ -366,14 +370,17 @@ public record AsyncCompletionSpecification(McpSchema.CompleteReference reference
 		 * @return an asynchronous wrapper of the provided sync specification, or
 		 * {@code null} if input is null
 		 */
-		static AsyncCompletionSpecification fromSync(SyncCompletionSpecification completion) {
+		static AsyncCompletionSpecification fromSync(SyncCompletionSpecification completion,
+				boolean immediateExecution) {
 			if (completion == null) {
 				return null;
 			}
-			return new AsyncCompletionSpecification(completion.referenceKey(),
-					(exchange, request) -> Mono.fromCallable(
-							() -> completion.completionHandler().apply(new McpSyncServerExchange(exchange), request))
-						.subscribeOn(Schedulers.boundedElastic()));
+			return new AsyncCompletionSpecification(completion.referenceKey(), (exchange, request) -> {
+				var completionResult = Mono.fromCallable(
+						() -> completion.completionHandler().apply(new McpSyncServerExchange(exchange), request));
+				return immediateExecution ? completionResult
+						: completionResult.subscribeOn(Schedulers.boundedElastic());
+			});
 		}
 	}
 

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

@@ -54,21 +54,36 @@ public class McpSyncServer {
 	 */
 	private final McpAsyncServer asyncServer;
 
+	private final boolean immediateExecution;
+
 	/**
 	 * Creates a new synchronous server that wraps the provided async server.
 	 * @param asyncServer The async server to wrap
 	 */
 	public McpSyncServer(McpAsyncServer asyncServer) {
+		this(asyncServer, false);
+	}
+
+	/**
+	 * Creates a new synchronous server that wraps the provided async server.
+	 * @param asyncServer The async server to wrap
+	 * @param immediateExecution Tools, prompts, and resources handlers execute work
+	 * without blocking code offloading.
+	 */
+	public McpSyncServer(McpAsyncServer asyncServer, boolean immediateExecution) {
 		Assert.notNull(asyncServer, "Async server must not be null");
 		this.asyncServer = asyncServer;
+		this.immediateExecution = immediateExecution;
 	}
 
 	/**
 	 * Add a new tool handler.
 	 * @param toolHandler The tool handler to add
 	 */
 	public void addTool(McpServerFeatures.SyncToolSpecification toolHandler) {
-		this.asyncServer.addTool(McpServerFeatures.AsyncToolSpecification.fromSync(toolHandler)).block();
+		this.asyncServer
+			.addTool(McpServerFeatures.AsyncToolSpecification.fromSync(toolHandler, this.immediateExecution))
+			.block();
 	}
 
 	/**
@@ -84,7 +99,10 @@ public void removeTool(String toolName) {
 	 * @param resourceHandler The resource handler to add
 	 */
 	public void addResource(McpServerFeatures.SyncResourceSpecification resourceHandler) {
-		this.asyncServer.addResource(McpServerFeatures.AsyncResourceSpecification.fromSync(resourceHandler)).block();
+		this.asyncServer
+			.addResource(
+					McpServerFeatures.AsyncResourceSpecification.fromSync(resourceHandler, this.immediateExecution))
+			.block();
 	}
 
 	/**
@@ -100,7 +118,10 @@ public void removeResource(String resourceUri) {
 	 * @param promptSpecification The prompt specification to add
 	 */
 	public void addPrompt(McpServerFeatures.SyncPromptSpecification promptSpecification) {
-		this.asyncServer.addPrompt(McpServerFeatures.AsyncPromptSpecification.fromSync(promptSpecification)).block();
+		this.asyncServer
+			.addPrompt(
+					McpServerFeatures.AsyncPromptSpecification.fromSync(promptSpecification, this.immediateExecution))
+			.block();
 	}
 
 	/**

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

@@ -155,6 +155,8 @@ void testRemoveTool() {
 		assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
 	}
 
+	// TODO: call tool? -> verify thread local still present
+
 	@Test
 	void testRemoveNonexistentTool() {
 		var mcpSyncServer = McpServer.sync(createMcpTransportProvider())