diff --git a/crates/rmcp/src/handler/server.rs b/crates/rmcp/src/handler/server.rs index 6dc7883ed..04c1bbc5d 100644 --- a/crates/rmcp/src/handler/server.rs +++ b/crates/rmcp/src/handler/server.rs @@ -322,20 +322,25 @@ macro_rules! server_handler_methods { ) -> impl Future> + MaybeSendFuture + '_ { context.peer.set_peer_info(request.clone()); let mut info = self.get_info(); - info.protocol_version = negotiate_protocol_version( + let Some(protocol_version) = negotiate_protocol_version( &request.protocol_version, info.protocol_version, &self.supported_protocol_versions(), - ); + ) else { + return std::future::ready(Err(McpError::method_not_found::< + InitializeResultMethod, + >())); + }; + info.protocol_version = protocol_version; std::future::ready(Ok(info)) } /// Return the protocol versions supported by this server. /// /// Defaults to every version this SDK knows. Override it to narrow the /// set to the revisions the server actually implements: the returned - /// list is advertised by [`Self::discover`], bounds what `initialize` - /// negotiation may agree to, and is what per-request versions are - /// validated against. + /// list is advertised by [`Self::discover`], bounds which legacy + /// revisions `initialize` negotiation may agree to, and is what + /// per-request versions are validated against. fn supported_protocol_versions(&self) -> Cow<'static, [ProtocolVersion]> { Cow::Borrowed(ProtocolVersion::KNOWN_VERSIONS) } diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 20fd2e981..8f5661b76 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -284,8 +284,9 @@ pub trait Service: Send + Sync + 'static { context: NotificationContext, ) -> impl Future> + MaybeSendFuture + '_; fn get_info(&self) -> R::Info; - /// The protocol versions this service can speak, bounding what `initialize` - /// negotiation may agree to. + /// The protocol versions this service can speak, bounding which legacy + /// revisions `initialize` negotiation may agree to and which revisions the + /// discover lifecycle may select. /// /// Servers normally override /// [`ServerHandler::supported_protocol_versions`] instead of this method; diff --git a/crates/rmcp/src/service/server.rs b/crates/rmcp/src/service/server.rs index 8efbd34be..5a88a0e27 100644 --- a/crates/rmcp/src/service/server.rs +++ b/crates/rmcp/src/service/server.rs @@ -15,13 +15,14 @@ use crate::{ model::{ CancelledNotification, CancelledNotificationParam, ClientInfo, ClientJsonRpcMessage, ClientNotification, ClientRequest, ClientResult, CreateMessageRequest, - CreateMessageRequestParams, CreateMessageResult, EmptyResult, ErrorData, ListRootsRequest, - ListRootsResult, LoggingMessageNotification, LoggingMessageNotificationParam, - ProgressNotification, ProgressNotificationParam, PromptListChangedNotification, - ProtocolVersion, ResourceListChangedNotification, ResourceUpdatedNotification, - ResourceUpdatedNotificationParam, ServerInfo, ServerNotification, ServerRequest, - ServerResult, SubscriptionFilter, SubscriptionsAcknowledgedNotification, - SubscriptionsAcknowledgedNotificationParams, ToolListChangedNotification, + CreateMessageRequestParams, CreateMessageResult, EmptyResult, ErrorData, + InitializeResultMethod, ListRootsRequest, ListRootsResult, LoggingMessageNotification, + LoggingMessageNotificationParam, ProgressNotification, ProgressNotificationParam, + PromptListChangedNotification, ProtocolVersion, ResourceListChangedNotification, + ResourceUpdatedNotification, ResourceUpdatedNotificationParam, ServerInfo, + ServerNotification, ServerRequest, ServerResult, SubscriptionFilter, + SubscriptionsAcknowledgedNotification, SubscriptionsAcknowledgedNotificationParams, + ToolListChangedNotification, }, transport::DynamicTransportError, }; @@ -460,26 +461,46 @@ where } } -/// Echoes the client-requested version if the server supports it; otherwise -/// returns `server_fallback`. +/// Negotiates the legacy protocol version used by an `initialize` handshake. /// -/// `server_supported` comes from [`Service::supported_protocol_versions`], so a -/// server that narrows that list is never made to answer `initialize` with a -/// version it cannot serve. +/// `2026-07-28` replaced `initialize` with the discover lifecycle. A client that +/// nevertheless opens with `initialize` is offering to use the legacy lifecycle, +/// so a dual-era server negotiates its preferred supported legacy revision. A +/// modern-only server returns `None` and rejects the method. pub(crate) fn negotiate_protocol_version( client_requested: &ProtocolVersion, server_fallback: ProtocolVersion, server_supported: &[ProtocolVersion], -) -> ProtocolVersion { - if server_supported.contains(client_requested) { - client_requested.clone() +) -> Option { + let requested_legacy = client_requested < &ProtocolVersion::V_2026_07_28; + if requested_legacy && server_supported.contains(client_requested) { + return Some(client_requested.clone()); + } + + let fallback = if server_fallback < ProtocolVersion::V_2026_07_28 + && server_supported.contains(&server_fallback) + { + Some(server_fallback) } else { + server_supported + .iter() + .filter(|version| *version < &ProtocolVersion::V_2026_07_28) + .max_by(|left, right| left.as_str().cmp(right.as_str())) + .cloned() + }; + if let Some(server_fallback) = fallback { tracing::warn!( client_requested = %client_requested, server_fallback = %server_fallback, - "client requested unsupported protocol version; falling back to server default" + "client requested a version unavailable to initialize; falling back to the server's preferred legacy version" + ); + Some(server_fallback) + } else { + tracing::warn!( + client_requested = %client_requested, + "server supports no legacy protocol version; rejecting initialize" ); - server_fallback + None } } @@ -584,11 +605,25 @@ where return Err(ServerInitializeError::InitializeFailed(e)); } }; - init_response.protocol_version = negotiate_protocol_version( + let negotiated_protocol_version = negotiate_protocol_version( &requested_protocol_version, init_response.protocol_version, &service.supported_protocol_versions(), ); + let Some(negotiated_protocol_version) = negotiated_protocol_version else { + let error = ErrorData::method_not_found::(); + transport + .send(ServerJsonRpcMessage::error(error.clone(), Some(id))) + .await + .map_err(|transport_error| { + ServerInitializeError::transport::( + transport_error, + "sending initialize error response", + ) + })?; + return Err(ServerInitializeError::InitializeFailed(error)); + }; + init_response.protocol_version = negotiated_protocol_version; // Update peer_info so context.protocol_version() reflects the negotiated // version in all subsequent request handlers. negotiated_peer_info.protocol_version = init_response.protocol_version.clone(); diff --git a/crates/rmcp/src/transport/streamable_http_server/tower.rs b/crates/rmcp/src/transport/streamable_http_server/tower.rs index f1fef585f..2d454d4ed 100644 --- a/crates/rmcp/src/transport/streamable_http_server/tower.rs +++ b/crates/rmcp/src/transport/streamable_http_server/tower.rs @@ -27,8 +27,8 @@ use crate::{ model::{ ClientCapabilities, ClientJsonRpcMessage, ClientNotification, ClientRequest, ErrorCode, ErrorData, GetExtensions, GetMeta, Implementation, InitializeRequest, - InitializeRequestParams, InitializedNotification, JsonObject, JsonRpcError, - ProtocolVersion, RequestId, ServerInfo, ServerJsonRpcMessage, ServerResult, + InitializeRequestParams, InitializeResultMethod, InitializedNotification, JsonObject, + JsonRpcError, ProtocolVersion, RequestId, ServerInfo, ServerJsonRpcMessage, ServerResult, }, serve_server, service::{ @@ -322,7 +322,8 @@ impl> Service for NegotiatingStatelessHttpSer &requested, result.protocol_version.clone(), &self.0.supported_protocol_versions(), - ); + ) + .ok_or_else(ErrorData::method_not_found::)?; if let Some(peer_info) = peer.peer_info() { let mut peer_info = (*peer_info).clone(); peer_info.protocol_version = result.protocol_version.clone(); @@ -353,9 +354,10 @@ impl> Service for NegotiatingStatelessHttpSer clippy::result_large_err, reason = "BoxResponse is intentionally large; matches other handlers in this file" )] -// SEP-2567: sessions are removed from the discover lifecycle. Validate -// protocol-version consistency, then classify the request with the shared -// lifecycle helper. +// SEP-2567: sessions are removed from the discover lifecycle. An initialize +// opener always selects the legacy lifecycle, even when its version offer is +// modern and will be negotiated down. Otherwise validate protocol-version +// consistency, then classify the request with the shared lifecycle helper. fn is_legacy_request( message: Option<&ClientJsonRpcMessage>, headers: &HeaderMap, @@ -401,10 +403,12 @@ fn is_legacy_request( .and_then(|s| serde_json::from_value(serde_json::Value::String(s.to_owned())).ok()) }) .unwrap_or(ProtocolVersion::V_2025_03_26); - Ok(uses_legacy_lifecycle( - Some(&version), - uses_discover_lifecycle, - )) + let is_initialize = matches!( + message, + Some(ClientJsonRpcMessage::Request(req)) + if matches!(&req.request, ClientRequest::InitializeRequest(_)) + ); + Ok(is_initialize || uses_legacy_lifecycle(Some(&version), uses_discover_lifecycle)) } fn method_not_allowed_response() -> BoxResponse { diff --git a/crates/rmcp/tests/test_protocol_version_negotiation.rs b/crates/rmcp/tests/test_protocol_version_negotiation.rs index e91ecf97a..6c3c15c84 100644 --- a/crates/rmcp/tests/test_protocol_version_negotiation.rs +++ b/crates/rmcp/tests/test_protocol_version_negotiation.rs @@ -1,6 +1,7 @@ -//! Tests for protocol version negotiation in the default ServerHandler::initialize impl. +//! Tests for legacy protocol version negotiation in ServerHandler::initialize. //! -//! Known versions are echoed back; unknown versions fall back to LATEST. +//! Known legacy versions are echoed back; modern and unknown offers fall back +//! to the server's preferred supported legacy revision. #![cfg(not(feature = "local"))] #![cfg(feature = "client")] @@ -8,8 +9,12 @@ use std::borrow::Cow; use rmcp::{ ClientHandler, ErrorData, RoleServer, ServerHandler, ServiceExt, - model::{ClientInfo, InitializeRequestParams, InitializeResult, ProtocolVersion, ServerInfo}, + model::{ + ClientInfo, ClientJsonRpcMessage, ClientRequest, InitializeRequestParams, InitializeResult, + ProtocolVersion, RequestId, ServerInfo, ServerJsonRpcMessage, ServerResult, + }, service::RequestContext, + transport::{IntoTransport, Transport}, }; #[derive(Debug, Clone, Default)] @@ -119,6 +124,9 @@ async fn negotiated_version_with( #[tokio::test] async fn known_version_echoed_back() { for version in ProtocolVersion::KNOWN_VERSIONS { + if version >= &ProtocolVersion::V_2026_07_28 { + continue; + } let negotiated = negotiated_version(version.clone()).await; assert_eq!( negotiated, *version, @@ -127,6 +135,37 @@ async fn known_version_echoed_back() { } } +#[tokio::test] +async fn modern_initialize_negotiates_the_preferred_legacy_version_on_a_direct_transport() { + let (server_transport, client_transport) = tokio::io::duplex(4096); + let server = tokio::spawn(async move { EchoServer.serve(server_transport).await }); + let mut client = IntoTransport::::into_transport(client_transport); + let request: ClientRequest = serde_json::from_value(serde_json::json!({ + "method": "initialize", + "params": { + "protocolVersion": "2026-07-28", + "capabilities": {}, + "clientInfo": { "name": "test", "version": "1.0" } + } + })) + .unwrap(); + client + .send(ClientJsonRpcMessage::request(request, RequestId::Number(1))) + .await + .unwrap(); + + let Some(ServerJsonRpcMessage::Response(response)) = client.receive().await else { + panic!("expected initialize response"); + }; + let ServerResult::InitializeResult(result) = response.result else { + panic!("expected initialize result"); + }; + assert_eq!(result.protocol_version, ProtocolVersion::V_2025_11_25); + + let running = server.await.unwrap().expect("server should initialize"); + running.cancel().await.expect("server should cancel"); +} + #[tokio::test] async fn unknown_version_falls_back_to_latest() { let unknown: ProtocolVersion = serde_json::from_str(r#""1999-01-01""#).unwrap(); @@ -151,7 +190,8 @@ async fn narrowed_server_still_echoes_versions_it_supports() { #[tokio::test] async fn narrowed_server_does_not_agree_to_version_it_excludes() { - let negotiated = negotiated_version_with(NarrowedServer, ProtocolVersion::V_2026_07_28).await; + let unsupported = serde_json::from_str(r#""1999-01-01""#).unwrap(); + let negotiated = negotiated_version_with(NarrowedServer, unsupported).await; assert_eq!( negotiated, ProtocolVersion::V_2025_11_25, @@ -161,8 +201,8 @@ async fn narrowed_server_does_not_agree_to_version_it_excludes() { #[tokio::test] async fn narrowed_server_caps_even_when_it_overrides_initialize() { - let negotiated = - negotiated_version_with(NarrowedOverridingServer, ProtocolVersion::V_2026_07_28).await; + let unsupported = serde_json::from_str(r#""1999-01-01""#).unwrap(); + let negotiated = negotiated_version_with(NarrowedOverridingServer, unsupported).await; assert_eq!( negotiated, ProtocolVersion::V_2025_11_25, diff --git a/crates/rmcp/tests/test_stateless_protocol_version.rs b/crates/rmcp/tests/test_stateless_protocol_version.rs index 02222ec2c..581005054 100644 --- a/crates/rmcp/tests/test_stateless_protocol_version.rs +++ b/crates/rmcp/tests/test_stateless_protocol_version.rs @@ -1,11 +1,12 @@ -//! Tests for protocol version negotiation in stateless HTTP mode. +//! Tests for legacy protocol version negotiation over streamable HTTP. //! -//! Supported versions are echoed back; unknown versions, and versions outside -//! the server's `supported_protocol_versions`, fall back to the handler's own -//! version. +//! Supported legacy versions are echoed back; modern and unknown offers, and +//! versions outside the server's `supported_protocol_versions`, fall back to +//! the handler's preferred supported legacy revision. #![cfg(not(feature = "local"))] use std::borrow::Cow; +use std::sync::Arc; use rmcp::{ ErrorData, RoleServer, ServerHandler, @@ -68,6 +69,31 @@ impl ServerHandler for NarrowedOverridingInitialize { } } +const MODERN_ONLY_VERSIONS: &[ProtocolVersion] = &[ProtocolVersion::V_2026_07_28]; + +#[derive(Clone, Default)] +struct ModernOnly; + +impl ServerHandler for ModernOnly { + fn get_info(&self) -> ServerInfo { + let mut info = ServerInfo::new(ServerCapabilities::default()); + info.protocol_version = ProtocolVersion::V_2026_07_28; + info + } + + fn supported_protocol_versions(&self) -> Cow<'static, [ProtocolVersion]> { + Cow::Borrowed(MODERN_ONLY_VERSIONS) + } + + async fn initialize( + &self, + _request: InitializeRequestParams, + _context: RequestContext, + ) -> Result { + Ok(self.get_info()) + } +} + fn stateless_sse_config() -> StreamableHttpServerConfig { StreamableHttpServerConfig::default() .with_legacy_session_mode(false) @@ -79,6 +105,13 @@ fn stateless_json_config() -> StreamableHttpServerConfig { stateless_sse_config().with_json_response(true) } +fn stateful_sse_config() -> StreamableHttpServerConfig { + StreamableHttpServerConfig::default() + .with_legacy_session_mode(true) + .with_sse_keep_alive(None) + .with_cancellation_token(CancellationToken::new()) +} + async fn spawn_server( config: StreamableHttpServerConfig, ) -> (reqwest::Client, String, CancellationToken) { @@ -139,10 +172,10 @@ async fn post_init(client: &reqwest::Client, url: &str, body_version: &str) -> s let body = resp.text().await.expect("read SSE body"); let data = body .lines() - .find_map(|line| line.strip_prefix("data:")) + .filter_map(|line| line.strip_prefix("data:")) .map(str::trim) - .filter(|data| !data.is_empty()) - .expect("SSE response contains data"); + .find(|data| !data.is_empty()) + .unwrap_or_else(|| panic!("SSE response contains data, got {body:?}")); serde_json::from_str(data).expect("parse SSE data") } } @@ -152,6 +185,9 @@ async fn stateless_json_init_echoes_known_versions_when_handler_overrides_initia let (client, url, ct) = spawn_server(stateless_json_config()).await; for version in ProtocolVersion::KNOWN_VERSIONS { + if version == &ProtocolVersion::V_2026_07_28 { + continue; + } let resp = post_init(&client, &url, version.as_str()).await; assert_eq!( resp["result"]["protocolVersion"], @@ -168,6 +204,9 @@ async fn stateless_sse_init_echoes_known_versions_when_handler_overrides_initial let (client, url, ct) = spawn_server(stateless_sse_config()).await; for version in ProtocolVersion::KNOWN_VERSIONS { + if version == &ProtocolVersion::V_2026_07_28 { + continue; + } let resp = post_init(&client, &url, version.as_str()).await; assert_eq!( resp["result"]["protocolVersion"], @@ -179,6 +218,52 @@ async fn stateless_sse_init_echoes_known_versions_when_handler_overrides_initial ct.cancel(); } +#[tokio::test] +async fn modern_initialize_negotiates_the_preferred_legacy_version_over_json_and_sse() { + for config in [stateless_json_config(), stateless_sse_config()] { + let (client, url, ct) = spawn_server(config).await; + let resp = post_init(&client, &url, ProtocolVersion::V_2026_07_28.as_str()).await; + assert_eq!( + resp["result"]["protocolVersion"], + ProtocolVersion::V_2025_11_25.as_str() + ); + ct.cancel(); + } +} + +#[tokio::test] +async fn modern_initialize_offer_uses_a_legacy_stateful_http_session() { + let manager = Arc::new(LocalSessionManager::default()); + let config = stateful_sse_config(); + let ct = config.cancellation_token.clone(); + let service: StreamableHttpService = + StreamableHttpService::new(|| Ok(OverridingInitialize), manager.clone(), config); + let router = axum::Router::new().nest_service("/mcp", service); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/mcp", listener.local_addr().unwrap()); + tokio::spawn({ + let ct = ct.clone(); + async move { + let _ = axum::serve(listener, router) + .with_graceful_shutdown(async move { ct.cancelled_owned().await }) + .await; + } + }); + + let resp = post_init( + &reqwest::Client::new(), + &url, + ProtocolVersion::V_2026_07_28.as_str(), + ) + .await; + assert_eq!( + resp["result"]["protocolVersion"], + ProtocolVersion::V_2025_11_25.as_str() + ); + assert_eq!(manager.sessions.read().await.len(), 1); + ct.cancel(); +} + #[tokio::test] async fn stateless_json_init_preserves_handler_fallback_for_unknown_version() { let (client, url, ct) = spawn_server(stateless_json_config()).await; @@ -211,7 +296,7 @@ async fn stateless_json_init_echoes_versions_the_server_narrowed_to() { } #[tokio::test] -async fn stateless_json_init_does_not_agree_to_version_outside_supported_list() { +async fn stateless_json_init_does_not_agree_to_modern_version() { let (client, url, ct) = spawn_server_of::(stateless_json_config()).await; @@ -219,8 +304,19 @@ async fn stateless_json_init_does_not_agree_to_version_outside_supported_list() assert_eq!( resp["result"]["protocolVersion"], ProtocolVersion::V_2025_11_25.as_str(), - "a version outside supported_protocol_versions should not be echoed back" + "initialize should negotiate the server's preferred legacy revision" ); ct.cancel(); } + +#[tokio::test] +async fn modern_only_server_rejects_initialize() { + let (client, url, ct) = spawn_server_of::(stateless_json_config()).await; + + let resp = post_init(&client, &url, ProtocolVersion::V_2026_07_28.as_str()).await; + assert_eq!(resp["error"]["code"], -32601); + assert_eq!(resp["error"]["message"], "initialize"); + + ct.cancel(); +}