turul_http_mcp_server/
server.rs

1//! HTTP MCP Server with SessionStorage integration
2//!
3//! This server provides MCP 2025-11-25 compliant HTTP transport with
4//! pluggable session storage backends and proper SSE resumability.
5
6use bytes::Bytes;
7use http_body_util::{BodyExt, Full};
8use hyper::server::conn::http1;
9use hyper::service::service_fn;
10use hyper::{Request, Response};
11use hyper_util::rt::TokioIo;
12use std::net::SocketAddr;
13use std::sync::Arc;
14use tokio::net::TcpListener;
15use tracing::{debug, error, info, warn};
16
17use turul_mcp_json_rpc_server::{JsonRpcDispatcher, JsonRpcHandler};
18use turul_mcp_protocol::McpError;
19use turul_mcp_session_storage::InMemorySessionStorage;
20
21use crate::streamable_http::{McpProtocolVersion, StreamableHttpHandler};
22use crate::{CorsLayer, Result, SessionMcpHandler, StreamConfig, StreamManager};
23
24/// Configuration for the HTTP MCP server
25#[derive(Debug, Clone)]
26pub struct ServerConfig {
27    /// Address to bind to
28    pub bind_address: SocketAddr,
29    /// Path for MCP endpoint
30    pub mcp_path: String,
31    /// Enable CORS
32    pub enable_cors: bool,
33    /// Maximum request body size
34    pub max_body_size: usize,
35    /// Enable GET SSE support (persistent event streams)
36    pub enable_get_sse: bool,
37    /// Enable POST SSE support (streaming tool call responses) - disabled by default for compatibility
38    pub enable_post_sse: bool,
39    /// Session expiry time in minutes (default: 30 minutes)
40    pub session_expiry_minutes: u64,
41    /// Allow ping requests without Mcp-Session-Id header (default: true)
42    ///
43    /// When true, the server accepts pre-initialization `ping` requests without
44    /// requiring a session. The full middleware stack still runs with `session=None`,
45    /// so rate-limiting middleware can still block unauthenticated pings.
46    ///
47    /// Set to false for hardened deployments that require session for all methods.
48    pub allow_unauthenticated_ping: bool,
49}
50
51impl Default for ServerConfig {
52    fn default() -> Self {
53        Self {
54            bind_address: "127.0.0.1:8000".parse().unwrap(),
55            mcp_path: "/mcp".to_string(),
56            enable_cors: true,
57            max_body_size: 1024 * 1024,            // 1MB
58            enable_get_sse: cfg!(feature = "sse"), // GET SSE enabled if "sse" feature is compiled
59            enable_post_sse: false, // Disabled by default for better client compatibility (e.g., MCP Inspector)
60            session_expiry_minutes: 30, // 30 minutes default
61            allow_unauthenticated_ping: true, // Allow pre-init pings per MCP spec
62        }
63    }
64}
65
66/// Builder for HTTP MCP server with pluggable storage
67pub struct HttpMcpServerBuilder {
68    config: ServerConfig,
69    dispatcher: JsonRpcDispatcher<McpError>,
70    session_storage: Option<Arc<turul_mcp_session_storage::BoxedSessionStorage>>,
71    stream_config: StreamConfig,
72    server_capabilities: Option<turul_mcp_protocol::ServerCapabilities>,
73    middleware_stack: Arc<crate::middleware::MiddlewareStack>,
74    route_registry: Arc<crate::routes::RouteRegistry>,
75    tool_fingerprint: Option<String>,
76    tool_notifier: Option<Arc<dyn crate::ToolChangeNotifier>>,
77}
78
79impl HttpMcpServerBuilder {
80    /// Create a new builder with in-memory storage (zero-configuration)
81    pub fn new() -> Self {
82        Self {
83            config: ServerConfig::default(),
84            dispatcher: JsonRpcDispatcher::<McpError>::new(),
85            session_storage: Some(Arc::new(InMemorySessionStorage::new())),
86            stream_config: StreamConfig::default(),
87            server_capabilities: None,
88            middleware_stack: Arc::new(crate::middleware::MiddlewareStack::new()),
89            route_registry: Arc::new(crate::routes::RouteRegistry::new()),
90            tool_fingerprint: None,
91            tool_notifier: None,
92        }
93    }
94}
95
96impl HttpMcpServerBuilder {
97    /// Create a new builder with specific session storage
98    pub fn with_storage(
99        session_storage: Arc<turul_mcp_session_storage::BoxedSessionStorage>,
100    ) -> Self {
101        Self {
102            config: ServerConfig::default(),
103            dispatcher: JsonRpcDispatcher::<McpError>::new(),
104            session_storage: Some(session_storage),
105            stream_config: StreamConfig::default(),
106            server_capabilities: None,
107            middleware_stack: Arc::new(crate::middleware::MiddlewareStack::new()),
108            route_registry: Arc::new(crate::routes::RouteRegistry::new()),
109            tool_fingerprint: None,
110            tool_notifier: None,
111        }
112    }
113
114    /// Set the middleware stack (for HTTP transport middleware support)
115    pub fn with_middleware_stack(
116        mut self,
117        middleware_stack: Arc<crate::middleware::MiddlewareStack>,
118    ) -> Self {
119        self.middleware_stack = middleware_stack;
120        self
121    }
122
123    /// Set the route registry for custom HTTP paths (e.g., `.well-known`)
124    pub fn route_registry(mut self, registry: Arc<crate::routes::RouteRegistry>) -> Self {
125        self.route_registry = registry;
126        self
127    }
128
129    /// Set tool fingerprint for session versioning across server restarts
130    pub fn tool_fingerprint(mut self, fingerprint: String) -> Self {
131        if fingerprint.is_empty() {
132            self.tool_fingerprint = None; // Static mode: no fingerprint check
133        } else {
134            self.tool_fingerprint = Some(fingerprint);
135        }
136        self
137    }
138
139    /// Set the tool change notifier for restart/redeploy fingerprint mismatch.
140    pub fn tool_notifier(mut self, notifier: Arc<dyn crate::ToolChangeNotifier>) -> Self {
141        self.tool_notifier = Some(notifier);
142        self
143    }
144
145    /// Set the bind address
146    pub fn bind_address(mut self, addr: SocketAddr) -> Self {
147        self.config.bind_address = addr;
148        self
149    }
150
151    /// Set the MCP endpoint path
152    pub fn mcp_path(mut self, path: impl Into<String>) -> Self {
153        self.config.mcp_path = path.into();
154        self
155    }
156
157    /// Enable or disable CORS
158    pub fn cors(mut self, enable: bool) -> Self {
159        self.config.enable_cors = enable;
160        self
161    }
162
163    /// Set maximum request body size
164    pub fn max_body_size(mut self, size: usize) -> Self {
165        self.config.max_body_size = size;
166        self
167    }
168
169    /// Enable or disable GET SSE for persistent event streams
170    pub fn get_sse(mut self, enable: bool) -> Self {
171        self.config.enable_get_sse = enable;
172        self
173    }
174
175    /// Enable or disable POST SSE for streaming tool call responses (disabled by default for compatibility)
176    pub fn post_sse(mut self, enable: bool) -> Self {
177        self.config.enable_post_sse = enable;
178        self
179    }
180
181    /// Enable or disable both GET and POST SSE (convenience method)
182    pub fn sse(mut self, enable: bool) -> Self {
183        self.config.enable_get_sse = enable;
184        self.config.enable_post_sse = enable;
185        self
186    }
187
188    /// Set session expiry time in minutes
189    pub fn session_expiry_minutes(mut self, minutes: u64) -> Self {
190        self.config.session_expiry_minutes = minutes;
191        self
192    }
193
194    /// Allow or disallow ping requests without Mcp-Session-Id header
195    ///
196    /// Default: `true` (sessionless pings allowed per MCP spec).
197    /// Set to `false` for hardened deployments requiring session for all methods.
198    pub fn allow_unauthenticated_ping(mut self, allow: bool) -> Self {
199        self.config.allow_unauthenticated_ping = allow;
200        self
201    }
202
203    /// Configure SSE streaming settings
204    pub fn stream_config(mut self, config: StreamConfig) -> Self {
205        self.stream_config = config;
206        self
207    }
208
209    /// Register a JSON-RPC handler for specific methods
210    pub fn register_handler<H>(mut self, methods: Vec<String>, handler: H) -> Self
211    where
212        H: JsonRpcHandler<Error = McpError> + 'static,
213    {
214        self.dispatcher.register_methods(methods, handler);
215        self
216    }
217
218    /// Register a default handler for unhandled methods
219    pub fn default_handler<H>(mut self, handler: H) -> Self
220    where
221        H: JsonRpcHandler<Error = McpError> + 'static,
222    {
223        self.dispatcher.set_default_handler(handler);
224        self
225    }
226
227    /// Set server capabilities
228    pub fn server_capabilities(
229        mut self,
230        capabilities: turul_mcp_protocol::ServerCapabilities,
231    ) -> Self {
232        self.server_capabilities = Some(capabilities);
233        self
234    }
235
236    /// Build the HTTP MCP server
237    pub fn build(self) -> HttpMcpServer {
238        let session_storage = self
239            .session_storage
240            .expect("Session storage must be provided");
241
242        // ✅ CORRECTED ARCHITECTURE: Create single shared StreamManager instance
243        let stream_manager = Arc::new(StreamManager::with_config(
244            Arc::clone(&session_storage),
245            self.stream_config.clone(),
246        ));
247
248        // Create shared dispatcher Arc
249        let dispatcher = Arc::new(self.dispatcher);
250
251        // Use middleware stack from builder
252        let middleware_stack = self.middleware_stack;
253
254        // Create StreamableHttpHandler for MCP 2025-11-25 support
255        let mut streamable_handler = StreamableHttpHandler::new(
256            Arc::new(self.config.clone()),
257            Arc::clone(&dispatcher),
258            Arc::clone(&session_storage),
259            Arc::clone(&stream_manager),
260            self.server_capabilities.unwrap_or_default(),
261            Arc::clone(&middleware_stack),
262            self.tool_fingerprint.clone(),
263        );
264        if let Some(ref notifier) = self.tool_notifier {
265            streamable_handler = streamable_handler.with_tool_notifier(Arc::clone(notifier));
266        }
267
268        HttpMcpServer {
269            config: self.config,
270            dispatcher,
271            session_storage,
272            stream_config: self.stream_config,
273            stream_manager,
274            streamable_handler,
275            route_registry: self.route_registry,
276            tool_fingerprint: self.tool_fingerprint,
277            tool_notifier: self.tool_notifier,
278        }
279    }
280}
281
282impl Default for HttpMcpServerBuilder {
283    fn default() -> Self {
284        Self::new()
285    }
286}
287
288/// HTTP MCP Server with SessionStorage integration
289#[derive(Clone)]
290pub struct HttpMcpServer {
291    config: ServerConfig,
292    dispatcher: Arc<JsonRpcDispatcher<McpError>>,
293    session_storage: Arc<turul_mcp_session_storage::BoxedSessionStorage>,
294    stream_config: StreamConfig,
295    // ✅ CORRECTED ARCHITECTURE: Single shared StreamManager instance
296    stream_manager: Arc<StreamManager>,
297    // StreamableHttpHandler for MCP 2025-11-25 clients
298    streamable_handler: StreamableHttpHandler,
299    // Custom route registry for paths like .well-known
300    route_registry: Arc<crate::routes::RouteRegistry>,
301    // Tool fingerprint for session versioning (shared with both handlers)
302    tool_fingerprint: Option<String>,
303    // Tool change notifier for restart/redeploy fingerprint mismatch
304    tool_notifier: Option<Arc<dyn crate::ToolChangeNotifier>>,
305}
306
307impl HttpMcpServer {
308    /// Create a new builder with default in-memory storage
309    pub fn builder() -> HttpMcpServerBuilder {
310        HttpMcpServerBuilder::new()
311    }
312}
313
314impl HttpMcpServer {
315    /// Create a new builder with specific session storage
316    pub fn builder_with_storage(
317        session_storage: Arc<turul_mcp_session_storage::BoxedSessionStorage>,
318    ) -> HttpMcpServerBuilder {
319        HttpMcpServerBuilder::with_storage(session_storage)
320    }
321
322    /// Get the shared StreamManager instance for event forwarding bridge
323    /// Returns reference to the same StreamManager used by HTTP server
324    pub fn get_stream_manager(&self) -> Arc<crate::StreamManager> {
325        Arc::clone(&self.stream_manager)
326    }
327
328    /// Run the server with session management
329    pub async fn run(&self) -> Result<()> {
330        // Start session cleanup task
331        self.start_session_cleanup().await;
332
333        let listener = TcpListener::bind(&self.config.bind_address).await?;
334        info!("HTTP MCP server listening on {}", self.config.bind_address);
335        info!("MCP endpoint available at: {}", self.config.mcp_path);
336        info!("Session storage: {}", self.session_storage.backend_name());
337
338        // ✅ CORRECTED ARCHITECTURE: Create single SessionMcpHandler instance outside the loop
339        // Use the same middleware stack as streamable_handler (both handlers share it)
340        let mut session_handler = SessionMcpHandler::with_shared_stream_manager(
341            self.config.clone(),
342            Arc::clone(&self.dispatcher),
343            Arc::clone(&self.session_storage),
344            self.stream_config.clone(),
345            Arc::clone(&self.stream_manager),
346            Arc::clone(&self.streamable_handler.middleware_stack),
347        )
348        .with_tool_fingerprint(self.tool_fingerprint.clone());
349        if let Some(ref notifier) = self.tool_notifier {
350            session_handler = session_handler.with_tool_notifier(Arc::clone(notifier));
351        }
352
353        // Create combined handler that routes based on protocol version
354        let handler = McpRequestHandler {
355            session_handler,
356            streamable_handler: self.streamable_handler.clone(),
357            route_registry: Arc::clone(&self.route_registry),
358        };
359
360        loop {
361            let (stream, peer_addr) = listener.accept().await?;
362            debug!("New connection from {}", peer_addr);
363
364            let handler_clone = handler.clone();
365            tokio::spawn(async move {
366                let io = TokioIo::new(stream);
367                let service = service_fn(move |req| handle_request(req, handler_clone.clone()));
368
369                if let Err(err) = http1::Builder::new().serve_connection(io, service).await {
370                    // Filter out common client disconnection errors that aren't actual problems
371                    let err_str = err.to_string();
372                    if err_str.contains("connection closed before message completed") {
373                        debug!("Client disconnected (normal): {}", err);
374                    } else {
375                        error!("Error serving connection: {}", err);
376                    }
377                }
378            });
379        }
380    }
381
382    /// Start background session cleanup task
383    async fn start_session_cleanup(&self) {
384        let storage = Arc::clone(&self.session_storage);
385        let session_expiry_minutes = self.config.session_expiry_minutes;
386        tokio::spawn(async move {
387            let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(60));
388            loop {
389                interval.tick().await;
390
391                let expire_time = std::time::SystemTime::now()
392                    - std::time::Duration::from_secs(session_expiry_minutes * 60);
393                match storage.expire_sessions(expire_time).await {
394                    Ok(expired) => {
395                        if !expired.is_empty() {
396                            info!("Expired {} sessions", expired.len());
397                            for session_id in expired {
398                                debug!("Expired session: {}", session_id);
399                            }
400                        }
401                    }
402                    Err(err) => {
403                        error!("Session cleanup error: {}", err);
404                    }
405                }
406            }
407        });
408    }
409
410    /// Get server statistics
411    pub async fn get_stats(&self) -> ServerStats {
412        let session_count = self.session_storage.session_count().await.unwrap_or(0);
413        let event_count = self.session_storage.event_count().await.unwrap_or(0);
414
415        ServerStats {
416            sessions: session_count,
417            events: event_count,
418            storage_type: self.session_storage.backend_name().to_string(),
419        }
420    }
421}
422
423/// Handle requests with MCP 2025-11-25 compliance
424/// Combined handler that routes based on MCP protocol version
425#[derive(Clone)]
426struct McpRequestHandler {
427    session_handler: SessionMcpHandler,
428    streamable_handler: StreamableHttpHandler,
429    route_registry: Arc<crate::routes::RouteRegistry>,
430}
431
432async fn handle_request(
433    req: Request<hyper::body::Incoming>,
434    handler: McpRequestHandler,
435) -> std::result::Result<
436    Response<http_body_util::combinators::UnsyncBoxBody<Bytes, hyper::Error>>,
437    hyper::Error,
438> {
439    let method = req.method().clone();
440    let uri = req.uri().clone();
441    let path = uri.path();
442
443    debug!("Handling {} {}", method, path);
444
445    // Route the request
446    debug!(
447        "HTTP server dispatch: path={}, expected_mcp_path={}",
448        path, handler.session_handler.config.mcp_path
449    );
450    let response = if path == handler.session_handler.config.mcp_path {
451        debug!("Path match: Request routed to MCP handler");
452        // Extract MCP protocol version from headers
453        let protocol_version_str = req
454            .headers()
455            .get("MCP-Protocol-Version")
456            .and_then(|h| h.to_str().ok())
457            .unwrap_or("2025-11-25"); // Default to latest version (we only support the latest protocol)
458        debug!("Protocol version: {}", protocol_version_str);
459
460        let protocol_version = McpProtocolVersion::parse_version(protocol_version_str)
461            .unwrap_or(McpProtocolVersion::V2025_11_25);
462
463        debug!(
464            "MCP request: protocol_version={}, method={}",
465            protocol_version.as_str(),
466            method
467        );
468
469        // Route based on protocol version - MCP 2025-11-25 uses Streamable HTTP, older versions use SessionMcpHandler
470        debug!(
471            "Routing decision: protocol_version={}, method={}, supports_streamable={}, handler={}",
472            protocol_version.as_str(),
473            method,
474            protocol_version.supports_streamable_http(),
475            if protocol_version.supports_streamable_http() {
476                "StreamableHttpHandler"
477            } else {
478                "SessionMcpHandler"
479            }
480        );
481
482        if protocol_version.supports_streamable_http() {
483            // Use StreamableHttpHandler for MCP 2025-11-25 clients
484            debug!(
485                "Calling streamable handler for protocol {}",
486                protocol_version.as_str()
487            );
488            let streamable_response = handler.streamable_handler.handle_request(req).await;
489            debug!("Streamable handler completed");
490            Ok(streamable_response)
491        } else {
492            // Use SessionMcpHandler for legacy clients (MCP 2024-11-05 and earlier)
493            match handler.session_handler.handle_mcp_request(req).await {
494                Ok(mcp_response) => Ok(mcp_response),
495                Err(err) => {
496                    error!("Request handling error: {}", err);
497                    Ok(Response::builder()
498                        .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
499                        .body(
500                            Full::new(Bytes::from(format!("Internal Server Error: {}", err)))
501                                .map_err(|never| match never {})
502                                .boxed_unsync(),
503                        )
504                        .unwrap())
505                }
506            }
507        }
508    } else {
509        // Check custom routes (e.g., .well-known)
510        match handler.route_registry.match_route(path) {
511            Ok(Some(route_handler)) => {
512                debug!("Custom route matched: {}", path);
513                // Convert Incoming body to type-erased RouteBody for handler portability
514                let (parts, body) = req.into_parts();
515                let boxed_req = Request::from_parts(parts, body.boxed_unsync());
516                Ok(route_handler.handle(boxed_req).await)
517            }
518            Ok(None) => {
519                // 404 for other paths
520                Ok(Response::builder()
521                    .status(hyper::StatusCode::NOT_FOUND)
522                    .body(
523                        Full::new(Bytes::from("Not Found"))
524                            .map_err(|never| match never {})
525                            .boxed_unsync(),
526                    )
527                    .unwrap())
528            }
529            Err(validation_err) => {
530                // Path failed security validation — 400 Bad Request
531                warn!(
532                    "Route validation failed for path '{}': {}",
533                    path, validation_err
534                );
535                Ok(validation_err.into_response())
536            }
537        }
538    };
539
540    // Apply CORS if enabled
541    match response {
542        Ok(mut final_response) => {
543            if handler.session_handler.config.enable_cors {
544                CorsLayer::apply_cors_headers(final_response.headers_mut());
545            }
546            Ok(final_response)
547        }
548        Err(e) => Err(e),
549    }
550}
551
552/// Server statistics
553#[derive(Debug, Clone)]
554pub struct ServerStats {
555    pub sessions: usize,
556    pub events: usize,
557    pub storage_type: String,
558}
559
560#[cfg(test)]
561mod tests {
562    use super::*;
563    use std::net::{IpAddr, Ipv4Addr};
564    use std::sync::Arc;
565    use turul_mcp_session_storage::InMemorySessionStorage;
566
567    #[test]
568    fn test_server_config_default() {
569        let config = ServerConfig::default();
570        assert_eq!(config.mcp_path, "/mcp");
571        assert!(config.enable_cors);
572        assert_eq!(config.max_body_size, 1024 * 1024);
573    }
574
575    #[test]
576    fn test_builder() {
577        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 3000);
578        let session_storage = Arc::new(InMemorySessionStorage::new());
579        let server = HttpMcpServer::builder_with_storage(session_storage)
580            .bind_address(addr)
581            .mcp_path("/api/mcp")
582            .cors(false)
583            .max_body_size(2048)
584            .build();
585
586        assert_eq!(server.config.bind_address, addr);
587        assert_eq!(server.config.mcp_path, "/api/mcp");
588        assert!(!server.config.enable_cors);
589        assert_eq!(server.config.max_body_size, 2048);
590    }
591
592    #[tokio::test]
593    async fn test_server_stats() {
594        let session_storage = Arc::new(InMemorySessionStorage::new());
595        let server = HttpMcpServer::builder_with_storage(session_storage).build();
596
597        let stats = server.get_stats().await;
598        assert_eq!(stats.sessions, 0);
599        assert_eq!(stats.events, 0);
600        assert_eq!(stats.storage_type, "InMemory");
601    }
602}
turul_http_mcp_server/server.rs

turul_http_mcp_server/
server.rs
