feat(transport): expose body-free HTTP status errors

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"]

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,19 @@ fn parse_json_rpc_error(body: &str) -> Option<ServerJsonRpcMessage> {
     }
 }
 
+// Retain only status and bounded Retry-After metadata for HTTP failures.
+fn http_status_error(status: StatusCode, headers: &HeaderMap) -> Option<HttpStatusError> {
+    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::new(status.as_u16(), retry_after))
+}
+
 impl StreamableHttpClient for reqwest::Client {
     type Error = reqwest::Error;
 
@@ -125,6 +141,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) => {
@@ -254,6 +273,23 @@ impl StreamableHttpClient for reqwest::Client {
             .headers()
             .get(reqwest::header::CONTENT_TYPE)
             .map(|ct| String::from_utf8_lossy(ct.as_bytes()).to_string());
+        let is_json_response = content_type
+            .as_deref()
+            .is_some_and(|ct| ct.as_bytes().starts_with(JSON_MIME_TYPE.as_bytes()));
+        // 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,
+        ) {
+            (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();
         let session_id = response
             .headers()
@@ -294,11 +330,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));
@@ -308,6 +340,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}"),
             )));
@@ -460,12 +495,50 @@ 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};
     use crate::{
         model::{ClientJsonRpcMessage, ClientRequest, JsonRpcMessage, PingRequest, RequestId},
         transport::streamable_http_client::{AuthRequiredError, InsufficientScopeError},
     };
 
+    #[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);
+            }
+        }
+    }
+
+    #[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<u8>,
+        #[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!(
+            error.retry_after,
+            retained.then(|| String::from_utf8(value).unwrap())
+        );
+    }
+
     #[test]
     fn auth_required_error_new() {
         let err = AuthRequiredError::new("Bearer realm=\"test\"".to_string());

crates/rmcp/src/transport/streamable_http_client.rs

@@ -176,6 +176,41 @@ impl InsufficientScopeError {
     }
 }
 
+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.
+/// 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}")]
+#[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<String>,
+}
+
+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<String>) -> Self {
+        Self {
+            status,
+            retry_after: retry_after.filter(|value| value.len() <= MAX_RETRY_AFTER_BYTES),
+        }
+    }
+}
+
 #[derive(Error, Debug)]
 #[non_exhaustive]
 pub enum StreamableHttpError<E: std::error::Error + Send + Sync + 'static> {
@@ -185,6 +220,9 @@ pub enum StreamableHttpError<E: std::error::Error + Send + Sync + 'static> {
     Io(#[from] std::io::Error),
     #[error("Client error: {0}")]
     Client(E),
+    /// HTTP failure metadata from a transport backend.
+    #[error(transparent)]
+    HttpStatus(#[from] HttpStatusError),
     #[error("unexpected end of stream")]
     UnexpectedEndOfStream,
     #[error("unexpected server response: {0}")]