fix: do not treat malformed JSON 200 as Accepted for requests

crates/rmcp/Cargo.toml

@@ -341,6 +341,15 @@ name = "test_streamable_http_4xx_error_body"
 required-features = ["transport-streamable-http-client", "transport-streamable-http-client-reqwest"]
 path = "tests/test_streamable_http_4xx_error_body.rs"
 
+[[test]]
+name = "test_streamable_http_malformed_json_request"
+required-features = [
+  "client",
+  "transport-streamable-http-client",
+  "transport-streamable-http-client-reqwest",
+]
+path = "tests/test_streamable_http_malformed_json_request.rs"
+
 
 [[test]]
 name = "test_custom_request"

crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs

@@ -303,17 +303,29 @@ impl StreamableHttpClient for reqwest::Client {
                 Ok(StreamableHttpPostResponse::Sse(event_stream, session_id))
             }
             Some(ct) if ct.as_bytes().starts_with(JSON_MIME_TYPE.as_bytes()) => {
-                // Try to parse as a valid JSON-RPC message. If the body is
-                // malformed (e.g. a 200 response to a notification that lacks
-                // an `id` field), treat it as accepted rather than failing.
-                match response.json::<ServerJsonRpcMessage>().await {
-                    Ok(message) => Ok(StreamableHttpPostResponse::Json(message, session_id)),
-                    Err(e) => {
+                // A notification/response/error POST does not await a JSON-RPC
+                // reply. Treat an unusable JSON body as Accepted. A request
+                // still needs a reply; the same body is an error so the worker
+                // does not wait on SSE forever.
+                let body = response.bytes().await?;
+                match serde_json::from_slice::<ServerJsonRpcMessage>(&body) {
+                    Ok(parsed) => Ok(StreamableHttpPostResponse::Json(parsed, session_id)),
+                    Err(e)
+                        if matches!(
+                            message,
+                            ClientJsonRpcMessage::Notification(_)
+                                | ClientJsonRpcMessage::Response(_)
+                                | ClientJsonRpcMessage::Error(_)
+                        ) =>
+                    {
                         tracing::warn!(
                             "could not parse JSON response as ServerJsonRpcMessage, treating as accepted: {e}"
                         );
                         Ok(StreamableHttpPostResponse::Accepted)
                     }
+                    Err(e) => Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned(
+                        json_rpc_parse_error_message(&e, &body),
+                    ))),
                 }
             }
             _ => {

crates/rmcp/src/transport/common/unix_socket.rs

@@ -317,13 +317,23 @@ impl StreamableHttpClient for UnixSocketHttpClient {
                     .map_err(|e| StreamableHttpError::Client(UnixSocketError::Hyper(e)))?
                     .to_bytes();
                 match serde_json::from_slice::<ServerJsonRpcMessage>(&body) {
-                    Ok(message) => Ok(StreamableHttpPostResponse::Json(message, session_id)),
-                    Err(e) => {
+                    Ok(parsed) => Ok(StreamableHttpPostResponse::Json(parsed, session_id)),
+                    Err(e)
+                        if matches!(
+                            message,
+                            ClientJsonRpcMessage::Notification(_)
+                                | ClientJsonRpcMessage::Response(_)
+                                | ClientJsonRpcMessage::Error(_)
+                        ) =>
+                    {
                         tracing::warn!(
                             "could not parse JSON response as ServerJsonRpcMessage, treating as accepted: {e}"
                         );
                         Ok(StreamableHttpPostResponse::Accepted)
                     }
+                    Err(e) => Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned(
+                        json_rpc_parse_error_message(&e, &body),
+                    ))),
                 }
             }
             _ => Err(StreamableHttpError::UnexpectedContentType(

crates/rmcp/src/transport/streamable_http_client.rs

@@ -209,6 +209,35 @@ pub enum StreamableHttpError<E: std::error::Error + Send + Sync + 'static> {
     SessionExpired,
 }
 
+/// Bytes of an unusable JSON body kept in
+/// [`StreamableHttpError::UnexpectedServerResponse`].
+pub(crate) const JSON_RPC_BODY_PREVIEW_LEN: usize = 256;
+
+/// Short, lossy view of a body that failed to parse as JSON-RPC.
+pub(crate) fn json_rpc_body_preview(body: &[u8]) -> Cow<'static, str> {
+    if body.is_empty() {
+        return Cow::Borrowed("<empty>");
+    }
+    let lossy = String::from_utf8_lossy(body);
+    if lossy.len() <= JSON_RPC_BODY_PREVIEW_LEN {
+        return Cow::Owned(lossy.into_owned());
+    }
+    let mut end = JSON_RPC_BODY_PREVIEW_LEN;
+    while end > 0 && !lossy.is_char_boundary(end) {
+        end -= 1;
+    }
+    Cow::Owned(lossy[..end].to_owned())
+}
+
+/// Request-POST parse failure, including a body preview so empty / HTML / `{}`
+/// are distinguishable from the serde diagnostic alone.
+pub(crate) fn json_rpc_parse_error_message(err: &serde_json::Error, body: &[u8]) -> String {
+    format!(
+        "could not parse JSON response as ServerJsonRpcMessage: {err}: {}",
+        json_rpc_body_preview(body)
+    )
+}
+
 impl<E: std::error::Error + Send + Sync + 'static> StreamableHttpError<E> {
     /// The `WWW-Authenticate` challenge carried by this error, when the
     /// server answered 401 ([`AuthRequired`](Self::AuthRequired)) or 403
