Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
174 changes: 131 additions & 43 deletions crates/rmcp/src/service/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,14 @@ pub enum ClientInitializeError {
#[error("conflict initialized response id: expected {0}, got {1}")]
ConflictInitResponseId(RequestId, RequestId),

#[error(
"uncorrelated error response: expected id {expected}, error response carried {received}"
)]
UncorrelatedErrorResponse {
expected: RequestId,
received: RequestId,
},

#[error("connection closed: {0}")]
ConnectionClosed(String),

Expand All @@ -74,6 +82,12 @@ pub enum ClientInitializeError {

#[error("Cancelled")]
Cancelled,

#[error("discover and legacy initialize both failed")]
LegacyFallbackFailed {
Comment thread
DaleSeo marked this conversation as resolved.
discover: Box<ClientInitializeError>,
fallback: Box<ClientInitializeError>,
},
}

impl ClientInitializeError {
Expand All @@ -96,8 +110,13 @@ impl ClientInitializeError {
pub fn auth_challenge(&self) -> Option<&str> {
use crate::transport::streamable_http_client::{AuthRequiredError, InsufficientScopeError};

let Self::TransportError { error, .. } = self else {
return None;
let error = match self {
Self::TransportError { error, .. } => error,
// A 401/403 in the fallback phase is still actionable.
Self::LegacyFallbackFailed { fallback, .. } => {
return fallback.auth_challenge();
}
_ => return None,
};
let mut source: Option<&(dyn std::error::Error + 'static)> = Some(error.error.as_ref());
while let Some(current) = source {
Expand All @@ -117,10 +136,11 @@ impl ClientInitializeError {
/// This covers both missing or expired local OAuth authorization and an HTTP
/// authorization challenge from the MCP server.
pub fn is_authorization_required(&self) -> bool {
matches!(
self,
Self::TransportError { error, .. } if error.is_authorization_required()
)
match self {
Self::TransportError { error, .. } => error.is_authorization_required(),
Self::LegacyFallbackFailed { fallback, .. } => fallback.is_authorization_required(),
_ => false,
}
}
}

Expand All @@ -138,27 +158,50 @@ where
.ok_or_else(|| ClientInitializeError::ConnectionClosed(context.to_string()))
}

/// Helper function to expect a response from the stream
/// Helper function to expect a response from the stream, correlated to
/// `expected_id`.
///
/// Both success and error responses are checked here: a mismatched id on a
/// success response is `ConflictInitResponseId`; on an error response (whose
/// `id` is optional per spec) it is `UncorrelatedErrorResponse`. The caller
/// never sees an uncorrelated response.
async fn expect_response<T, S>(
transport: &mut T,
context: &str,
service: &S,
peer: Peer<RoleClient>,
) -> Result<(ServerResult, RequestId), ClientInitializeError>
expected_id: &RequestId,
) -> Result<ServerResult, ClientInitializeError>
where
T: Transport<RoleClient>,
S: Service<RoleClient>,
{
loop {
let message = expect_next_message(transport, context).await?;
match message {
// Expected message to complete the initialization
ServerJsonRpcMessage::Response(JsonRpcResponse { id, result, .. }) => {
break Ok((result, id));
if !expected_id.matches_response_id(&id) {
return Err(ClientInitializeError::ConflictInitResponseId(
expected_id.clone(),
id,
));
}
return Ok(result);
}
// Handle JSON-RPC error responses
ServerJsonRpcMessage::Error(error) => {
break Err(ClientInitializeError::JsonRpcError(error.error));
return Err(match &error.id {
Some(id) if expected_id.matches_response_id(id) => {
ClientInitializeError::JsonRpcError(error.error)
}
Comment thread
DaleSeo marked this conversation as resolved.
// Spec: error id is optional; a server that cannot read
// the request id omits it. The error is still a response
// to our request, so it remains available to the caller.
None => ClientInitializeError::JsonRpcError(error.error),
Some(id) => ClientInitializeError::UncorrelatedErrorResponse {
expected: expected_id.clone(),
received: id.clone(),
},
});
}
// Server could send logging messages before handshake
ServerJsonRpcMessage::Notification(mut notification) => {
Expand Down Expand Up @@ -714,40 +757,50 @@ where
legacy_startup(&service, &mut transport, &id_provider, &peer, client_info).await?;
}
ClientLifecycleMode::Discover { preferred_versions } => {
discover_startup(
match discover_startup(
&service,
&mut transport,
&id_provider,
&peer,
&client_info,
preferred_versions,
)
.await?;
.await?
{
DiscoverOutcome::Modern => {}
// Discover mode does not fall back; a legacy server is an error.
DiscoverOutcome::Legacy(error) => return Err(*error),
}
}
ClientLifecycleMode::Auto {
preferred_versions,
legacy_version,
} => {
let discover_result = discover_startup(
match discover_startup(
&service,
&mut transport,
&id_provider,
&peer,
&client_info,
preferred_versions,
)
.await;
match discover_result {
Ok(()) => {}
Err(ClientInitializeError::JsonRpcError(error))
if error.code == crate::model::ErrorCode::METHOD_NOT_FOUND =>
{
.await
{
Ok(DiscoverOutcome::Modern) => {}
Ok(DiscoverOutcome::Legacy(discover_error)) => {
let mut legacy_info = client_info;
if let Some(version) = legacy_version {
legacy_info.protocol_version = version;
}
legacy_startup(&service, &mut transport, &id_provider, &peer, legacy_info)
.await?;
if let Err(fallback_error) =
legacy_startup(&service, &mut transport, &id_provider, &peer, legacy_info)
.await
{
return Err(ClientInitializeError::LegacyFallbackFailed {
discover: discover_error,
fallback: Box::new(fallback_error),
});
}
}
Err(error) => return Err(error),
}
Expand All @@ -756,6 +809,41 @@ where
Ok(serve_inner(service, transport, peer, peer_rx, ct))
}

/// Modern-era JSON-RPC error codes a server can return from `server/discover`
/// without being legacy. Version negotiation (`UNSUPPORTED_PROTOCOL_VERSION`)
/// is handled by `discover_startup`'s own retry loop and never reaches the
/// classification below.
///
/// `ErrorCode` is an open integer type, so this cannot be exhaustive: if a
/// future revision adds another modern-era rejection code, add it here.
fn is_modern_rejection_code(code: crate::model::ErrorCode) -> bool {
matches!(
code,
crate::model::ErrorCode::MISSING_REQUIRED_CLIENT_CAPABILITY
| crate::model::ErrorCode::HEADER_MISMATCH
)
}

/// The outcome of a `server/discover` probe, classified at the point where all
/// the context (request id, response correlation, transport state) is still
/// available.
///
/// `Legacy` is returned only when the probe produced a complete, correlated
/// JSON-RPC error whose code is not a modern-era rejection — i.e. the
/// transport is in a known-good state and the error identifies the peer as
/// legacy per the 2026-07-28 backward-compatibility guidance. Every other
/// failure (transport error, uncorrelated response, modern rejection, etc.)
/// becomes `Err` so the caller surfaces it instead of retrying.
enum DiscoverOutcome {
/// The server speaks the modern protocol; discovery succeeded.
Modern,
/// The server is legacy: discovery received a correlated, non-modern
/// JSON-RPC error. The transport is still usable for a legacy `initialize`
/// handshake. The original error is preserved so a failed fallback can
/// report both phases.
Legacy(Box<ClientInitializeError>),
}

async fn legacy_startup<S, T>(
service: &S,
transport: &mut T,
Expand Down Expand Up @@ -784,15 +872,8 @@ where
context: "send initialize request".into(),
})?;

let (response, response_id) =
expect_response(transport, "initialize response", service, peer.clone()).await?;

if !id.matches_response_id(&response_id) {
return Err(ClientInitializeError::ConflictInitResponseId(
id,
response_id,
));
}
let response =
expect_response(transport, "initialize response", service, peer.clone(), &id).await?;

let ServerResult::InitializeResult(initialize_result) = response else {
return Err(ClientInitializeError::ExpectedInitResult(Some(response)));
Expand All @@ -819,7 +900,7 @@ async fn discover_startup<S, T>(
peer: &Peer<RoleClient>,
client_info: &ClientInfo,
preferred_versions: Vec<ProtocolVersion>,
) -> Result<(), ClientInitializeError>
) -> Result<DiscoverOutcome, ClientInitializeError>
where
S: Service<RoleClient>,
T: Transport<RoleClient> + 'static,
Expand Down Expand Up @@ -851,14 +932,8 @@ where
ClientInitializeError::transport::<T>(error, "send discover request")
})?;

match expect_response(transport, "discover response", service, peer.clone()).await {
Ok((ServerResult::DiscoverResult(result), response_id)) => {
if !id.matches_response_id(&response_id) {
return Err(ClientInitializeError::ConflictInitResponseId(
id,
response_id,
));
}
match expect_response(transport, "discover response", service, peer.clone(), &id).await {
Ok(ServerResult::DiscoverResult(result)) => {
let Some(selected) =
select_protocol_version(&preferred_versions, &result.supported_versions)
else {
Expand All @@ -876,9 +951,9 @@ where
client_info: client_info.client_info.clone(),
client_capabilities: client_info.capabilities.clone(),
});
return Ok(());
return Ok(DiscoverOutcome::Modern);
}
Ok((response, _)) => {
Ok(response) => {
return Err(ClientInitializeError::ExpectedInitResult(Some(response)));
}
Err(ClientInitializeError::JsonRpcError(error))
Expand Down Expand Up @@ -912,6 +987,19 @@ where
};
candidate = next;
}
// A correlated JSON-RPC error that is not a modern-era rejection
// and not a version-negotiation signal: the server is legacy.
// The transport delivered a complete response, so a legacy
// `initialize` can follow on the same connection.
Err(error)
if matches!(
&error,
ClientInitializeError::JsonRpcError(data)
if !is_modern_rejection_code(data.code)
) =>
{
return Ok(DiscoverOutcome::Legacy(Box::new(error)));
}
Err(error) => return Err(error),
}
}
Expand Down
11 changes: 9 additions & 2 deletions crates/rmcp/tests/test_client_initialization.rs
Original file line number Diff line number Diff line change
Expand Up @@ -125,11 +125,18 @@ async fn test_client_init_handles_jsonrpc_error() {
});

tokio::spawn(async move {
let _init_request = server.receive().await;
let request = server.receive().await;
// Echo the request's own id back on the error so it correlates: an
// uncorrelated id would surface as `UncorrelatedErrorResponse`
// instead of the `JsonRpcError` this test exercises.
let request_id = request
.and_then(|message| message.into_request())
.map(|(_, id)| id)
.expect("client sent an initialize request");

let error_msg = ServerJsonRpcMessage::Error(JsonRpcError {
jsonrpc: JsonRpcVersion2_0,
id: Some(RequestId::Number(1)),
id: Some(request_id),
error: ErrorData {
code: ErrorCode(-32600),
message: Cow::Borrowed("Invalid Request"),
Expand Down
Loading