From 1174dcf836896567f11e1b2ed8217700722ba4d8 Mon Sep 17 00:00:00 2001 From: theamodhshetty <153991321+theamodhshetty@users.noreply.github.com> Date: Fri, 23 Oct 2026 00:56:44 +0530 Subject: [PATCH 1/4] feat(transport): expose body-free HTTP status errors --- crates/rmcp/CHANGELOG.md | 4 + crates/rmcp/Cargo.toml | 5 + .../common/reqwest/streamable_http_client.rs | 83 +++++++- .../src/transport/streamable_http_client.rs | 22 ++ .../test_streamable_http_status_metadata.rs | 196 ++++++++++++++++++ 5 files changed, 308 insertions(+), 2 deletions(-) create mode 100644 crates/rmcp/tests/test_streamable_http_status_metadata.rs diff --git a/crates/rmcp/CHANGELOG.md b/crates/rmcp/CHANGELOG.md index 13a7c9d26..d40ca418b 100644 --- a/crates/rmcp/CHANGELOG.md +++ b/crates/rmcp/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Return body-free HTTP status and bounded `Retry-After` metadata for 429/5xx responses from the built-in reqwest GET/POST client, before parsing JSON-RPC or SSE bodies. Other HTTP statuses retain their existing handling. + ## [3.6.0](https://github.com/modelcontextprotocol/rust-sdk/compare/rmcp-v3.5.1...rmcp-v3.6.0) - 2026-10-19 ### Added diff --git a/crates/rmcp/Cargo.toml b/crates/rmcp/Cargo.toml index 31586d17f..7da38d364 100644 --- a/crates/rmcp/Cargo.toml +++ b/crates/rmcp/Cargo.toml @@ -501,3 +501,8 @@ path = "tests/test_streamable_http_disconnect_cancel.rs" name = "test_sep_2260_stream_routing" required-features = ["server", "elicitation", "transport-streamable-http-server", "reqwest"] path = "tests/test_sep_2260_stream_routing.rs" + +[[test]] +name = "test_streamable_http_status_metadata" +path = "tests/test_streamable_http_status_metadata.rs" +required-features = ["transport-streamable-http-client", "transport-streamable-http-client-reqwest"] 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 80a29f34f..0e861c551 100644 --- a/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs +++ b/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs @@ -1,7 +1,10 @@ use std::{borrow::Cow, collections::HashMap, sync::Arc}; use futures::stream::BoxStream; -use http::{HeaderName, HeaderValue, header::WWW_AUTHENTICATE}; +use http::{ + HeaderMap, HeaderName, HeaderValue, StatusCode, + header::{RETRY_AFTER, WWW_AUTHENTICATE}, +}; use reqwest::header::ACCEPT; use sse_stream::Sse; @@ -46,6 +49,24 @@ fn parse_json_rpc_error(body: &str) -> Option { } } +const MAX_RETRY_AFTER_BYTES: usize = 128; + +// Classify from headers alone: error bodies may be private, malformed, or unfinished. +fn http_status_error(status: StatusCode, headers: &HeaderMap) -> Option { + if status != StatusCode::TOO_MANY_REQUESTS && !status.is_server_error() { + return None; + } + let retry_after = headers + .get(RETRY_AFTER) + .filter(|value| value.as_bytes().len() <= MAX_RETRY_AFTER_BYTES) + .and_then(|value| std::str::from_utf8(value.as_bytes()).ok()) + .map(str::to_owned); + Some(HttpStatusError { + status: status.as_u16(), + retry_after, + }) +} + impl StreamableHttpClient for reqwest::Client { type Error = reqwest::Error; @@ -125,6 +146,9 @@ impl StreamableHttpClient for reqwest::Client { }, )); } + if let Some(error) = http_status_error(response.status(), response.headers()) { + return Err(error.into()); + } let response = response.error_for_status()?; match response.headers().get(reqwest::header::CONTENT_TYPE) { Some(ct) => { @@ -250,6 +274,9 @@ impl StreamableHttpClient for reqwest::Client { if status == reqwest::StatusCode::NOT_FOUND && session_was_attached { return Err(StreamableHttpError::SessionExpired); } + if let Some(error) = http_status_error(status, response.headers()) { + return Err(error.into()); + } let content_type = response .headers() .get(reqwest::header::CONTENT_TYPE) @@ -460,7 +487,59 @@ fn same_origin_redirect_policy() -> reqwest::redirect::Policy { mod tests { use rstest::rstest; - use super::parse_json_rpc_error; + use super::{http_status_error, parse_json_rpc_error}; + + #[test] + fn http_status_error_classifies_only_429_and_5xx() { + let headers = http::HeaderMap::new(); + for code in 100..=599 { + let error = http_status_error(http::StatusCode::from_u16(code).unwrap(), &headers); + assert_eq!(error.is_some(), code == 429 || code >= 500, "HTTP {code}"); + if let Some(error) = error { + assert_eq!(error.status, code); + assert_eq!(error.retry_after, None); + } + } + } + + #[test] + fn http_status_error_retains_only_bounded_utf8_retry_after() { + fn retained(value: http::HeaderValue) -> Option { + let mut headers = http::HeaderMap::new(); + headers.insert(http::header::RETRY_AFTER, value); + headers.insert( + "x-private-response", + http::HeaderValue::from_static("private"), + ); + http_status_error(http::StatusCode::SERVICE_UNAVAILABLE, &headers) + .unwrap() + .retry_after + } + for value in ["17", "Fri, 23 Oct 2026 12:00:00 GMT", "", "é"] { + assert_eq!( + retained(http::HeaderValue::from_bytes(value.as_bytes()).unwrap()), + Some(value.to_owned()) + ); + } + let boundary = "a".repeat(128); + assert_eq!( + retained(http::HeaderValue::from_str(&boundary).unwrap()), + Some(boundary) + ); + assert_eq!( + retained(http::HeaderValue::from_str(&"a".repeat(129)).unwrap()), + None + ); + // Bound bytes rather than Unicode scalar values. + assert_eq!( + retained(http::HeaderValue::from_bytes("é".repeat(65).as_bytes()).unwrap()), + None + ); + assert_eq!( + retained(http::HeaderValue::from_bytes(b"\xff").unwrap()), + None + ); + } use crate::{ model::{ClientJsonRpcMessage, ClientRequest, JsonRpcMessage, PingRequest, RequestId}, transport::streamable_http_client::{AuthRequiredError, InsufficientScopeError}, diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index ae946a33f..9d0fc1145 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -176,6 +176,25 @@ impl InsufficientScopeError { } } +/// Body-free HTTP failure metadata returned by the built-in reqwest client's +/// GET and POST methods for HTTP 429 and 5xx responses. +/// +/// These statuses are classified before decoding any JSON-RPC or SSE body. +/// No response body, request URL, credentials, or other headers are retained. +/// This does not add automatic retries or change session DELETE handling. +#[derive(Debug, Error)] +#[error("HTTP {status}")] +#[non_exhaustive] +pub struct HttpStatusError { + /// HTTP status code (429 or 500..=599 for the built-in reqwest client). + pub status: u16, + /// First raw `Retry-After` value, if valid UTF-8 and at most 128 bytes. + /// + /// This is not parsed as a delay or date; callers decide whether and when + /// to retry. Missing, oversized, or invalid UTF-8 values become `None`. + pub retry_after: Option, +} + #[derive(Error, Debug)] #[non_exhaustive] pub enum StreamableHttpError { @@ -185,6 +204,9 @@ pub enum StreamableHttpError { Io(#[from] std::io::Error), #[error("Client error: {0}")] Client(E), + /// HTTP 429 or 5xx from a built-in reqwest GET/POST, classified before body parsing. + #[error(transparent)] + HttpStatus(#[from] HttpStatusError), #[error("unexpected end of stream")] UnexpectedEndOfStream, #[error("unexpected server response: {0}")] diff --git a/crates/rmcp/tests/test_streamable_http_status_metadata.rs b/crates/rmcp/tests/test_streamable_http_status_metadata.rs new file mode 100644 index 000000000..4953284fd --- /dev/null +++ b/crates/rmcp/tests/test_streamable_http_status_metadata.rs @@ -0,0 +1,196 @@ +#![cfg(all( + feature = "transport-streamable-http-client", + feature = "transport-streamable-http-client-reqwest", + not(feature = "local") +))] + +use std::{ + collections::HashMap, + convert::Infallible, + sync::{Arc, Mutex}, + time::Duration, +}; + +use axum::{ + Router, + body::{Body, Bytes}, + http::Response, + routing::any, +}; +use futures::{StreamExt, stream}; +use rmcp::{ + ClientLifecycleMode, ClientServiceExt, RoleClient, + model::{ + ClientConfig, ClientJsonRpcMessage, ClientRequest, PingRequest, ProtocolVersion, RequestId, + }, + service::ClientInitializeError, + transport::{ + StreamableHttpClientTransport, + streamable_http_client::{ + StreamableHttpClient, StreamableHttpClientTransportConfig, StreamableHttpError, + }, + }, +}; +use tokio::task::JoinHandle; + +struct MockServer { + uri: Arc, + task: JoinHandle<()>, + methods: Arc>>, +} + +impl Drop for MockServer { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn mock_server(status: u16, unfinished: bool) -> MockServer { + mock_response(status, unfinished, "private-response-body").await +} + +async fn mock_response(status: u16, unfinished: bool, text: &'static str) -> MockServer { + let methods = Arc::new(Mutex::new(Vec::new())); + let captured = methods.clone(); + let router = Router::new().route( + "/mcp", + any(move |body: Bytes| { + let methods = captured.clone(); + async move { + if let Ok(message) = serde_json::from_slice::(&body) + && let Some(method) = message["method"].as_str() + { + methods.lock().unwrap().push(method.to_owned()); + } + let body = if unfinished { + Body::from_stream( + stream::once(async move { Ok::<_, Infallible>(text) }) + .chain(stream::pending()), + ) + } else { + Body::from(text) + }; + Response::builder() + .status(status) + .header("retry-after", "17") + .header("content-type", "application/json") + .header("x-private-response", "private-response-header") + .body(body) + .unwrap() + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let task = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + MockServer { + uri: Arc::from(format!("http://{address}/mcp?private-query")), + task, + methods, + } +} + +fn ping() -> ClientJsonRpcMessage { + ClientJsonRpcMessage::request( + ClientRequest::PingRequest(PingRequest::default()), + RequestId::Number(1), + ) +} + +fn assert_status_error(error: StreamableHttpError, status: u16) { + let rendered = format!("{error:?} {error}"); + assert!( + !rendered.contains("private-"), + "retained private response/request state: {rendered}" + ); + match error { + StreamableHttpError::HttpStatus(metadata) => { + assert_eq!(metadata.status, status); + assert_eq!(metadata.retry_after.as_deref(), Some("17")); + } + other => panic!("expected body-free HTTP status metadata, got {other:?}"), + } +} + +#[tokio::test] +async fn post_rate_limit_and_server_errors_return_before_the_body_finishes() { + for status in [429, 500, 503, 599] { + let server = mock_server(status, true).await; + let client = reqwest::Client::new(); + let result = tokio::time::timeout( + Duration::from_secs(2), + client.post_message( + server.uri.clone(), + ping(), + Some(Arc::from("private-session")), + Some("private-token".to_owned()), + HashMap::new(), + ), + ) + .await + .expect("HTTP status classification must not wait for the error body to finish"); + assert_status_error(result.unwrap_err(), status); + } +} + +#[tokio::test] +async fn get_errors_keep_only_status_and_retry_after() { + for status in [429, 503] { + let server = mock_server(status, false).await; + let error = reqwest::Client::new() + .get_stream( + server.uri.clone(), + Some(Arc::from("private-session")), + Some("private-event-id".to_owned()), + Some("private-token".to_owned()), + HashMap::new(), + ) + .await + .err() + .expect("HTTP status must fail"); + assert_status_error(error, status); + } +} + +#[tokio::test] +async fn selected_status_takes_precedence_over_a_json_rpc_error_body() { + // This is a deliberate difference from other HTTP errors (e.g. 400): + // 429/5xx are classified without reading even a valid JSON-RPC error body. + let text = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32603,"message":"private-error"}}"#; + let server = mock_response(503, false, text).await; + let result = reqwest::Client::new() + .post_message(server.uri.clone(), ping(), None, None, HashMap::new()) + .await; + assert_status_error(result.unwrap_err(), 503); +} + +#[tokio::test] +async fn rate_limited_auto_discovery_preserves_metadata_without_legacy_fallback() { + let server = mock_server(429, false).await; + let transport = StreamableHttpClientTransport::with_client( + reqwest::Client::new(), + StreamableHttpClientTransportConfig::with_uri(server.uri.clone()), + ); + let result = tokio::time::timeout( + Duration::from_secs(2), + ClientConfig::default().serve_with_lifecycle( + transport, + ClientLifecycleMode::Auto { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + legacy_version: Some(ProtocolVersion::V_2025_11_25), + }, + ), + ) + .await + .expect("rate-limited discovery should fail promptly"); + match result.expect_err("rate-limited discovery must fail") { + ClientInitializeError::TransportError { error, .. } => { + let error = error + .downcast::, RoleClient>() + .expect("transport error type must be preserved"); + assert_status_error(error, 429); + } + other => panic!("expected HTTP transport error, got {other:?}"), + } + assert_eq!(*server.methods.lock().unwrap(), ["server/discover"]); +} From 2e8420cf6fd197f08812f3142937ff18a800197b Mon Sep 17 00:00:00 2001 From: theamodhshetty <153991321+theamodhshetty@users.noreply.github.com> Date: Fri, 23 Oct 2026 01:43:41 +0530 Subject: [PATCH 2/4] fix(transport): preserve JSON-RPC errors with HTTP metadata --- crates/rmcp/CHANGELOG.md | 4 - .../common/reqwest/streamable_http_client.rs | 95 ++++++++----------- .../src/transport/streamable_http_client.rs | 25 ++++- .../test_streamable_http_status_metadata.rs | 74 +++++++++++++-- 4 files changed, 126 insertions(+), 72 deletions(-) diff --git a/crates/rmcp/CHANGELOG.md b/crates/rmcp/CHANGELOG.md index d40ca418b..13a7c9d26 100644 --- a/crates/rmcp/CHANGELOG.md +++ b/crates/rmcp/CHANGELOG.md @@ -7,10 +7,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -### Added - -- Return body-free HTTP status and bounded `Retry-After` metadata for 429/5xx responses from the built-in reqwest GET/POST client, before parsing JSON-RPC or SSE bodies. Other HTTP statuses retain their existing handling. - ## [3.6.0](https://github.com/modelcontextprotocol/rust-sdk/compare/rmcp-v3.5.1...rmcp-v3.6.0) - 2026-10-19 ### Added 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 0e861c551..0f486b0f9 100644 --- a/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs +++ b/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs @@ -49,9 +49,7 @@ fn parse_json_rpc_error(body: &str) -> Option { } } -const MAX_RETRY_AFTER_BYTES: usize = 128; - -// Classify from headers alone: error bodies may be private, malformed, or unfinished. +// Retain only status and bounded Retry-After metadata for HTTP failures. fn http_status_error(status: StatusCode, headers: &HeaderMap) -> Option { if status != StatusCode::TOO_MANY_REQUESTS && !status.is_server_error() { return None; @@ -61,10 +59,7 @@ fn http_status_error(status: StatusCode, headers: &HeaderMap) -> Option return Err(error.into()), + (error, _) => error, + }; let content_length = response.content_length(); let session_id = response .headers() @@ -321,11 +325,7 @@ impl StreamableHttpClient for reqwest::Client { status, reqwest::StatusCode::UNAUTHORIZED | reqwest::StatusCode::FORBIDDEN ); - if !is_auth_failure - && content_type - .as_deref() - .is_some_and(|ct| ct.as_bytes().starts_with(JSON_MIME_TYPE.as_bytes())) - { + if !is_auth_failure && is_json_response { match parse_json_rpc_error(&body) { Some(message) => { return Ok(StreamableHttpPostResponse::Json(message, session_id)); @@ -335,6 +335,9 @@ impl StreamableHttpClient for reqwest::Client { ), } } + if let Some(error) = status_error { + return Err(error.into()); + } return Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned( format!("HTTP {status}: {body}"), ))); @@ -488,6 +491,10 @@ mod tests { use rstest::rstest; use super::{http_status_error, parse_json_rpc_error}; + use crate::{ + model::{ClientJsonRpcMessage, ClientRequest, JsonRpcMessage, PingRequest, RequestId}, + transport::streamable_http_client::{AuthRequiredError, InsufficientScopeError}, + }; #[test] fn http_status_error_classifies_only_429_and_5xx() { @@ -502,48 +509,30 @@ mod tests { } } - #[test] - fn http_status_error_retains_only_bounded_utf8_retry_after() { - fn retained(value: http::HeaderValue) -> Option { - let mut headers = http::HeaderMap::new(); - headers.insert(http::header::RETRY_AFTER, value); - headers.insert( - "x-private-response", - http::HeaderValue::from_static("private"), - ); - http_status_error(http::StatusCode::SERVICE_UNAVAILABLE, &headers) - .unwrap() - .retry_after - } - for value in ["17", "Fri, 23 Oct 2026 12:00:00 GMT", "", "é"] { - assert_eq!( - retained(http::HeaderValue::from_bytes(value.as_bytes()).unwrap()), - Some(value.to_owned()) - ); - } - let boundary = "a".repeat(128); - assert_eq!( - retained(http::HeaderValue::from_str(&boundary).unwrap()), - Some(boundary) - ); - assert_eq!( - retained(http::HeaderValue::from_str(&"a".repeat(129)).unwrap()), - None - ); - // Bound bytes rather than Unicode scalar values. - assert_eq!( - retained(http::HeaderValue::from_bytes("é".repeat(65).as_bytes()).unwrap()), - None + #[rstest] + #[case::delay_seconds(b"17".to_vec(), true)] + #[case::http_date(b"Fri, 23 Oct 2026 12:00:00 GMT".to_vec(), true)] + #[case::empty(Vec::new(), true)] + #[case::non_ascii_utf8("é".as_bytes().to_vec(), true)] + #[case::at_limit("a".repeat(128).into_bytes(), true)] + #[case::over_limit("a".repeat(129).into_bytes(), false)] + #[case::limit_counts_bytes_not_chars("é".repeat(65).into_bytes(), false)] + #[case::invalid_utf8(b"\xff".to_vec(), false)] + fn http_status_error_retains_only_bounded_utf8_retry_after( + #[case] value: Vec, + #[case] retained: bool, + ) { + let mut headers = http::HeaderMap::new(); + headers.insert( + http::header::RETRY_AFTER, + http::HeaderValue::from_bytes(&value).unwrap(), ); + let error = http_status_error(http::StatusCode::SERVICE_UNAVAILABLE, &headers).unwrap(); assert_eq!( - retained(http::HeaderValue::from_bytes(b"\xff").unwrap()), - None + error.retry_after, + retained.then(|| String::from_utf8(value).unwrap()) ); } - use crate::{ - model::{ClientJsonRpcMessage, ClientRequest, JsonRpcMessage, PingRequest, RequestId}, - transport::streamable_http_client::{AuthRequiredError, InsufficientScopeError}, - }; #[test] fn auth_required_error_new() { diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 9d0fc1145..32857bcb5 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -176,12 +176,13 @@ impl InsufficientScopeError { } } -/// Body-free HTTP failure metadata returned by the built-in reqwest client's -/// GET and POST methods for HTTP 429 and 5xx responses. +pub(crate) const MAX_RETRY_AFTER_BYTES: usize = 128; + +/// Body-free HTTP failure metadata. /// -/// These statuses are classified before decoding any JSON-RPC or SSE body. +/// The built-in reqwest client returns this for HTTP 429 and 5xx responses. +/// POST responses containing a valid JSON-RPC error retain that error instead. /// No response body, request URL, credentials, or other headers are retained. -/// This does not add automatic retries or change session DELETE handling. #[derive(Debug, Error)] #[error("HTTP {status}")] #[non_exhaustive] @@ -195,6 +196,20 @@ pub struct HttpStatusError { pub retry_after: Option, } +impl HttpStatusError { + /// Create HTTP failure metadata for a custom transport backend. + /// + /// `retry_after` is retained without parsing if it is at most 128 bytes; + /// oversized values are discarded. No status-code restriction is imposed + /// on custom backends. + pub fn new(status: u16, retry_after: Option) -> Self { + Self { + status, + retry_after: retry_after.filter(|value| value.len() <= MAX_RETRY_AFTER_BYTES), + } + } +} + #[derive(Error, Debug)] #[non_exhaustive] pub enum StreamableHttpError { @@ -204,7 +219,7 @@ pub enum StreamableHttpError { Io(#[from] std::io::Error), #[error("Client error: {0}")] Client(E), - /// HTTP 429 or 5xx from a built-in reqwest GET/POST, classified before body parsing. + /// HTTP failure metadata from a transport backend. #[error(transparent)] HttpStatus(#[from] HttpStatusError), #[error("unexpected end of stream")] diff --git a/crates/rmcp/tests/test_streamable_http_status_metadata.rs b/crates/rmcp/tests/test_streamable_http_status_metadata.rs index 4953284fd..816782e6e 100644 --- a/crates/rmcp/tests/test_streamable_http_status_metadata.rs +++ b/crates/rmcp/tests/test_streamable_http_status_metadata.rs @@ -21,13 +21,15 @@ use futures::{StreamExt, stream}; use rmcp::{ ClientLifecycleMode, ClientServiceExt, RoleClient, model::{ - ClientConfig, ClientJsonRpcMessage, ClientRequest, PingRequest, ProtocolVersion, RequestId, + ClientConfig, ClientJsonRpcMessage, ClientRequest, JsonRpcMessage, PingRequest, + ProtocolVersion, RequestId, }, service::ClientInitializeError, transport::{ StreamableHttpClientTransport, streamable_http_client::{ - StreamableHttpClient, StreamableHttpClientTransportConfig, StreamableHttpError, + HttpStatusError, StreamableHttpClient, StreamableHttpClientTransportConfig, + StreamableHttpError, }, }, }; @@ -46,10 +48,15 @@ impl Drop for MockServer { } async fn mock_server(status: u16, unfinished: bool) -> MockServer { - mock_response(status, unfinished, "private-response-body").await + mock_response(status, unfinished, "private-response-body", "text/plain").await } -async fn mock_response(status: u16, unfinished: bool, text: &'static str) -> MockServer { +async fn mock_response( + status: u16, + unfinished: bool, + text: &'static str, + content_type: &'static str, +) -> MockServer { let methods = Arc::new(Mutex::new(Vec::new())); let captured = methods.clone(); let router = Router::new().route( @@ -73,7 +80,7 @@ async fn mock_response(status: u16, unfinished: bool, text: &'static str) -> Moc Response::builder() .status(status) .header("retry-after", "17") - .header("content-type", "application/json") + .header("content-type", content_type) .header("x-private-response", "private-response-header") .body(body) .unwrap() @@ -152,12 +159,37 @@ async fn get_errors_keep_only_status_and_retry_after() { } } +#[rstest::rstest] +#[case(429)] +#[case(500)] +#[case(503)] #[tokio::test] -async fn selected_status_takes_precedence_over_a_json_rpc_error_body() { - // This is a deliberate difference from other HTTP errors (e.g. 400): - // 429/5xx are classified without reading even a valid JSON-RPC error body. - let text = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32603,"message":"private-error"}}"#; - let server = mock_response(503, false, text).await; +async fn json_rpc_errors_preserve_the_server_code_and_message(#[case] status: u16) { + let text = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32603,"message":"server-error"}}"#; + let server = mock_response(status, false, text, "application/json").await; + let result = reqwest::Client::new() + .post_message(server.uri.clone(), ping(), None, None, HashMap::new()) + .await + .expect("a JSON-RPC error should reach the caller"); + match result { + rmcp::transport::streamable_http_client::StreamableHttpPostResponse::Json( + JsonRpcMessage::Error(error), + _, + ) => { + assert_eq!(error.id, Some(RequestId::Number(1))); + assert_eq!(error.error.code.0, -32603); + assert_eq!(error.error.message, "server-error"); + } + other => panic!("expected the original JSON-RPC error, got {other:?}"), + } +} + +#[rstest::rstest] +#[case("private-invalid-json")] +#[case(r#"{"jsonrpc":"2.0","id":1,"result":{}}"#)] +#[tokio::test] +async fn unusable_json_error_bodies_fall_back_to_status_metadata(#[case] text: &'static str) { + let server = mock_response(503, false, text, "application/json").await; let result = reqwest::Client::new() .post_message(server.uri.clone(), ping(), None, None, HashMap::new()) .await; @@ -194,3 +226,25 @@ async fn rate_limited_auto_discovery_preserves_metadata_without_legacy_fallback( } assert_eq!(*server.methods.lock().unwrap(), ["server/discover"]); } + +#[test] +fn custom_backends_can_construct_bounded_status_metadata() { + // This integration test is compiled outside rmcp, where a non-exhaustive + // struct cannot be constructed with a struct literal. + for value in [ + None, + Some("17".to_owned()), + Some("a".repeat(128)), + Some("é".repeat(65)), + ] { + let expected = value.clone().filter(|value| value.len() <= 128); + let error: StreamableHttpError = HttpStatusError::new(503, value).into(); + match error { + StreamableHttpError::HttpStatus(metadata) => { + assert_eq!(metadata.status, 503); + assert_eq!(metadata.retry_after, expected); + } + other => panic!("expected custom backend status metadata, got {other:?}"), + } + } +} From cf40e5b136e7a70a04cf36996b34cab0d0105ff6 Mon Sep 17 00:00:00 2001 From: theamodhshetty <153991321+theamodhshetty@users.noreply.github.com> Date: Fri, 23 Oct 2026 02:19:09 +0530 Subject: [PATCH 3/4] fix(transport): classify JSON rate limits before reading bodies --- .../common/reqwest/streamable_http_client.rs | 10 +++++++--- .../src/transport/streamable_http_client.rs | 3 ++- .../test_streamable_http_status_metadata.rs | 17 ++++++++++++++++- 3 files changed, 25 insertions(+), 5 deletions(-) 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 0f486b0f9..b474ca7cf 100644 --- a/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs +++ b/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs @@ -276,13 +276,17 @@ impl StreamableHttpClient for reqwest::Client { let is_json_response = content_type .as_deref() .is_some_and(|ct| ct.as_bytes().starts_with(JSON_MIME_TYPE.as_bytes())); - // Preserve JSON-RPC errors carried by JSON responses. Non-JSON failures - // can be reported without waiting for a possibly unfinished body. + // Rate limits must not wait for a body. Preserve JSON-RPC errors from + // JSON server failures; non-JSON failures also return immediately. let status_error = match ( http_status_error(status, response.headers()), is_json_response, ) { - (Some(error), false) => return Err(error.into()), + (Some(error), is_json) + if !is_json || status == reqwest::StatusCode::TOO_MANY_REQUESTS => + { + return Err(error.into()); + } (error, _) => error, }; let content_length = response.content_length(); diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 32857bcb5..14005f645 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -181,7 +181,8 @@ pub(crate) const MAX_RETRY_AFTER_BYTES: usize = 128; /// Body-free HTTP failure metadata. /// /// The built-in reqwest client returns this for HTTP 429 and 5xx responses. -/// POST responses containing a valid JSON-RPC error retain that error instead. +/// HTTP 429 is classified without reading the body. JSON POST 5xx responses +/// containing a valid JSON-RPC error retain that error instead. /// No response body, request URL, credentials, or other headers are retained. #[derive(Debug, Error)] #[error("HTTP {status}")] diff --git a/crates/rmcp/tests/test_streamable_http_status_metadata.rs b/crates/rmcp/tests/test_streamable_http_status_metadata.rs index 816782e6e..e00cd5559 100644 --- a/crates/rmcp/tests/test_streamable_http_status_metadata.rs +++ b/crates/rmcp/tests/test_streamable_http_status_metadata.rs @@ -160,7 +160,6 @@ async fn get_errors_keep_only_status_and_retry_after() { } #[rstest::rstest] -#[case(429)] #[case(500)] #[case(503)] #[tokio::test] @@ -184,6 +183,22 @@ async fn json_rpc_errors_preserve_the_server_code_and_message(#[case] status: u1 } } +#[rstest::rstest] +#[case::unfinished(true)] +#[case::complete(false)] +#[tokio::test] +async fn json_rate_limits_return_without_reading_the_body(#[case] unfinished: bool) { + let text = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32603,"message":"private-error"}}"#; + let server = mock_response(429, unfinished, text, "application/json").await; + let result = tokio::time::timeout( + Duration::from_secs(2), + reqwest::Client::new().post_message(server.uri.clone(), ping(), None, None, HashMap::new()), + ) + .await + .expect("JSON rate limits must not wait for the body to finish"); + assert_status_error(result.unwrap_err(), 429); +} + #[rstest::rstest] #[case("private-invalid-json")] #[case(r#"{"jsonrpc":"2.0","id":1,"result":{}}"#)] From 5ab959d9411d7787109c9120c28ffbae7ea21735 Mon Sep 17 00:00:00 2001 From: theamodhshetty <153991321+theamodhshetty@users.noreply.github.com> Date: Fri, 23 Oct 2026 06:35:54 +0530 Subject: [PATCH 4/4] docs(transport): clarify JSON server error body handling --- .../src/transport/common/reqwest/streamable_http_client.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) 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 b474ca7cf..dbd6ae7de 100644 --- a/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs +++ b/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs @@ -276,8 +276,9 @@ impl StreamableHttpClient for reqwest::Client { let is_json_response = content_type .as_deref() .is_some_and(|ct| ct.as_bytes().starts_with(JSON_MIME_TYPE.as_bytes())); - // Rate limits must not wait for a body. Preserve JSON-RPC errors from - // JSON server failures; non-JSON failures also return immediately. + // Rate limits and non-JSON failures return without reading the body. JSON + // server failures are read to keep their JSON-RPC error, so an unfinished + // body still blocks here. let status_error = match ( http_status_error(status, response.headers()), is_json_response,