diff --git a/crates/rmcp/src/service/client.rs b/crates/rmcp/src/service/client.rs index b23a1496d..86c064118 100644 --- a/crates/rmcp/src/service/client.rs +++ b/crates/rmcp/src/service/client.rs @@ -127,13 +127,18 @@ where .ok_or_else(|| ClientInitializeError::ConnectionClosed(context.to_string())) } +enum StartupResponse { + Response(Box, RequestId), + Error(ErrorData, Option), +} + /// Helper function to expect a response from the stream async fn expect_response( transport: &mut T, context: &str, service: &S, peer: Peer, -) -> Result<(ServerResult, RequestId), ClientInitializeError> +) -> Result where T: Transport, S: Service, @@ -143,11 +148,11 @@ where match message { // Expected message to complete the initialization ServerJsonRpcMessage::Response(JsonRpcResponse { id, result, .. }) => { - break Ok((result, id)); + break Ok(StartupResponse::Response(Box::new(result), id)); } // Handle JSON-RPC error responses ServerJsonRpcMessage::Error(error) => { - break Err(ClientInitializeError::JsonRpcError(error.error)); + break Ok(StartupResponse::Error(error.error, error.id)); } // Server could send logging messages before handshake ServerJsonRpcMessage::Notification(mut notification) => { @@ -575,6 +580,301 @@ pub enum ClientLifecycleMode { }, } +#[derive(Debug)] +struct DiscoverStartupError { + error: ClientInitializeError, + failure: DiscoverProbeFailure, +} + +struct ProbeHttpResponse { + status: u16, + body: String, +} + +#[derive(Debug)] +enum DiscoverProbeFailure { + Other, + IncompatibleVersions(Vec), + Http { + status: u16, + body: String, + requested_version: ProtocolVersion, + request_id: RequestId, + }, + Rpc { + error: ErrorData, + requested_version: ProtocolVersion, + request_id: RequestId, + response_id: Option, + }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ProbeDisposition { + RetryLegacy, + Fail, +} + +fn classify_probe_failure(failure: &DiscoverProbeFailure) -> ProbeDisposition { + match failure { + DiscoverProbeFailure::IncompatibleVersions(versions) + if exclusively_historical_protocol_versions(versions) => + { + ProbeDisposition::RetryLegacy + } + DiscoverProbeFailure::Http { + status, + body, + requested_version, + request_id, + } if http_error_proves_legacy(*status, body, requested_version, request_id) => { + ProbeDisposition::RetryLegacy + } + DiscoverProbeFailure::Rpc { + error, + requested_version, + request_id, + response_id, + } if rpc_error_proves_legacy( + error, + requested_version, + request_id, + response_id.as_ref(), + ) => + { + ProbeDisposition::RetryLegacy + } + _ => ProbeDisposition::Fail, + } +} + +fn http_error_proves_legacy( + status: u16, + body: &str, + requested_version: &ProtocolVersion, + request_id: &RequestId, +) -> bool { + match status { + 404 | 405 => match serde_json::from_str::(body) { + Ok(ServerJsonRpcMessage::Error(error)) => error + .id + .as_ref() + .is_none_or(|response_id| request_id.matches_response_id(response_id)), + _ => true, + }, + 400 => { + let Ok(ServerJsonRpcMessage::Error(error)) = + serde_json::from_str::(body) + else { + return false; + }; + rpc_error_proves_legacy( + &error.error, + requested_version, + request_id, + error.id.as_ref(), + ) + } + _ => false, + } +} + +fn rpc_error_proves_legacy( + error: &ErrorData, + requested_version: &ProtocolVersion, + request_id: &RequestId, + response_id: Option<&RequestId>, +) -> bool { + let correlated = + response_id.is_some_and(|response_id| request_id.matches_response_id(response_id)); + if response_id.is_some() && !correlated { + return false; + } + + if error.code == crate::model::ErrorCode::METHOD_NOT_FOUND { + return correlated; + } + + if matches!( + error.code, + crate::model::ErrorCode::INVALID_REQUEST | crate::model::ErrorCode::INVALID_PARAMS + ) { + return correlated; + } + + if !matches!( + error.code, + crate::model::ErrorCode::UNSUPPORTED_PROTOCOL_VERSION | crate::model::ErrorCode(-32000) + ) { + return false; + } + + let message = error.message.trim().to_ascii_lowercase(); + let normalized_message = message.strip_prefix("bad request: ").unwrap_or(&message); + if normalized_message == "no valid session id provided" { + return true; + } + + if !normalized_message.contains("unsupported protocol version") + || !normalized_message.contains(requested_version.as_str()) + { + return false; + } + + let supported_data = error.data.as_ref().and_then(|data| data.get("supported")); + let supported_from_data = match supported_data { + Some(value) => { + let Ok(versions) = serde_json::from_value::>(value.clone()) else { + return false; + }; + Some(versions) + } + None => None, + }; + let supported_from_message = historical_versions_from_message(normalized_message); + if normalized_message.contains("supported versions:") && supported_from_message.is_none() { + return false; + } + + let supported = match (supported_from_data, supported_from_message) { + (Some(data), Some(message)) + if data.len() == message.len() + && data.iter().all(|version| message.contains(version)) => + { + Some(data) + } + (Some(_), Some(_)) => None, + (Some(versions), None) | (None, Some(versions)) => Some(versions), + (None, None) => None, + }; + + match supported { + Some(versions) => exclusively_historical_protocol_versions(&versions), + None => supported_data.is_none() && !normalized_message.contains("supported versions:"), + } +} + +impl From for DiscoverStartupError { + fn from(error: ClientInitializeError) -> Self { + let failure = match &error { + ClientInitializeError::NoCompatibleProtocolVersion { + server_supported, .. + } => DiscoverProbeFailure::IncompatibleVersions(server_supported.clone()), + _ => DiscoverProbeFailure::Other, + }; + Self { error, failure } + } +} + +impl DiscoverStartupError { + fn transport( + error: T::Error, + context: impl Into>, + requested_version: ProtocolVersion, + request_id: RequestId, + ) -> Self + where + T: Transport + 'static, + { + let error = ClientInitializeError::transport::(error, context); + let failure = probe_http_status(&error) + .map(|http| DiscoverProbeFailure::Http { + status: http.status, + body: http.body, + requested_version, + request_id, + }) + .unwrap_or(DiscoverProbeFailure::Other); + Self { error, failure } + } + + fn json_rpc( + error: ErrorData, + requested_version: ProtocolVersion, + request_id: RequestId, + response_id: Option, + ) -> Self { + let failure = DiscoverProbeFailure::Rpc { + error: error.clone(), + requested_version, + request_id, + response_id, + }; + Self { + error: ClientInitializeError::JsonRpcError(error), + failure, + } + } + + fn disposition(&self) -> ProbeDisposition { + classify_probe_failure(&self.failure) + } +} + +#[cfg(feature = "transport-streamable-http-client")] +fn probe_http_status(error: &ClientInitializeError) -> Option { + let ClientInitializeError::TransportError { error, .. } = error else { + return None; + }; + let mut source: Option<&(dyn std::error::Error + 'static)> = Some(error.error.as_ref()); + while let Some(current) = source { + if let Some(http) = + current.downcast_ref::() + { + return Some(ProbeHttpResponse { + status: http.status, + body: http.body.to_string(), + }); + } + source = current.source(); + } + None +} + +#[cfg(not(feature = "transport-streamable-http-client"))] +fn probe_http_status(_error: &ClientInitializeError) -> Option { + None +} + +fn exclusively_historical_protocol_versions(versions: &[ProtocolVersion]) -> bool { + !versions.is_empty() + && versions.iter().all(|version| { + (ProtocolVersion::KNOWN_VERSIONS.contains(version) + && version < &ProtocolVersion::V_2026_07_28) + // Some deployed legacy servers also advertise this pre-release version. + || version.as_str() == "2024-10-07" + }) +} + +fn historical_versions_from_message(message: &str) -> Option> { + let (_, supported_versions) = message.split_once("supported versions:")?; + let supported_versions = supported_versions.split(')').next()?; + let mut versions = Vec::new(); + for candidate in supported_versions.split(',') { + let candidate = candidate + .trim() + .trim_matches(|character| matches!(character, '[' | ']' | '"' | '\'')); + let bytes = candidate.as_bytes(); + if bytes.len() != 10 + || bytes.get(4) != Some(&b'-') + || bytes.get(7) != Some(&b'-') + || bytes + .iter() + .enumerate() + .any(|(index, byte)| index != 4 && index != 7 && !byte.is_ascii_digit()) + { + return None; + } + let version = serde_json::from_value::(serde_json::Value::String( + candidate.to_owned(), + )) + .ok()?; + versions.push(version); + } + + (!versions.is_empty()).then_some(versions) +} + /// Client-specific lifecycle entry points. pub trait ClientServiceExt: Service + Sized { fn serve_with_lifecycle( @@ -701,7 +1001,8 @@ where &client_info, preferred_versions, ) - .await?; + .await + .map_err(|error| error.error)?; } ClientLifecycleMode::Auto { preferred_versions, @@ -718,9 +1019,7 @@ where .await; match discover_result { Ok(()) => {} - Err(ClientInitializeError::JsonRpcError(error)) - if error.code == crate::model::ErrorCode::METHOD_NOT_FOUND => - { + Err(error) if error.disposition() == ProbeDisposition::RetryLegacy => { let mut legacy_info = client_info; if let Some(version) = legacy_version { legacy_info.protocol_version = version; @@ -728,7 +1027,7 @@ where legacy_startup(&service, &mut transport, &id_provider, &peer, legacy_info) .await?; } - Err(error) => return Err(error), + Err(error) => return Err(error.error), } } } @@ -764,7 +1063,12 @@ where })?; let (response, response_id) = - expect_response(transport, "initialize response", service, peer.clone()).await?; + match expect_response(transport, "initialize response", service, peer.clone()).await? { + StartupResponse::Response(response, response_id) => (*response, response_id), + StartupResponse::Error(error, _) => { + return Err(ClientInitializeError::JsonRpcError(error)); + } + }; if !id.matches_response_id(&response_id) { return Err(ClientInitializeError::ConflictInitResponseId( @@ -798,13 +1102,13 @@ async fn discover_startup( peer: &Peer, client_info: &ClientInfo, preferred_versions: Vec, -) -> Result<(), ClientInitializeError> +) -> Result<(), DiscoverStartupError> where S: Service, T: Transport + 'static, { if preferred_versions.is_empty() { - return Err(ClientInitializeError::NoPreferredProtocolVersion); + return Err(ClientInitializeError::NoPreferredProtocolVersion.into()); } let mut attempted = Vec::new(); @@ -827,16 +1131,28 @@ where )) .await .map_err(|error| { - ClientInitializeError::transport::(error, "send discover request") + DiscoverStartupError::transport::( + error, + "send discover request", + candidate.clone(), + id.clone(), + ) })?; - match expect_response(transport, "discover response", service, peer.clone()).await { - Ok((ServerResult::DiscoverResult(result), response_id)) => { + match expect_response(transport, "discover response", service, peer.clone()).await? { + StartupResponse::Response(response, response_id) => { + let result = match *response { + ServerResult::DiscoverResult(result) => result, + response => { + return Err( + ClientInitializeError::ExpectedInitResult(Some(response)).into() + ); + } + }; if !id.matches_response_id(&response_id) { - return Err(ClientInitializeError::ConflictInitResponseId( - id, - response_id, - )); + return Err( + ClientInitializeError::ConflictInitResponseId(id, response_id).into(), + ); } let Some(selected) = select_protocol_version(&preferred_versions, &result.supported_versions) @@ -844,7 +1160,8 @@ where return Err(ClientInitializeError::NoCompatibleProtocolVersion { client_supported: preferred_versions, server_supported: result.supported_versions, - }); + } + .into()); }; peer.set_peer_info(ServerInfo { protocol_version: selected.clone(), @@ -860,12 +1177,28 @@ where }); return Ok(()); } - Ok((response, _)) => { - return Err(ClientInitializeError::ExpectedInitResult(Some(response))); - } - Err(ClientInitializeError::JsonRpcError(error)) - if error.code == crate::model::ErrorCode::UNSUPPORTED_PROTOCOL_VERSION => - { + StartupResponse::Error(error, response_id) => { + if let Some(response_id) = response_id.as_ref() + && !id.matches_response_id(response_id) + { + return Err(ClientInitializeError::ConflictInitResponseId( + id, + response_id.clone(), + ) + .into()); + } + + if error.code != crate::model::ErrorCode::UNSUPPORTED_PROTOCOL_VERSION + || response_id.is_none() + { + return Err(DiscoverStartupError::json_rpc( + error, + candidate, + id, + response_id, + )); + } + let supported = error .data .as_ref() @@ -890,11 +1223,11 @@ where return Err(ClientInitializeError::NoCompatibleProtocolVersion { client_supported: preferred_versions, server_supported: supported, - }); + } + .into()); }; candidate = next; } - Err(error) => return Err(error), } } } @@ -2066,6 +2399,244 @@ where mod tests { use super::*; + fn rpc_probe_failure( + code: crate::model::ErrorCode, + message: &str, + response_id: Option, + ) -> DiscoverProbeFailure { + rpc_probe_failure_with_data(code, message, response_id, None) + } + + fn rpc_probe_failure_with_data( + code: crate::model::ErrorCode, + message: &str, + response_id: Option, + data: Option, + ) -> DiscoverProbeFailure { + DiscoverProbeFailure::Rpc { + error: ErrorData::new(code, message.to_owned(), data), + requested_version: ProtocolVersion::V_2026_07_28, + request_id: RequestId::Number(7), + response_id, + } + } + + fn http_probe_failure(status: u16, body: &str) -> DiscoverProbeFailure { + DiscoverProbeFailure::Http { + status, + body: body.to_owned(), + requested_version: ProtocolVersion::V_2026_07_28, + request_id: RequestId::Number(7), + } + } + + #[test] + fn probe_classifier_uses_explicit_downgrade_evidence() { + let correlated_id = Some(RequestId::Number(7)); + let cases = vec![ + ( + "historical discovery result", + DiscoverProbeFailure::IncompatibleVersions(vec![ + ProtocolVersion::V_2025_11_25, + ProtocolVersion::V_2025_06_18, + ]), + ProbeDisposition::RetryLegacy, + ), + ( + "future discovery result", + DiscoverProbeFailure::IncompatibleVersions(vec![ + serde_json::from_value(serde_json::json!("2099-01-01")).unwrap(), + ]), + ProbeDisposition::Fail, + ), + ( + "correlated method not found", + rpc_probe_failure( + crate::model::ErrorCode::METHOD_NOT_FOUND, + "Method not found", + correlated_id.clone(), + ), + ProbeDisposition::RetryLegacy, + ), + ( + "correlated invalid request", + rpc_probe_failure( + crate::model::ErrorCode::INVALID_REQUEST, + "Invalid Request", + correlated_id.clone(), + ), + ProbeDisposition::RetryLegacy, + ), + ( + "correlated invalid params", + rpc_probe_failure( + crate::model::ErrorCode::INVALID_PARAMS, + "Invalid params", + correlated_id.clone(), + ), + ProbeDisposition::RetryLegacy, + ), + ( + "uncorrelated invalid request", + rpc_probe_failure( + crate::model::ErrorCode::INVALID_REQUEST, + "Invalid Request", + None, + ), + ProbeDisposition::Fail, + ), + ( + "unsupported attempted version without supported list", + rpc_probe_failure( + crate::model::ErrorCode::UNSUPPORTED_PROTOCOL_VERSION, + "Unsupported protocol version: 2026-07-28", + correlated_id.clone(), + ), + ProbeDisposition::RetryLegacy, + ), + ( + "legacy prevalidation without supported list", + rpc_probe_failure( + crate::model::ErrorCode(-32000), + "Bad Request: Unsupported protocol version: 2026-07-28", + None, + ), + ProbeDisposition::RetryLegacy, + ), + ( + "legacy prevalidation with historical versions", + rpc_probe_failure( + crate::model::ErrorCode(-32000), + "Bad Request: Unsupported protocol version: 2026-07-28 \ + (supported versions: 2025-11-25, 2025-06-18)", + None, + ), + ProbeDisposition::RetryLegacy, + ), + ( + "known missing-session prevalidation", + rpc_probe_failure( + crate::model::ErrorCode(-32000), + "Bad Request: No valid session ID provided", + None, + ), + ProbeDisposition::RetryLegacy, + ), + ( + "unrelated response id", + rpc_probe_failure( + crate::model::ErrorCode::INVALID_PARAMS, + "Invalid params", + Some(RequestId::Number(99)), + ), + ProbeDisposition::Fail, + ), + ( + "arbitrary server error", + rpc_probe_failure( + crate::model::ErrorCode(-32000), + "Bad Request: database unavailable", + None, + ), + ProbeDisposition::Fail, + ), + ( + "unsupported different version", + rpc_probe_failure( + crate::model::ErrorCode(-32000), + "Bad Request: Unsupported protocol version: 2025-11-25", + None, + ), + ProbeDisposition::Fail, + ), + ( + "mixed historical and future versions", + rpc_probe_failure( + crate::model::ErrorCode(-32000), + "Bad Request: Unsupported protocol version: 2026-07-28 \ + (supported versions: 2025-06-18, 2027-01-01)", + None, + ), + ProbeDisposition::Fail, + ), + ( + "unknown supported version data", + rpc_probe_failure_with_data( + crate::model::ErrorCode::UNSUPPORTED_PROTOCOL_VERSION, + "Unsupported protocol version: 2026-07-28", + correlated_id, + Some(serde_json::json!({ "supported": ["next-draft"] })), + ), + ProbeDisposition::Fail, + ), + ( + "initial HTTP 404", + http_probe_failure(404, "not found"), + ProbeDisposition::RetryLegacy, + ), + ( + "initial HTTP 405", + http_probe_failure(405, "method not allowed"), + ProbeDisposition::RetryLegacy, + ), + ( + "initial HTTP 404 with unrelated response id", + http_probe_failure( + 404, + r#"{"jsonrpc":"2.0","id":99,"error":{"code":-32601,"message":"Method not found"}}"#, + ), + ProbeDisposition::Fail, + ), + ( + "initial HTTP 400 recognized null-id error", + http_probe_failure( + 400, + r#"{"jsonrpc":"2.0","id":null,"error":{"code":-32000,"message":"Bad Request: Unsupported protocol version: 2026-07-28"}}"#, + ), + ProbeDisposition::RetryLegacy, + ), + ( + "initial HTTP 400 unrelated id", + http_probe_failure( + 400, + r#"{"jsonrpc":"2.0","id":99,"error":{"code":-32602,"message":"Invalid params"}}"#, + ), + ProbeDisposition::Fail, + ), + ( + "initial HTTP 400 arbitrary body", + http_probe_failure(400, "Bad Request"), + ProbeDisposition::Fail, + ), + ( + "HTTP 401", + http_probe_failure(401, "authentication required"), + ProbeDisposition::Fail, + ), + ( + "HTTP 403", + http_probe_failure(403, "forbidden"), + ProbeDisposition::Fail, + ), + ( + "HTTP 500 with legacy-shaped body", + http_probe_failure( + 500, + r#"{"jsonrpc":"2.0","id":null,"error":{"code":-32000,"message":"Bad Request: No valid session ID provided"}}"#, + ), + ProbeDisposition::Fail, + ), + ]; + + for (name, failure, expected) in cases { + assert_eq!( + classify_probe_failure(&failure), + expected, + "unexpected disposition for {name}" + ); + } + } + fn disconnected_peer() -> Peer { let (peer, receiver) = Peer::::new(Arc::new(AtomicU32RequestIdProvider::default()), None); diff --git a/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs b/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs index e2eeebb4a..614da9b32 100644 --- a/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs +++ b/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs @@ -165,6 +165,11 @@ impl StreamableHttpClient for reqwest::Client { custom_headers: HashMap, max_sse_event_size: usize, ) -> Result> { + let is_discover_request = matches!( + &message, + ClientJsonRpcMessage::Request(request) + if matches!(&request.request, crate::model::ClientRequest::DiscoverRequest(_)) + ); let mut request = self .post(uri.as_ref()) .header(ACCEPT, [EVENT_STREAM_MIME_TYPE, JSON_MIME_TYPE].join(", ")); @@ -219,6 +224,21 @@ impl StreamableHttpClient for reqwest::Client { if status == reqwest::StatusCode::NOT_FOUND && session_was_attached { return Err(StreamableHttpError::SessionExpired); } + if matches!( + status, + reqwest::StatusCode::UNAUTHORIZED + | reqwest::StatusCode::FORBIDDEN + | reqwest::StatusCode::NOT_FOUND + | reqwest::StatusCode::METHOD_NOT_ALLOWED + ) { + let body = response + .text() + .await + .unwrap_or_else(|_| "".to_owned()); + return Err(StreamableHttpError::UnexpectedHttpStatus( + HttpStatusError::new(status.as_u16(), body), + )); + } let content_type = response .headers() .get(reqwest::header::CONTENT_TYPE) @@ -249,6 +269,11 @@ impl StreamableHttpClient for reqwest::Client { .text() .await .unwrap_or_else(|_| "".to_owned()); + if is_discover_request { + return Err(StreamableHttpError::UnexpectedHttpStatus( + HttpStatusError::new(status.as_u16(), body), + )); + } if content_type .as_deref() .is_some_and(|ct| ct.as_bytes().starts_with(JSON_MIME_TYPE.as_bytes())) @@ -262,9 +287,9 @@ impl StreamableHttpClient for reqwest::Client { ), } } - return Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned( - format!("HTTP {status}: {body}"), - ))); + return Err(StreamableHttpError::UnexpectedHttpStatus( + HttpStatusError::new(status.as_u16(), body), + )); } match content_type.as_deref() { Some(ct) if ct.as_bytes().starts_with(EVENT_STREAM_MIME_TYPE.as_bytes()) => { @@ -472,7 +497,9 @@ mod tests { use tokio::sync::Mutex; use super::StreamableHttpClientTransport; - use crate::transport::streamable_http_client::{StreamableHttpClient, StreamableHttpError}; + use crate::transport::streamable_http_client::{ + HttpStatusError, StreamableHttpClient, StreamableHttpError, + }; const API_KEY_HEADER: &str = "x-api-key"; const API_KEY_VALUE: &str = "secret"; @@ -575,7 +602,10 @@ mod tests { assert!( matches!( result, - Err(StreamableHttpError::UnexpectedServerResponse(_)) + Err(StreamableHttpError::UnexpectedHttpStatus(HttpStatusError { + status: 307, + .. + })) ), "redirect response should be returned to the transport, got {result:?}" ); diff --git a/crates/rmcp/src/transport/common/unix_socket.rs b/crates/rmcp/src/transport/common/unix_socket.rs index 5f995db2b..efed98200 100644 --- a/crates/rmcp/src/transport/common/unix_socket.rs +++ b/crates/rmcp/src/transport/common/unix_socket.rs @@ -274,9 +274,9 @@ impl StreamableHttpClient for UnixSocketHttpClient { .await .map(|c| String::from_utf8_lossy(&c.to_bytes()).into_owned()) .unwrap_or_else(|_| "".to_owned()); - return Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned( - format!("HTTP {status}: {body}"), - ))); + return Err(StreamableHttpError::UnexpectedHttpStatus( + HttpStatusError::new(status.as_u16(), body), + )); } let content_type = response.headers().get(http::header::CONTENT_TYPE).cloned(); diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index c43fb7dd5..b3ce621ad 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -20,8 +20,8 @@ use crate::{ RoleClient, model::{ ClientJsonRpcMessage, ClientNotification, ClientRequest, ErrorData, GetMeta, - InitializedNotification, JsonObject, ProtocolVersion, RequestId, ServerJsonRpcMessage, - ServerResult, + InitializedNotification, JsonObject, JsonRpcMessage, ProtocolVersion, RequestId, + ServerJsonRpcMessage, ServerResult, }, transport::{ common::{client_side_sse::SseAutoReconnectStream, mcp_headers}, @@ -168,6 +168,35 @@ impl InsufficientScopeError { } } +/// A non-success HTTP response, including the status and response body. +/// +/// Keeping this error independent of the HTTP client implementation lets +/// lifecycle negotiation inspect the response consistently for reqwest, +/// custom [`StreamableHttpClient`] implementations, and Unix-socket clients. +#[derive(Error, Debug)] +#[error("unexpected HTTP status {status}: {body}")] +#[non_exhaustive] +pub struct HttpStatusError { + pub status: u16, + pub body: Cow<'static, str>, +} + +impl HttpStatusError { + pub fn new(status: u16, body: impl Into>) -> Self { + Self { + status, + body: body.into(), + } + } + + fn json_rpc_error(&self) -> Option { + match serde_json::from_str::(&self.body) { + Ok(message @ JsonRpcMessage::Error(_)) => Some(message), + _ => None, + } + } +} + #[derive(Error, Debug)] #[non_exhaustive] pub enum StreamableHttpError { @@ -181,6 +210,8 @@ pub enum StreamableHttpError { UnexpectedEndOfStream, #[error("unexpected server response: {0}")] UnexpectedServerResponse(Cow<'static, str>), + #[error("{0}")] + UnexpectedHttpStatus(#[source] HttpStatusError), #[error("Unexpected content type: {0:?}")] UnexpectedContentType(Option), #[error("Server does not support SSE")] @@ -324,6 +355,12 @@ impl StreamableHttpPostResponse { /// [`Self::post_message_with_max_sse_event_size`] and /// [`Self::get_stream_with_max_sse_event_size`] to enforce the transport's /// configured event-size limit. +/// +/// For a non-success response to an initial `server/discover` request, +/// implementations should return +/// [`StreamableHttpError::UnexpectedHttpStatus`] with the original status and +/// body. [`ClientLifecycleMode::Auto`](crate::service::ClientLifecycleMode::Auto) +/// uses that typed response to decide whether a legacy retry is safe. pub trait StreamableHttpClient: Clone + Send + 'static { type Error: std::error::Error + Send + Sync + 'static; fn post_message( @@ -833,54 +870,100 @@ impl Worker for StreamableHttpClientWorker { let config = self.config.clone(); let transport_task_ct = context.cancellation_token.clone(); let _drop_guard = transport_task_ct.clone().drop_guard(); - let WorkerSendRequest { - responder, - message: startup_request, - } = context.recv_from_handler().await?; - let is_legacy_startup = matches!( - &startup_request, - ClientJsonRpcMessage::Request(request) - if matches!(&request.request, ClientRequest::InitializeRequest(_)) - ); - let mut saved_init_request = is_legacy_startup.then(|| startup_request.clone()); let empty_tool_cache = HashMap::new(); - let (bootstrap_version, bootstrap_headers) = if is_legacy_startup { - (ProtocolVersion::default(), config.custom_headers.clone()) - } else { - request_version_headers( - &config.custom_headers, + let ( + startup_request, + is_legacy_startup, + bootstrap_version, + bootstrap_headers, + message, + session_id, + ) = loop { + let WorkerSendRequest { + responder, + message: startup_request, + } = context.recv_from_handler().await?; + let is_legacy_startup = matches!( &startup_request, - &ProtocolVersion::default(), - &empty_tool_cache, - ) - }; - let (message, session_id) = match self - .client - .post_message_with_max_sse_event_size( - config.uri.clone(), - startup_request, - None, - config.auth_header.clone(), - bootstrap_headers.clone(), - config.max_sse_event_size, - ) - .await - { - Ok(res) => { - let _ = responder.send(Ok(())); - res.expect_initialized::().await.map_err( - WorkerQuitReason::fatal_context("process initialize response"), - )? - } - Err(err) => { - let msg = format!("{:?}", err); - let _ = responder.send(Err(err)); - return Err(WorkerQuitReason::fatal( - StreamableHttpError::TransportChannelClosed, - msg, - )); + ClientJsonRpcMessage::Request(request) + if matches!(&request.request, ClientRequest::InitializeRequest(_)) + ); + let is_discover_startup = matches!( + &startup_request, + ClientJsonRpcMessage::Request(request) + if matches!(&request.request, ClientRequest::DiscoverRequest(_)) + ); + let (bootstrap_version, bootstrap_headers) = if is_legacy_startup { + (ProtocolVersion::default(), config.custom_headers.clone()) + } else { + request_version_headers( + &config.custom_headers, + &startup_request, + &ProtocolVersion::default(), + &empty_tool_cache, + ) + }; + match self + .client + .post_message_with_max_sse_event_size( + config.uri.clone(), + startup_request.clone(), + None, + config.auth_header.clone(), + bootstrap_headers.clone(), + config.max_sse_event_size, + ) + .await + { + Ok(response) => { + let _ = responder.send(Ok(())); + let (message, session_id) = + response.expect_initialized::().await.map_err( + WorkerQuitReason::fatal_context("process initialize response"), + )?; + break ( + startup_request, + is_legacy_startup, + bootstrap_version, + bootstrap_headers, + message, + session_id, + ); + } + Err(StreamableHttpError::UnexpectedHttpStatus(error)) if is_discover_startup => { + if error.status == 400 + && let Some(message) = error.json_rpc_error() + { + // An HTTP 400 can still carry a valid JSON-RPC + // discovery error. Route it through normal negotiation + // so a correlated unsupported-version response can + // select or retry another modern version. + let _ = responder.send(Ok(())); + break ( + startup_request, + is_legacy_startup, + bootstrap_version, + bootstrap_headers, + message, + None, + ); + } + + // Surface raw prevalidation failures to Auto mode but keep + // the worker available for a possible legacy initialize. + let _ = responder.send(Err(StreamableHttpError::UnexpectedHttpStatus(error))); + } + Err(error) => { + let message = format!("{error:?}"); + let _ = responder.send(Err(error)); + return Err(WorkerQuitReason::fatal( + StreamableHttpError::TransportChannelClosed, + message, + )); + } } }; + let mut saved_init_request = is_legacy_startup.then(|| startup_request.clone()); let mut uses_modern_http = !is_legacy_startup; let mut session_id: Option> = if uses_modern_http { None diff --git a/crates/rmcp/tests/test_client_lifecycle_modes.rs b/crates/rmcp/tests/test_client_lifecycle_modes.rs index 375364e88..c46d00ba2 100644 --- a/crates/rmcp/tests/test_client_lifecycle_modes.rs +++ b/crates/rmcp/tests/test_client_lifecycle_modes.rs @@ -7,7 +7,7 @@ use rmcp::{ Implementation, InitializeResult, ProtocolVersion, RequestId, ServerCapabilities, ServerJsonRpcMessage, ServerResult, }, - service::PeerRequestOptions, + service::{ClientInitializeError, PeerRequestOptions}, transport::{IntoTransport, Transport}, }; @@ -229,6 +229,273 @@ async fn auto_startup_falls_back_after_discover_method_not_found() { server_task.await.expect("server task"); } +async fn assert_auto_startup_falls_back( + rejection: impl FnOnce(RequestId) -> ServerJsonRpcMessage + Send + 'static, +) { + let (server_transport, client_transport) = tokio::io::duplex(4096); + let mut server = IntoTransport::::into_transport(server_transport); + let server_task = tokio::spawn(async move { + let ClientJsonRpcMessage::Request(discover) = + server.receive().await.expect("expected discover request") + else { + panic!("expected discover request"); + }; + assert!(matches!( + discover.request, + ClientRequest::DiscoverRequest(_) + )); + server + .send(rejection(discover.id)) + .await + .expect("send legacy discovery rejection"); + + let ClientJsonRpcMessage::Request(initialize) = server + .receive() + .await + .expect("expected fallback initialize request") + else { + panic!("expected fallback initialize request"); + }; + let ClientRequest::InitializeRequest(request) = initialize.request else { + panic!("expected initialize request"); + }; + assert_eq!( + request.params.protocol_version, + ProtocolVersion::V_2025_06_18 + ); + server + .send(ServerJsonRpcMessage::response( + ServerResult::InitializeResult( + InitializeResult::new(ServerCapabilities::default()), + ), + initialize.id, + )) + .await + .expect("send initialize response"); + assert!(matches!( + server.receive().await, + Some(ClientJsonRpcMessage::Notification(_)) + )); + }); + + let client = DiscoverClient + .serve_with_lifecycle( + client_transport, + ClientLifecycleMode::Auto { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + legacy_version: Some(ProtocolVersion::V_2025_06_18), + }, + ) + .await + .expect("Auto mode should fall back for a recognized legacy-only rejection"); + client.cancel().await.expect("cancel client"); + server_task.await.expect("server task"); +} + +#[tokio::test] +async fn auto_startup_falls_back_when_server_only_supports_historical_versions() { + assert_auto_startup_falls_back(|id| { + ServerJsonRpcMessage::error( + ErrorData::unsupported_protocol_version( + ProtocolVersion::V_2026_07_28, + &[ProtocolVersion::V_2025_11_25, ProtocolVersion::V_2025_06_18], + ), + Some(id), + ) + }) + .await; +} + +#[tokio::test] +async fn auto_startup_falls_back_when_discovery_only_advertises_historical_versions() { + assert_auto_startup_falls_back(|id| { + ServerJsonRpcMessage::response( + ServerResult::DiscoverResult(DiscoverResult::new( + vec![ProtocolVersion::V_2025_06_18], + ServerCapabilities::default(), + Implementation::new("legacy-server", "1.0.0"), + )), + id, + ) + }) + .await; +} + +#[tokio::test] +async fn auto_startup_falls_back_for_uncorrelated_legacy_protocol_prevalidation() { + assert_auto_startup_falls_back(|_| { + ServerJsonRpcMessage::error( + ErrorData::new( + ErrorCode(-32000), + "Bad Request: Unsupported protocol version: 2026-07-28 \ + (supported versions: 2025-11-25, 2025-06-18, 2025-03-26, \ + 2024-11-05, 2024-10-07)", + None, + ), + None, + ) + }) + .await; +} + +#[tokio::test] +async fn auto_startup_falls_back_for_uncorrelated_missing_session_prevalidation() { + assert_auto_startup_falls_back(|_| { + ServerJsonRpcMessage::error( + ErrorData::new( + ErrorCode(-32000), + "Bad Request: No valid session ID provided", + None, + ), + None, + ) + }) + .await; +} + +async fn assert_auto_startup_does_not_fall_back( + rejection: impl FnOnce(RequestId) -> ServerJsonRpcMessage + Send + 'static, +) -> ClientInitializeError { + let (server_transport, client_transport) = tokio::io::duplex(4096); + let mut server = IntoTransport::::into_transport(server_transport); + let server_task = tokio::spawn(async move { + let ClientJsonRpcMessage::Request(discover) = + server.receive().await.expect("expected discover request") + else { + panic!("expected discover request"); + }; + server + .send(rejection(discover.id)) + .await + .expect("send discovery rejection"); + assert!( + server.receive().await.is_none(), + "unsafe discovery rejection must not trigger initialize" + ); + }); + + let error = match DiscoverClient + .serve_with_lifecycle( + client_transport, + ClientLifecycleMode::Auto { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + legacy_version: Some(ProtocolVersion::V_2025_06_18), + }, + ) + .await + { + Ok(_) => panic!("unsafe discovery rejection must not trigger fallback"), + Err(error) => error, + }; + server_task.await.expect("server task"); + error +} + +#[tokio::test] +async fn auto_startup_does_not_fall_back_for_unrelated_error_response_id() { + let error = assert_auto_startup_does_not_fall_back(|_| { + ServerJsonRpcMessage::error( + ErrorData::new(ErrorCode::METHOD_NOT_FOUND, "Method not found", None), + Some(RequestId::Number(999)), + ) + }) + .await; + assert!(matches!( + error, + ClientInitializeError::ConflictInitResponseId(_, _) + )); +} + +#[tokio::test] +async fn auto_startup_does_not_fall_back_for_uncorrelated_method_not_found() { + let error = assert_auto_startup_does_not_fall_back(|_| { + ServerJsonRpcMessage::error( + ErrorData::new(ErrorCode::METHOD_NOT_FOUND, "Method not found", None), + None, + ) + }) + .await; + assert!(matches!(error, ClientInitializeError::JsonRpcError(_))); +} + +#[tokio::test] +async fn auto_startup_does_not_fall_back_for_arbitrary_uncorrelated_errors() { + let error = assert_auto_startup_does_not_fall_back(|_| { + ServerJsonRpcMessage::error( + ErrorData::new(ErrorCode(-32000), "Bad Request: database unavailable", None), + None, + ) + }) + .await; + assert!(matches!(error, ClientInitializeError::JsonRpcError(_))); +} + +#[tokio::test] +async fn auto_startup_does_not_fall_back_for_unknown_or_future_protocol_versions() { + let error = assert_auto_startup_does_not_fall_back(|_| { + ServerJsonRpcMessage::error( + ErrorData::new( + ErrorCode(-32000), + "Bad Request: Unsupported protocol version: 2026-07-28 \ + (supported versions: 2025-06-18, 2027-01-01)", + None, + ), + None, + ) + }) + .await; + assert!(matches!(error, ClientInitializeError::JsonRpcError(_))); +} + +#[tokio::test] +async fn auto_startup_does_not_fall_back_for_unknown_protocol_version_tokens() { + let error = assert_auto_startup_does_not_fall_back(|_| { + ServerJsonRpcMessage::error( + ErrorData::new( + ErrorCode(-32000), + "Bad Request: Unsupported protocol version: 2026-07-28 \ + (supported versions: 2025-06-18, next-draft)", + None, + ), + None, + ) + }) + .await; + assert!(matches!(error, ClientInitializeError::JsonRpcError(_))); +} + +#[tokio::test] +async fn auto_startup_does_not_fall_back_for_contradictory_supported_versions() { + let error = assert_auto_startup_does_not_fall_back(|_| { + ServerJsonRpcMessage::error( + ErrorData::new( + ErrorCode(-32000), + "Bad Request: Unsupported protocol version: 2026-07-28 \ + (supported versions: 2025-06-18, 2027-01-01)", + Some(serde_json::json!({ "supported": ["2025-06-18"] })), + ), + None, + ) + }) + .await; + assert!(matches!(error, ClientInitializeError::JsonRpcError(_))); +} + +#[tokio::test] +async fn auto_startup_does_not_fall_back_for_uncorrelated_protocol_version_error() { + let error = assert_auto_startup_does_not_fall_back(|_| { + ServerJsonRpcMessage::error( + ErrorData::unsupported_protocol_version( + ProtocolVersion::V_2026_07_28, + &[ProtocolVersion::V_2025_06_18], + ), + None, + ) + }) + .await; + assert!(matches!(error, ClientInitializeError::JsonRpcError(_))); +} + #[tokio::test] async fn discover_startup_retries_a_mutually_supported_version() { let unsupported: ProtocolVersion = diff --git a/crates/rmcp/tests/test_discover_http_client_startup.rs b/crates/rmcp/tests/test_discover_http_client_startup.rs index c6051a047..bbcff428b 100644 --- a/crates/rmcp/tests/test_discover_http_client_startup.rs +++ b/crates/rmcp/tests/test_discover_http_client_startup.rs @@ -5,15 +5,35 @@ feature = "transport-streamable-http-server" ))] -use std::borrow::Cow; +use std::{ + borrow::Cow, + collections::HashMap, + sync::{Arc, Mutex}, +}; +use axum::{ + body::Bytes, + extract::State, + http::{StatusCode, header}, + response::{IntoResponse, Response}, + routing::post, +}; +use futures::{StreamExt, stream}; +use http::{HeaderName, HeaderValue}; use rmcp::{ ClientLifecycleMode, ClientServiceExt, ServerHandler, - model::{ClientInfo, DiscoverResult, ErrorCode, ErrorData, ProtocolVersion}, + model::{ + ClientInfo, ClientJsonRpcMessage, ClientRequest, DiscoverResult, ErrorCode, ErrorData, + Implementation, InitializeResult, ProtocolVersion, ServerCapabilities, + ServerJsonRpcMessage, ServerResult, + }, service::{MaybeSendFuture, RequestContext, RoleServer}, transport::{ StreamableHttpClientTransport, - streamable_http_client::StreamableHttpClientTransportConfig, + streamable_http_client::{ + HttpStatusError, StreamableHttpClient, StreamableHttpClientTransportConfig, + StreamableHttpError, StreamableHttpPostResponse, + }, streamable_http_server::{ StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager, }, @@ -135,3 +155,546 @@ async fn auto_http_client_falls_back_to_stateful_legacy_startup() { ct.cancel(); server.await.expect("server task"); } + +#[derive(Debug, Clone, Copy)] +enum LegacyDiscoveryRejection { + UnsupportedProtocol, + UnsupportedProtocolWithoutVersionList, + MissingSession, + InvalidRequest, + InvalidParams, + NotFound, + MethodNotAllowed, + Unauthorized, + Forbidden, + UnauthorizedJson, + ForbiddenJson, + UnrelatedResponseId, + NotFoundWithUnrelatedResponseId, + ArbitraryBadRequest, + MixedFutureVersions, + InternalServerErrorWithLegacyBody, +} + +#[derive(Clone)] +struct LegacyPrevalidationState { + rejection: LegacyDiscoveryRejection, + methods: Arc>>, +} + +async fn legacy_prevalidation_handler( + State(state): State, + body: Bytes, +) -> Response { + let message: serde_json::Value = serde_json::from_slice(&body).expect("JSON-RPC request body"); + let method = message + .get("method") + .and_then(serde_json::Value::as_str) + .expect("JSON-RPC request method"); + state + .methods + .lock() + .expect("methods lock") + .push(method.into()); + + match method { + "server/discover" => match state.rejection { + LegacyDiscoveryRejection::UnsupportedProtocol => json_response( + StatusCode::BAD_REQUEST, + serde_json::json!({ + "jsonrpc": "2.0", + "id": null, + "error": { + "code": -32000, + "message": "Bad Request: Unsupported protocol version: 2026-07-28 \ + (supported versions: 2025-11-25, 2025-06-18, 2025-03-26, \ + 2024-11-05, 2024-10-07)", + }, + }), + ), + LegacyDiscoveryRejection::UnsupportedProtocolWithoutVersionList => json_response( + StatusCode::BAD_REQUEST, + serde_json::json!({ + "jsonrpc": "2.0", + "id": null, + "error": { + "code": -32000, + "message": "Bad Request: Unsupported protocol version: 2026-07-28", + }, + }), + ), + LegacyDiscoveryRejection::MissingSession => json_response( + StatusCode::BAD_REQUEST, + serde_json::json!({ + "jsonrpc": "2.0", + "id": null, + "error": { + "code": -32000, + "message": "Bad Request: No valid session ID provided", + }, + }), + ), + LegacyDiscoveryRejection::InvalidRequest => json_response( + StatusCode::BAD_REQUEST, + serde_json::json!({ + "jsonrpc": "2.0", + "id": message.get("id"), + "error": { + "code": -32600, + "message": "Invalid Request", + }, + }), + ), + LegacyDiscoveryRejection::InvalidParams => json_response( + StatusCode::BAD_REQUEST, + serde_json::json!({ + "jsonrpc": "2.0", + "id": message.get("id"), + "error": { + "code": -32602, + "message": "Invalid params", + }, + }), + ), + LegacyDiscoveryRejection::NotFound => { + (StatusCode::NOT_FOUND, "legacy endpoint not found").into_response() + } + LegacyDiscoveryRejection::MethodNotAllowed => { + (StatusCode::METHOD_NOT_ALLOWED, "legacy method not allowed").into_response() + } + LegacyDiscoveryRejection::Unauthorized => { + (StatusCode::UNAUTHORIZED, "authentication required").into_response() + } + LegacyDiscoveryRejection::Forbidden => { + (StatusCode::FORBIDDEN, "access forbidden").into_response() + } + LegacyDiscoveryRejection::UnauthorizedJson => json_response( + StatusCode::UNAUTHORIZED, + serde_json::json!({ + "jsonrpc": "2.0", + "id": null, + "error": { + "code": -32000, + "message": "Bad Request: No valid session ID provided", + }, + }), + ), + LegacyDiscoveryRejection::ForbiddenJson => json_response( + StatusCode::FORBIDDEN, + serde_json::json!({ + "jsonrpc": "2.0", + "id": null, + "error": { + "code": -32000, + "message": "Bad Request: Unsupported protocol version: 2026-07-28 \ + (supported versions: 2025-11-25, 2025-06-18)", + }, + }), + ), + LegacyDiscoveryRejection::UnrelatedResponseId => json_response( + StatusCode::BAD_REQUEST, + serde_json::json!({ + "jsonrpc": "2.0", + "id": 999, + "error": { + "code": -32602, + "message": "Invalid params", + }, + }), + ), + LegacyDiscoveryRejection::NotFoundWithUnrelatedResponseId => json_response( + StatusCode::NOT_FOUND, + serde_json::json!({ + "jsonrpc": "2.0", + "id": 999, + "error": { + "code": -32601, + "message": "Method not found", + }, + }), + ), + LegacyDiscoveryRejection::ArbitraryBadRequest => json_response( + StatusCode::BAD_REQUEST, + serde_json::json!({ + "jsonrpc": "2.0", + "id": null, + "error": { + "code": -32000, + "message": "Bad Request: database unavailable", + }, + }), + ), + LegacyDiscoveryRejection::MixedFutureVersions => json_response( + StatusCode::BAD_REQUEST, + serde_json::json!({ + "jsonrpc": "2.0", + "id": null, + "error": { + "code": -32000, + "message": "Bad Request: Unsupported protocol version: 2026-07-28 \ + (supported versions: 2025-06-18, 2027-01-01)", + }, + }), + ), + LegacyDiscoveryRejection::InternalServerErrorWithLegacyBody => json_response( + StatusCode::INTERNAL_SERVER_ERROR, + serde_json::json!({ + "jsonrpc": "2.0", + "id": null, + "error": { + "code": -32000, + "message": "Bad Request: No valid session ID provided", + }, + }), + ), + }, + "initialize" => { + assert_eq!( + message + .get("params") + .and_then(|params| params.get("protocolVersion")), + Some(&serde_json::json!("2025-06-18")) + ); + let mut result = InitializeResult::new(ServerCapabilities::default()); + result.protocol_version = ProtocolVersion::V_2025_06_18; + json_response( + StatusCode::OK, + serde_json::json!({ + "jsonrpc": "2.0", + "id": message.get("id"), + "result": result, + }), + ) + } + "notifications/initialized" => StatusCode::ACCEPTED.into_response(), + "tools/list" => json_response( + StatusCode::OK, + serde_json::json!({ + "jsonrpc": "2.0", + "id": message.get("id"), + "result": { "tools": [] }, + }), + ), + _ => (StatusCode::BAD_REQUEST, "unexpected request").into_response(), + } +} + +fn json_response(status: StatusCode, value: serde_json::Value) -> Response { + ( + status, + [(header::CONTENT_TYPE, "application/json")], + value.to_string(), + ) + .into_response() +} + +async fn assert_http_legacy_fallback(rejection: LegacyDiscoveryRejection) { + let methods = Arc::new(Mutex::new(Vec::new())); + let router = axum::Router::new() + .route("/mcp", post(legacy_prevalidation_handler)) + .with_state(LegacyPrevalidationState { + rejection, + methods: methods.clone(), + }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("listener should bind"); + let address = listener.local_addr().expect("listener address"); + let cancellation = CancellationToken::new(); + let server = tokio::spawn({ + let cancellation = cancellation.clone(); + async move { + axum::serve(listener, router) + .with_graceful_shutdown(cancellation.cancelled_owned()) + .await + .expect("serve legacy HTTP endpoint"); + } + }); + + let transport = StreamableHttpClientTransport::from_config( + StreamableHttpClientTransportConfig::with_uri(format!("http://{address}/mcp")), + ); + let client = ClientInfo::default() + .serve_with_lifecycle( + transport, + ClientLifecycleMode::Auto { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + legacy_version: Some(ProtocolVersion::V_2025_06_18), + }, + ) + .await + .expect("Auto mode should recognize the deployed legacy HTTP response"); + client.list_tools(None).await.expect("list legacy tools"); + client.cancel().await.expect("cancel client"); + + assert_eq!( + *methods.lock().expect("methods lock"), + [ + "server/discover", + "initialize", + "notifications/initialized", + "tools/list", + ] + ); + cancellation.cancel(); + server.await.expect("server task"); +} + +#[tokio::test] +async fn auto_http_client_falls_back_for_recognized_legacy_rejections() { + for rejection in [ + LegacyDiscoveryRejection::UnsupportedProtocol, + LegacyDiscoveryRejection::UnsupportedProtocolWithoutVersionList, + LegacyDiscoveryRejection::MissingSession, + LegacyDiscoveryRejection::InvalidRequest, + LegacyDiscoveryRejection::InvalidParams, + LegacyDiscoveryRejection::NotFound, + LegacyDiscoveryRejection::MethodNotAllowed, + ] { + assert_http_legacy_fallback(rejection).await; + } +} + +async fn assert_http_rejection_does_not_downgrade(rejection: LegacyDiscoveryRejection) { + let methods = Arc::new(Mutex::new(Vec::new())); + let router = axum::Router::new() + .route("/mcp", post(legacy_prevalidation_handler)) + .with_state(LegacyPrevalidationState { + rejection, + methods: methods.clone(), + }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("listener should bind"); + let address = listener.local_addr().expect("listener address"); + let cancellation = CancellationToken::new(); + let server = tokio::spawn({ + let cancellation = cancellation.clone(); + async move { + axum::serve(listener, router) + .with_graceful_shutdown(cancellation.cancelled_owned()) + .await + .expect("serve auth-rejecting HTTP endpoint"); + } + }); + + let transport = StreamableHttpClientTransport::from_config( + StreamableHttpClientTransportConfig::with_uri(format!("http://{address}/mcp")), + ); + let result = ClientInfo::default() + .serve_with_lifecycle( + transport, + ClientLifecycleMode::Auto { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + legacy_version: Some(ProtocolVersion::V_2025_06_18), + }, + ) + .await; + assert!( + result.is_err(), + "authentication failures must not downgrade" + ); + assert_eq!(*methods.lock().expect("methods lock"), ["server/discover"]); + cancellation.cancel(); + server.await.expect("server task"); +} + +#[tokio::test] +async fn auto_http_client_rejects_unsafe_downgrade_signals() { + for rejection in [ + LegacyDiscoveryRejection::Unauthorized, + LegacyDiscoveryRejection::Forbidden, + LegacyDiscoveryRejection::UnauthorizedJson, + LegacyDiscoveryRejection::ForbiddenJson, + LegacyDiscoveryRejection::UnrelatedResponseId, + LegacyDiscoveryRejection::NotFoundWithUnrelatedResponseId, + LegacyDiscoveryRejection::ArbitraryBadRequest, + LegacyDiscoveryRejection::MixedFutureVersions, + LegacyDiscoveryRejection::InternalServerErrorWithLegacyBody, + ] { + assert_http_rejection_does_not_downgrade(rejection).await; + } +} + +#[derive(Debug, thiserror::Error)] +#[error("mock HTTP client error")] +struct MockHttpClientError; + +#[derive(Clone, Default)] +struct TypedProbeFailureClient { + methods: Arc>>, + retry_modern: bool, +} + +impl StreamableHttpClient for TypedProbeFailureClient { + type Error = MockHttpClientError; + + async fn post_message( + &self, + _uri: Arc, + message: ClientJsonRpcMessage, + _session_id: Option>, + _auth_header: Option, + _custom_headers: HashMap, + ) -> Result> { + let method = match &message { + ClientJsonRpcMessage::Request(request) => match &request.request { + ClientRequest::DiscoverRequest(_) => "server/discover", + ClientRequest::InitializeRequest(_) => "initialize", + other => panic!("unexpected request: {other:?}"), + }, + ClientJsonRpcMessage::Notification(_) => "notifications/initialized", + other => panic!("unexpected client message: {other:?}"), + }; + let discover_attempt = { + let mut methods = self.methods.lock().expect("methods lock"); + methods.push(method.to_owned()); + methods + .iter() + .filter(|method| method.as_str() == "server/discover") + .count() + }; + + match message { + ClientJsonRpcMessage::Request(request) + if matches!(&request.request, ClientRequest::DiscoverRequest(_)) => + { + if self.retry_modern && discover_attempt == 1 { + Err(StreamableHttpError::UnexpectedHttpStatus( + HttpStatusError::new( + 400, + serde_json::json!({ + "jsonrpc": "2.0", + "id": request.id, + "error": { + "code": -32022, + "message": "Unsupported protocol version", + "data": { + "supported": ["2026-07-28"], + "requested": "2026-07-28", + }, + }, + }) + .to_string(), + ), + )) + } else if self.retry_modern { + Ok(StreamableHttpPostResponse::Json( + ServerJsonRpcMessage::response( + ServerResult::DiscoverResult(DiscoverResult::new( + vec![ProtocolVersion::V_2026_07_28], + ServerCapabilities::default(), + Implementation::new("modern-server", "1.0.0"), + )), + request.id, + ), + None, + )) + } else { + Err(StreamableHttpError::UnexpectedHttpStatus( + HttpStatusError::new( + 400, + r#"{"jsonrpc":"2.0","id":null,"error":{"code":-32000,"message":"Bad Request: Unsupported protocol version: 2026-07-28"}}"#, + ), + )) + } + } + ClientJsonRpcMessage::Request(request) + if matches!(&request.request, ClientRequest::InitializeRequest(_)) => + { + Ok(StreamableHttpPostResponse::Json( + ServerJsonRpcMessage::response( + ServerResult::InitializeResult( + InitializeResult::new(ServerCapabilities::default()) + .with_protocol_version(ProtocolVersion::V_2025_06_18), + ), + request.id, + ), + Some("legacy-session".to_owned()), + )) + } + ClientJsonRpcMessage::Notification(_) => Ok(StreamableHttpPostResponse::Accepted), + other => panic!("unexpected client message: {other:?}"), + } + } + + async fn delete_session( + &self, + _uri: Arc, + _session_id: Arc, + _auth_header: Option, + _custom_headers: HashMap, + ) -> Result<(), StreamableHttpError> { + Ok(()) + } + + async fn get_stream( + &self, + _uri: Arc, + _session_id: Option>, + _last_event_id: Option, + _auth_header: Option, + _custom_headers: HashMap, + ) -> Result< + futures::stream::BoxStream<'static, Result>, + StreamableHttpError, + > { + Ok(stream::pending().boxed()) + } +} + +#[tokio::test] +async fn auto_mode_consumes_typed_probe_failures_from_custom_http_clients() { + let http_client = TypedProbeFailureClient::default(); + let methods = http_client.methods.clone(); + let transport = StreamableHttpClientTransport::with_client( + http_client, + StreamableHttpClientTransportConfig::with_uri("http://custom.invalid/mcp"), + ); + + let client = ClientInfo::default() + .serve_with_lifecycle( + transport, + ClientLifecycleMode::Auto { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + legacy_version: Some(ProtocolVersion::V_2025_06_18), + }, + ) + .await + .expect("Auto mode should consume the typed custom-client probe failure"); + client.cancel().await.expect("cancel client"); + + assert_eq!( + *methods.lock().expect("methods lock"), + ["server/discover", "initialize", "notifications/initialized"] + ); +} + +#[tokio::test] +async fn discover_mode_consumes_json_rpc_errors_from_typed_http_400_responses() { + let http_client = TypedProbeFailureClient { + retry_modern: true, + ..Default::default() + }; + let methods = http_client.methods.clone(); + let transport = StreamableHttpClientTransport::with_client( + http_client, + StreamableHttpClientTransportConfig::with_uri("http://custom.invalid/mcp"), + ); + + let client = ClientInfo::default() + .serve_with_lifecycle( + transport, + ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + }, + ) + .await + .expect("Discover mode should retry a modern version from a typed HTTP 400 response"); + client.cancel().await.expect("cancel client"); + + assert_eq!( + *methods.lock().expect("methods lock"), + ["server/discover", "server/discover"] + ); +} diff --git a/crates/rmcp/tests/test_streamable_http_4xx_error_body.rs b/crates/rmcp/tests/test_streamable_http_4xx_error_body.rs index ea49a4172..9fc3ea578 100644 --- a/crates/rmcp/tests/test_streamable_http_4xx_error_body.rs +++ b/crates/rmcp/tests/test_streamable_http_4xx_error_body.rs @@ -74,10 +74,10 @@ async fn http_4xx_json_rpc_error_body_is_surfaced_as_json_response() { } } -/// HTTP 4xx with non-JSON content-type must still return `UnexpectedServerResponse` +/// HTTP 4xx with non-JSON content-type must retain the status and response body. /// (no regression on the original error path). #[tokio::test] -async fn http_4xx_non_json_body_returns_unexpected_server_response() { +async fn http_4xx_non_json_body_returns_typed_status_error() { let url = spawn_mock_server(400, "text/plain", "Bad Request").await; let client = reqwest::Client::new(); @@ -92,15 +92,15 @@ async fn http_4xx_non_json_body_returns_unexpected_server_response() { .await; match result { - Err(StreamableHttpError::UnexpectedServerResponse(_)) => {} - other => panic!("expected UnexpectedServerResponse, got: {other:?}"), + Err(StreamableHttpError::UnexpectedHttpStatus(error)) if error.status == 400 => {} + other => panic!("expected UnexpectedHttpStatus, got: {other:?}"), } } /// HTTP 4xx with Content-Type: application/json but a body that is NOT a valid -/// JSON-RPC message must fall back to `UnexpectedServerResponse`. +/// JSON-RPC message must retain the status and response body. #[tokio::test] -async fn http_4xx_malformed_json_body_falls_back_to_unexpected_server_response() { +async fn http_4xx_malformed_json_body_returns_typed_status_error() { let url = spawn_mock_server(400, "application/json", r#"{"error":"not jsonrpc"}"#).await; let client = reqwest::Client::new(); @@ -115,7 +115,7 @@ async fn http_4xx_malformed_json_body_falls_back_to_unexpected_server_response() .await; match result { - Err(StreamableHttpError::UnexpectedServerResponse(_)) => {} - other => panic!("expected UnexpectedServerResponse, got: {other:?}"), + Err(StreamableHttpError::UnexpectedHttpStatus(error)) if error.status == 400 => {} + other => panic!("expected UnexpectedHttpStatus, got: {other:?}"), } } diff --git a/crates/rmcp/tests/test_unix_socket_transport.rs b/crates/rmcp/tests/test_unix_socket_transport.rs index 4c4ad52f1..410531483 100644 --- a/crates/rmcp/tests/test_unix_socket_transport.rs +++ b/crates/rmcp/tests/test_unix_socket_transport.rs @@ -12,7 +12,8 @@ use axum::{ use http::{HeaderName, HeaderValue}; use hyper_util::rt::TokioIo; use rmcp::{ - ServiceExt, + ClientLifecycleMode, ClientServiceExt, ServiceExt, + model::{ClientInfo, ProtocolVersion}, transport::{ StreamableHttpClientTransport, UnixSocketHttpClient, streamable_http_client::StreamableHttpClientTransportConfig, @@ -296,3 +297,100 @@ async fn test_unix_socket_convenience_constructor() -> anyhow::Result<()> { Ok(()) } + +#[derive(Clone, Default)] +struct LegacyDiscoveryState { + methods: Arc>>, +} + +async fn legacy_discovery_handler( + State(state): State, + body: Bytes, +) -> axum::response::Response { + let request: serde_json::Value = serde_json::from_slice(&body).expect("valid JSON-RPC request"); + let method = request + .get("method") + .and_then(serde_json::Value::as_str) + .expect("request method"); + state.methods.lock().await.push(method.to_owned()); + + match method { + "server/discover" => ( + StatusCode::BAD_REQUEST, + [(http::header::CONTENT_TYPE, "application/json")], + json!({ + "jsonrpc": "2.0", + "id": null, + "error": { + "code": -32000, + "message": "Bad Request: Unsupported protocol version: 2026-07-28", + }, + }) + .to_string(), + ) + .into_response(), + "initialize" => ( + StatusCode::OK, + [(http::header::CONTENT_TYPE, "application/json")], + json!({ + "jsonrpc": "2.0", + "id": request.get("id"), + "result": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "serverInfo": { + "name": "legacy-unix-server", + "version": "1.0.0", + }, + }, + }) + .to_string(), + ) + .into_response(), + "notifications/initialized" => StatusCode::ACCEPTED.into_response(), + other => panic!("unexpected method: {other}"), + } +} + +#[tokio::test] +async fn auto_mode_falls_back_from_typed_unix_socket_http_failure() -> anyhow::Result<()> { + let dir = std::env::temp_dir().join(format!("rmcp-test-auto-{}", std::process::id())); + std::fs::create_dir_all(&dir)?; + let socket_path = dir.join("mcp.sock"); + let _ = std::fs::remove_file(&socket_path); + + let state = LegacyDiscoveryState::default(); + let app = Router::new() + .route("/mcp", post(legacy_discovery_handler)) + .with_state(state.clone()); + let listener = tokio::net::UnixListener::bind(&socket_path)?; + let server_handle = spawn_unix_server(listener, app); + + let socket_str = socket_path.to_str().expect("UTF-8 socket path"); + let uri = "http://legacy-unix-server.internal/mcp"; + let transport = StreamableHttpClientTransport::with_client( + UnixSocketHttpClient::new(socket_str, uri), + StreamableHttpClientTransportConfig::with_uri(uri), + ); + let client = ClientInfo::default() + .serve_with_lifecycle( + transport, + ClientLifecycleMode::Auto { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + legacy_version: Some(ProtocolVersion::V_2025_06_18), + }, + ) + .await + .expect("Auto mode should fall back over a Unix socket"); + client.cancel().await.expect("cancel client"); + + assert_eq!( + *state.methods.lock().await, + ["server/discover", "initialize", "notifications/initialized"] + ); + + server_handle.abort(); + let _ = std::fs::remove_file(&socket_path); + let _ = std::fs::remove_dir(&dir); + Ok(()) +}