modelcontextprotocol/rust-sdk
Watch
1
0
Fork
You've already forked rust-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

fix: accept stringified numeric response IDs (#1021)

Browse source
This commit is contained in:
Dale Seo 2026-07-21 16:01:22 -04:00 • committed by GitHub
commit 8803d39034
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 214 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

16
crates/rmcp/src/model.rs
View file

@ -240,6 +240,22 @@ impl NumberOrString {
NumberOrString::String(s) => Value::String(s.to_string()),
}
}
pub(crate) fn numeric_string_value(&self) -> Option<i64> {
match self {
Self::String(id) => id.parse().ok(),
Self::Number(_) => None,
}
}
pub(crate) fn matches_response_id(&self, response_id: &Self) -> bool {
self == response_id
|| matches!(
self,
Self::Number(request_id)
if response_id.numeric_string_value() == Some(*request_id)
)
}
}
impl std::fmt::Display for NumberOrString {

19
crates/rmcp/src/service.rs
View file

@ -305,6 +305,17 @@ pub trait ProgressTokenProvider: Send + Sync + 'static {
pub type AtomicU32RequestIdProvider = AtomicU32Provider;
pub type AtomicU32ProgressTokenProvider = AtomicU32Provider;
pub(crate) fn remove_pending_request<T>(
pending_requests: &mut HashMap<RequestId, T>,
response_id: &RequestId,
) -> Option<T> {
pending_requests.remove(response_id).or_else(|| {
response_id
.numeric_string_value()
.and_then(|id| pending_requests.remove(&RequestId::Number(id)))
})
}
#[derive(Debug, Default)]
pub struct AtomicU32Provider {
id: AtomicU64,
@ -1481,7 +1492,9 @@ where
id,
..
})) => {
if let Some(responder) = local_responder_pool.remove(&id) {
if let Some(responder) =
remove_pending_request(&mut local_responder_pool, &id)
{
let response_result = responder.send(Ok(result));
if let Err(_error) = response_result {
tracing::warn!(%id, "Error sending response");
@ -1495,7 +1508,9 @@ where
tracing::debug!(?error, "received id-less peer error");
continue;
};
if let Some(responder) = local_responder_pool.remove(&id) {
if let Some(responder) =
remove_pending_request(&mut local_responder_pool, &id)
{
let service_error = if error.is_transport_closed() {
ServiceError::TransportClosed
} else {

4
crates/rmcp/src/service/client.rs
View file

@ -687,7 +687,7 @@ where
let (response, response_id) =
expect_response(transport, "initialize response", service, peer.clone()).await?;
if id != response_id {
if !id.matches_response_id(&response_id) {
return Err(ClientInitializeError::ConflictInitResponseId(
id,
response_id,
@ -753,7 +753,7 @@ where
match expect_response(transport, "discover response", service, peer.clone()).await {
Ok((ServerResult::DiscoverResult(result), response_id)) => {
if response_id != id {
if !id.matches_response_id(&response_id) {
return Err(ClientInitializeError::ConflictInitResponseId(
id,
response_id,

50
crates/rmcp/src/transport/streamable_http_client.rs
View file

@ -499,8 +499,14 @@ impl<C: StreamableHttpClient> StreamableHttpClientWorker<C> {
pending_stream_response_ids: &mut HashSet<RequestId>,
message: &ServerJsonRpcMessage,
) {
if let Some(id) = Self::server_response_id(message) {
pending_stream_response_ids.remove(id);
let Some(response_id) = Self::server_response_id(message) else {
return;
};
if pending_stream_response_ids.remove(response_id) {
return;
}
if let Some(id) = response_id.numeric_string_value() {
pending_stream_response_ids.remove(&RequestId::Number(id));
}
}
@ -1382,7 +1388,10 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
}
Event::ServerMessage(mut json_rpc_message) => {
if let Some(response_id) = Self::server_response_id(&json_rpc_message)
&& let Some(stream_ct) = request_stream_cancellations.remove(response_id)
&& let Some(stream_ct) = crate::service::remove_pending_request(
&mut request_stream_cancellations,
response_id,
)
{
stream_ct.cancel();
}
@ -1848,4 +1857,39 @@ mod tests {
vec!["legacy"]
);
}
#[cfg(feature = "transport-streamable-http-client-reqwest")]
#[test]
fn clear_stream_response_pending_accepts_stringified_numeric_id() {
let mut pending = HashSet::from([NumberOrString::Number(1)]);
let response = ServerJsonRpcMessage::response(
ServerResult::ListToolsResult(ListToolsResult::default()),
NumberOrString::String("1".into()),
);
StreamableHttpClientWorker::<reqwest::Client>::clear_stream_response_pending(
&mut pending,
&response,
);
assert!(pending.is_empty());
}
#[cfg(feature = "transport-streamable-http-client-reqwest")]
#[test]
fn clear_stream_response_pending_prefers_exact_string_id() {
let string_id = NumberOrString::String("1".into());
let mut pending = HashSet::from([NumberOrString::Number(1), string_id.clone()]);
let response = ServerJsonRpcMessage::response(
ServerResult::ListToolsResult(ListToolsResult::default()),
string_id,
);
StreamableHttpClientWorker::<reqwest::Client>::clear_stream_response_pending(
&mut pending,
&response,
);
assert_eq!(pending, HashSet::from([NumberOrString::Number(1)]));
}
}

93
crates/rmcp/tests/test_client_initialization.rs
View file

@ -9,11 +9,102 @@ use common::handlers::TestClientHandler;
use rmcp::{
ServiceExt,
model::{
ErrorCode, ErrorData, JsonRpcError, JsonRpcVersion2_0, RequestId, ServerJsonRpcMessage,
ClientJsonRpcMessage, ErrorCode, ErrorData, InitializeResult, JsonRpcError,
JsonRpcVersion2_0, RequestId, ServerCapabilities, ServerJsonRpcMessage, ServerResult,
},
transport::{IntoTransport, Transport},
};
fn stringify_numeric_id(id: RequestId) -> RequestId {
let RequestId::Number(id) = id else {
panic!("expected a numeric request ID");
};
RequestId::String(id.to_string().into())
}
#[tokio::test]
async fn client_initialization_accepts_stringified_numeric_response_id() {
let (server_transport, client_transport) = tokio::io::duplex(1024);
let mut server = IntoTransport::<rmcp::RoleServer, _, _>::into_transport(server_transport);
let server_task = tokio::spawn(async move {
let ClientJsonRpcMessage::Request(request) =
server.receive().await.expect("expected initialize request")
else {
panic!("expected initialize request");
};
server
.send(ServerJsonRpcMessage::response(
ServerResult::InitializeResult(
InitializeResult::new(ServerCapabilities::default()),
),
stringify_numeric_id(request.id),
))
.await
.expect("send initialize response");
assert!(matches!(
server.receive().await,
Some(ClientJsonRpcMessage::Notification(_))
));
});
let client = TestClientHandler::new(true, true)
.serve(client_transport)
.await
.expect("client should accept stringified initialize response ID");
client.cancel().await.expect("cancel client");
server_task.await.expect("server task");
}
#[tokio::test]
async fn client_correlates_stringified_numeric_response_id() {
let (server_transport, client_transport) = tokio::io::duplex(1024);
let mut server = IntoTransport::<rmcp::RoleServer, _, _>::into_transport(server_transport);
let server_task = tokio::spawn(async move {
let ClientJsonRpcMessage::Request(initialize) =
server.receive().await.expect("expected initialize request")
else {
panic!("expected initialize request");
};
server
.send(ServerJsonRpcMessage::response(
ServerResult::InitializeResult(
InitializeResult::new(ServerCapabilities::default()),
),
initialize.id,
))
.await
.expect("send initialize response");
assert!(matches!(
server.receive().await,
Some(ClientJsonRpcMessage::Notification(_))
));
let ClientJsonRpcMessage::Request(request) =
server.receive().await.expect("expected tools/list request")
else {
panic!("expected tools/list request");
};
server
.send(ServerJsonRpcMessage::response(
ServerResult::ListToolsResult(Default::default()),
stringify_numeric_id(request.id),
))
.await
.expect("send tools/list response");
});
let client = TestClientHandler::new(true, true)
.serve(client_transport)
.await
.expect("initialize client");
client
.list_tools(None)
.await
.expect("client should correlate stringified response ID");
client.cancel().await.expect("cancel client");
server_task.await.expect("server task");
}
#[tokio::test]
async fn test_client_init_handles_jsonrpc_error() {
let (server_transport, client_transport) = tokio::io::duplex(1024);

41
crates/rmcp/tests/test_client_lifecycle_modes.rs
View file

@ -4,7 +4,7 @@ use rmcp::{
ClientHandler, ClientLifecycleMode, ClientServiceExt, ServerHandler, ServiceExt,
model::{
ClientJsonRpcMessage, ClientRequest, DiscoverResult, ErrorCode, ErrorData, GetMeta,
Implementation, InitializeResult, ProtocolVersion, ServerCapabilities,
Implementation, InitializeResult, ProtocolVersion, RequestId, ServerCapabilities,
ServerJsonRpcMessage, ServerResult,
},
service::PeerRequestOptions,
@ -21,6 +21,45 @@ struct StatelessServer;
impl ServerHandler for StatelessServer {}
#[tokio::test]
async fn discover_startup_accepts_stringified_numeric_response_id() {
let (server_transport, client_transport) = tokio::io::duplex(4096);
let mut server = IntoTransport::<rmcp::RoleServer, _, _>::into_transport(server_transport);
let server_task = tokio::spawn(async move {
let ClientJsonRpcMessage::Request(request) =
server.receive().await.expect("expected discover request")
else {
panic!("expected discover request");
};
let RequestId::Number(response_id) = request.id else {
panic!("expected a numeric request ID");
};
server
.send(ServerJsonRpcMessage::response(
ServerResult::DiscoverResult(DiscoverResult::new(
vec![ProtocolVersion::V_2026_07_28],
ServerCapabilities::default(),
Implementation::new("discover-server", "1.0.0"),
)),
RequestId::String(response_id.to_string().into()),
))
.await
.expect("send discover response");
});
let client = DiscoverClient
.serve_with_lifecycle(
client_transport,
ClientLifecycleMode::Discover {
preferred_versions: vec![ProtocolVersion::V_2026_07_28],
},
)
.await
.expect("client should accept stringified discover response ID");
client.cancel().await.expect("cancel client");
server_task.await.expect("server task");
}
#[tokio::test]
async fn high_level_server_accepts_discover_startup_without_initialize() {
let (server_transport, client_transport) = tokio::io::duplex(4096);