fix(transport): fall back after sessionless HTTP discover rejections (#1211)

* fix(transport): fall back after HTTP discover rejection

* fix(transport): limit legacy fallback to sessionless probes

* style: apply nightly rustfmt import ordering
This commit is contained in:
King Star 2026-08-26 00:16:49 +08:00 • committed by GitHub
commit 02decfb682
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 184 additions and 3 deletions

View file

@ -293,6 +293,11 @@ impl StreamableHttpClient for reqwest::Client {
), ),
} }
} }
if let Some(response) =
legacy_discover_response(&message, session_was_attached, status, &body)
{
return Ok(response);
}
return Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned( return Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned(
format!("HTTP {status}: {body}"), format!("HTTP {status}: {body}"),
))); )));

View file

@ -274,6 +274,11 @@ impl StreamableHttpClient for UnixSocketHttpClient {
.await .await
.map(|c| String::from_utf8_lossy(&c.to_bytes()).into_owned()) .map(|c| String::from_utf8_lossy(&c.to_bytes()).into_owned())
.unwrap_or_else(|_| "<failed to read response body>".to_owned()); .unwrap_or_else(|_| "<failed to read response body>".to_owned());
if let Some(response) =
legacy_discover_response(&message, session_was_attached, status, &body)
{
return Ok(response);
}
return Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned( return Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned(
format!("HTTP {status}: {body}"), format!("HTTP {status}: {body}"),
))); )));

View file

@ -10,7 +10,7 @@ use futures::{
future::BoxFuture, future::BoxFuture,
stream::{BoxStream, FuturesUnordered}, stream::{BoxStream, FuturesUnordered},
}; };
use http::{HeaderName, HeaderValue}; use http::{HeaderName, HeaderValue, StatusCode};
pub use sse_stream::Error as SseError; pub use sse_stream::Error as SseError;
use sse_stream::Sse; use sse_stream::Sse;
use thiserror::Error; use thiserror::Error;
@ -332,6 +332,43 @@ impl StreamableHttpPostResponse {
} }
} }
/// Convert a sessionless discovery rejection into a response the lifecycle
/// layer can classify as a legacy-server signal.
///
/// Some legacy streamable-HTTP servers reject `server/discover` in middleware
/// before it reaches JSON-RPC dispatch. Their response may be an empty or
/// plain-text 4xx, so there is no JSON-RPC error for the client to forward.
/// Keep authentication failures and server errors on their original paths.
pub(super) fn legacy_discover_response(
message: &ClientJsonRpcMessage,
session_was_attached: bool,
status: StatusCode,
body: &str,
) -> Option<StreamableHttpPostResponse> {
if session_was_attached
|| !status.is_client_error()
|| matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN)
{
return None;
}
let ClientJsonRpcMessage::Request(request) = message else {
return None;
};
if !matches!(request.request, ClientRequest::DiscoverRequest(_)) {
return None;
}
let error = ErrorData::invalid_request(
format!("server/discover rejected with HTTP {status}: {body}"),
None,
);
Some(StreamableHttpPostResponse::Json(
ServerJsonRpcMessage::error(error, Some(request.id.clone())),
None,
))
}
/// HTTP backend used by [`StreamableHttpClientTransport`]. /// HTTP backend used by [`StreamableHttpClientTransport`].
/// ///
/// Custom implementations that parse SSE responses must override /// Custom implementations that parse SSE responses must override

View file

