fix(server): serve draft-version requests statelessly per SEP-2567 (#999)

* fix(server): serve draft-version requests statelessly per SEP-2567

* refactor: address PR feedback

* refactor!: rename stateful_mode to legacy_session_mode

* fix: address PR feedback

crates/rmcp/src/transport/streamable_http_server/session.rs

@@ -13,7 +13,7 @@
 //!
 //! * [`local::LocalSessionManager`] — in-memory session store (default).
 //! * [`never::NeverSessionManager`] — rejects all session operations, used
-//!   when stateful mode is disabled.
+//!   when legacy session mode is disabled.
 //!
 //! # Custom session managers
 //!

crates/rmcp/src/transport/streamable_http_server/tower.rs

@@ -59,8 +59,13 @@ pub struct StreamableHttpServerConfig {
     pub sse_retry: Option<Duration>,
     /// If true, the server will create a session for each request and keep it alive.
     /// When enabled, SSE priming events are sent to enable client reconnection.
-    pub stateful_mode: bool,
-    /// When true and `stateful_mode` is false, the server prefers
+    ///
+    /// Only applies to legacy protocol versions (`< 2026-07-28`). Per SEP-2567,
+    /// sessions are removed from the `2026-07-28` draft version, so requests
+    /// negotiating that version are always served statelessly regardless of
+    /// this setting.
+    pub legacy_session_mode: bool,
+    /// When true and `legacy_session_mode` is false, the server prefers
     /// `Content-Type: application/json` for simple request-response tools.
     /// If the handler emits a notification or request before the final response,
     /// the server falls back to `text/event-stream` so no message is lost.
@@ -130,7 +135,7 @@ impl Default for StreamableHttpServerConfig {
         Self {
             sse_keep_alive: Some(Duration::from_secs(15)),
             sse_retry: Some(Duration::from_secs(3)),
-            stateful_mode: true,
+            legacy_session_mode: true,
             json_response: false,
             cancellation_token: CancellationToken::new(),
             allowed_hosts: vec!["localhost".into(), "127.0.0.1".into(), "::1".into()],
@@ -176,8 +181,8 @@ impl StreamableHttpServerConfig {
         self
     }
 
-    pub fn with_stateful_mode(mut self, stateful: bool) -> Self {
-        self.stateful_mode = stateful;
+    pub fn with_legacy_session_mode(mut self, legacy_session_mode: bool) -> Self {
+        self.legacy_session_mode = legacy_session_mode;
         self
     }
 
@@ -250,6 +255,57 @@ fn message_has_per_request_protocol_version(message: &ClientJsonRpcMessage) -> b
     }
 }
 
+#[expect(
+    clippy::result_large_err,
+    reason = "BoxResponse is intentionally large; matches other handlers in this file"
+)]
+// SEP-2567: sessions are removed from 2026-07-28; older versions are legacy.
+// Validates protocol-version consistency and returns `Ok(true)` only for a valid legacy request.
+fn is_legacy_request(
+    message: Option<&ClientJsonRpcMessage>,
+    headers: &HeaderMap,
+) -> Result<bool, BoxResponse> {
+    let has_per_request_version = message.is_some_and(message_has_per_request_protocol_version);
+    validate_protocol_version_header(headers, has_per_request_version)?;
+    if let Some(message) = message {
+        if let ClientJsonRpcMessage::Request(req) = message {
+            if let ClientRequest::InitializeRequest(init) = &req.request {
+                validate_header_matches_init_body(
+                    headers,
+                    init.params.protocol_version.as_str(),
+                    Some(req.id.clone()),
+                )?;
+            }
+        }
+        validate_request_protocol_version_meta(headers, message)?;
+    }
+
+    let from_body = match message {
+        Some(ClientJsonRpcMessage::Request(req)) => match &req.request {
+            ClientRequest::InitializeRequest(init) => Some(init.params.protocol_version.clone()),
+            _ => req.request.get_meta().protocol_version(),
+        },
+        _ => None,
+    };
+    let version = from_body
+        .or_else(|| {
+            headers
+                .get(HEADER_MCP_PROTOCOL_VERSION)
+                .and_then(|value| value.to_str().ok())
+                .and_then(|s| serde_json::from_value(serde_json::Value::String(s.to_owned())).ok())
+        })
+        .unwrap_or(ProtocolVersion::V_2025_03_26);
+    Ok(version < ProtocolVersion::V_2026_07_28)
+}
+
+fn method_not_allowed_response() -> BoxResponse {
+    Response::builder()
+        .status(http::StatusCode::METHOD_NOT_ALLOWED)
+        .header(ALLOW, "POST")
+        .body(Full::new(Bytes::from("Method Not Allowed")).boxed())
+        .expect("valid response")
+}
+
 fn invalid_request_jsonrpc_response(
     id: Option<RequestId>,
     message: impl Into<Cow<'static, str>>,
@@ -662,7 +718,7 @@ fn validate_origin_header(
 ///
 /// ## Session management
 ///
-/// When [`StreamableHttpServerConfig::stateful_mode`] is `true` (the default),
+/// When [`StreamableHttpServerConfig::legacy_session_mode`] is `true` (the default),
 /// the server creates a session for each client that sends an `initialize`
 /// request. The session ID is returned in the `Mcp-Session-Id` response header
 /// and the client must include it on all subsequent requests.
@@ -1158,13 +1214,13 @@ where
             return response;
         }
         let method = request.method().clone();
-        let allowed_methods = match self.config.stateful_mode {
+        let allowed_methods = match self.config.legacy_session_mode {
             true => "GET, POST, DELETE",
             false => "POST",
         };
-        let result = match (method, self.config.stateful_mode) {
+        let result = match (method, self.config.legacy_session_mode) {
             (Method::POST, _) => self.handle_post(request).await,
-            // if we're not in stateful mode, we don't support GET or DELETE because there is no session
+            // if legacy session mode is disabled, we don't support GET or DELETE because there is no session
             (Method::GET, true) => self.handle_get(request).await,
             (Method::DELETE, true) => self.handle_delete(request).await,
             _ => {
@@ -1187,6 +1243,9 @@ where
         B: Body + Send + 'static,
         B::Error: Display,
     {
+        if !is_legacy_request(None, request.headers())? {
+            return Ok(method_not_allowed_response());
+        }
         // check accept header
         if !request
             .headers()
@@ -1340,7 +1399,10 @@ where
             Err(response) => return Ok(response),
         };
 
-        if self.config.stateful_mode {
+        let use_session =
+            self.config.legacy_session_mode && is_legacy_request(Some(&message), &part.headers)?;
+
+        if use_session {
             // do we have a session id?
             let session_id = part
                 .headers
@@ -1673,6 +1735,9 @@ where
         B: Body + Send + 'static,
         B::Error: Display,
     {
+        if !is_legacy_request(None, request.headers())? {
+            return Ok(method_not_allowed_response());
+        }
         // check session id
         let session_id = request
             .headers()