From 4b1566662d8c9c90b8b21d873ef7be2a6b760e03 Mon Sep 17 00:00:00 2001 From: tsouth89 Date: Sun, 28 Jun 2026 14:08:11 -0400 Subject: [PATCH 1/4] fix: don't respond to unparseable messages --- crates/rmcp/src/transport/async_rw.rs | 54 +++++++++++---------------- 1 file changed, 22 insertions(+), 32 deletions(-) diff --git a/crates/rmcp/src/transport/async_rw.rs b/crates/rmcp/src/transport/async_rw.rs index 2ef0aae25..8d5476bdd 100644 --- a/crates/rmcp/src/transport/async_rw.rs +++ b/crates/rmcp/src/transport/async_rw.rs @@ -13,10 +13,7 @@ use tokio_util::{ }; use super::{IntoTransport, Transport}; -use crate::{ - model::ErrorData, - service::{RxJsonRpcMessage, ServiceRole, TxJsonRpcMessage}, -}; +use crate::service::{RxJsonRpcMessage, ServiceRole, TxJsonRpcMessage}; #[non_exhaustive] pub enum TransportAdapterAsyncRW {} @@ -143,16 +140,12 @@ where Ok(Some(msg)) => return Some(msg), Ok(None) => continue, Err(JsonRpcMessageCodecError::Serde(e)) => { - tracing::debug!("Parse error on incoming message: {e}"); - let mut write = self.write.lock().await; - let framed = write.as_mut()?; - let response = TxJsonRpcMessage::::error( - ErrorData::parse_error("Parse error", None), - None, - ); - if framed.send(response).await.is_err() { - return None; - } + // Don't respond to unparseable input. The message id is unknown, so a response + // can't be correlated to a request, and replying to invalid data can trigger an + // error storm if the peer echoes the response back as more invalid data. This + // matches the other official MCP SDKs, which ignore unparseable messages. + // See https://github.com/modelcontextprotocol/rust-sdk/issues/938 + tracing::debug!("Ignoring unparseable incoming message: {e}"); } Err(e) => { tracing::error!("Error reading from stream: {}", e); @@ -618,8 +611,8 @@ mod test { #[cfg(feature = "server")] #[tokio::test] - async fn receive_recovers_from_parse_error() { - use tokio::io::AsyncWriteExt; + async fn receive_ignores_parse_error() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; use crate::{RoleServer, transport::Transport}; @@ -638,28 +631,25 @@ mod test { .await .unwrap(); + // The unparseable line is skipped and the next valid message is still yielded. let received = transport .receive() .await - .expect("transport should recover and yield the next valid message"); - - // Read one line back from the peer side and parse as JSON. - let mut reply_buf = Vec::new(); - let mut peer = tokio::io::BufReader::new(&mut client_r); - peer.read_until(b'\n', &mut reply_buf).await.unwrap(); - let reply: serde_json::Value = serde_json::from_slice(&reply_buf).unwrap(); - - // Per MCP 2025-11-25: id is omitted when the server can't read the request id. - assert_eq!( - reply, - serde_json::json!({ - "jsonrpc": "2.0", - "error": {"code": -32700, "message": "Parse error"}, - }) - ); + .expect("transport should skip the invalid line and yield the next valid message"); assert_eq!( serde_json::to_value(&received).unwrap()["method"], "notifications/initialized", ); + + // No response is sent back for the unparseable message (issue #938). Dropping the + // transport closes its write side, so the peer reads to EOF and should see no bytes. + drop(transport); + let mut reply_buf = Vec::new(); + client_r.read_to_end(&mut reply_buf).await.unwrap(); + assert!( + reply_buf.is_empty(), + "expected no response to an unparseable message, got: {}", + String::from_utf8_lossy(&reply_buf), + ); } } From 6507c0ed9cf64d73f807bb68cc3e9ecfc7e02ab4 Mon Sep 17 00:00:00 2001 From: tsouth89 Date: Sun, 28 Jun 2026 15:07:15 -0400 Subject: [PATCH 2/4] docs: spell 'unparsable' to satisfy typos linter --- crates/rmcp/src/transport/async_rw.rs | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/crates/rmcp/src/transport/async_rw.rs b/crates/rmcp/src/transport/async_rw.rs index 8d5476bdd..1d32a36bd 100644 --- a/crates/rmcp/src/transport/async_rw.rs +++ b/crates/rmcp/src/transport/async_rw.rs @@ -140,12 +140,12 @@ where Ok(Some(msg)) => return Some(msg), Ok(None) => continue, Err(JsonRpcMessageCodecError::Serde(e)) => { - // Don't respond to unparseable input. The message id is unknown, so a response + // Don't respond to unparsable input. The message id is unknown, so a response // can't be correlated to a request, and replying to invalid data can trigger an // error storm if the peer echoes the response back as more invalid data. This - // matches the other official MCP SDKs, which ignore unparseable messages. + // matches the other official MCP SDKs, which ignore unparsable messages. // See https://github.com/modelcontextprotocol/rust-sdk/issues/938 - tracing::debug!("Ignoring unparseable incoming message: {e}"); + tracing::debug!("Ignoring unparsable incoming message: {e}"); } Err(e) => { tracing::error!("Error reading from stream: {}", e); @@ -631,7 +631,7 @@ mod test { .await .unwrap(); - // The unparseable line is skipped and the next valid message is still yielded. + // The unparsable line is skipped and the next valid message is still yielded. let received = transport .receive() .await @@ -641,14 +641,14 @@ mod test { "notifications/initialized", ); - // No response is sent back for the unparseable message (issue #938). Dropping the + // No response is sent back for the unparsable message (issue #938). Dropping the // transport closes its write side, so the peer reads to EOF and should see no bytes. drop(transport); let mut reply_buf = Vec::new(); client_r.read_to_end(&mut reply_buf).await.unwrap(); assert!( reply_buf.is_empty(), - "expected no response to an unparseable message, got: {}", + "expected no response to an unparsable message, got: {}", String::from_utf8_lossy(&reply_buf), ); } From ff1dd33fe4c6a262a4cb8a69ff7f63c123facdc8 Mon Sep 17 00:00:00 2001 From: tsouth89 Date: Tue, 30 Jun 2026 23:01:10 -0400 Subject: [PATCH 3/4] fix: only ignore unparsable JSON, keep protocol errors visible Classify the serde error in the receive loop: syntax/EOF errors are unparsable input with no correlatable id (issue #938) and stay silent, while data errors (valid JSON that doesn't match the message shape) are real protocol errors and get an error response instead of being dropped. Add a test covering the protocol-error path. --- crates/rmcp/src/transport/async_rw.rs | 81 ++++++++++++++++++++++++--- 1 file changed, 74 insertions(+), 7 deletions(-) diff --git a/crates/rmcp/src/transport/async_rw.rs b/crates/rmcp/src/transport/async_rw.rs index 1d32a36bd..62ab86b47 100644 --- a/crates/rmcp/src/transport/async_rw.rs +++ b/crates/rmcp/src/transport/async_rw.rs @@ -13,7 +13,10 @@ use tokio_util::{ }; use super::{IntoTransport, Transport}; -use crate::service::{RxJsonRpcMessage, ServiceRole, TxJsonRpcMessage}; +use crate::{ + model::ErrorData, + service::{RxJsonRpcMessage, ServiceRole, TxJsonRpcMessage}, +}; #[non_exhaustive] pub enum TransportAdapterAsyncRW {} @@ -140,12 +143,31 @@ where Ok(Some(msg)) => return Some(msg), Ok(None) => continue, Err(JsonRpcMessageCodecError::Serde(e)) => { - // Don't respond to unparsable input. The message id is unknown, so a response - // can't be correlated to a request, and replying to invalid data can trigger an - // error storm if the peer echoes the response back as more invalid data. This - // matches the other official MCP SDKs, which ignore unparsable messages. - // See https://github.com/modelcontextprotocol/rust-sdk/issues/938 - tracing::debug!("Ignoring unparsable incoming message: {e}"); + match e.classify() { + serde_json::error::Category::Syntax | serde_json::error::Category::Eof => { + // The input isn't valid JSON, so there's no message id to correlate a + // response to, and replying to invalid data can trigger an error storm + // if the peer echoes the response back as more invalid data. This + // matches the other official MCP SDKs, which ignore unparsable input. + // See https://github.com/modelcontextprotocol/rust-sdk/issues/938 + tracing::debug!("Ignoring unparsable incoming message: {e}"); + } + serde_json::error::Category::Data | serde_json::error::Category::Io => { + // Valid JSON that doesn't match the expected message shape is a real + // protocol error rather than unparsable input, so surface it with a + // response instead of silently dropping it. + tracing::debug!("Protocol error on incoming message: {e}"); + let mut write = self.write.lock().await; + let framed = write.as_mut()?; + let response = TxJsonRpcMessage::::error( + ErrorData::parse_error("Parse error", None), + None, + ); + if framed.send(response).await.is_err() { + return None; + } + } + } } Err(e) => { tracing::error!("Error reading from stream: {}", e); @@ -652,4 +674,49 @@ mod test { String::from_utf8_lossy(&reply_buf), ); } + + #[cfg(feature = "server")] + #[tokio::test] + async fn receive_responds_to_protocol_error() { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + + use crate::{RoleServer, transport::Transport}; + + let (server_io, client_io) = tokio::io::duplex(4096); + let (server_r, server_w) = tokio::io::split(server_io); + let (client_r, mut client_w) = tokio::io::split(client_io); + + let mut transport = AsyncRwTransport::::new(server_r, server_w); + + // Well-formed JSON that does not match the JSON-RPC message shape, followed by a + // valid notification. Unlike unparsable bytes, this is a protocol error: the + // transport should reply to it and still yield the next valid message. + client_w + .write_all( + b"{\"foo\":\"bar\"}\n{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}\n", + ) + .await + .unwrap(); + + let received = transport.receive().await.expect( + "transport should reply to the protocol error and yield the next valid message", + ); + assert_eq!( + serde_json::to_value(&received).unwrap()["method"], + "notifications/initialized", + ); + + // A protocol error gets an error response back (id omitted since it can't be read). + let mut reply_buf = Vec::new(); + let mut peer = BufReader::new(client_r); + peer.read_until(b'\n', &mut reply_buf).await.unwrap(); + let reply: serde_json::Value = serde_json::from_slice(&reply_buf).unwrap(); + assert_eq!( + reply, + serde_json::json!({ + "jsonrpc": "2.0", + "error": {"code": -32700, "message": "Parse error"}, + }), + ); + } } From 8e7773d87ebf0eaeb6e006c759bec74facc33fff Mon Sep 17 00:00:00 2001 From: tsouth89 Date: Wed, 1 Jul 2026 15:40:04 -0400 Subject: [PATCH 4/4] fix: respond with Invalid Request for malformed protocol messages --- crates/rmcp/src/transport/async_rw.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/crates/rmcp/src/transport/async_rw.rs b/crates/rmcp/src/transport/async_rw.rs index 62ab86b47..bb5418350 100644 --- a/crates/rmcp/src/transport/async_rw.rs +++ b/crates/rmcp/src/transport/async_rw.rs @@ -153,14 +153,14 @@ where tracing::debug!("Ignoring unparsable incoming message: {e}"); } serde_json::error::Category::Data | serde_json::error::Category::Io => { - // Valid JSON that doesn't match the expected message shape is a real - // protocol error rather than unparsable input, so surface it with a - // response instead of silently dropping it. + // Well-formed JSON that doesn't match the expected message shape is a + // real protocol error rather than unparsable input, so surface it with + // an Invalid Request response instead of silently dropping it. tracing::debug!("Protocol error on incoming message: {e}"); let mut write = self.write.lock().await; let framed = write.as_mut()?; let response = TxJsonRpcMessage::::error( - ErrorData::parse_error("Parse error", None), + ErrorData::invalid_request("Invalid request", None), None, ); if framed.send(response).await.is_err() { @@ -715,7 +715,7 @@ mod test { reply, serde_json::json!({ "jsonrpc": "2.0", - "error": {"code": -32700, "message": "Parse error"}, + "error": {"code": -32600, "message": "Invalid request"}, }), ); }