diff --git a/crates/rmcp/src/model.rs b/crates/rmcp/src/model.rs index 6531e6ee..e6a39573 100644 --- a/crates/rmcp/src/model.rs +++ b/crates/rmcp/src/model.rs @@ -4419,6 +4419,61 @@ impl ServerResult { ServerResult::EmptyResult(EmptyResult {}) } + /// Serialize this result using the wire format required by `protocol_version`. + /// + /// MCP 2026-07-28 requires completed results to carry `resultType`, but + /// earlier protocol versions did not define that field. Preserve the + /// ordinary serializer for modern sessions and remove only the completed + /// result discriminator for legacy sessions. + pub fn to_value_for_protocol( + &self, + protocol_version: &ProtocolVersion, + ) -> Result { + let mut value = serde_json::to_value(self)?; + if !matches!(self, ServerResult::DiscoverResult(_)) + && protocol_version < &ProtocolVersion::V_2026_07_28 + && let Some(result) = value.as_object_mut() + && result + .get("resultType") + .and_then(Value::as_str) + .is_some_and(|result_type| result_type == ResultType::COMPLETE.as_str()) + { + result.remove("resultType"); + } + Ok(value) + } + + /// Adapt this result to the negotiated protocol before sending it. + /// + /// Legacy results that carry `resultType: "complete"` become a custom JSON + /// result without that discriminator. All other result variants, including + /// modern results and custom discriminators, retain their original type. + pub fn into_result_for_protocol( + self, + protocol_version: &ProtocolVersion, + ) -> Result { + if protocol_version >= &ProtocolVersion::V_2026_07_28 + || matches!(self, ServerResult::DiscoverResult(_)) + { + return Ok(self); + } + + let mut value = serde_json::to_value(&self)?; + let Some(result) = value.as_object_mut() else { + return Ok(self); + }; + if result + .get("resultType") + .and_then(Value::as_str) + .is_none_or(|result_type| result_type != ResultType::COMPLETE.as_str()) + { + return Ok(self); + } + + result.remove("resultType"); + Ok(ServerResult::CustomResult(CustomResult::new(value))) + } + /// Empty `tasks/update` / `tasks/cancel` acknowledgement carrying the /// SEP-2322 `resultType: "complete"` discriminator (SEP-2663). pub fn task_ack(_: ()) -> ServerResult { diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 217464ff..4340d8a5 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -136,6 +136,22 @@ pub trait ServiceRole: std::fmt::Debug + Send + Sync + 'static + Copy + Clone { fn peer_cancelled_params(_notification: &Self::PeerNot) -> Option<&CancelledNotificationParam> { None } + + /// Select the protocol version used when serializing a response. + #[doc(hidden)] + fn response_protocol_version(_context: &RequestContext) -> Option { + None + } + + /// Adapt a response to the negotiated protocol before it reaches the transport. + #[doc(hidden)] + fn prepare_response( + response: Self::Resp, + _protocol_version: Option<&ProtocolVersion>, + ) -> Result { + Ok(response) + } + /// Invalidate any response cache affected by an inbound peer notification. /// /// The serve loop calls this for every notification *before* subscription @@ -1486,12 +1502,16 @@ where meta, extensions, }; + let response_protocol_version = R::response_protocol_version(&context); let current_span = tracing::Span::current(); let handler_id = id.clone(); spawn_service_task(async move { let result = ORIGINATING_REQUEST .scope(handler_id, service.handle_request(request, context)) - .await; + .await + .and_then(|result| { + R::prepare_response(result, response_protocol_version.as_ref()) + }); let response = match result { Ok(result) => { tracing::debug!(%id, ?result, "response message"); diff --git a/crates/rmcp/src/service/server.rs b/crates/rmcp/src/service/server.rs index 2b84a943..6e06f674 100644 --- a/crates/rmcp/src/service/server.rs +++ b/crates/rmcp/src/service/server.rs @@ -50,6 +50,34 @@ impl ServiceRole for RoleServer { } } + fn response_protocol_version(context: &RequestContext) -> Option { + if !context.peer.request_metadata_required() + && let Some(peer_info) = context.peer.peer_info() + { + return Some(peer_info.protocol_version.clone()); + } + + context.protocol_version() + } + + fn prepare_response( + response: Self::Resp, + protocol_version: Option<&ProtocolVersion>, + ) -> Result { + let Some(protocol_version) = protocol_version else { + return Ok(response); + }; + + response + .into_result_for_protocol(protocol_version) + .map_err(|error| { + ErrorData::internal_error( + format!("failed to serialize result for negotiated protocol: {error}"), + None, + ) + }) + } + fn enforce_request_association( request: &Self::Req, peer_info: Option<&Self::PeerInfo>, @@ -554,7 +582,13 @@ where extensions: std::mem::take(request.extensions_mut()), peer: peer.clone(), }; - let response = match service.handle_request(request, context).await { + let response_protocol_version = context.protocol_version(); + let response = match service + .handle_request(request, context) + .await + .and_then(|result| { + RoleServer::prepare_response(result, response_protocol_version.as_ref()) + }) { Ok(result) => ServerJsonRpcMessage::response(result, id), Err(error) => ServerJsonRpcMessage::error(error, Some(id)), }; diff --git a/crates/rmcp/tests/test_cache_hints.rs b/crates/rmcp/tests/test_cache_hints.rs index 2b6aecd8..8b4b78bd 100644 --- a/crates/rmcp/tests/test_cache_hints.rs +++ b/crates/rmcp/tests/test_cache_hints.rs @@ -1,4 +1,8 @@ -use rmcp::model::{CacheScope, ListToolsResult, ReadResourceResult, ResourceContents}; +use rmcp::model::{ + CacheScope, CallToolResult, CustomResult, DiscoverResult, Implementation, ListToolsResult, + ProtocolVersion, ReadResourceResult, ResourceContents, ResultType, ServerCapabilities, + ServerResult, +}; use serde_json::json; #[test] @@ -77,3 +81,91 @@ fn cache_scope_round_trips() { CacheScope::Private ); } + +#[test] +fn legacy_protocol_omits_complete_result_discriminator() { + let results = [ + ServerResult::CallToolResult(CallToolResult::success(Vec::new())), + ServerResult::ListToolsResult(ListToolsResult::with_all_items(Vec::new())), + ServerResult::ReadResourceResult(ReadResourceResult::new(Vec::new())), + ]; + + for result in results { + let actual = result + .to_value_for_protocol(&ProtocolVersion::V_2025_06_18) + .expect("legacy result should serialize"); + assert!( + actual.get("resultType").is_none(), + "legacy result unexpectedly included resultType: {actual}" + ); + + let adapted = result + .into_result_for_protocol(&ProtocolVersion::V_2025_06_18) + .expect("legacy result should adapt"); + let wire_value = serde_json::to_value(adapted).expect("adapted result should serialize"); + assert!(wire_value.get("resultType").is_none()); + } +} + +#[test] +fn modern_protocol_preserves_complete_result_discriminator() { + let result = ServerResult::CallToolResult(CallToolResult::success(Vec::new())); + + let actual = result + .to_value_for_protocol(&ProtocolVersion::V_2026_07_28) + .expect("modern result should serialize"); + assert_eq!(actual["resultType"], json!("complete")); + + let adapted = result + .into_result_for_protocol(&ProtocolVersion::V_2026_07_28) + .expect("modern result should retain its type"); + assert!(matches!(adapted, ServerResult::CallToolResult(_))); +} + +#[test] +fn legacy_protocol_preserves_non_complete_custom_discriminators() { + let result = ServerResult::CustomResult(CustomResult::new(json!({ + "resultType": "custom-extension", + "value": true + }))); + + let actual = result + .to_value_for_protocol(&ProtocolVersion::V_2025_11_25) + .expect("custom result should serialize"); + assert_eq!(actual["resultType"], json!("custom-extension")); + + let adapted = result + .into_result_for_protocol(&ProtocolVersion::V_2025_11_25) + .expect("custom result should retain its type"); + assert!(matches!(adapted, ServerResult::CustomResult(_))); +} + +#[test] +fn legacy_results_without_discriminator_deserialize_as_complete() { + let result: CallToolResult = serde_json::from_value(json!({ + "content": [], + "isError": false + })) + .expect("legacy tool result should deserialize"); + + assert_eq!(result.result_type, ResultType::COMPLETE); +} + +#[test] +fn discovery_preserves_result_discriminator_for_legacy_version_candidates() { + let result = ServerResult::DiscoverResult(DiscoverResult::new( + vec![ProtocolVersion::V_2025_11_25], + ServerCapabilities::default(), + Implementation::new("discovery-server", "1.0.0"), + )); + + let actual = result + .to_value_for_protocol(&ProtocolVersion::V_2025_11_25) + .expect("discovery result should serialize"); + assert_eq!(actual["resultType"], "complete"); + + let adapted = result + .into_result_for_protocol(&ProtocolVersion::V_2025_11_25) + .expect("discovery result should preserve its type"); + assert!(matches!(adapted, ServerResult::DiscoverResult(_))); +} diff --git a/crates/rmcp/tests/test_mrtr_behavior.rs b/crates/rmcp/tests/test_mrtr_behavior.rs index 3cb5da31..24e02347 100644 --- a/crates/rmcp/tests/test_mrtr_behavior.rs +++ b/crates/rmcp/tests/test_mrtr_behavior.rs @@ -17,6 +17,7 @@ use rmcp::{ service::{RequestContext, RoleClient, RoleServer, ServiceError, serve_directly}, }; use serde_json::json; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; /// A `requestState` value with characters that must survive a byte-exact echo: /// dots (the codec delimiter), base64 punctuation, whitespace, and quotes. @@ -327,10 +328,99 @@ where .await } +async fn raw_tool_result( + protocol_version: ProtocolVersion, + request_protocol_version: Option, +) -> anyhow::Result { + tokio::task::LocalSet::new() + .run_until(async move { + let (server_transport, client_transport) = tokio::io::duplex(8192); + let server = serve_directly::( + MrtrServer::default(), + server_transport, + Some(client_info(protocol_version)), + ); + + let (reader, mut writer) = tokio::io::split(client_transport); + let mut request = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { "name": "noop", "arguments": {} } + }); + if let Some(request_protocol_version) = request_protocol_version { + request["params"]["_meta"] = json!({ + "io.modelcontextprotocol/protocolVersion": request_protocol_version, + "io.modelcontextprotocol/clientInfo": { + "name": "raw-client", + "version": "1.0.0" + }, + "io.modelcontextprotocol/clientCapabilities": {} + }); + } + writer.write_all(request.to_string().as_bytes()).await?; + writer.write_all(b"\n").await?; + + let mut response = String::new(); + BufReader::new(reader).read_line(&mut response).await?; + let response: serde_json::Value = serde_json::from_str(&response)?; + server.cancel().await?; + + Ok(response["result"].clone()) + }) + .await +} + // ============================================================================= // Tests // ============================================================================= +#[tokio::test(flavor = "current_thread")] +async fn legacy_protocol_omits_complete_result_type_on_the_wire() -> anyhow::Result<()> { + let result = raw_tool_result(ProtocolVersion::V_2025_06_18, None).await?; + + assert_eq!(result["content"][0]["text"], "noop"); + assert!(result.get("resultType").is_none()); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn modern_protocol_includes_complete_result_type_on_the_wire() -> anyhow::Result<()> { + let result = raw_tool_result(ProtocolVersion::V_2026_07_28, None).await?; + + assert_eq!(result["content"][0]["text"], "noop"); + assert_eq!(result["resultType"], "complete"); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn legacy_session_ignores_modern_per_request_version_for_result_wire_format() +-> anyhow::Result<()> { + let result = raw_tool_result( + ProtocolVersion::V_2025_06_18, + Some(ProtocolVersion::V_2026_07_28), + ) + .await?; + + assert_eq!(result["content"][0]["text"], "noop"); + assert!(result.get("resultType").is_none()); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn modern_session_ignores_legacy_per_request_version_for_result_wire_format() +-> anyhow::Result<()> { + let result = raw_tool_result( + ProtocolVersion::V_2026_07_28, + Some(ProtocolVersion::V_2025_06_18), + ) + .await?; + + assert_eq!(result["content"][0]["text"], "noop"); + assert_eq!(result["resultType"], "complete"); + Ok(()) +} + #[tokio::test(flavor = "current_thread")] async fn client_auto_fulfills_input_required_tool_call() -> anyhow::Result<()> { let server = MrtrServer::default(); diff --git a/crates/rmcp/tests/test_server_discover_client.rs b/crates/rmcp/tests/test_server_discover_client.rs index adbe577e..bb109638 100644 --- a/crates/rmcp/tests/test_server_discover_client.rs +++ b/crates/rmcp/tests/test_server_discover_client.rs @@ -1,7 +1,7 @@ #![cfg(all(feature = "client", not(feature = "local")))] use rmcp::{ - ClientHandler, ServerHandler, ServiceExt, + ClientHandler, ClientLifecycleMode, ClientServiceExt, ServerHandler, ServiceExt, model::{ ClientCapabilities, Implementation, ProtocolVersion, RequestMetaObject, ServerCapabilities, ServerInfo, @@ -75,3 +75,35 @@ async fn client_discover_helper_returns_typed_result() { ); client.cancel().await.expect("client should cancel"); } + +#[tokio::test] +async fn discover_startup_accepts_historical_preferred_protocol_version() { + let (server_transport, client_transport) = tokio::io::duplex(4096); + let server_task = tokio::spawn(async move { + DiscoveryServer + .serve(server_transport) + .await + .expect("server should accept discovery") + }); + + let client = DiscoveryClient + .serve_with_lifecycle( + client_transport, + ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2025_11_25], + }, + ) + .await + .expect("discovery should preserve its required result discriminator"); + + assert_eq!( + client + .peer_info() + .expect("discovery should store server information") + .protocol_version, + ProtocolVersion::V_2025_11_25 + ); + client.cancel().await.expect("client should cancel"); + let server = server_task.await.expect("server task should complete"); + server.cancel().await.expect("server should cancel"); +}