Fixed backward-compatibility issues

Author: Zachary German

mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/server/transport/StreamableHttpTransportIntegrationTest.java

@@ -61,7 +61,7 @@ void setUp() {
 		serverTransportProvider
 			.setStreamableHttpSessionFactory(sessionId -> new io.modelcontextprotocol.spec.McpServerSession(sessionId,
 					java.time.Duration.ofSeconds(30),
-					request -> reactor.core.publisher.Mono.just(new McpSchema.InitializeResult("2025-06-18",
+					request -> reactor.core.publisher.Mono.just(new McpSchema.InitializeResult("2024-11-05",
 							serverCapabilities, new McpSchema.Implementation("Test Server", "1.0.0"), null)),
 					() -> reactor.core.publisher.Mono.empty(), java.util.Map.of(), java.util.Map.of()));
 
@@ -134,7 +134,7 @@ void shouldCallImmediateToolSuccessfully() {
 		serverTransportProvider
 			.setStreamableHttpSessionFactory(sessionId -> new io.modelcontextprotocol.spec.McpServerSession(sessionId,
 					java.time.Duration.ofSeconds(30),
-					request -> reactor.core.publisher.Mono.just(new McpSchema.InitializeResult("2025-06-18",
+					request -> reactor.core.publisher.Mono.just(new McpSchema.InitializeResult("2024-11-05",
 							serverCapabilities, new McpSchema.Implementation("Test Server", "1.0.0"), null)),
 					() -> reactor.core.publisher.Mono.empty(),
 					java.util.Map.of("tools/call",
@@ -176,7 +176,7 @@ void shouldCallStreamingToolSuccessfully() {
 		serverTransportProvider
 			.setStreamableHttpSessionFactory(sessionId -> new io.modelcontextprotocol.spec.McpServerSession(sessionId,
 					java.time.Duration.ofSeconds(30),
-					request -> reactor.core.publisher.Mono.just(new McpSchema.InitializeResult("2025-06-18",
+					request -> reactor.core.publisher.Mono.just(new McpSchema.InitializeResult("2024-11-05",
 							serverCapabilities, new McpSchema.Implementation("Test Server", "1.0.0"), null)),
 					() -> reactor.core.publisher.Mono.empty(), java.util.Map.of("tools/call",
 							(io.modelcontextprotocol.spec.McpServerSession.StreamingRequestHandler<CallToolResult>) new io.modelcontextprotocol.spec.McpServerSession.StreamingRequestHandler<CallToolResult>() {
@@ -215,7 +215,7 @@ void shouldReceiveNotificationThroughGetStream() throws InterruptedException {
 		AtomicReference<String> sessionId = new AtomicReference<>();
 
 		WebClient webClient = WebClient.create("http://localhost:" + PORT);
-		String initMessage = "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":\"2025-06-18\",\"capabilities\":{},\"clientInfo\":{\"name\":\"Test\",\"version\":\"1.0\"}}}";
+		String initMessage = "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":\"2024-11-05\",\"capabilities\":{},\"clientInfo\":{\"name\":\"Test\",\"version\":\"1.0\"}}}";
 
 		// Initialize and get session ID
 		webClient.post()

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

@@ -88,12 +88,16 @@ public class McpAsyncServer {
 
 	private final McpSchema.ServerCapabilities serverCapabilities;
 
+	private final boolean isStreamableHttp;
+
 	private final McpSchema.Implementation serverInfo;
 
 	private final String instructions;
 
 	private final CopyOnWriteArrayList<McpServerFeatures.AsyncToolSpecification> tools = new CopyOnWriteArrayList<>();
 
+	private final CopyOnWriteArrayList<McpServerFeatures.AsyncStreamingToolSpecification> streamTools = new CopyOnWriteArrayList<>();
+
 	private final CopyOnWriteArrayList<McpSchema.ResourceTemplate> resourceTemplates = new CopyOnWriteArrayList<>();
 
 	private final ConcurrentHashMap<String, McpServerFeatures.AsyncResourceSpecification> resources = new ConcurrentHashMap<>();
@@ -119,7 +123,7 @@ public class McpAsyncServer {
 	 */
 	McpAsyncServer(McpServerTransportProvider mcpTransportProvider, ObjectMapper objectMapper,
 			McpServerFeatures.Async features, Duration requestTimeout,
-			McpUriTemplateManagerFactory uriTemplateManagerFactory) {
+			McpUriTemplateManagerFactory uriTemplateManagerFactory, boolean isStreamableHttp) {
 		this.mcpTransportProvider = mcpTransportProvider;
 		this.objectMapper = objectMapper;
 		this.serverInfo = features.serverInfo();
@@ -131,6 +135,7 @@ public class McpAsyncServer {
 		this.prompts.putAll(features.prompts());
 		this.completions.putAll(features.completions());
 		this.uriTemplateManagerFactory = uriTemplateManagerFactory;
+		this.isStreamableHttp = isStreamableHttp;
 
 		Map<String, McpServerSession.RequestHandler<?>> requestHandlers = new HashMap<>();
 
@@ -188,6 +193,13 @@ public class McpAsyncServer {
 				notificationHandlers));
 	}
 
+	// Alternate constructor for HTTP+SSE servers (past spec)
+	McpAsyncServer(McpServerTransportProvider mcpTransportProvider, ObjectMapper objectMapper,
+			McpServerFeatures.Async features, Duration requestTimeout,
+			McpUriTemplateManagerFactory uriTemplateManagerFactory) {
+		this(mcpTransportProvider, objectMapper, features, requestTimeout, uriTemplateManagerFactory, false);
+	}
+
 	// ---------------------------------------
 	// Lifecycle Management
 	// ---------------------------------------
@@ -329,6 +341,69 @@ public Mono<Void> removeTool(String toolName) {
 		});
 	}
 
+	/**
+	 * Add a new tool specification at runtime.
+	 * @param toolSpecification The tool specification to add
+	 * @return Mono that completes when clients have been notified of the change
+	 */
+	public Mono<Void> addStreamTool(McpServerFeatures.AsyncStreamingToolSpecification toolSpecification) {
+		if (toolSpecification == null) {
+			return Mono.error(new McpError("Tool specification must not be null"));
+		}
+		if (toolSpecification.tool() == null) {
+			return Mono.error(new McpError("Tool must not be null"));
+		}
+		if (toolSpecification.call() == null) {
+			return Mono.error(new McpError("Tool call handler must not be null"));
+		}
+		if (this.serverCapabilities.tools() == null) {
+			return Mono.error(new McpError("Server must be configured with tool capabilities"));
+		}
+
+		return Mono.defer(() -> {
+			// Check for duplicate tool names
+			if (this.streamTools.stream().anyMatch(th -> th.tool().name().equals(toolSpecification.tool().name()))) {
+				return Mono
+					.error(new McpError("Tool with name '" + toolSpecification.tool().name() + "' already exists"));
+			}
+
+			this.streamTools.add(toolSpecification);
+			logger.debug("Added tool handler: {}", toolSpecification.tool().name());
+
+			if (this.serverCapabilities.tools().listChanged()) {
+				return notifyToolsListChanged();
+			}
+			return Mono.empty();
+		});
+	}
+
+	/**
+	 * Remove a tool handler at runtime.
+	 * @param toolName The name of the tool handler to remove
+	 * @return Mono that completes when clients have been notified of the change
+	 */
+	public Mono<Void> removeStreamTool(String toolName) {
+		if (toolName == null) {
+			return Mono.error(new McpError("Tool name must not be null"));
+		}
+		if (this.serverCapabilities.tools() == null) {
+			return Mono.error(new McpError("Server must be configured with tool capabilities"));
+		}
+
+		return Mono.defer(() -> {
+			boolean removed = this.tools
+				.removeIf(toolSpecification -> toolSpecification.tool().name().equals(toolName));
+			if (removed) {
+				logger.debug("Removed tool handler: {}", toolName);
+				if (this.serverCapabilities.tools().listChanged()) {
+					return notifyToolsListChanged();
+				}
+				return Mono.empty();
+			}
+			return Mono.error(new McpError("Tool with name '" + toolName + "' not found"));
+		});
+	}
+
 	/**
 	 * Notifies clients that the list of available tools has changed.
 	 * @return A Mono that completes when all clients have been notified
@@ -339,29 +414,97 @@ public Mono<Void> notifyToolsListChanged() {
 
 	private McpServerSession.RequestHandler<McpSchema.ListToolsResult> toolsListRequestHandler() {
 		return (exchange, params) -> {
-			List<Tool> tools = this.tools.stream().map(McpServerFeatures.AsyncToolSpecification::tool).toList();
+			List<Tool> tools = new ArrayList<>();
+			tools.addAll(this.tools.stream().map(McpServerFeatures.AsyncToolSpecification::tool).toList());
+			tools.addAll(
+					this.streamTools.stream().map(McpServerFeatures.AsyncStreamingToolSpecification::tool).toList());
 
 			return Mono.just(new McpSchema.ListToolsResult(tools, null));
 		};
 	}
 
 	private McpServerSession.RequestHandler<CallToolResult> toolsCallRequestHandler() {
-		return (exchange, params) -> {
-			McpSchema.CallToolRequest callToolRequest = objectMapper.convertValue(params,
-					new TypeReference<McpSchema.CallToolRequest>() {
-					});
+		if (isStreamableHttp) {
+			return new McpServerSession.StreamingRequestHandler<CallToolResult>() {
+				@Override
+				public Mono<CallToolResult> handle(McpAsyncServerExchange exchange, Object params) {
+					var callToolRequest = objectMapper.convertValue(params, McpSchema.CallToolRequest.class);
+
+					// Check regular tools first
+					var regularTool = tools.stream()
+						.filter(tool -> callToolRequest.name().equals(tool.tool().name()))
+						.findFirst();
+
+					if (regularTool.isPresent()) {
+						return regularTool.get().call().apply(exchange, callToolRequest.arguments());
+					}
+
+					// Check streaming tools (take first result)
+					var streamingTool = streamTools.stream()
+						.filter(tool -> callToolRequest.name().equals(tool.tool().name()))
+						.findFirst();
+
+					if (streamingTool.isPresent()) {
+						return streamingTool.get().call().apply(exchange, callToolRequest.arguments()).next();
+					}
+
+					return Mono.error(new McpError("Tool not found: " + callToolRequest.name()));
+				}
 
-			Optional<McpServerFeatures.AsyncToolSpecification> toolSpecification = this.tools.stream()
-				.filter(tr -> callToolRequest.name().equals(tr.tool().name()))
-				.findAny();
+				@Override
+				public Flux<CallToolResult> handleStreaming(McpAsyncServerExchange exchange, Object params) {
+					var callToolRequest = objectMapper.convertValue(params, McpSchema.CallToolRequest.class);
 
-			if (toolSpecification.isEmpty()) {
-				return Mono.error(new McpError("Tool not found: " + callToolRequest.name()));
-			}
+					// Check streaming tools first (preferred for streaming)
+					var streamingTool = streamTools.stream()
+						.filter(tool -> callToolRequest.name().equals(tool.tool().name()))
+						.findFirst();
 
-			return toolSpecification.map(tool -> tool.call().apply(exchange, callToolRequest.arguments()))
-				.orElse(Mono.error(new McpError("Tool not found: " + callToolRequest.name())));
-		};
+					if (streamingTool.isPresent()) {
+						return streamingTool.get().call().apply(exchange, callToolRequest.arguments());
+					}
+
+					// Fallback to regular tools (convert Mono to Flux)
+					var regularTool = tools.stream()
+						.filter(tool -> callToolRequest.name().equals(tool.tool().name()))
+						.findFirst();
+
+					if (regularTool.isPresent()) {
+						return regularTool.get().call().apply(exchange, callToolRequest.arguments()).flux();
+					}
+
+					return Flux.error(new McpError("Tool not found: " + callToolRequest.name()));
+				}
+			};
+		}
+		else {
+			return (exchange, params) -> {
+				McpSchema.CallToolRequest callToolRequest = objectMapper.convertValue(params,
+						new TypeReference<McpSchema.CallToolRequest>() {
+						});
+
+				// Check regular tools first
+				Optional<McpServerFeatures.AsyncToolSpecification> toolSpecification = this.tools.stream()
+					.filter(tr -> callToolRequest.name().equals(tr.tool().name()))
+					.findAny();
+
+				if (toolSpecification.isPresent()) {
+					return toolSpecification.get().call().apply(exchange, callToolRequest.arguments());
+				}
+
+				// Check streaming tools (take first result)
+				Optional<McpServerFeatures.AsyncStreamingToolSpecification> streamToolSpecification = this.streamTools
+					.stream()
+					.filter(tr -> callToolRequest.name().equals(tr.tool().name()))
+					.findAny();
+
+				if (streamToolSpecification.isPresent()) {
+					return streamToolSpecification.get().call().apply(exchange, callToolRequest.arguments()).next();
+				}
+
+				return Mono.error(new McpError("Tool not found: " + callToolRequest.name()));
+			};
+		}
 	}
 
 	// ---------------------------------------

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

@@ -26,19 +26,43 @@
 import reactor.core.publisher.Mono;
 
 /**
- * Streamable HTTP-based MCP server implementation that uses a single session class to
- * manage all streams and transports efficiently.
+ * The Model Context Protocol (MCP) server implementation that provides asynchronous
+ * communication using Project Reactor's Mono and Flux types.
  *
  * <p>
- * Featuring:
+ * This server implements the MCP specification, enabling AI models to expose tools,
+ * resources, and prompts through a standardized interface. Key features include:
  * <ul>
- * <li>Single session class manages all transport streams</li>
- * <li>Simplified transport registration and management</li>
- * <li>Clear separation of concerns between session and transport</li>
- * <li>Efficient resource management</li>
- * <li>API for handling both immediate and streaming responses</li>
+ * <li>Asynchronous communication using reactive programming patterns
+ * <li>Dynamic tool registration and management
+ * <li>Resource handling with URI-based addressing
+ * <li>Prompt template management
+ * <li>Real-time client notifications for state changes
+ * <li>Structured logging with configurable severity levels
+ * <li>Support for client-side AI model sampling
  * </ul>
  *
+ * <p>
+ * The server follows a lifecycle:
+ * <ol>
+ * <li>Initialization - Accepts client connections and negotiates capabilities
+ * <li>Normal Operation - Handles client requests and sends notifications
+ * <li>Graceful Shutdown - Ensures clean connection termination
+ * </ol>
+ *
+ * <p>
+ * This implementation uses Project Reactor for non-blocking operations, making it
+ * suitable for high-throughput scenarios and reactive applications. All operations return
+ * Mono or Flux types that can be composed into reactive pipelines.
+ *
+ * <p>
+ * The server supports runtime modification of its capabilities through methods like
+ * {@link #addTool}, {@link #addResource}, and {@link #addPrompt}, automatically notifying
+ * connected clients of changes when configured to do so.
+ *
+ * @author Christian Tzolov
+ * @author Dariusz Jędrzejczyk
+ * @author Jihoon Kim
  * @author Zachary German
  */
 public class McpAsyncStreamableHttpServer {

mcp/src/main/java/io/modelcontextprotocol/server/transport/HttpServletSseServerTransportProvider.java

@@ -215,8 +215,8 @@ protected void doGet(HttpServletRequest request, HttpServletResponse response)
 		PrintWriter writer = response.getWriter();
 
 		// Create a new session transport
-		HttpServletMcpSessionTransport sessionTransport = new HttpServletMcpSessionTransport(sessionId, asyncContext,
-				writer);
+		HttpServletMcpSessionTransport sessionTransport = new HttpServletMcpSessionTransport(
+				McpServerSession.LISTENING_TRANSPORT, asyncContext, writer);
 
 		// Create a new session using the session factory
 		McpServerSession session = sessionFactory.create(sessionTransport);

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

@@ -15,7 +15,7 @@ public McpError(JSONRPCError jsonRpcError) {
 	}
 
 	public McpError(Object error) {
-		super(error.toString());
+		super(String.valueOf(error));
 	}
 
 	public JSONRPCError getJsonRpcError() {

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

@@ -27,7 +27,7 @@
 /**
  * Based on the <a href="http://www.jsonrpc.org/specification">JSON-RPC 2.0
  * specification</a> and the <a href=
- * "https://github.com/modelcontextprotocol/specification/blob/main/schema/2025-06-18/schema.ts">Model
+ * "https://github.com/modelcontextprotocol/specification/blob/main/schema/2024-11-05/schema.ts">Model
  * Context Protocol Schema</a>.
  *
  * @author Christian Tzolov
@@ -40,7 +40,7 @@ public final class McpSchema {
 	private McpSchema() {
 	}
 
-	public static final String LATEST_PROTOCOL_VERSION = "2025-06-18";
+	public static final String LATEST_PROTOCOL_VERSION = "2024-11-05";
 
 	public static final String JSONRPC_VERSION = "2.0";