@@ -1797,6 +1826,32 @@ mod tests {
         service::InboundStreamOrigin,
     };
 
+    #[test]
+    fn json_rpc_body_preview_empty_html_and_object() {
+        assert_eq!(json_rpc_body_preview(b""), "<empty>");
+        assert_eq!(json_rpc_body_preview(b"{}"), "{}");
+        assert_eq!(
+            json_rpc_body_preview(b"<html>not json</html>"),
+            "<html>not json</html>"
+        );
+    }
+
+    #[test]
+    fn json_rpc_body_preview_truncates_long_bodies() {
+        let long = vec![b'x'; JSON_RPC_BODY_PREVIEW_LEN + 40];
+        let preview = json_rpc_body_preview(&long);
+        assert_eq!(preview.len(), JSON_RPC_BODY_PREVIEW_LEN);
+        assert!(preview.chars().all(|c| c == 'x'));
+    }
+
+    #[test]
+    fn json_rpc_parse_error_message_includes_preview() {
+        let err = serde_json::from_slice::<ServerJsonRpcMessage>(b"{}").unwrap_err();
+        let msg = json_rpc_parse_error_message(&err, b"{}");
+        assert!(msg.contains("could not parse JSON response as ServerJsonRpcMessage"));
+        assert!(msg.contains("{}"), "missing body preview: {msg}");
+    }
+
     #[expect(
         deprecated,
         reason = "Sampling is deprecated by SEP-2577 but remains the canonical restricted request"

crates/rmcp/tests/test_streamable_http_malformed_json_request.rs

