backup

Author: Mohammod Al Amin Ashik

Cargo.lock

@@ -102,14 +102,14 @@ impl ServerHandler for McpMuxGatewayHandler {
             protocol_version: Default::default(),
             capabilities: ServerCapabilities::builder()
                 .enable_tools_with(ToolsCapability {
-                    list_changed: Some(false), // Stateless mode - no notifications
+                    list_changed: Some(true),
                 })
                 .enable_prompts_with(PromptsCapability {
-                    list_changed: Some(false), // Stateless mode - no notifications
+                    list_changed: Some(true),
                 })
                 .enable_resources_with(ResourcesCapability {
                     subscribe: Some(false),
-                    list_changed: Some(false), // Stateless mode - no notifications
+                    list_changed: Some(true),
                 })
                 .build(),
             server_info: Implementation {
@@ -150,19 +150,34 @@ impl ServerHandler for McpMuxGatewayHandler {
     }
 
     async fn on_initialized(&self, context: NotificationContext<RoleServer>) {
-        // Silently process - entry already logged in oauth_middleware
-        let _oauth_ctx = match self.get_oauth_context(&context.extensions) {
+        let oauth_ctx = match self.get_oauth_context(&context.extensions) {
             Ok(ctx) => ctx,
             Err(e) => {
-                warn!("Failed to extract OAuth context: {}", e);
+                warn!("Failed to extract OAuth context on_initialized: {}", e);
                 return;
             }
         };
 
-        // In stateless mode:
-        // - No session tracking
-        // - No notification registration
-        // - Each request is independent
+        // Register peer with MCPNotifier for list_changed notification delivery
+        let peer = std::sync::Arc::new(context.peer);
+        self.notification_bridge
+            .register_peer(oauth_ctx.client_id.clone(), peer);
+
+        // Mark the client stream as active immediately - RMCP's session transport
+        // handles SSE streaming and message caching internally
+        self.notification_bridge
+            .mark_client_stream_active(&oauth_ctx.client_id);
+
+        // Pre-populate feature hashes to prevent spurious first notifications
+        self.notification_bridge
+            .prime_hashes_for_space(oauth_ctx.space_id)
+            .await;
+
+        info!(
+            client_id = %oauth_ctx.client_id,
+            space_id = %oauth_ctx.space_id,
+            "Client initialized - peer registered for notifications"
+        );
     }
 
     async fn list_tools(

crates/mcpmux-gateway/src/mcp/handler.rs

@@ -240,29 +240,27 @@ impl GatewayServer {
         let handler =
             McpMuxGatewayHandler::new(Arc::new(self.services.clone()), notification_bridge.clone());
 
-        // Create STATELESS MCP service
-        // stateful_mode: false means:
-        // - No Mcp-Session-Id header
-        // - GET/DELETE return 405 automatically (no notification streams)
-        // - Each POST is independent
-        // - Avoids the stream management issues that caused connection loops
-        // Trade-off: Cannot send list_changed notifications
+        // Create STATEFUL MCP service (full Streamable HTTP per spec 2025-11-25)
+        // stateful_mode: true means:
+        // - Mcp-Session-Id header for session management
+        // - GET endpoint for SSE streams (server-initiated notifications)
+        // - DELETE endpoint for session termination
+        // - list_changed notifications delivered via SSE
         let mcp_service = StreamableHttpService::new(
             move || {
-                debug!("[Gateway] Creating handler instance for MCP request");
+                debug!("[Gateway] Creating handler instance for MCP session");
                 Ok(handler.clone())
             },
             LocalSessionManager::default().into(),
             StreamableHttpServerConfig {
-                stateful_mode: false,
+                stateful_mode: true,
                 sse_keep_alive: Some(std::time::Duration::from_secs(30)),
-                sse_retry: None,
+                sse_retry: Some(std::time::Duration::from_secs(3)),
                 cancellation_token: CancellationToken::new(),
             },
         );
 
         // Wrap MCP service with OAuth middleware
-        // In stateless mode, no session healing needed - rmcp handles 405 for GET/DELETE
         let mcp_routes =
             Router::new()
                 .nest_service("/mcp", mcp_service)

crates/mcpmux-gateway/src/server/mod.rs

@@ -49,6 +49,16 @@ url = "2.5"
 # Sync primitives for tests
 parking_lot = "0.12"
 
+# RMCP for streamable HTTP transport tests
+rmcp = { version = "0.14.0", features = [
+    "client",
+    "server",
+    "transport-streamable-http-server",
+    "transport-streamable-http-client-reqwest",
+] }
+tokio-util = { version = "0.7", features = ["rt"] }
+axum = "0.8"
+
 [lib]
 path = "src/lib.rs"
 
@@ -71,3 +81,7 @@ path = "tests/oauth/mod.rs"
 [[test]]
 name = "integration"
 path = "tests/integration/mod.rs"
+
+[[test]]
+name = "streamable_http"
+path = "tests/streamable_http/mod.rs"

tests/rust/Cargo.toml

@@ -0,0 +1,8 @@
+//! Streamable HTTP Transport Integration Tests
+//!
+//! Tests the full stateful Streamable HTTP transport with:
+//! - Session management (Mcp-Session-Id)
+//! - Server-initiated notifications (list_changed via SSE)
+//! - Proper protocol negotiation
+
+mod notifications;