fix: do not treat malformed JSON 200 as Accepted for requests (#1208)

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

Streamable HTTP mapped any JSON body that failed to parse as
ServerJsonRpcMessage to Accepted, which is correct for notification
POSTs. For a request the worker then waits on SSE forever, so
call_tool hangs when the server returns 200 application/json with
an empty or non-JSON-RPC body.

Signed-off-by: Sebastien Tardif <sebtardif@ncf.ca>

* fix: include a short body preview in JSON-RPC parse errors

Request POST parse failures now include a 256-byte prefix of the
unusable body so empty, HTML, and `{}` are distinguishable from the
serde diagnostic. HTTP reads bytes() then from_slice, matching Unix.

Signed-off-by: Sebastien Tardif <sebtardif@ncf.ca>

---------

Signed-off-by: Sebastien Tardif <sebtardif@ncf.ca>

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

@@ -308,17 +308,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

@@ -322,13 +322,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

@@ -222,6 +222,35 @@ pub enum StreamableHttpError<E: std::error::Error + Send + Sync + 'static> {
     ControlRequestTimeout,
 }
 
+/// 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
@@ -2155,6 +2184,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"