@ -5,8 +5,15 @@
feature = "transport-streamable-http-server" feature = "transport-streamable-http-server"
))] ))]
use std::borrow::Cow; use std::{borrow::Cow, sync::Arc};
use axum::{
Router,
body::{Body, Bytes},
extract::State,
http::{Response, StatusCode},
routing::post,
};
use rmcp::{ use rmcp::{
ClientLifecycleMode, ClientServiceExt, ServerHandler, ClientLifecycleMode, ClientServiceExt, ServerHandler,
model::{ClientInfo, DiscoverResult, ErrorCode, ErrorData, ProtocolVersion}, model::{ClientInfo, DiscoverResult, ErrorCode, ErrorData, ProtocolVersion},
@ -19,6 +26,8 @@ use rmcp::{
}, },
}, },
}; };
use serde_json::json;
use tokio::sync::Mutex;
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
#[derive(Clone, Default)] #[derive(Clone, Default)]
@ -46,6 +55,54 @@ impl ServerHandler for LegacyHttpServer {
} }
} }
#[derive(Clone, Default)]
struct PlainTextLegacyHttpState {
methods: Arc<Mutex<Vec<String>>>,
}
async fn plain_text_legacy_http_handler(
State(state): State<PlainTextLegacyHttpState>,
body: Bytes,
) -> Response<Body> {
let request: serde_json::Value = serde_json::from_slice(&body).expect("valid JSON-RPC body");
let method = request["method"]
.as_str()
.expect("request method")
.to_owned();
state.methods.lock().await.push(method.clone());
if method == "server/discover" {
return Response::builder()
.status(StatusCode::UNPROCESSABLE_ENTITY)
.body(Body::from("Unexpected message, expect initialize request"))
.expect("build rejection response");
}
if method == "initialize" {
return Response::builder()
.status(StatusCode::OK)
.header("content-type", "application/json")
.body(Body::from(
json!({
"jsonrpc": "2.0",
"id": request["id"],
"result": {
"protocolVersion": "2025-11-25",
"capabilities": {},
"serverInfo": {"name": "legacy", "version": "1.0"}
}
})
.to_string(),
))
.expect("build initialize response");
}
Response::builder()
.status(StatusCode::ACCEPTED)
.body(Body::empty())
.expect("build notification response")
}
#[tokio::test] #[tokio::test]
async fn discover_http_client_bootstraps_headers_without_initialize() { async fn discover_http_client_bootstraps_headers_without_initialize() {
let ct = CancellationToken::new(); let ct = CancellationToken::new();
@ -135,3 +192,47 @@ async fn auto_http_client_falls_back_to_stateful_legacy_startup() {
ct.cancel(); ct.cancel();
server.await.expect("server task"); server.await.expect("server task");
} }
#[tokio::test]
async fn auto_http_client_falls_back_after_plain_text_4xx_rejection() {
let ct = CancellationToken::new();
let state = PlainTextLegacyHttpState::default();
let methods = state.methods.clone();
let router = Router::new()
.route("/mcp", post(plain_text_legacy_http_handler))
.with_state(state);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("listener should bind");
let address = listener.local_addr().expect("listener address");
let server = tokio::spawn({
let ct = ct.clone();
async move {
let _ = axum::serve(listener, router)
.with_graceful_shutdown(async move { ct.cancelled_owned().await })
.await;
}
});
let transport = StreamableHttpClientTransport::from_config(
StreamableHttpClientTransportConfig::with_uri(format!("http://{address}/mcp")),
);
let client = ClientInfo::default()
.serve_with_lifecycle(
transport,
ClientLifecycleMode::Auto {
preferred_versions: vec![ProtocolVersion::V_2026_07_28],
legacy_version: Some(ProtocolVersion::V_2025_11_25),
},
)
.await
.expect("auto HTTP client should fall back after a transport-level rejection");
client.cancel().await.expect("cancel client");
assert_eq!(
methods.lock().await.as_slice(),
&["server/discover", "initialize", "notifications/initialized"]
);
ct.cancel();
server.await.expect("server task");
}

View file

@ -7,7 +7,10 @@
use std::{collections::HashMap, sync::Arc}; use std::{collections::HashMap, sync::Arc};
use rmcp::{ use rmcp::{
model::{ClientJsonRpcMessage, ClientRequest, PingRequest, RequestId}, model::{
ClientJsonRpcMessage, ClientRequest, DiscoverRequest, DiscoverRequestParams, PingRequest,
RequestId,
},
transport::streamable_http_client::{ transport::streamable_http_client::{
StreamableHttpClient, StreamableHttpError, StreamableHttpPostResponse, StreamableHttpClient, StreamableHttpError, StreamableHttpPostResponse,
}, },
@ -45,6 +48,13 @@ fn ping_message() -> ClientJsonRpcMessage {
) )
} }
fn discover_message() -> ClientJsonRpcMessage {
ClientJsonRpcMessage::request(
ClientRequest::DiscoverRequest(DiscoverRequest::new(DiscoverRequestParams {})),
RequestId::Number(1),
)
}
/// HTTP 4xx with Content-Type: application/json and a valid JSON-RPC error body /// HTTP 4xx with Content-Type: application/json and a valid JSON-RPC error body
/// must be surfaced as `StreamableHttpPostResponse::Json`, not swallowed as a /// must be surfaced as `StreamableHttpPostResponse::Json`, not swallowed as a
/// transport error. /// transport error.
@ -119,3 +129,26 @@ async fn http_4xx_malformed_json_body_falls_back_to_unexpected_server_response()
other => panic!("expected UnexpectedServerResponse, got: {other:?}"), other => panic!("expected UnexpectedServerResponse, got: {other:?}"),
} }
} }
/// A discovery request attached to an existing session must retain the
/// transport error path; legacy fallback only applies to sessionless probes.
#[tokio::test]
async fn sessionful_discover_4xx_does_not_trigger_legacy_fallback() {
let url = spawn_mock_server(422, "text/plain", "Session-bound rejection").await;
let client = reqwest::Client::new();
let result = client
.post_message(
Arc::from(url.as_str()),
discover_message(),
Some(Arc::from("session-1")),
None,
HashMap::new(),
)
.await;
match result {
Err(StreamableHttpError::UnexpectedServerResponse(_)) => {}
other => panic!("expected UnexpectedServerResponse, got: {other:?}"),
}
}