Update rmcp to 3.1.2 (#39101)

## What changed

- Upgrade `rmcp` from 3.0.0 to 3.1.2.
- Use `rmcp`'s native JSON-RPC decoding and remove the local compatibility layer for multi-round-trip tool results.
- Preserve response metadata on `input_required` SSE results.
- Support OAuth protected-resource metadata discovery and include the discovered resource audience when refreshing tokens.

## Testing

- Add coverage for metadata preservation on modern SSE `input_required` responses.
- Add an OAuth startup scenario that discovers protected-resource metadata and verifies the refresh-token resource audience.

GitOrigin-RevId: 689f009b4a33552b640304111b74585b4009eeb8
This commit is contained in:
thomas
2026-08-17 23:04:09 +00:00
committed by copyberry
parent 050aa077b5
commit 7500ab4c8d
11 changed files with 182 additions and 262 deletions
+6 -2
View File
File diff suppressed because one or more lines are too long
+51 -6
View File
@@ -5377,6 +5377,16 @@ dependencies = [
"darling_macro 0.23.0",
]
[[package]]
name = "darling"
version = "0.24.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "88490bf1b990d87eaaa7ac8aa887f629a08e7359765b4911faf63c3763347d23"
dependencies = [
"darling_core 0.24.0",
"darling_macro 0.24.0",
]
[[package]]
name = "darling_core"
version = "0.20.11"
@@ -5418,6 +5428,19 @@ dependencies = [
"syn 2.0.117",
]
[[package]]
name = "darling_core"
version = "0.24.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "084e274f91c482280130e1e34e0b8d6e66776a060d7b6de7b84289ca778868c4"
dependencies = [
"ident_case",
"proc-macro2",
"quote",
"strsim 0.11.1",
"syn 3.0.3",
]
[[package]]
name = "darling_macro"
version = "0.20.11"
@@ -5451,6 +5474,17 @@ dependencies = [
"syn 2.0.117",
]
[[package]]
name = "darling_macro"
version = "0.24.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "68f5792fa0d41cd2325ce0ffa64f0a340eaebd4971a3a0c5e1ffd2cc488a355e"
dependencies = [
"darling_core 0.24.0",
"quote",
"syn 3.0.3",
]
[[package]]
name = "dashmap"
version = "6.1.0"
@@ -11810,9 +11844,9 @@ dependencies = [
[[package]]
name = "rmcp"
version = "3.0.0"
version = "3.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fcd2b6dd3b18129368955f32661a7718969e8c152c7d8866434c09cf15a512e0"
checksum = "c8dddc5b1924b9a59fba420166160ca2c4663a4e01803e52eda33070f56d63c8"
dependencies = [
"async-trait",
"base64 0.23.0",
@@ -11845,15 +11879,15 @@ dependencies = [
[[package]]
name = "rmcp-macros"
version = "3.0.0"
version = "3.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e1aa4b9345795260a43fc23d6d05e096407c8b953903f673af7c4404b49fb2d6"
checksum = "6898e24cd16342b59bfa8a53c2c04b9cf62fc8a2cfea57b9c038b09984bfc521"
dependencies = [
"darling 0.23.0",
"darling 0.24.0",
"proc-macro2",
"quote",
"serde_json",
"syn 2.0.117",
"syn 3.0.3",
]
[[package]]
@@ -13514,6 +13548,17 @@ dependencies = [
"unicode-ident",
]
[[package]]
name = "syn"
version = "3.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3"
dependencies = [
"proc-macro2",
"quote",
"unicode-ident",
]
[[package]]
name = "sync_wrapper"
version = "1.0.2"
+1 -1
View File
@@ -396,7 +396,7 @@ rcgen = { version = "0.14.7", default-features = false, features = [
regex = "1.12.3"
regex-lite = "0.1.8"
reqwest = { version = "0.12", features = ["cookies"] }
rmcp = { version = "=3.0.0", default-features = false }
rmcp = { version = "=3.1.2", default-features = false }
runfiles = { git = "https://github.com/dzbarsky/rules_rust", rev = "b56cbaa8465e74127f1ea216f813cd377295ad81" }
rustix = { version = "1.1.4", features = ["net"] }
rustls = { version = "0.23", default-features = false, features = [
@@ -45,8 +45,6 @@ use tracing::debug;
use tracing::info;
use tracing::warn;
use crate::incoming_jsonrpc::deserialize_incoming_jsonrpc_message;
static PROCESS_COUNTER: AtomicUsize = AtomicUsize::new(1);
// Tool results can make valid MCP responses large, so keep the protocol
// ceiling well above ordinary messages while still bounding hostile input.
@@ -452,7 +450,7 @@ impl ExecutorProcessTransport {
None => return None,
};
let line = Self::trim_trailing_carriage_return(line);
match deserialize_incoming_jsonrpc_message(&line) {
match serde_json::from_slice(&line) {
Ok(message) => return Some(message),
Err(error) => {
debug!(
@@ -57,8 +57,6 @@ use tokio::sync::oneshot;
use crate::event_notification_transport::MAX_EVENT_NOTIFICATION_BYTES;
use crate::http_client_redirect::SameOriginRedirectHttpClient;
use crate::incoming_jsonrpc::deserialize_incoming_jsonrpc_message;
use crate::incoming_jsonrpc::normalize_sse_jsonrpc_message;
use crate::local_stdio_transport::MAX_MCP_STDIO_LINE_BYTES;
mod www_authenticate;
@@ -357,11 +355,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
}
match content_type.as_deref() {
Some(content_type) if content_type.starts_with(EVENT_STREAM_MIME_TYPE) => {
let mut event_stream = sse_stream_from_body(
body_stream,
maximum_response_bytes,
is_discovery_request || uses_modern_protocol,
);
let mut event_stream = sse_stream_from_body(body_stream, maximum_response_bytes);
if mcp_method.as_deref() == Some(DiscoverRequestMethod::VALUE) {
while let Some(event) = event_stream.next().await {
let event = event.map_err(StreamableHttpError::Sse)?;
@@ -375,7 +369,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
continue;
}
let response = deserialize_incoming_jsonrpc_message(data.as_bytes())
let response = serde_json::from_slice(data.as_bytes())
.map_err(StreamableHttpError::Deserialize)?;
let response = legacy_discovery_fallback_response(
&message, response, /*allow_uncorrelated_http_rejection*/ false,
@@ -425,8 +419,8 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
}
Some(content_type) if content_type.starts_with(JSON_MIME_TYPE) => {
let body = collect_body(&mut body_stream, maximum_response_bytes).await?;
let response_message = deserialize_incoming_jsonrpc_message(&body)
.map_err(StreamableHttpError::Deserialize)?;
let response_message =
serde_json::from_slice(&body).map_err(StreamableHttpError::Deserialize)?;
Ok(StreamableHttpPostResponse::Json(
legacy_discovery_fallback_response(
&message,
@@ -593,11 +587,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
.and_then(|value| value.to_str().ok())
.is_some_and(|version| version == ProtocolVersion::V_2026_07_28.as_str());
let maximum_response_bytes = uses_modern_protocol.then_some(MAX_MCP_STDIO_LINE_BYTES);
Ok(sse_stream_from_body(
body_stream,
maximum_response_bytes,
uses_modern_protocol,
))
Ok(sse_stream_from_body(body_stream, maximum_response_bytes))
}
}
@@ -948,7 +938,6 @@ async fn collect_body(
fn sse_stream_from_body(
body_stream: HttpResponseBodyStream,
maximum_event_bytes: Option<usize>,
modern_session: bool,
) -> BoxStream<'static, std::result::Result<Sse, sse_stream::Error>> {
SseStream::from_bytes_stream(stream::unfold(
(body_stream, SseEventSizeLimit::new(maximum_event_bytes)),
@@ -966,16 +955,6 @@ fn sse_stream_from_body(
}
},
))
.map(move |event| {
event.map(|mut event| {
if let Some(payload) = event.data.as_deref()
&& let Some(normalized) = normalize_sse_jsonrpc_message(payload, modern_session)
{
event.data = Some(normalized);
}
event
})
})
.boxed()
}
@@ -1,58 +0,0 @@
//! Compatibility decoding for multi-round-trip tool results that rmcp's
//! untagged result union does not yet represent correctly.
use rmcp::model::InputRequiredResult;
use rmcp::model::JsonRpcResponse;
use rmcp::model::ServerJsonRpcMessage;
use rmcp::model::ServerResult;
use serde_json::Value;
/// Decode at the transport boundary, before rmcp's untagged `ServerResult`
/// deserializer can mistake a modern result for a completed tool call.
pub(crate) fn deserialize_incoming_jsonrpc_message(
bytes: &[u8],
) -> serde_json::Result<ServerJsonRpcMessage> {
let message: Value = serde_json::from_slice(bytes)?;
if message
.pointer("/result/resultType")
.and_then(Value::as_str)
== Some("input_required")
{
let response: JsonRpcResponse<InputRequiredResult> = serde_json::from_value(message)?;
return Ok(ServerJsonRpcMessage::response(
ServerResult::InputRequiredResult(response.result),
response.id,
));
}
serde_json::from_value(message)
}
/// Normalize an SSE payload before rmcp parses it internally. SSE does not
/// expose a typed-message hook, so response-level MRTR metadata cannot be
/// retained until the upstream untagged-result decoder is fixed.
pub(crate) fn normalize_sse_jsonrpc_message(payload: &str, modern_session: bool) -> Option<String> {
if !modern_session {
return None;
}
let mut message: Value = serde_json::from_str(payload).ok()?;
if message
.pointer("/result/resultType")
.and_then(Value::as_str)
!= Some("input_required")
{
return None;
}
message
.get_mut("result")
.and_then(Value::as_object_mut)?
.remove("_meta")?;
serde_json::to_string(&message).ok()
}
#[cfg(test)]
#[path = "incoming_jsonrpc_tests.rs"]
mod tests;
@@ -1,151 +0,0 @@
use pretty_assertions::assert_eq;
use rmcp::model::JsonRpcMessage;
use rmcp::model::ServerResult;
use serde_json::json;
use super::deserialize_incoming_jsonrpc_message;
use super::normalize_sse_jsonrpc_message;
#[test]
fn discovery_accepts_metadata_namespaced_server_identity() {
let message = json!({
"jsonrpc": "2.0",
"id": 1,
"result": {
"resultType": "complete",
"supportedVersions": ["2026-07-28"],
"capabilities": {"tools": {}},
"ttlMs": 0,
"cacheScope": "private",
"_meta": {
"io.modelcontextprotocol/serverInfo": {
"name": "modern-server",
"version": "1.0.0",
},
"retained": "metadata",
},
},
});
let decoded = deserialize_incoming_jsonrpc_message(&serde_json::to_vec(&message).unwrap())
.expect("metadata-only identity must decode as discovery");
let JsonRpcMessage::Response(response) = decoded else {
panic!("expected a discovery response");
};
let ServerResult::DiscoverResult(result) = response.result else {
panic!("metadata-only identity must not become a completed tool result");
};
assert_eq!(
result.server_info().map(|server_info| server_info.name),
Some("modern-server".to_owned())
);
assert_eq!(
result.meta.and_then(|meta| meta.get("retained").cloned()),
Some(json!("metadata"))
);
}
#[test]
fn extension_result_with_discovery_like_fields_remains_a_custom_result() {
let result = json!({
"supportedVersions": ["custom-extension-v1"],
"capabilities": {"custom": true},
});
let message = json!({
"jsonrpc": "2.0",
"id": 2,
"result": result,
});
let decoded = deserialize_incoming_jsonrpc_message(&serde_json::to_vec(&message).unwrap())
.expect("legacy extension result must deserialize normally");
let JsonRpcMessage::Response(response) = decoded else {
panic!("expected a legacy extension response");
};
let ServerResult::CustomResult(custom) = response.result else {
panic!("legacy extension response must remain a custom result");
};
assert_eq!(custom.0, result);
}
#[test]
fn input_required_discriminator_wins_over_tool_result_metadata() {
let message = json!({
"jsonrpc": "2.0",
"id": "tool-call",
"result": {
"resultType": "input_required",
"requestState": "opaque-server-state",
"_meta": {"retained": "response-metadata"},
},
});
let decoded = deserialize_incoming_jsonrpc_message(&serde_json::to_vec(&message).unwrap())
.expect("input-required result must decode");
let JsonRpcMessage::Response(response) = decoded else {
panic!("expected an input-required response");
};
let ServerResult::InputRequiredResult(result) = response.result else {
panic!("input_required must not become an empty completed tool result");
};
assert_eq!(result.request_state.as_deref(), Some("opaque-server-state"));
assert_eq!(
result.meta.and_then(|meta| meta.get("retained").cloned()),
Some(json!("response-metadata"))
);
}
#[test]
fn sse_discovery_preserves_metadata_namespaced_identity_without_normalization() {
let message = json!({
"jsonrpc": "2.0",
"id": 1,
"result": {
"resultType": "complete",
"supportedVersions": ["2026-07-28"],
"capabilities": {},
"ttlMs": 0,
"cacheScope": "private",
"_meta": {
"io.modelcontextprotocol/serverInfo": {
"name": "sse-server",
"version": "1.0.0",
},
},
},
});
assert_eq!(
normalize_sse_jsonrpc_message(&message.to_string(), /*modern_session*/ true),
None
);
let decoded = deserialize_incoming_jsonrpc_message(&serde_json::to_vec(&message).unwrap())
.expect("rmcp must natively decode metadata-only SSE discovery");
let JsonRpcMessage::Response(response) = decoded else {
panic!("expected a discovery response");
};
let ServerResult::DiscoverResult(result) = response.result else {
panic!("metadata-only identity must decode as discovery without normalization");
};
assert_eq!(
result.server_info().map(|server_info| server_info.name),
Some("sse-server".to_owned())
);
}
#[test]
fn legacy_sse_payload_preserves_intermediate_response_metadata() {
let message = json!({
"jsonrpc": "2.0",
"id": "legacy-extension",
"result": {
"resultType": "input_required",
"_meta": {"preserved": "legacy-extension-metadata"},
},
});
assert_eq!(
normalize_sse_jsonrpc_message(&message.to_string(), /*modern_session*/ false),
None
);
}
-1
View File
@@ -6,7 +6,6 @@ mod http_client_adapter;
mod http_client_redirect;
mod http_headers;
mod in_process_transport;
mod incoming_jsonrpc;
mod local_stdio_transport;
mod logging_client_handler;
mod oauth;
@@ -23,8 +23,6 @@ use tokio::sync::Mutex;
use tracing::debug;
use tracing::warn;
use crate::incoming_jsonrpc::deserialize_incoming_jsonrpc_message;
/// Match the existing executor stdio transport and the production MCP limit.
pub(crate) const MAX_MCP_STDIO_LINE_BYTES: usize = 8 * 1024 * 1024;
@@ -118,7 +116,7 @@ impl LocalStdioTransport {
fn decode_pending_message(&mut self) -> Option<RxJsonRpcMessage<RoleClient>> {
let line = std::mem::take(&mut self.pending_line);
let line = line.strip_suffix(b"\r").unwrap_or(&line);
match deserialize_incoming_jsonrpc_message(line) {
match serde_json::from_slice(line) {
Ok(message) => Some(message),
Err(error) => {
debug!(
@@ -20,6 +20,7 @@ use rmcp::model::Implementation;
use rmcp::model::InitializeRequestParams;
use rmcp::model::ProtocolVersion;
use rmcp::model::ReadResourceRequestParams;
use rmcp::model::ServerResult;
use rmcp::model::UrlElicitationCapability;
use serde_json::Value;
use serde_json::json;
@@ -170,6 +171,50 @@ async fn create_client(
Ok(client)
}
#[tokio::test]
async fn modern_sse_input_required_preserves_response_metadata() -> anyhow::Result<()> {
let server = MockServer::start().await;
let expected_result = json!({
"resultType": "input_required",
"requestState": OPAQUE_STATE,
"_meta": {"responseContext": "preserved"},
});
let result = expected_result.clone();
Mock::given(method("POST"))
.and(path("/mcp"))
.respond_with(move |request: &Request| {
let body: Value = request.body_json().expect("valid JSON-RPC request");
match body["method"].as_str() {
Some("server/discover") => discover_response(&body),
Some("tools/call") => sse_result_response(&body, result.clone()),
other => panic!("unexpected MRTR request: {other:?}"),
}
})
.expect(2)
.mount(&server)
.await;
let client = create_client(&server, Arc::new(Mutex::new(Vec::new()))).await?;
let result = client
.send_custom_request_with_timeout(
"tools/call",
Some(json!({"name": "confirm", "arguments": {}})),
Some(Duration::from_secs(5)),
)
.await?;
let ServerResult::InputRequiredResult(result) = result else {
anyhow::bail!("SSE response must remain an input_required result");
};
assert_eq!(serde_json::to_value(result)?, expected_result);
client.shutdown().await;
server.verify().await;
Ok(())
}
#[tokio::test]
async fn modern_tool_mrtr_drives_form_and_url_elicitation_and_preserves_metadata()
-> anyhow::Result<()> {
@@ -55,28 +55,82 @@ const UNREFRESHABLE_SERVER_URL: &str = "https://unrefreshable.example/mcp";
const UNEXPIRED_SERVER_URL: &str = "https://unexpired.example/mcp";
const REFRESHABLE_SERVER_URL: &str = "https://refreshable.example/mcp";
#[derive(Clone, Copy)]
enum OAuthStartupScenario {
DirectAuthorizationMetadata,
GatewayHeadersHelper,
ProtectedResourceMetadata,
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn refreshes_expired_persisted_token_before_initialize() -> anyhow::Result<()> {
assert_expired_token_refresh(/*with_headers_helper*/ false).await
assert_expired_token_refresh(OAuthStartupScenario::DirectAuthorizationMetadata).await
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn refreshes_oauth_with_gateway_headers_helper() -> anyhow::Result<()> {
assert_expired_token_refresh(/*with_headers_helper*/ true).await
assert_expired_token_refresh(OAuthStartupScenario::GatewayHeadersHelper).await
}
async fn assert_expired_token_refresh(with_headers_helper: bool) -> anyhow::Result<()> {
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn refresh_uses_discovered_protected_resource_audience() -> anyhow::Result<()> {
assert_expired_token_refresh(OAuthStartupScenario::ProtectedResourceMetadata).await
}
async fn assert_expired_token_refresh(scenario: OAuthStartupScenario) -> anyhow::Result<()> {
let server = MockServer::start().await;
let authorization_server = MockServer::start().await;
let resource_url = format!("{}/mcp", server.uri());
let (server_url, mcp_path, authorization_metadata_path) = match scenario {
OAuthStartupScenario::DirectAuthorizationMetadata
| OAuthStartupScenario::GatewayHeadersHelper => (
resource_url.clone(),
"/mcp",
"/.well-known/oauth-authorization-server/mcp",
),
OAuthStartupScenario::ProtectedResourceMetadata => (
format!("{resource_url}/"),
"/mcp/",
"/.well-known/oauth-authorization-server",
),
};
if matches!(scenario, OAuthStartupScenario::ProtectedResourceMetadata) {
let resource_metadata_url = format!("{}/resource-metadata", server.uri());
Mock::given(method("GET"))
.and(path(mcp_path))
.respond_with(ResponseTemplate::new(401).insert_header(
"www-authenticate",
format!("Bearer resource_metadata=\"{resource_metadata_url}\""),
))
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/resource-metadata"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"resource": resource_url,
"authorization_servers": [server.uri()],
})))
.expect(1)
.mount(&server)
.await;
}
let mut authorization_metadata = json!({
"authorization_endpoint": format!("{}/oauth/authorize", authorization_server.uri()),
"token_endpoint": format!("{}/oauth/token", authorization_server.uri()),
"scopes_supported": [""],
});
if matches!(scenario, OAuthStartupScenario::ProtectedResourceMetadata) {
authorization_metadata["issuer"] = json!(server.uri());
}
Mock::given(method("GET"))
.and(path("/.well-known/oauth-authorization-server/mcp"))
.and(path(authorization_metadata_path))
.and(header("user-agent", RESOURCE_USER_AGENT))
.and(header("x-api-key", RESOURCE_API_KEY))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"authorization_endpoint": format!("{}/oauth/authorize", authorization_server.uri()),
"token_endpoint": format!("{}/oauth/token", authorization_server.uri()),
"scopes_supported": [""],
})))
.respond_with(ResponseTemplate::new(200).set_body_json(authorization_metadata))
.expect(1)
.mount(&server)
.await;
@@ -87,6 +141,13 @@ async fn assert_expired_token_refresh(with_headers_helper: bool) -> anyhow::Resu
.and(body_string_contains(format!(
"refresh_token={REFRESH_TOKEN}"
)))
.and({
let expected_resource = resource_url.clone();
move |request: &Request| {
url::form_urlencoded::parse(&request.body)
.any(|(name, value)| name == "resource" && value == expected_resource)
}
})
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"access_token": REFRESHED_ACCESS_TOKEN,
"token_type": "Bearer",
@@ -97,7 +158,7 @@ async fn assert_expired_token_refresh(with_headers_helper: bool) -> anyhow::Resu
.mount(&authorization_server)
.await;
Mock::given(method("POST"))
.and(path("/mcp"))
.and(path(mcp_path))
.and(header("user-agent", RESOURCE_USER_AGENT))
.and(header("x-api-key", RESOURCE_API_KEY))
.and(header(
@@ -137,7 +198,7 @@ async fn assert_expired_token_refresh(with_headers_helper: bool) -> anyhow::Resu
.await;
let codex_home = TempDir::new()?;
let server_url = format!("{}/mcp", server.uri());
let with_headers_helper = matches!(scenario, OAuthStartupScenario::GatewayHeadersHelper);
// Credential storage resolves CODEX_HOME from the process environment.
// Run the client half of the test in an ignored helper test so it can use