Integrated dedicated GET/listening stream and multi-transport management

Author: Zachary German

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

@@ -58,12 +58,12 @@ void setUp() {
 		// Set up session factory with proper server capabilities
 		McpSchema.ServerCapabilities serverCapabilities = new McpSchema.ServerCapabilities(null, null, null, null, null,
 				null);
-		serverTransportProvider.setStreamableHttpSessionFactory(
-				sessionId -> new io.modelcontextprotocol.spec.McpStreamableHttpServerSession(sessionId,
-						java.time.Duration.ofSeconds(30),
-						request -> reactor.core.publisher.Mono.just(new McpSchema.InitializeResult("2025-06-18",
-								serverCapabilities, new McpSchema.Implementation("Test Server", "1.0.0"), null)),
-						() -> reactor.core.publisher.Mono.empty(), java.util.Map.of(), java.util.Map.of()));
+		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",
+							serverCapabilities, new McpSchema.Implementation("Test Server", "1.0.0"), null)),
+					() -> reactor.core.publisher.Mono.empty(), java.util.Map.of(), java.util.Map.of()));
 
 		tomcat = TomcatTestUtil.createTomcatServer("", PORT, serverTransportProvider);
 		try {
@@ -132,14 +132,14 @@ void shouldCallImmediateToolSuccessfully() {
 		McpSchema.ServerCapabilities serverCapabilities = new McpSchema.ServerCapabilities(null, null, null, null, null,
 				new McpSchema.ServerCapabilities.ToolCapabilities(true));
 		serverTransportProvider
-			.setStreamableHttpSessionFactory(sessionId -> new io.modelcontextprotocol.spec.McpStreamableHttpServerSession(
-					sessionId, java.time.Duration.ofSeconds(30),
+			.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",
 							serverCapabilities, new McpSchema.Implementation("Test Server", "1.0.0"), null)),
 					() -> reactor.core.publisher.Mono.empty(),
 					java.util.Map.of("tools/call",
-							(io.modelcontextprotocol.spec.McpStreamableHttpServerSession.RequestHandler<CallToolResult>) (
-									exchange, params) -> tool.call().apply(exchange, (Map<String, Object>) params)),
+							(io.modelcontextprotocol.spec.McpServerSession.RequestHandler<CallToolResult>) (exchange,
+									params) -> tool.call().apply(exchange, (Map<String, Object>) params)),
 					java.util.Map.of()));
 
 		var mcpClient = clientBuilder.build();