@@ -0,0 +1,195 @@
+#![cfg(all(
+    feature = "transport-streamable-http-client",
+    feature = "transport-streamable-http-client-reqwest",
+    feature = "client",
+    not(feature = "local")
+))]
+
+use std::{collections::HashMap, sync::Arc, time::Duration};
+
+use axum::{Router, body::Bytes, http::StatusCode, response::IntoResponse, routing::post};
+use rmcp::{
+    ServiceError, ServiceExt,
+    model::{
+        CallToolRequestParams, ClientJsonRpcMessage, ClientNotification, ClientRequest,
+        InitializedNotification, PingRequest, RequestId,
+    },
+    transport::{
+        StreamableHttpClientTransport,
+        streamable_http_client::{
+            StreamableHttpClient, StreamableHttpError, StreamableHttpPostResponse,
+        },
+    },
+};
+use rstest::rstest;
+
+/// Initialize succeeds; every later POST (including `tools/call`) returns
+/// HTTP 200 + `application/json` with a body that is not a JSON-RPC message.
+async fn spawn_malformed_json_request_server(call_body: &'static str) -> String {
+    let router = Router::new().route(
+        "/mcp",
+        post(move |body: Bytes| async move {
+            let parsed: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default();
+            let method = parsed.get("method").and_then(|m| m.as_str()).unwrap_or("");
+            if method == "initialize" {
+                let id = parsed.get("id").cloned().unwrap_or(serde_json::json!(1));
+                return (
+                    StatusCode::OK,
+                    [
+                        (http::header::CONTENT_TYPE, "application/json"),
+                        (
+                            http::HeaderName::from_static("mcp-session-id"),
+                            "test-session",
+                        ),
+                    ],
+                    serde_json::json!({
+                        "jsonrpc": "2.0",
+                        "id": id,
+                        "result": {
+                            "protocolVersion": "2025-03-26",
+                            "capabilities": { "tools": {} },
+                            "serverInfo": { "name": "mock", "version": "0.0.1" }
+                        }
+                    })
+                    .to_string(),
+                )
+                    .into_response();
+            }
+            if method == "notifications/initialized" {
+                return StatusCode::ACCEPTED.into_response();
+            }
+            (
+                StatusCode::OK,
+                [(http::header::CONTENT_TYPE, "application/json")],
+                call_body,
+            )
+                .into_response()
+        }),
+    );
+
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        axum::serve(listener, router).await.unwrap();
+    });
+    format!("http://{addr}/mcp")
+}
+
+async fn spawn_json_body_server(body: &'static str) -> String {
+    let router = Router::new().route(
+        "/mcp",
+        post(move || async move {
+            (
+                StatusCode::OK,
+                [(http::header::CONTENT_TYPE, "application/json")],
+                body,
+            )
+        }),
+    );
+
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        axum::serve(listener, router).await.unwrap();
+    });
+    format!("http://{addr}/mcp")
+}
+
+/// Public-API repro: `call_tool` must return a transport error instead of
+/// hanging when the server answers a request POST with 200 JSON that is not
+/// a JSON-RPC message.
+#[rstest]
+#[case::empty_object("{}")]
+#[case::empty_body("")]
+#[case::invalid_json("{")]
+#[case::html("<html>not json</html>")]
+#[tokio::test]
+async fn call_tool_errors_on_malformed_json_200(#[case] call_body: &'static str) {
+    let url = spawn_malformed_json_request_server(call_body).await;
+    let transport = StreamableHttpClientTransport::from_uri(url);
+    let client = ().serve(transport).await.expect("initialize should succeed");
+    let peer = client.peer().clone();
+
+    let result = tokio::time::timeout(
+        Duration::from_secs(2),
+        peer.call_tool(CallToolRequestParams::new("anything")),
+    )
+    .await
+    .expect("call_tool must return instead of hanging on malformed JSON 200");
+
+    match result {
+        Err(ServiceError::TransportSend(ref dyn_err)) => {
+            let err_msg = format!("{dyn_err}");
+            assert!(
+                err_msg.contains("unexpected server response"),
+                "expected UnexpectedServerResponse, got: {err_msg}"
+            );
+            let preview = if call_body.is_empty() {
+                "<empty>"
+            } else {
+                call_body
+            };
+            assert!(
+                err_msg.contains(preview),
+                "expected body preview {preview:?} in error, got: {err_msg}"
+            );
+        }
+        other => panic!("expected TransportSend(UnexpectedServerResponse), got: {other:?}"),
+    }
+
+    let _ = client.cancel().await;
+}
+
+/// Direct `post_message` check: a request POST cannot treat a JSON parse
+/// failure as Accepted.
+#[tokio::test]
+async fn post_request_malformed_json_is_unexpected_server_response() {
+    let url = spawn_json_body_server("{}").await;
+    let client = reqwest::Client::new();
+    let result = client
+        .post_message(
+            Arc::from(url.as_str()),
+            ClientJsonRpcMessage::request(
+                ClientRequest::PingRequest(PingRequest::default()),
+                RequestId::Number(1),
+            ),
+            None,
+            None,
+            HashMap::new(),
+        )
+        .await;
+
+    match result {
+        Err(StreamableHttpError::UnexpectedServerResponse(ref msg)) => {
+            assert!(
+                msg.contains("{}"),
+                "expected body preview in error, got: {msg}"
+            );
+        }
+        other => panic!("expected UnexpectedServerResponse, got: {other:?}"),
+    }
+}
+
+/// Notification POSTs still treat an unusable JSON body as Accepted.
+/// That fallback is for messages that do not await a reply.
+#[tokio::test]
+async fn post_notification_malformed_json_is_still_accepted() {
+    let url = spawn_json_body_server("{}").await;
+    let client = reqwest::Client::new();
+    let result = client
+        .post_message(
+            Arc::from(url.as_str()),
+            ClientJsonRpcMessage::notification(ClientNotification::InitializedNotification(
+                InitializedNotification::default(),
+            )),
+            None,
+            None,
+            HashMap::new(),
+        )
+        .await;
+
+    match result {
+        Ok(StreamableHttpPostResponse::Accepted) => {}
+        other => panic!("expected Accepted, got: {other:?}"),
+    }
+}