@@ -174,12 +174,12 @@ void shouldCallStreamingToolSuccessfully() {
 		McpSchema.ServerCapabilities serverCapabilities = new McpSchema.ServerCapabilities(null, null, null, null, null,
 				new McpSchema.ServerCapabilities.ToolCapabilities(true));
 		serverTransportProvider
-			.setStreamableHttpSessionFactory(sessionId -> new io.modelcontextprotocol.spec.McpStreamableHttpServerSession(
-					sessionId, java.time.Duration.ofSeconds(30),
+			.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",
 							serverCapabilities, new McpSchema.Implementation("Test Server", "1.0.0"), null)),
 					() -> reactor.core.publisher.Mono.empty(), java.util.Map.of("tools/call",
-							(io.modelcontextprotocol.spec.McpStreamableHttpServerSession.StreamingRequestHandler<CallToolResult>) new io.modelcontextprotocol.spec.McpStreamableHttpServerSession.StreamingRequestHandler<CallToolResult>() {
+							(io.modelcontextprotocol.spec.McpServerSession.StreamingRequestHandler<CallToolResult>) new io.modelcontextprotocol.spec.McpServerSession.StreamingRequestHandler<CallToolResult>() {
 								@Override
 								public Mono<CallToolResult> handle(
 										io.modelcontextprotocol.server.McpAsyncServerExchange exchange, Object params) {

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

@@ -183,9 +183,9 @@ public class McpAsyncServer {
 		notificationHandlers.put(McpSchema.METHOD_NOTIFICATION_ROOTS_LIST_CHANGED,
 				asyncRootsListChangedNotificationHandler(rootsChangeConsumers));
 
-		mcpTransportProvider.setSessionFactory(
-				transport -> new McpServerSession(UUID.randomUUID().toString(), requestTimeout, transport,
-						this::asyncInitializeRequestHandler, Mono::empty, requestHandlers, notificationHandlers));
+		mcpTransportProvider.setSessionFactory(listeningTransport -> new McpServerSession(UUID.randomUUID().toString(),
+				requestTimeout, listeningTransport, this::asyncInitializeRequestHandler, Mono::empty, requestHandlers,
+				notificationHandlers));
 	}
 
 	// ---------------------------------------

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

@@ -13,7 +13,6 @@
 import io.modelcontextprotocol.spec.McpSchema.LoggingLevel;
 import io.modelcontextprotocol.spec.McpSchema.LoggingMessageNotification;
 import io.modelcontextprotocol.spec.McpSession;
-import io.modelcontextprotocol.spec.McpStreamableHttpServerSession;
 import io.modelcontextprotocol.spec.McpServerSession;
 import io.modelcontextprotocol.util.Assert;
 import reactor.core.publisher.Mono;
@@ -61,20 +60,6 @@ public McpAsyncServerExchange(McpServerSession session, McpSchema.ClientCapabili
 		this.clientInfo = clientInfo;
 	}
 
-	/**
-	 * Create a new asynchronous exchange with the client.
-	 * @param session The server session representing a 1-1 interaction.
-	 * @param clientCapabilities The client capabilities that define the supported
-	 * features and functionality.
-	 * @param clientInfo The client implementation information.
-	 */
-	public McpAsyncServerExchange(McpStreamableHttpServerSession session,
-			McpSchema.ClientCapabilities clientCapabilities, McpSchema.Implementation clientInfo) {
-		this.session = session;
-		this.clientCapabilities = clientCapabilities;
-		this.clientInfo = clientInfo;
-	}
-
 	/**
 	 * Get the client capabilities that define the supported features and functionality.
 	 * @return The client capabilities

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

@@ -15,7 +15,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import io.modelcontextprotocol.server.transport.StreamableHttpServerTransportProvider;
-import io.modelcontextprotocol.spec.McpStreamableHttpServerSession;
+import io.modelcontextprotocol.spec.McpServerSession;
 import io.modelcontextprotocol.spec.McpSchema;
 import io.modelcontextprotocol.util.Assert;
 import io.modelcontextprotocol.util.DeafaultMcpUriTemplateManagerFactory;
@@ -86,7 +86,7 @@ public class McpAsyncStreamableHttpServer {
 	 * Sets up the request handlers for standard MCP methods.
 	 */
 	private void setupRequestHandlers() {
-		Map<String, McpStreamableHttpServerSession.RequestHandler<?>> requestHandlers = new HashMap<>();
+		Map<String, McpServerSession.RequestHandler<?>> requestHandlers = new HashMap<>();
 
 		// Ping handler
 		requestHandlers.put(McpSchema.METHOD_PING, (exchange, params) -> Mono.just(Map.of()));
@@ -123,15 +123,15 @@ private void setupRequestHandlers() {
 		this.requestHandlers = requestHandlers;
 	}
 
-	private Map<String, McpStreamableHttpServerSession.RequestHandler<?>> requestHandlers;
+	private Map<String, McpServerSession.RequestHandler<?>> requestHandlers;
 
-	private Map<String, McpStreamableHttpServerSession.NotificationHandler> notificationHandlers;
+	private Map<String, McpServerSession.NotificationHandler> notificationHandlers;
 
 	/**
 	 * Sets up notification handlers.
 	 */
 	private void setupNotificationHandlers() {
-		Map<String, McpStreamableHttpServerSession.NotificationHandler> handlers = new HashMap<>();
+		Map<String, McpServerSession.NotificationHandler> handlers = new HashMap<>();
 
 		handlers.put(McpSchema.METHOD_NOTIFICATION_INITIALIZED, (exchange, params) -> {
 			logger.info("[INIT] Received initialized notification - initialization complete!");
@@ -150,7 +150,7 @@ private void setupNotificationHandlers() {
 	private void setupSessionFactory() {
 		setupNotificationHandlers();
 
-		httpTransportProvider.setStreamableHttpSessionFactory(sessionId -> new McpStreamableHttpServerSession(sessionId,
+		httpTransportProvider.setStreamableHttpSessionFactory(sessionId -> new McpServerSession(sessionId,
 				requestTimeout, this::handleInitializeRequest, Mono::empty, requestHandlers, notificationHandlers));
 	}
 
@@ -181,7 +181,7 @@ private Mono<McpSchema.InitializeResult> handleInitializeRequest(McpSchema.Initi
 	}
 
 	// Request handler creation methods
-	private McpStreamableHttpServerSession.RequestHandler<McpSchema.ListToolsResult> createToolsListHandler() {
+	private McpServerSession.RequestHandler<McpSchema.ListToolsResult> createToolsListHandler() {
 		return (exchange, params) -> {
 			var regularTools = features.tools().stream().map(McpServerFeatures.AsyncToolSpecification::tool).toList();
 			var streamingTools = features.streamTools()
@@ -194,8 +194,8 @@ private McpStreamableHttpServerSession.RequestHandler<McpSchema.ListToolsResult>
 		};
 	}
 
-	private McpStreamableHttpServerSession.RequestHandler<McpSchema.CallToolResult> createToolsCallHandler() {
-		return new McpStreamableHttpServerSession.StreamingRequestHandler<McpSchema.CallToolResult>() {
+	private McpServerSession.RequestHandler<McpSchema.CallToolResult> createToolsCallHandler() {
+		return new McpServerSession.StreamingRequestHandler<McpSchema.CallToolResult>() {
 			@Override
 			public Mono<McpSchema.CallToolResult> handle(McpAsyncServerExchange exchange, Object params) {
 				var callToolRequest = objectMapper.convertValue(params, McpSchema.CallToolRequest.class);
@@ -252,7 +252,7 @@ public Flux<McpSchema.CallToolResult> handleStreaming(McpAsyncServerExchange exc
 		};
 	}
 
-	private McpStreamableHttpServerSession.RequestHandler<McpSchema.ListResourcesResult> createResourcesListHandler() {
+	private McpServerSession.RequestHandler<McpSchema.ListResourcesResult> createResourcesListHandler() {
 		return (exchange, params) -> {
 			var resources = features.resources()
 				.values()
@@ -263,7 +263,7 @@ private McpStreamableHttpServerSession.RequestHandler<McpSchema.ListResourcesRes
 		};
 	}
 
-	private McpStreamableHttpServerSession.RequestHandler<McpSchema.ReadResourceResult> createResourcesReadHandler() {
+	private McpServerSession.RequestHandler<McpSchema.ReadResourceResult> createResourcesReadHandler() {
 		return (exchange, params) -> {
 			var resourceRequest = objectMapper.convertValue(params, McpSchema.ReadResourceRequest.class);
 			var resourceUri = resourceRequest.uri();
@@ -278,12 +278,12 @@ private McpStreamableHttpServerSession.RequestHandler<McpSchema.ReadResourceResu
 		};
 	}
 
-	private McpStreamableHttpServerSession.RequestHandler<McpSchema.ListResourceTemplatesResult> createResourceTemplatesListHandler() {
+	private McpServerSession.RequestHandler<McpSchema.ListResourceTemplatesResult> createResourceTemplatesListHandler() {
 		return (exchange, params) -> Mono
 			.just(new McpSchema.ListResourceTemplatesResult(features.resourceTemplates(), null));
 	}
 
-	private McpStreamableHttpServerSession.RequestHandler<McpSchema.ListPromptsResult> createPromptsListHandler() {
+	private McpServerSession.RequestHandler<McpSchema.ListPromptsResult> createPromptsListHandler() {
 		return (exchange, params) -> {
 			var prompts = features.prompts()
 				.values()
@@ -294,7 +294,7 @@ private McpStreamableHttpServerSession.RequestHandler<McpSchema.ListPromptsResul
 		};
 	}
 
-	private McpStreamableHttpServerSession.RequestHandler<McpSchema.GetPromptResult> createPromptsGetHandler() {
+	private McpServerSession.RequestHandler<McpSchema.GetPromptResult> createPromptsGetHandler() {
 		return (exchange, params) -> {
 			var promptRequest = objectMapper.convertValue(params, McpSchema.GetPromptRequest.class);
 
@@ -308,15 +308,15 @@ private McpStreamableHttpServerSession.RequestHandler<McpSchema.GetPromptResult>
 		};
 	}
 
-	private McpStreamableHttpServerSession.RequestHandler<Object> createLoggingSetLevelHandler() {
+	private McpServerSession.RequestHandler<Object> createLoggingSetLevelHandler() {
 		return (exchange, params) -> {
 			var setLevelRequest = objectMapper.convertValue(params, McpSchema.SetLevelRequest.class);
 			exchange.setMinLoggingLevel(setLevelRequest.level());
 			return Mono.just(Map.of());
 		};
 	}
 
-	private McpStreamableHttpServerSession.RequestHandler<McpSchema.CompleteResult> createCompletionCompleteHandler() {
+	private McpServerSession.RequestHandler<McpSchema.CompleteResult> createCompletionCompleteHandler() {
 		return (exchange, params) -> {
 			var completeRequest = objectMapper.convertValue(params, McpSchema.CompleteRequest.class);
 
@@ -330,7 +330,7 @@ private McpStreamableHttpServerSession.RequestHandler<McpSchema.CompleteResult>
 		};
 	}
 
-	private McpStreamableHttpServerSession.NotificationHandler createRootsListChangedHandler() {
+	private McpServerSession.NotificationHandler createRootsListChangedHandler() {
 		return (exchange, params) -> {
 			var rootsChangeConsumers = features.rootsChangeConsumers();
 			if (rootsChangeConsumers.isEmpty()) {
@@ -418,8 +418,8 @@ public static Builder builder() {
 	 * transport.
 	 *
 	 * <p>
-	 * This builder provides a fluent API for configuring Streamable HTTP MCP
-	 * servers with enhanced features:
+	 * This builder provides a fluent API for configuring Streamable HTTP MCP servers with
+	 * enhanced features:
 	 * <ul>
 	 * <li>Single session class managing all transport streams</li>
 	 * <li>Resource management and lifecycle handling</li>