From 5a1ead3f104c74686e71dcfd55dea1915fba98f0 Mon Sep 17 00:00:00 2001 From: Dale Seo <5466341+DaleSeo@users.noreply.github.com> Date: Thu, 16 Jul 2026 20:48:21 -0400 Subject: [PATCH] feat: add subscription listen streams --- .github/workflows/conformance.yml | 73 +- README.md | 101 +-- conformance/src/bin/server.rs | 182 ++++- crates/rmcp/src/handler/client.rs | 19 + crates/rmcp/src/handler/server.rs | 117 ++- crates/rmcp/src/model.rs | 440 ++++++++++- crates/rmcp/src/model/meta.rs | 2 + crates/rmcp/src/service.rs | 184 ++++- crates/rmcp/src/service/client.rs | 409 ++++++++++- crates/rmcp/src/service/server.rs | 290 +++++++- .../src/transport/common/server_side_http.rs | 29 +- .../src/transport/streamable_http_client.rs | 121 ++- .../transport/streamable_http_server/tower.rs | 51 +- .../client_json_rpc_message_schema.json | 112 ++- ...lient_json_rpc_message_schema_current.json | 112 ++- .../server_json_rpc_message_schema.json | 90 +++ ...erver_json_rpc_message_schema_current.json | 90 +++ crates/rmcp/tests/test_mrtr_behavior.rs | 9 +- crates/rmcp/tests/test_notification.rs | 1 + crates/rmcp/tests/test_subscriptions.rs | 691 ++++++++++++++++++ crates/rmcp/tests/test_subscriptions_model.rs | 237 ++++++ .../test_subscriptions_streamable_http.rs | 310 ++++++++ examples/clients/Cargo.toml | 4 + examples/clients/README.md | 7 + .../clients/src/subscriptions_streamhttp.rs | 49 ++ examples/servers/Cargo.toml | 4 + examples/servers/README.md | 7 + .../servers/src/subscriptions_streamhttp.rs | 83 +++ 28 files changed, 3603 insertions(+), 221 deletions(-) create mode 100644 crates/rmcp/tests/test_subscriptions.rs create mode 100644 crates/rmcp/tests/test_subscriptions_model.rs create mode 100644 crates/rmcp/tests/test_subscriptions_streamable_http.rs create mode 100644 examples/clients/src/subscriptions_streamhttp.rs create mode 100644 examples/servers/src/subscriptions_streamhttp.rs diff --git a/.github/workflows/conformance.yml b/.github/workflows/conformance.yml index 817c7d4ed..d070ee81a 100644 --- a/.github/workflows/conformance.yml +++ b/.github/workflows/conformance.yml @@ -81,81 +81,10 @@ jobs: echo "draft conformance server did not become ready" >&2 exit 1 - # Run discovery separately until #985 enables the full draft suite. - - name: Run SEP-2575 discovery contract - run: | - endpoint=http://127.0.0.1:8002/mcp - common_headers=( - -H "Content-Type: application/json" - -H "Accept: application/json, text/event-stream" - -H "Mcp-Method: server/discover" - ) - - discover_response="$( - curl --fail-with-body --silent --show-error \ - "${common_headers[@]}" \ - -H "MCP-Protocol-Version: 2026-07-28" \ - --data '{ - "jsonrpc": "2.0", - "id": "discover", - "method": "server/discover", - "params": { - "_meta": { - "io.modelcontextprotocol/protocolVersion": "2026-07-28", - "io.modelcontextprotocol/clientInfo": { - "name": "conformance-workflow", - "version": "1.0.0" - }, - "io.modelcontextprotocol/clientCapabilities": {} - } - } - }' \ - "$endpoint" - )" - jq -e ' - .result.resultType == "complete" and - (.result.supportedVersions | index("2026-07-28") != null) and - (.result.capabilities | type == "object") and - (.result.serverInfo.name | type == "string") and - .result.ttlMs == 0 and - .result.cacheScope == "private" - ' <<<"$discover_response" - - status="$( - curl --silent --show-error \ - --output /tmp/unsupported-version.json \ - --write-out "%{http_code}" \ - "${common_headers[@]}" \ - -H "MCP-Protocol-Version: 2099-01-01" \ - --data '{ - "jsonrpc": "2.0", - "id": "unsupported", - "method": "server/discover", - "params": { - "_meta": { - "io.modelcontextprotocol/protocolVersion": "2099-01-01", - "io.modelcontextprotocol/clientInfo": { - "name": "conformance-workflow", - "version": "1.0.0" - }, - "io.modelcontextprotocol/clientCapabilities": {} - } - } - }' \ - "$endpoint" - )" - test "$status" = "400" - jq -e ' - .id == "unsupported" and - .error.code == -32022 and - .error.data.requested == "2099-01-01" and - (.error.data.supported | index("2026-07-28") != null) - ' /tmp/unsupported-version.json - - # Keep this explicit list until the full draft suite is enabled by #985. - name: Run supported draft server scenarios run: | for scenario in \ + server-stateless \ sep-2164-resource-not-found \ caching \ http-header-validation \ diff --git a/README.md b/README.md index 0f8367377..7586d5c58 100644 --- a/README.md +++ b/README.md @@ -882,89 +882,90 @@ context.peer.notify_resource_list_changed().await?; ## Subscriptions -Clients can subscribe to specific resources. When a subscribed resource changes, the server sends a notification and the client can re-read it. +Protocol `2026-07-28` replaces `resources/subscribe`, `resources/unsubscribe`, and +the standalone HTTP GET stream with the transport-neutral, long-lived +`subscriptions/listen` request. Each requested notification category is opt-in. -**MCP Spec:** [Resources - Subscriptions](https://modelcontextprotocol.io/specification/2025-11-25/server/resources#subscriptions) +**MCP Spec:** [Subscriptions](https://modelcontextprotocol.io/specification/draft/basic/patterns/subscriptions) ### Server-side -Enable subscriptions in the resources capability and implement the `subscribe()` / `unsubscribe()` handlers: +Declare the notification capabilities you serve, return the accepted subset, +and use the filter-enforcing subscription sink: ```rust -use rmcp::{ErrorData as McpError, ServerHandler, model::*, service::RequestContext, RoleServer}; -use std::sync::Arc; -use tokio::sync::Mutex; -use std::collections::HashSet; - -#[derive(Clone)] -struct MyServer { - subscriptions: Arc>>, -} +use rmcp::{ + ErrorData, ServerHandler, + model::*, + service::SubscriptionContext, +}; impl ServerHandler for MyServer { fn get_info(&self) -> ServerInfo { ServerInfo::new( ServerCapabilities::builder() - .enable_resources() - .enable_resources_subscribe() + .enable_tools() + .enable_tool_list_changed() .build(), ) } - async fn subscribe( + fn accepted_subscription_filter( &self, - request: SubscribeRequestParams, - _context: RequestContext, - ) -> Result<(), McpError> { - self.subscriptions.lock().await.insert(request.uri); - Ok(()) + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.clone()) } - async fn unsubscribe( - &self, - request: UnsubscribeRequestParams, - _context: RequestContext, - ) -> Result<(), McpError> { - self.subscriptions.lock().await.remove(&request.uri); + async fn listen(&self, context: SubscriptionContext) -> Result<(), ErrorData> { + if context.accepted().tools_list_changed == Some(true) { + context.sink().notify_tool_list_changed().await + .map_err(|error| ErrorData::internal_error(error.to_string(), None))?; + } + context.cancelled().await; Ok(()) } } ``` -When a subscribed resource changes, notify the client: - -```rust -// Check if the resource has subscribers, then notify -context.peer.notify_resource_updated( - ResourceUpdatedNotificationParam::new("file:///config.json"), -).await?; -``` +The SDK intersects the handler's filter with the requested categories and the +capabilities advertised by `get_info()`. It sends the acknowledgment before +`listen`, tags every sink notification with the listen request ID, and rejects +categories or resource URIs outside the accepted filter. ### Client-side ```rust use rmcp::model::*; -// Subscribe to updates for a resource -client.subscribe(SubscribeRequestParams::new("file:///config.json")).await?; +let mut subscription = client.listen( + SubscriptionFilter::builder() + .tools_list_changed() + .resource_subscription("file:///config.json") + .build(), +).await?; + +println!("accepted: {:?}", subscription.acknowledged()); +while let Some(notification) = subscription.next().await? { + println!("notification: {notification:?}"); +} -// Unsubscribe when no longer needed -client.unsubscribe(UnsubscribeRequestParams::new("file:///config.json")).await?; +subscription.cancel().await?; ``` -Handle update notifications in `ClientHandler`: +`listen()` buffers up to 64 notifications per subscription. Use +`listen_with_capacity()` to choose a different non-zero capacity; if a consumer +falls behind, `Subscription::end()` reports `SubscriptionEnd::Lagged`. -```rust -impl ClientHandler for MyClient { - async fn on_resource_updated( - &self, - params: ResourceUpdatedNotificationParam, - _context: NotificationContext, - ) { - // Re-read the resource at params.uri - } -} -``` +For older negotiated protocol versions, the deprecated `subscribe()` and +`unsubscribe()` APIs retain their legacy wire behavior. Modern Streamable HTTP +uses the listen POST response stream directly and does not use sessions, GET, +DELETE, or `Last-Event-ID`. After an abrupt transport close, call `listen` +again; subscription state is not resumed across HTTP or stdio reconnects. + +See the +[modern subscription server](examples/servers/src/subscriptions_streamhttp.rs) +and [client](examples/clients/src/subscriptions_streamhttp.rs) examples. --- diff --git a/conformance/src/bin/server.rs b/conformance/src/bin/server.rs index 4df4812d6..9da7a26da 100644 --- a/conformance/src/bin/server.rs +++ b/conformance/src/bin/server.rs @@ -1,10 +1,16 @@ #![allow(deprecated)] -use std::{collections::HashSet, sync::Arc}; +use std::{ + collections::{HashMap, HashSet}, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, +}; use rmcp::{ ErrorData, RoleServer, ServerHandler, model::*, - service::RequestContext, + service::{RequestContext, SubscriptionContext, SubscriptionSink}, transport::{ StreamableHttpServerConfig, StreamableHttpService, streamable_http_server::session::local::LocalSessionManager, @@ -48,7 +54,9 @@ const REQUEST_STATE_KEY: &[u8] = b"rust-sdk-conformance-request-state-key!!"; #[derive(Clone)] struct ConformanceServer { - subscriptions: Arc>>, + legacy_resource_subscriptions: Arc>>, + subscriptions: Arc>>, + next_subscription: Arc, log_level: Arc>, request_state_codec: RequestStateCodec, } @@ -56,7 +64,9 @@ struct ConformanceServer { impl ConformanceServer { fn new() -> Self { Self { - subscriptions: Arc::new(Mutex::new(HashSet::new())), + legacy_resource_subscriptions: Arc::new(Mutex::new(HashSet::new())), + subscriptions: Arc::new(Mutex::new(HashMap::new())), + next_subscription: Arc::new(AtomicU64::new(0)), log_level: Arc::new(Mutex::new(LoggingLevel::Debug)), request_state_codec: RequestStateCodec::new(REQUEST_STATE_KEY), } @@ -379,22 +389,51 @@ impl ServerHandler for ConformanceServer { (name == "test_custom_header").then(custom_header_tool) } - async fn initialize( - &self, - request: InitializeRequestParams, - _cx: RequestContext, - ) -> Result { - Ok(InitializeResult::new( + fn get_info(&self) -> ServerInfo { + ServerInfo::new( ServerCapabilities::builder() .enable_prompts() + .enable_prompts_list_changed() .enable_resources() + .enable_resources_subscribe() + .enable_resources_list_changed() .enable_tools() + .enable_tool_list_changed() .enable_logging() .build(), ) - .with_protocol_version(request.protocol_version) .with_server_info(Implementation::new("rust-conformance-server", "0.1.0")) - .with_instructions("Rust MCP conformance test server")) + .with_instructions("Rust MCP conformance test server") + } + + async fn initialize( + &self, + request: InitializeRequestParams, + _cx: RequestContext, + ) -> Result { + let info = self.get_info(); + Ok(InitializeResult::new(info.capabilities) + .with_protocol_version(request.protocol_version) + .with_server_info(info.server_info) + .with_instructions(info.instructions.unwrap_or_default())) + } + + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.supported_by(&self.get_info().capabilities)) + } + + async fn listen(&self, context: SubscriptionContext) -> Result<(), ErrorData> { + let key = self.next_subscription.fetch_add(1, Ordering::Relaxed); + self.subscriptions + .lock() + .await + .insert(key, context.sink().clone()); + context.cancelled().await; + self.subscriptions.lock().await.remove(&key); + Ok(()) } async fn ping(&self, _cx: RequestContext) -> Result<(), ErrorData> { @@ -540,6 +579,46 @@ impl ServerHandler for ConformanceServer { })), ), custom_header_tool(), + Tool::new( + "test_trigger_tool_change", + "Triggers a tools/list_changed notification on matching subscriptions", + json_object(json!({ + "type": "object", + "properties": {} + })), + ), + Tool::new( + "test_trigger_prompt_change", + "Triggers a prompts/list_changed notification on matching subscriptions", + json_object(json!({ + "type": "object", + "properties": {} + })), + ), + Tool::new( + "test_missing_capability", + "Requires the sampling client capability", + json_object(json!({ + "type": "object", + "properties": {} + })), + ), + Tool::new( + "test_streaming_elicitation", + "Returns an input_required result containing an elicitation request", + json_object(json!({ + "type": "object", + "properties": {} + })), + ), + Tool::new( + "test_logging_tool", + "Emits notifications/message only when logLevel is requested", + json_object(json!({ + "type": "object", + "properties": {} + })), + ), ]; // SEP-2322 MRTR test tools; all take no arguments. let mrtr_tools = [ @@ -922,6 +1001,81 @@ impl ServerHandler for ConformanceServer { )])) } + "test_trigger_tool_change" => { + let subscriptions = self + .subscriptions + .lock() + .await + .values() + .cloned() + .collect::>(); + for subscription in subscriptions { + let _ = subscription.notify_tool_list_changed().await; + } + Ok(CallToolResult::success(vec![ContentBlock::text( + "Tool list change triggered", + )])) + } + + "test_trigger_prompt_change" => { + let subscriptions = self + .subscriptions + .lock() + .await + .values() + .cloned() + .collect::>(); + for subscription in subscriptions { + let _ = subscription.notify_prompt_list_changed().await; + } + Ok(CallToolResult::success(vec![ContentBlock::text( + "Prompt list change triggered", + )])) + } + + "test_missing_capability" => { + let capabilities = cx.meta.client_capabilities().unwrap_or_default(); + if capabilities.sampling.is_none() { + return Err(ErrorData::missing_required_client_capability( + ClientCapabilities::builder().enable_sampling().build(), + )); + } + Ok(CallToolResult::success(vec![ContentBlock::text( + "Required capability declared", + )])) + } + + "test_streaming_elicitation" => { + let mut requests = InputRequests::new(); + requests.insert( + "streaming_elicitation".into(), + mrtr_elicitation_request( + "Provide a value", + json!({ "value": { "type": "string" } }), + json!(["value"]), + ), + ); + return Ok(InputRequiredResult::from_input_requests(requests).into()); + } + + "test_logging_tool" => { + if let Some(level) = cx.meta.log_level() { + let _ = cx + .peer + .notify_logging_message( + LoggingMessageNotificationParam::new( + level, + json!("logLevel was requested"), + ) + .with_logger("conformance-server"), + ) + .await; + } + Ok(CallToolResult::success(vec![ContentBlock::text( + "Logging tool completed", + )])) + } + _ => Err(ErrorData::invalid_params( format!("Unknown tool: {}", request.name), None, @@ -1029,7 +1183,7 @@ impl ServerHandler for ConformanceServer { request: SubscribeRequestParams, _cx: RequestContext, ) -> Result<(), ErrorData> { - let mut subs = self.subscriptions.lock().await; + let mut subs = self.legacy_resource_subscriptions.lock().await; subs.insert(request.uri.to_string()); Ok(()) } @@ -1039,7 +1193,7 @@ impl ServerHandler for ConformanceServer { request: UnsubscribeRequestParams, _cx: RequestContext, ) -> Result<(), ErrorData> { - let mut subs = self.subscriptions.lock().await; + let mut subs = self.legacy_resource_subscriptions.lock().await; subs.remove(request.uri.as_str()); Ok(()) } diff --git a/crates/rmcp/src/handler/client.rs b/crates/rmcp/src/handler/client.rs index c9097e241..99d099d65 100644 --- a/crates/rmcp/src/handler/client.rs +++ b/crates/rmcp/src/handler/client.rs @@ -66,6 +66,10 @@ impl Service for H { ServerNotification::PromptListChangedNotification(_notification_no_param) => { self.on_prompt_list_changed(context).await } + ServerNotification::SubscriptionsAcknowledgedNotification(notification) => { + self.on_subscriptions_acknowledged(notification.params, context) + .await + } ServerNotification::TaskStatusNotification(notification) => { self.on_task_status(notification.params, context).await } @@ -237,6 +241,13 @@ pub trait ClientHandler: Sized + Send + Sync + 'static { ) -> impl Future + MaybeSendFuture + '_ { std::future::ready(()) } + fn on_subscriptions_acknowledged( + &self, + params: SubscriptionsAcknowledgedNotificationParams, + context: NotificationContext, + ) -> impl Future + MaybeSendFuture + '_ { + std::future::ready(()) + } fn on_task_status( &self, @@ -365,6 +376,14 @@ macro_rules! impl_client_handler_for_wrapper { (**self).on_prompt_list_changed(context) } + fn on_subscriptions_acknowledged( + &self, + params: SubscriptionsAcknowledgedNotificationParams, + context: NotificationContext, + ) -> impl Future + MaybeSendFuture + '_ { + (**self).on_subscriptions_acknowledged(params, context) + } + fn on_task_status( &self, params: TaskStatusNotificationParam, diff --git a/crates/rmcp/src/handler/server.rs b/crates/rmcp/src/handler/server.rs index 689bcfc78..6526c12e4 100644 --- a/crates/rmcp/src/handler/server.rs +++ b/crates/rmcp/src/handler/server.rs @@ -7,7 +7,7 @@ use crate::{ model::{TaskSupport, *}, service::{ MaybeSendFuture, NotificationContext, RequestContext, RoleServer, Service, ServiceRole, - negotiate_protocol_version, + SubscriptionContext, negotiate_protocol_version, }, }; @@ -79,7 +79,11 @@ impl Service for H { .await .map(ServerResult::DiscoverResult), ClientRequest::PingRequest(_request) => { - self.ping(context).await.map(ServerResult::empty) + if mrtr_supported { + Err(McpError::method_not_found::()) + } else { + self.ping(context).await.map(ServerResult::empty) + } } ClientRequest::CompleteRequest(request) => self .complete(request.params, context) @@ -109,14 +113,60 @@ impl Service for H { .read_resource(request.params, context) .await .map(ServerResult::from), - ClientRequest::SubscribeRequest(request) => self - .subscribe(request.params, context) - .await - .map(ServerResult::empty), - ClientRequest::UnsubscribeRequest(request) => self - .unsubscribe(request.params, context) - .await - .map(ServerResult::empty), + ClientRequest::SubscriptionsListenRequest(request) => { + if !mrtr_supported { + Err(McpError::method_not_found::()) + } else { + let requested = request.params.notifications; + let Some(candidate) = self.accepted_subscription_filter(&requested) else { + return Err( + McpError::method_not_found::(), + ); + }; + let advertised = requested.supported_by(&self.get_info().capabilities); + let handler_accepted = requested.intersection(&candidate); + let accepted = handler_accepted.intersection(&advertised); + if accepted != handler_accepted { + tracing::debug!( + requested_resource_count = requested + .resource_subscriptions + .as_ref() + .map_or(0, Vec::len), + accepted_resource_count = + accepted.resource_subscriptions.as_ref().map_or(0, Vec::len), + "subscription filter reduced to advertised server capabilities" + ); + } + let subscription_id = context.id.clone(); + let subscription = + SubscriptionContext::establish(context, requested, accepted).await?; + // The integrated draft schema defines a final result for graceful + // server teardown; explicit stdio cancellation remains a notification. + self.listen(subscription).await.map(|()| { + ServerResult::SubscriptionsListenResult( + SubscriptionsListenResult::complete(subscription_id), + ) + }) + } + } + ClientRequest::SubscribeRequest(request) => { + if mrtr_supported { + Err(McpError::method_not_found::()) + } else { + self.subscribe(request.params, context) + .await + .map(ServerResult::empty) + } + } + ClientRequest::UnsubscribeRequest(request) => { + if mrtr_supported { + Err(McpError::method_not_found::()) + } else { + self.unsubscribe(request.params, context) + .await + .map(ServerResult::empty) + } + } ClientRequest::CallToolRequest(request) => { let is_task = request.params.task.is_some(); @@ -335,6 +385,36 @@ macro_rules! server_handler_methods { McpError::method_not_found::(), )) } + /// Return the subset of a requested notification filter this server accepts. + /// + /// Returning `None` leaves `subscriptions/listen` unimplemented. The SDK + /// intersects the returned filter with both `requested` and the notification + /// capabilities advertised by [`Self::get_info`] before acknowledging it. + /// Categories that were not requested or advertised are always removed. + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + None + } + /// Run one established subscription until it is cancelled or closed gracefully. + /// + /// The SDK sends the acknowledgment before invoking this method. Returning + /// `Ok(())` sends the final [`SubscriptionsListenResult`] defined by the + /// integrated draft schema, marking graceful server teardown. Explicit + /// stdio cancellation uses `notifications/cancelled` instead. + fn listen( + &self, + context: SubscriptionContext, + ) -> impl Future> + MaybeSendFuture + '_ { + async move { + context.cancelled().await; + Ok(()) + } + } + #[deprecated( + note = "resources/subscribe is legacy-only; implement accepted_subscription_filter and listen for protocol version 2026-07-28" + )] fn subscribe( &self, request: SubscribeRequestParams, @@ -342,6 +422,9 @@ macro_rules! server_handler_methods { ) -> impl Future> + MaybeSendFuture + '_ { std::future::ready(Err(McpError::method_not_found::())) } + #[deprecated( + note = "resources/unsubscribe is legacy-only; subscriptions/listen is cancelled through its request lifecycle" + )] fn unsubscribe( &self, request: UnsubscribeRequestParams, @@ -604,6 +687,20 @@ macro_rules! impl_server_handler_for_wrapper { (**self).read_resource(request, context) } + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + (**self).accepted_subscription_filter(requested) + } + + fn listen( + &self, + context: SubscriptionContext, + ) -> impl Future> + MaybeSendFuture + '_ { + (**self).listen(context) + } + fn subscribe( &self, request: SubscribeRequestParams, diff --git a/crates/rmcp/src/model.rs b/crates/rmcp/src/model.rs index 8da14fe53..97273d9ae 100644 --- a/crates/rmcp/src/model.rs +++ b/crates/rmcp/src/model.rs @@ -3,8 +3,10 @@ #![expect(deprecated)] use std::{ borrow::Cow, + collections::hash_map::RandomState, + hash::{BuildHasher, Hasher}, ops::{Deref, DerefMut}, - sync::Arc, + sync::{Arc, OnceLock}, }; mod annotated; mod capabilities; @@ -558,6 +560,8 @@ pub struct ErrorData { } impl ErrorData { + const TRANSPORT_CLOSED_MARKER: &str = "io.modelcontextprotocol/transportClosed"; + pub fn new( code: ErrorCode, message: impl Into>, @@ -616,6 +620,33 @@ impl ErrorData { pub fn internal_error(message: impl Into>, data: Option) -> Self { Self::new(ErrorCode::INTERNAL_ERROR, message, data) } + + #[cfg(feature = "transport-streamable-http-client")] + pub(crate) fn transport_closed(message: impl Into>) -> Self { + let mut data = JsonObject::new(); + data.insert( + Self::TRANSPORT_CLOSED_MARKER.to_owned(), + Value::from(Self::transport_closed_token()), + ); + Self::internal_error(message, Some(Value::Object(data))) + } + + pub(crate) fn is_transport_closed(&self) -> bool { + self.data + .as_ref() + .and_then(|data| data.get(Self::TRANSPORT_CLOSED_MARKER)) + .and_then(Value::as_u64) + == Some(Self::transport_closed_token()) + } + + fn transport_closed_token() -> u64 { + static TOKEN: OnceLock = OnceLock::new(); + *TOKEN.get_or_init(|| { + let mut hasher = RandomState::new().build_hasher(); + hasher.write(b"rmcp transport-closed marker"); + hasher.finish() + }) + } } /// Represents any JSON-RPC message that can be sent or received. @@ -1661,6 +1692,9 @@ impl RequestParamsMeta for SubscribeRequestParams { pub type SubscribeRequestParam = SubscribeRequestParams; /// Request to subscribe to resource updates +#[deprecated( + note = "resources/subscribe is legacy-only; use subscriptions/listen for protocol version 2026-07-28" +)] pub type SubscribeRequest = Request; const_string!(UnsubscribeRequestMethod = "resources/unsubscribe"); @@ -1701,6 +1735,9 @@ impl RequestParamsMeta for UnsubscribeRequestParams { pub type UnsubscribeRequestParam = UnsubscribeRequestParams; /// Request to unsubscribe from resource updates +#[deprecated( + note = "resources/unsubscribe is legacy-only; cancel the subscriptions/listen request for protocol version 2026-07-28" +)] pub type UnsubscribeRequest = Request; const_string!(ResourceUpdatedNotificationMethod = "notifications/resources/updated"); @@ -1730,6 +1767,390 @@ impl ResourceUpdatedNotificationParam { pub type ResourceUpdatedNotification = Notification; +// ============================================================================= +// SUBSCRIPTIONS +// ============================================================================= + +/// Notification categories a client opts in to on a `subscriptions/listen` stream. +#[derive(Debug, Default, Serialize, Deserialize, Clone, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))] +#[non_exhaustive] +pub struct SubscriptionFilter { + #[serde(default, skip_serializing_if = "Option::is_none")] + #[cfg_attr(feature = "schemars", schemars(with = "bool"))] + pub tools_list_changed: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[cfg_attr(feature = "schemars", schemars(with = "bool"))] + pub prompts_list_changed: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[cfg_attr(feature = "schemars", schemars(with = "bool"))] + pub resources_list_changed: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[cfg_attr(feature = "schemars", schemars(with = "Vec"))] + pub resource_subscriptions: Option>, +} + +impl SubscriptionFilter { + /// Create an empty filter that opts in to no notifications. + pub fn new() -> Self { + Self::default() + } + + /// Create a builder for a subscription filter. + pub fn builder() -> SubscriptionFilterBuilder { + SubscriptionFilterBuilder::default() + } + + /// Return the subset present in both filters. + pub fn intersection(&self, other: &Self) -> Self { + let resource_subscriptions = self + .resource_subscriptions + .as_ref() + .and_then(|requested| { + other.resource_subscriptions.as_ref().map(|accepted| { + requested + .iter() + .filter(|uri| accepted.contains(uri)) + .cloned() + .collect() + }) + }) + .filter(|uris: &Vec| !uris.is_empty()); + Self { + tools_list_changed: (self.tools_list_changed == Some(true) + && other.tools_list_changed == Some(true)) + .then_some(true), + prompts_list_changed: (self.prompts_list_changed == Some(true) + && other.prompts_list_changed == Some(true)) + .then_some(true), + resources_list_changed: (self.resources_list_changed == Some(true) + && other.resources_list_changed == Some(true)) + .then_some(true), + resource_subscriptions, + } + } + + /// Return whether this filter accepts only notifications requested by `other`. + pub fn is_subset_of(&self, other: &Self) -> bool { + let booleans_are_subset = [ + (self.tools_list_changed, other.tools_list_changed), + (self.prompts_list_changed, other.prompts_list_changed), + (self.resources_list_changed, other.resources_list_changed), + ] + .into_iter() + .all(|(accepted, requested)| accepted != Some(true) || requested == Some(true)); + let resources_are_subset = self.resource_subscriptions.as_ref().is_none_or(|accepted| { + accepted.iter().all(|uri| { + other + .resource_subscriptions + .as_ref() + .is_some_and(|requested| requested.contains(uri)) + }) + }); + booleans_are_subset && resources_are_subset + } + + /// Return the requested notification types advertised by server capabilities. + pub fn supported_by(&self, capabilities: &ServerCapabilities) -> Self { + Self { + tools_list_changed: (self.tools_list_changed == Some(true) + && capabilities + .tools + .as_ref() + .is_some_and(|tools| tools.list_changed == Some(true))) + .then_some(true), + prompts_list_changed: (self.prompts_list_changed == Some(true) + && capabilities + .prompts + .as_ref() + .is_some_and(|prompts| prompts.list_changed == Some(true))) + .then_some(true), + resources_list_changed: (self.resources_list_changed == Some(true) + && capabilities + .resources + .as_ref() + .is_some_and(|resources| resources.list_changed == Some(true))) + .then_some(true), + resource_subscriptions: capabilities + .resources + .as_ref() + .is_some_and(|resources| resources.subscribe == Some(true)) + .then(|| self.resource_subscriptions.clone()) + .flatten(), + } + } +} + +/// Builder for [`SubscriptionFilter`]. +#[derive(Debug, Default)] +#[non_exhaustive] +pub struct SubscriptionFilterBuilder { + filter: SubscriptionFilter, +} + +impl SubscriptionFilterBuilder { + /// Opt in to `notifications/tools/list_changed`. + pub fn tools_list_changed(mut self) -> Self { + self.filter.tools_list_changed = Some(true); + self + } + + /// Opt in to `notifications/prompts/list_changed`. + pub fn prompts_list_changed(mut self) -> Self { + self.filter.prompts_list_changed = Some(true); + self + } + + /// Opt in to `notifications/resources/list_changed`. + pub fn resources_list_changed(mut self) -> Self { + self.filter.resources_list_changed = Some(true); + self + } + + /// Opt in to updates for all supplied resource URIs. + pub fn resource_subscriptions( + mut self, + uris: impl IntoIterator>, + ) -> Self { + self.filter.resource_subscriptions = Some(uris.into_iter().map(Into::into).collect()); + self + } + + /// Add one resource URI to the update subscription set. + pub fn resource_subscription(mut self, uri: impl Into) -> Self { + self.filter + .resource_subscriptions + .get_or_insert_default() + .push(uri.into()); + self + } + + /// Build the filter. + pub fn build(self) -> SubscriptionFilter { + self.filter + } +} + +const_string!(SubscriptionsListenRequestMethod = "subscriptions/listen"); + +#[cfg(feature = "schemars")] +fn subscriptions_listen_request_meta_schema( + generator: &mut schemars::SchemaGenerator, +) -> schemars::Schema { + let progress_token = generator.subschema_for::(); + let client_info = generator.subschema_for::(); + let client_capabilities = generator.subschema_for::(); + let log_level = generator.subschema_for::(); + schemars::json_schema!({ + "type": "object", + "properties": { + "progressToken": progress_token, + "io.modelcontextprotocol/protocolVersion": { + "type": "string", + }, + "io.modelcontextprotocol/clientInfo": client_info, + "io.modelcontextprotocol/clientCapabilities": client_capabilities, + "io.modelcontextprotocol/logLevel": log_level, + }, + "required": RequestMetaObject::DRAFT_REQUIRED_KEYS, + "additionalProperties": true, + }) +} + +/// Parameters for opening a long-lived notification subscription. +#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))] +#[non_exhaustive] +pub struct SubscriptionsListenRequestParams { + /// Protocol-level metadata. Required by the draft wire schema. + #[serde(rename = "_meta", skip_serializing_if = "Option::is_none")] + #[cfg_attr( + feature = "schemars", + schemars(required, schema_with = "subscriptions_listen_request_meta_schema") + )] + pub meta: Option, + /// Notification categories requested for this stream. + pub notifications: SubscriptionFilter, +} + +impl SubscriptionsListenRequestParams { + /// Create listen parameters for a notification filter. + pub fn new(notifications: SubscriptionFilter) -> Self { + Self { + meta: None, + notifications, + } + } + + /// Set protocol-level request metadata. + pub fn with_meta(mut self, meta: RequestMetaObject) -> Self { + self.meta = Some(meta); + self + } +} + +impl RequestParamsMeta for SubscriptionsListenRequestParams { + fn meta(&self) -> Option<&RequestMetaObject> { + self.meta.as_ref() + } + + fn meta_mut(&mut self) -> &mut Option { + &mut self.meta + } +} + +/// Request that opens a long-lived notification subscription. +pub type SubscriptionsListenRequest = + Request; + +const SUBSCRIPTION_ID_META_KEY: &str = "io.modelcontextprotocol/subscriptionId"; + +/// Metadata on the final result of a `subscriptions/listen` request. +#[derive(Debug, Serialize, Clone, PartialEq)] +#[serde(transparent)] +#[non_exhaustive] +pub struct SubscriptionsListenResultMeta(MetaObject); + +impl SubscriptionsListenResultMeta { + /// Create result metadata for the originating listen request. + pub fn new(subscription_id: RequestId) -> Self { + let mut meta = MetaObject::new(); + meta.insert( + SUBSCRIPTION_ID_META_KEY.to_owned(), + subscription_id.into_json_value(), + ); + Self(meta) + } + + /// Return the originating listen request ID, if the metadata remains valid. + pub fn subscription_id(&self) -> Option { + self.0 + .get(SUBSCRIPTION_ID_META_KEY) + .and_then(|value| RequestId::deserialize(value).ok()) + } + + /// Replace the originating listen request ID. + pub fn set_subscription_id(&mut self, subscription_id: RequestId) { + self.0.insert( + SUBSCRIPTION_ID_META_KEY.to_owned(), + subscription_id.into_json_value(), + ); + } +} + +impl<'de> Deserialize<'de> for SubscriptionsListenResultMeta { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + let meta = MetaObject::deserialize(deserializer)?; + let Some(value) = meta.get(SUBSCRIPTION_ID_META_KEY) else { + return Err(serde::de::Error::missing_field(SUBSCRIPTION_ID_META_KEY)); + }; + RequestId::deserialize(value).map_err(serde::de::Error::custom)?; + Ok(Self(meta)) + } +} + +impl std::ops::Deref for SubscriptionsListenResultMeta { + type Target = MetaObject; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl std::ops::DerefMut for SubscriptionsListenResultMeta { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.0 + } +} + +#[cfg(feature = "schemars")] +impl schemars::JsonSchema for SubscriptionsListenResultMeta { + fn schema_name() -> Cow<'static, str> { + Cow::Borrowed("SubscriptionsListenResultMeta") + } + + fn json_schema(generator: &mut schemars::SchemaGenerator) -> schemars::Schema { + let subscription_id = generator.subschema_for::(); + schemars::json_schema!({ + "type": "object", + "properties": { + "io.modelcontextprotocol/subscriptionId": subscription_id, + }, + "required": ["io.modelcontextprotocol/subscriptionId"], + "additionalProperties": true, + }) + } +} + +/// Final response indicating that a subscription ended gracefully. +#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))] +#[non_exhaustive] +pub struct SubscriptionsListenResult { + pub result_type: ResultType, + #[serde(rename = "_meta")] + pub meta: SubscriptionsListenResultMeta, +} + +impl SubscriptionsListenResult { + /// Create a completed subscription result. + pub fn new(meta: SubscriptionsListenResultMeta) -> Self { + Self { + result_type: ResultType::COMPLETE, + meta, + } + } + + /// Create a completed result for the originating listen request. + pub fn complete(subscription_id: RequestId) -> Self { + Self::new(SubscriptionsListenResultMeta::new(subscription_id)) + } +} + +const_string!( + SubscriptionsAcknowledgedNotificationMethod = "notifications/subscriptions/acknowledged" +); + +/// Parameters reporting the accepted subset of a subscription filter. +#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))] +#[non_exhaustive] +pub struct SubscriptionsAcknowledgedNotificationParams { + #[serde(rename = "_meta", default, skip_serializing_if = "Option::is_none")] + #[cfg_attr(feature = "schemars", schemars(with = "NotificationMetaObject"))] + pub meta: Option, + pub notifications: SubscriptionFilter, +} + +impl SubscriptionsAcknowledgedNotificationParams { + /// Create acknowledgment parameters for the accepted filter. + pub fn new(notifications: SubscriptionFilter) -> Self { + Self { + meta: None, + notifications, + } + } + + /// Set notification metadata. + pub fn with_meta(mut self, meta: NotificationMetaObject) -> Self { + self.meta = Some(meta); + self + } +} + +/// First notification sent on an established subscription stream. +pub type SubscriptionsAcknowledgedNotification = Notification< + SubscriptionsAcknowledgedNotificationMethod, + SubscriptionsAcknowledgedNotificationParams, +>; + // ============================================================================= // PROMPT MANAGEMENT // ============================================================================= @@ -3937,6 +4358,7 @@ ts_union!( | ListResourcesRequest | ListResourceTemplatesRequest | ReadResourceRequest + | SubscriptionsListenRequest | SubscribeRequest | UnsubscribeRequest | CallToolRequest @@ -3961,6 +4383,7 @@ impl ClientRequest { ClientRequest::ListResourcesRequest(r) => r.method.as_str(), ClientRequest::ListResourceTemplatesRequest(r) => r.method.as_str(), ClientRequest::ReadResourceRequest(r) => r.method.as_str(), + ClientRequest::SubscriptionsListenRequest(r) => r.method.as_str(), ClientRequest::SubscribeRequest(r) => r.method.as_str(), ClientRequest::UnsubscribeRequest(r) => r.method.as_str(), ClientRequest::CallToolRequest(r) => r.method.as_str(), @@ -4019,6 +4442,7 @@ ts_union!( | ResourceListChangedNotification | ToolListChangedNotification | PromptListChangedNotification + | SubscriptionsAcknowledgedNotification | TaskStatusNotification | CustomNotification; ); @@ -4033,6 +4457,7 @@ ts_union!( | ListResourcesResult | ListResourceTemplatesResult | ReadResourceResult + | SubscriptionsListenResult | ListToolsResult | ElicitResult | CreateTaskResult @@ -4096,6 +4521,19 @@ mod tests { let _: ResourceReference = ResourceTemplateReference::new("res://x"); } + #[cfg(feature = "transport-streamable-http-client")] + #[test] + fn transport_closed_marker_accepts_only_the_process_local_token() { + let local = ErrorData::transport_closed("closed"); + let spoofed = ErrorData::internal_error( + "spoofed", + Some(json!({ "io.modelcontextprotocol/transportClosed": true })), + ); + + assert!(local.is_transport_closed()); + assert!(!spoofed.is_transport_closed()); + } + #[test] fn cancelled_notification_request_id_is_optional_on_wire() { // None → requestId 생략 diff --git a/crates/rmcp/src/model/meta.rs b/crates/rmcp/src/model/meta.rs index c51fc6d7c..d86156aa9 100644 --- a/crates/rmcp/src/model/meta.rs +++ b/crates/rmcp/src/model/meta.rs @@ -197,6 +197,7 @@ variant_extension! { ListResourcesRequest ListResourceTemplatesRequest ReadResourceRequest + SubscriptionsListenRequest SubscribeRequest UnsubscribeRequest CallToolRequest @@ -239,6 +240,7 @@ variant_extension! { ResourceListChangedNotification ToolListChangedNotification PromptListChangedNotification + SubscriptionsAcknowledgedNotification TaskStatusNotification CustomNotification } diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 69c3f2105..89c6a1d67 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -85,6 +85,8 @@ pub enum ServiceError { TransportClosed, #[error("Unexpected response type")] UnexpectedResponse, + #[error("subscription consumer lagged behind its {capacity}-message buffer")] + SubscriptionLagged { capacity: usize }, #[error("task cancelled for reason {}", reason.as_deref().unwrap_or(""))] Cancelled { reason: Option }, #[error("request timeout after {}", chrono::Duration::from_std(*timeout).unwrap_or_default())] @@ -128,6 +130,12 @@ pub trait ServiceRole: std::fmt::Debug + Send + Sync + 'static + Copy + Clone { const IS_CLIENT: bool; type Info: TransferObject; type PeerInfo: TransferObject; + #[doc(hidden)] + fn configure_direct_peer(_peer: &Peer, _info: &Self::Info) {} + #[doc(hidden)] + fn peer_cancelled_params(_notification: &Self::PeerNot) -> Option<&CancelledNotificationParam> { + None + } } pub type TxJsonRpcMessage = @@ -340,6 +348,8 @@ impl ProgressNotificationToken for ServerNotification { type Responder = tokio::sync::oneshot::Sender; type ProgressTimeoutWatchers = Arc>>>; +type SubscriptionChannel = (mpsc::Sender, usize); +type SubscriptionChannelMap = HashMap>; /// A handle to a remote request /// @@ -400,10 +410,12 @@ impl RequestHandle { has_progress_reset_rx, ) .await; + self.peer.unregister_subscription(&self.id); result } async fn send_timeout_cancel_notification(&self, reason: &str) { + self.peer.unregister_subscription(&self.id); let notification = CancelledNotification { params: CancelledNotificationParam { request_id: Some(self.id.clone()), @@ -478,6 +490,7 @@ impl RequestHandle { self.progress_reset_rx.is_some(), ) .await; + self.peer.unregister_subscription(&self.id); let notification = CancelledNotification { params: CancelledNotificationParam { request_id: Some(self.id), @@ -530,7 +543,6 @@ pub(crate) struct ClientRequestMetadata { /// For general purpose, call [`Peer::send_request`] or [`Peer::send_notification`] to send message to remote peer. /// /// To create a cancellable request, call [`Peer::send_request_with_option`]. -#[derive(Clone)] pub struct Peer { tx: mpsc::Sender>, request_id_provider: Arc, @@ -539,6 +551,25 @@ pub struct Peer { info: Arc>>>, client_request_metadata: Arc>, request_metadata_required: Arc, + subscription_channels: Arc>>, +} + +impl Clone for Peer +where + R::PeerInfo: Clone, +{ + fn clone(&self) -> Peer { + Self { + tx: self.tx.clone(), + request_id_provider: self.request_id_provider.clone(), + progress_token_provider: self.progress_token_provider.clone(), + progress_timeout_watchers: self.progress_timeout_watchers.clone(), + info: self.info.clone(), + client_request_metadata: self.client_request_metadata.clone(), + request_metadata_required: self.request_metadata_required.clone(), + subscription_channels: self.subscription_channels.clone(), + } + } } impl std::fmt::Debug for Peer { @@ -610,6 +641,7 @@ impl Peer { info: Arc::new(std::sync::RwLock::new(peer_info.map(Arc::new))), client_request_metadata: Default::default(), request_metadata_required: Default::default(), + subscription_channels: Default::default(), }, rx, ) @@ -641,9 +673,19 @@ impl Peer { } pub async fn send_request_with_option( + &self, + request: R::Req, + options: PeerRequestOptions, + ) -> Result, ServiceError> { + self.send_request_with_option_and_subscription(request, options, None) + .await + } + + async fn send_request_with_option_and_subscription( &self, mut request: R::Req, options: PeerRequestOptions, + subscription_sender: Option>, ) -> Result, ServiceError> { let id = self.request_id_provider.next_request_id(); let progress_token = self.progress_token_provider.next_progress_token(); @@ -670,6 +712,10 @@ impl Peer { } else { None }; + if let Some(channel) = subscription_sender { + self.subscription_channels_write() + .insert(id.clone(), channel); + } if self .tx .send(PeerSinkMessage::Request { @@ -686,6 +732,7 @@ impl Peer { .await .remove(&progress_token); } + self.unregister_subscription(&id); return Err(ServiceError::TransportClosed); } Ok(RequestHandle { @@ -698,6 +745,67 @@ impl Peer { }) } + #[cfg(feature = "client")] + pub(crate) async fn send_subscription_request( + &self, + request: R::Req, + options: PeerRequestOptions, + channel_capacity: usize, + ) -> Result<(RequestHandle, mpsc::Receiver), ServiceError> { + let (sender, receiver) = mpsc::channel(channel_capacity); + let handle = self + .send_request_with_option_and_subscription( + request, + options, + Some((sender, channel_capacity)), + ) + .await?; + Ok((handle, receiver)) + } + + fn subscription_sender(&self, id: &RequestId) -> Option> { + self.subscription_channels_read().get(id).cloned() + } + + pub(crate) fn unregister_subscription(&self, id: &RequestId) { + self.subscription_channels_write().remove(id); + } + + fn subscription_channels_read( + &self, + ) -> std::sync::RwLockReadGuard<'_, SubscriptionChannelMap> { + match self.subscription_channels.read() { + Ok(channels) => channels, + Err(poisoned) => poisoned.into_inner(), + } + } + + fn subscription_channels_write( + &self, + ) -> std::sync::RwLockWriteGuard<'_, SubscriptionChannelMap> { + match self.subscription_channels.write() { + Ok(channels) => channels, + Err(poisoned) => poisoned.into_inner(), + } + } + + pub(crate) fn try_cancel_request(&self, id: RequestId, reason: Option) { + let notification = CancelledNotification { + params: CancelledNotificationParam { + request_id: Some(id), + reason, + meta: None, + }, + method: crate::model::CancelledNotificationMethod, + extensions: Default::default(), + }; + let (responder, _receiver) = tokio::sync::oneshot::channel(); + let _ = self.tx.try_send(PeerSinkMessage::Notification { + notification: notification.into(), + responder, + }); + } + async fn notify_progress_timeout_watcher(&self, progress_token: &ProgressToken) { let sender = self .progress_timeout_watchers @@ -999,6 +1107,7 @@ where E: std::error::Error + Send + Sync + 'static, { let (peer, peer_rx) = Peer::new(Arc::new(AtomicU32RequestIdProvider::default()), peer_info); + R::configure_direct_peer(&peer, &service.get_info()); serve_inner(service, transport.into_transport(), peer, peer_rx, ct) } @@ -1274,19 +1383,67 @@ where .. })) => { tracing::info!(?notification, "received notification"); - // catch cancelled notification - let mut notification = match notification.try_into() { - Ok::(cancelled) => { - if let Some(request_id) = &cancelled.params.request_id { - if let Some(ct) = local_ct_pool.remove(request_id) { - tracing::info!(id = %request_id, reason = cancelled.params.reason, "cancelled"); + let cancellation_request_id = + if let Some(cancelled) = R::peer_cancelled_params(¬ification) { + let request_id = cancelled.request_id.clone(); + if let Some(request_id) = request_id.as_ref() { + if R::IS_CLIENT { + if let Some(responder) = + local_responder_pool.remove(request_id) + { + let _ = responder.send(Err(ServiceError::Cancelled { + reason: cancelled.reason.clone(), + })); + } + } else if let Some(ct) = local_ct_pool.remove(request_id) { + tracing::info!(id = %request_id, reason = cancelled.reason, "cancelled"); ct.cancel(); } } - cancelled.into() + request_id + } else { + None + }; + let subscription_id = notification + .get_meta() + .subscription_id() + .or(cancellation_request_id); + if let Some(subscription_id) = subscription_id + && let Some((sender, capacity)) = + peer.subscription_sender(&subscription_id) + { + match sender.try_send(notification) { + Ok(()) => {} + Err(mpsc::error::TrySendError::Full(_)) => { + tracing::warn!( + id = %subscription_id, + capacity, + "subscription notification buffer full" + ); + if R::IS_CLIENT + && let Some(responder) = + local_responder_pool.remove(&subscription_id) + { + let _ = responder + .send(Err(ServiceError::SubscriptionLagged { capacity })); + } + peer.unregister_subscription(&subscription_id); + peer.try_cancel_request( + subscription_id, + Some("subscription notification buffer full".to_owned()), + ); + } + Err(mpsc::error::TrySendError::Closed(_)) => { + peer.unregister_subscription(&subscription_id); + peer.try_cancel_request( + subscription_id, + Some("subscription notification receiver closed".to_owned()), + ); + } } - Err(notification) => notification, - }; + continue; + } + let mut notification = notification; if let Some(progress_token) = notification.progress_token() { peer.notify_progress_timeout_watcher(progress_token).await; } @@ -1331,7 +1488,12 @@ where continue; }; if let Some(responder) = local_responder_pool.remove(&id) { - let _response_result = responder.send(Err(ServiceError::McpError(error))); + let service_error = if error.is_transport_closed() { + ServiceError::TransportClosed + } else { + ServiceError::McpError(error) + }; + let _response_result = responder.send(Err(service_error)); if let Err(_error) = _response_result { tracing::warn!(%id, "Error sending response"); } diff --git a/crates/rmcp/src/service/client.rs b/crates/rmcp/src/service/client.rs index 9ba3024d0..35deb38c8 100644 --- a/crates/rmcp/src/service/client.rs +++ b/crates/rmcp/src/service/client.rs @@ -1,6 +1,6 @@ // Sampling/Roots/Logging are SEP-2577-deprecated; internal references are expected. #![expect(deprecated)] -use std::{borrow::Cow, sync::Arc, time::Duration}; +use std::{borrow::Cow, num::NonZeroUsize, sync::Arc, time::Duration}; use thiserror::Error; @@ -21,8 +21,9 @@ use crate::{ ReadResourceRequest, ReadResourceRequestParams, ReadResourceResponse, ReadResourceResult, Reference, RequestId, RequestMetaObject, RootsListChangedNotification, ServerInfo, ServerJsonRpcMessage, ServerNotification, ServerRequest, ServerResult, SetLevelRequest, - SetLevelRequestParams, SubscribeRequest, SubscribeRequestParams, UnsubscribeRequest, - UnsubscribeRequestParams, + SetLevelRequestParams, SubscribeRequest, SubscribeRequestParams, SubscriptionFilter, + SubscriptionsListenRequest, SubscriptionsListenRequestParams, SubscriptionsListenResult, + UnsubscribeRequest, UnsubscribeRequestParams, }, transport::DynamicTransportError, }; @@ -184,10 +185,298 @@ impl ServiceRole for RoleClient { type PeerInfo = ServerInfo; type InitializeError = ClientInitializeError; const IS_CLIENT: bool = true; + + fn configure_direct_peer(peer: &Peer, info: &Self::Info) { + let Some(server_info) = peer.peer_info() else { + return; + }; + if server_info.protocol_version.as_str() < ProtocolVersion::V_2026_07_28.as_str() { + return; + } + peer.set_client_request_metadata(ClientRequestMetadata { + protocol_version: server_info.protocol_version.clone(), + client_info: info.client_info.clone(), + client_capabilities: info.capabilities.clone(), + }); + } + + fn peer_cancelled_params(notification: &Self::PeerNot) -> Option<&CancelledNotificationParam> { + match notification { + ServerNotification::CancelledNotification(notification) => Some(¬ification.params), + _ => None, + } + } } pub type ServerSink = Peer; +/// Default number of notifications buffered for one subscription. +pub const DEFAULT_SUBSCRIPTION_CHANNEL_CAPACITY: usize = 64; + +/// How a client-side subscription stream ended. +#[derive(Debug, Clone, PartialEq)] +#[non_exhaustive] +pub enum SubscriptionEnd { + /// The server returned a final `SubscriptionsListenResult`. + Graceful(SubscriptionsListenResult), + /// The transport closed without a final result. Call `Peer::listen` again + /// after reconnecting; subscription streams are not resumable. + Abrupt, + /// The subscription was explicitly cancelled by either peer. + Cancelled, + /// The consumer did not drain notifications before the channel filled. + Lagged { capacity: usize }, +} + +/// Handle for one active `subscriptions/listen` request. +#[derive(Debug)] +pub struct Subscription { + id: RequestId, + acknowledged: SubscriptionFilter, + notifications: tokio::sync::mpsc::Receiver, + request: Option>, + end: Option, +} + +type SubscriptionResponse = + Result, tokio::sync::oneshot::error::RecvError>; + +struct PendingSubscriptionRequest { + handle: Option>, +} + +impl PendingSubscriptionRequest { + fn new(handle: RequestHandle) -> Self { + Self { + handle: Some(handle), + } + } + + async fn recv(&mut self) -> Option { + let handle = self.handle.as_mut()?; + Some((&mut handle.rx).await) + } + + fn take(&mut self) -> Option> { + self.handle.take() + } + + fn unregister(&self, id: &RequestId) { + if let Some(handle) = self.handle.as_ref() { + handle.peer.unregister_subscription(id); + } + } + + fn disarm(&mut self) { + self.handle.take(); + } + + async fn cancel(&mut self, reason: &'static str) { + if let Some(handle) = self.handle.take() { + let _ = handle.cancel(Some(reason.to_owned())).await; + } + } +} + +impl Drop for PendingSubscriptionRequest { + fn drop(&mut self) { + let Some(handle) = self.handle.take() else { + return; + }; + handle.peer.unregister_subscription(&handle.id); + handle.peer.try_cancel_request( + handle.id, + Some("subscription establishment cancelled".to_owned()), + ); + } +} + +impl Subscription { + /// Return the originating listen request ID. + pub fn id(&self) -> &RequestId { + &self.id + } + + /// Return the notification filter accepted by the server. + pub fn acknowledged(&self) -> &SubscriptionFilter { + &self.acknowledged + } + + /// Return the terminal state after this subscription has ended. + pub fn end(&self) -> Option<&SubscriptionEnd> { + self.end.as_ref() + } + + /// Receive the next notification, or `None` after the subscription ends. + /// + /// A graceful final result and an abrupt transport close are distinguished + /// through [`Self::end`]. + /// + /// # Errors + /// + /// Returns a service or protocol error when the stream carries an invalid + /// message, an unexpected final result, or another request failure. + pub async fn next(&mut self) -> Result, ServiceError> { + if self.end.is_some() { + return Ok(None); + } + let Some(request) = self.request.as_mut() else { + self.end = Some(SubscriptionEnd::Abrupt); + return Ok(None); + }; + + tokio::select! { + biased; + notification = self.notifications.recv() => { + let Some(notification) = notification else { + let response = (&mut request.rx).await; + return self.handle_response(response); + }; + if let ServerNotification::CancelledNotification(cancelled) = ¬ification { + if cancelled.params.request_id.as_ref() != Some(&self.id) { + self.cancel_as_abrupt("subscription cancellation ID mismatch") + .await; + return Err(ServiceError::UnexpectedResponse); + } + self.finish(SubscriptionEnd::Cancelled); + return Ok(None); + } + if notification.get_meta().subscription_id().as_ref() != Some(&self.id) { + self.cancel_as_abrupt("subscription notification ID mismatch") + .await; + return Err(ServiceError::UnexpectedResponse); + } + if !self.accepts(¬ification) { + self.cancel_as_abrupt( + "subscription notification was outside the acknowledged filter", + ) + .await; + return Err(ServiceError::UnexpectedResponse); + } + Ok(Some(notification)) + } + response = &mut request.rx => { + self.handle_response(response) + } + } + } + + /// Cancel this subscription. + /// + /// # Errors + /// + /// Returns a transport error when the cancellation signal cannot be sent. + pub async fn cancel(&mut self) -> Result<(), ServiceError> { + self.cancel_with_reason(None).await + } + + /// Cancel this subscription with a diagnostic reason. + /// + /// # Errors + /// + /// Returns a transport error when the cancellation signal cannot be sent. + pub async fn cancel_with_reason(&mut self, reason: Option) -> Result<(), ServiceError> { + let Some(request) = self.request.take() else { + return Ok(()); + }; + request.cancel(reason).await?; + self.end = Some(SubscriptionEnd::Cancelled); + Ok(()) + } + + fn finish(&mut self, end: SubscriptionEnd) { + if let Some(request) = self.request.take() { + request.peer.unregister_subscription(&self.id); + } + self.end = Some(end); + } + + async fn cancel_as_abrupt(&mut self, reason: &'static str) { + if let Some(request) = self.request.take() { + let _ = request.cancel(Some(reason.to_owned())).await; + } + self.end = Some(SubscriptionEnd::Abrupt); + } + + fn accepts(&self, notification: &ServerNotification) -> bool { + match notification { + ServerNotification::ToolListChangedNotification(_) => { + self.acknowledged.tools_list_changed == Some(true) + } + ServerNotification::PromptListChangedNotification(_) => { + self.acknowledged.prompts_list_changed == Some(true) + } + ServerNotification::ResourceListChangedNotification(_) => { + self.acknowledged.resources_list_changed == Some(true) + } + ServerNotification::ResourceUpdatedNotification(update) => self + .acknowledged + .resource_subscriptions + .as_ref() + .is_some_and(|uris| uris.contains(&update.params.uri)), + ServerNotification::SubscriptionsAcknowledgedNotification(_) + | ServerNotification::CancelledNotification(_) + | ServerNotification::ProgressNotification(_) + | ServerNotification::LoggingMessageNotification(_) + | ServerNotification::TaskStatusNotification(_) + | ServerNotification::CustomNotification(_) => false, + } + } + + fn handle_response( + &mut self, + response: SubscriptionResponse, + ) -> Result, ServiceError> { + let response = match response { + Ok(response) => response, + Err(_) => { + self.finish(SubscriptionEnd::Abrupt); + return Ok(None); + } + }; + let response = match response { + Ok(response) => response, + Err(ServiceError::TransportClosed) => { + self.finish(SubscriptionEnd::Abrupt); + return Ok(None); + } + Err(ServiceError::SubscriptionLagged { capacity }) => { + self.finish(SubscriptionEnd::Lagged { capacity }); + return Ok(None); + } + Err(error) => { + self.finish(SubscriptionEnd::Abrupt); + return Err(error); + } + }; + let ServerResult::SubscriptionsListenResult(result) = response else { + self.finish(SubscriptionEnd::Abrupt); + return Err(ServiceError::UnexpectedResponse); + }; + if !result.result_type.is_complete() + || result.meta.subscription_id().as_ref() != Some(&self.id) + { + self.finish(SubscriptionEnd::Abrupt); + return Err(ServiceError::UnexpectedResponse); + } + self.finish(SubscriptionEnd::Graceful(result)); + Ok(None) + } +} + +impl Drop for Subscription { + fn drop(&mut self) { + let Some(request) = self.request.take() else { + return; + }; + request.peer.unregister_subscription(&self.id); + request.peer.try_cancel_request( + self.id.clone(), + Some("subscription handle dropped".to_owned()), + ); + } +} + /// Selects how a client establishes its MCP lifecycle. /// /// Existing [`ServiceExt::serve`] behavior remains legacy initialization. @@ -621,6 +910,106 @@ macro_rules! method { } impl Peer { + /// Open a long-lived notification subscription and wait for its acknowledgment. + /// + /// Notifications routed to the returned [`Subscription`] are not also + /// delivered through [`ClientHandler`](crate::ClientHandler) callbacks. + /// + /// # Errors + /// + /// Returns a service, transport, or protocol error when the request cannot + /// be established or the acknowledgment is invalid. + pub async fn listen( + &self, + notifications: SubscriptionFilter, + ) -> Result { + self.listen_with_channel_capacity_inner( + notifications, + DEFAULT_SUBSCRIPTION_CHANNEL_CAPACITY, + ) + .await + } + + /// Open a subscription with an explicit notification buffer capacity. + /// + /// Notifications routed to the returned [`Subscription`] are not also + /// delivered through [`ClientHandler`](crate::ClientHandler) callbacks. + /// + /// # Errors + /// + /// Returns a service, transport, or protocol error when the request cannot + /// be established or the acknowledgment is invalid. + pub async fn listen_with_capacity( + &self, + notifications: SubscriptionFilter, + channel_capacity: NonZeroUsize, + ) -> Result { + self.listen_with_channel_capacity_inner(notifications, channel_capacity.get()) + .await + } + + async fn listen_with_channel_capacity_inner( + &self, + notifications: SubscriptionFilter, + channel_capacity: usize, + ) -> Result { + let request = ClientRequest::SubscriptionsListenRequest(SubscriptionsListenRequest::new( + SubscriptionsListenRequestParams::new(notifications.clone()), + )); + let (handle, mut subscription_notifications) = self + .send_subscription_request(request, PeerRequestOptions::no_options(), channel_capacity) + .await?; + let id = handle.id.clone(); + let mut pending = PendingSubscriptionRequest::new(handle); + + tokio::select! { + biased; + notification = subscription_notifications.recv() => { + let Some(notification) = notification else { + pending.cancel("subscription stream closed before acknowledgment").await; + return Err(ServiceError::TransportClosed); + }; + if notification.get_meta().subscription_id().as_ref() != Some(&id) { + pending.cancel("subscription acknowledgment ID mismatch").await; + return Err(ServiceError::UnexpectedResponse); + } + let ServerNotification::SubscriptionsAcknowledgedNotification( + acknowledgment, + ) = notification else { + pending.cancel("notification received before subscription acknowledgment").await; + return Err(ServiceError::UnexpectedResponse); + }; + let accepted = acknowledgment.params.notifications; + if !accepted.is_subset_of(¬ifications) { + pending.cancel("subscription acknowledged an unrequested filter").await; + return Err(ServiceError::UnexpectedResponse); + } + let Some(handle) = pending.take() else { + return Err(ServiceError::TransportClosed); + }; + Ok(Subscription { + id, + acknowledged: accepted, + notifications: subscription_notifications, + request: Some(handle), + end: None, + }) + } + response = pending.recv() => { + pending.unregister(&id); + pending.disarm(); + let Some(response) = response else { + return Err(ServiceError::TransportClosed); + }; + match response { + Ok(Err(error)) => Err(error), + Ok(Ok(_)) => Err(ServiceError::UnexpectedResponse), + Err(_) => Err(ServiceError::TransportClosed), + } + } + } + } + /// Discover the server's supported protocol versions and capabilities. /// /// The high-level client currently exposes this peer only after initialization; @@ -716,8 +1105,18 @@ impl Peer { method!(peer_req list_resources ListResourcesRequest(PaginatedRequestParams)? => ListResourcesResult); method!(peer_req list_resource_templates ListResourceTemplatesRequest(PaginatedRequestParams)? => ListResourceTemplatesResult); method!(peer_req read_resource ReadResourceRequest(ReadResourceRequestParams) => ReadResourceResult); - method!(peer_req subscribe SubscribeRequest(SubscribeRequestParams) ); - method!(peer_req unsubscribe UnsubscribeRequest(UnsubscribeRequestParams)); + method!( + #[deprecated( + note = "resources/subscribe is legacy-only; use Peer::listen for protocol version 2026-07-28" + )] + peer_req subscribe SubscribeRequest(SubscribeRequestParams) + ); + method!( + #[deprecated( + note = "resources/unsubscribe is legacy-only; cancel the Subscription handle instead" + )] + peer_req unsubscribe UnsubscribeRequest(UnsubscribeRequestParams) + ); method!(peer_req call_tool CallToolRequest(CallToolRequestParams) => CallToolResult); method!(peer_req list_tools ListToolsRequest(PaginatedRequestParams)? => ListToolsResult); diff --git a/crates/rmcp/src/service/server.rs b/crates/rmcp/src/service/server.rs index e24ae085f..b7aa6b968 100644 --- a/crates/rmcp/src/service/server.rs +++ b/crates/rmcp/src/service/server.rs @@ -1,8 +1,8 @@ // Sampling/Roots/Logging are SEP-2577-deprecated; internal references are expected. #![expect(deprecated)] -use std::borrow::Cow; #[cfg(feature = "elicitation")] use std::collections::HashSet; +use std::{borrow::Cow, sync::Arc}; use thiserror::Error; #[cfg(feature = "elicitation")] @@ -20,7 +20,8 @@ use crate::{ ProgressNotification, ProgressNotificationParam, PromptListChangedNotification, ProtocolVersion, ResourceListChangedNotification, ResourceUpdatedNotification, ResourceUpdatedNotificationParam, ServerInfo, ServerNotification, ServerRequest, - ServerResult, ToolListChangedNotification, + ServerResult, SubscriptionFilter, SubscriptionsAcknowledgedNotification, + SubscriptionsAcknowledgedNotificationParams, ToolListChangedNotification, }, transport::DynamicTransportError, }; @@ -41,6 +42,13 @@ impl ServiceRole for RoleServer { type InitializeError = ServerInitializeError; const IS_CLIENT: bool = false; + + fn peer_cancelled_params(notification: &Self::PeerNot) -> Option<&CancelledNotificationParam> { + match notification { + ClientNotification::CancelledNotification(notification) => Some(¬ification.params), + _ => None, + } + } } /// It represents the error that may occur when serving the server. @@ -98,6 +106,284 @@ impl ServerInitializeError { } pub type ClientSink = Peer; +/// Failure to send a notification through a [`SubscriptionSink`]. +#[derive(Debug, Error)] +#[non_exhaustive] +pub enum SubscriptionSendError { + #[error("subscription is no longer active")] + SubscriptionClosed, + #[error("notification is not allowed on a subscription stream: {0}")] + UnsupportedNotification(&'static str), + #[error("notification was not accepted for this subscription: {0}")] + NotificationNotAccepted(&'static str), + #[error(transparent)] + Service(#[from] ServiceError), +} + +/// A server-side notification sink scoped to one `subscriptions/listen` request. +/// +/// The sink applies the accepted filter and adds the subscription request ID to +/// every notification it sends. +#[derive(Debug, Clone)] +pub struct SubscriptionSink { + peer: Peer, + id: RequestId, + accepted: Arc, + active: CancellationToken, +} + +impl SubscriptionSink { + fn new( + peer: Peer, + id: RequestId, + accepted: Arc, + active: CancellationToken, + ) -> Self { + Self { + peer, + id, + accepted, + active, + } + } + + /// Return the JSON-RPC ID of the originating listen request. + pub fn id(&self) -> &RequestId { + &self.id + } + + /// Return the filter accepted for this subscription. + pub fn accepted(&self) -> &SubscriptionFilter { + self.accepted.as_ref() + } + + /// Send an allowed change notification with subscription metadata attached. + /// + /// # Errors + /// + /// Returns [`SubscriptionSendError::SubscriptionClosed`] after the request + /// ends, a filter error for disallowed notifications, or a transport error. + pub async fn send( + &self, + mut notification: ServerNotification, + ) -> Result<(), SubscriptionSendError> { + if self.active.is_cancelled() { + return Err(SubscriptionSendError::SubscriptionClosed); + } + match ¬ification { + ServerNotification::ToolListChangedNotification(_) => { + if self.accepted.tools_list_changed != Some(true) { + return Err(SubscriptionSendError::NotificationNotAccepted( + "notifications/tools/list_changed", + )); + } + } + ServerNotification::PromptListChangedNotification(_) => { + if self.accepted.prompts_list_changed != Some(true) { + return Err(SubscriptionSendError::NotificationNotAccepted( + "notifications/prompts/list_changed", + )); + } + } + ServerNotification::ResourceListChangedNotification(_) => { + if self.accepted.resources_list_changed != Some(true) { + return Err(SubscriptionSendError::NotificationNotAccepted( + "notifications/resources/list_changed", + )); + } + } + ServerNotification::ResourceUpdatedNotification(update) => { + let accepted = self + .accepted + .resource_subscriptions + .as_ref() + .is_some_and(|uris| uris.contains(&update.params.uri)); + if !accepted { + return Err(SubscriptionSendError::NotificationNotAccepted( + "notifications/resources/updated", + )); + } + } + ServerNotification::SubscriptionsAcknowledgedNotification(_) => { + return Err(SubscriptionSendError::UnsupportedNotification( + "notifications/subscriptions/acknowledged", + )); + } + ServerNotification::CancelledNotification(_) => { + return Err(SubscriptionSendError::UnsupportedNotification( + "notifications/cancelled", + )); + } + ServerNotification::ProgressNotification(_) => { + return Err(SubscriptionSendError::UnsupportedNotification( + "notifications/progress", + )); + } + ServerNotification::LoggingMessageNotification(_) => { + return Err(SubscriptionSendError::UnsupportedNotification( + "notifications/message", + )); + } + ServerNotification::TaskStatusNotification(_) => { + return Err(SubscriptionSendError::UnsupportedNotification( + "notifications/tasks/status", + )); + } + ServerNotification::CustomNotification(_) => { + return Err(SubscriptionSendError::UnsupportedNotification( + "custom notification", + )); + } + } + + notification + .get_meta_mut() + .set_subscription_id(self.id.clone()); + self.peer.send_notification(notification).await?; + Ok(()) + } + + /// Send `notifications/tools/list_changed`. + /// + /// # Errors + /// + /// See [`Self::send`]. + pub async fn notify_tool_list_changed(&self) -> Result<(), SubscriptionSendError> { + self.send(ServerNotification::ToolListChangedNotification( + ToolListChangedNotification { + method: Default::default(), + extensions: Default::default(), + }, + )) + .await + } + + /// Send `notifications/prompts/list_changed`. + /// + /// # Errors + /// + /// See [`Self::send`]. + pub async fn notify_prompt_list_changed(&self) -> Result<(), SubscriptionSendError> { + self.send(ServerNotification::PromptListChangedNotification( + PromptListChangedNotification { + method: Default::default(), + extensions: Default::default(), + }, + )) + .await + } + + /// Send `notifications/resources/list_changed`. + /// + /// # Errors + /// + /// See [`Self::send`]. + pub async fn notify_resource_list_changed(&self) -> Result<(), SubscriptionSendError> { + self.send(ServerNotification::ResourceListChangedNotification( + ResourceListChangedNotification { + method: Default::default(), + extensions: Default::default(), + }, + )) + .await + } + + /// Send `notifications/resources/updated` for an accepted URI. + /// + /// # Errors + /// + /// See [`Self::send`]. + pub async fn notify_resource_updated( + &self, + uri: impl Into, + ) -> Result<(), SubscriptionSendError> { + self.send(ServerNotification::ResourceUpdatedNotification( + ResourceUpdatedNotification::new(ResourceUpdatedNotificationParam::new(uri)), + )) + .await + } +} + +/// Context for one established server-side notification subscription. +/// +/// The acknowledgment has already been sent before this context is handed to +/// [`ServerHandler::listen`](crate::ServerHandler::listen). +#[derive(Debug)] +pub struct SubscriptionContext { + request: RequestContext, + requested: SubscriptionFilter, + accepted: Arc, + sink: SubscriptionSink, + _active_guard: DropGuard, +} + +impl SubscriptionContext { + pub(crate) async fn establish( + request: RequestContext, + requested: SubscriptionFilter, + accepted: SubscriptionFilter, + ) -> Result { + let active = request.ct.child_token(); + let accepted = Arc::new(accepted); + let sink = SubscriptionSink::new( + request.peer.clone(), + request.id.clone(), + accepted.clone(), + active.clone(), + ); + let mut acknowledgment = SubscriptionsAcknowledgedNotification::new( + SubscriptionsAcknowledgedNotificationParams::new(accepted.as_ref().clone()), + ); + let mut meta = NotificationMetaObject::new(); + meta.set_subscription_id(request.id.clone()); + acknowledgment.extensions.insert(meta); + request + .peer + .send_notification(ServerNotification::SubscriptionsAcknowledgedNotification( + acknowledgment, + )) + .await + .map_err(|error| { + ErrorData::internal_error( + format!("failed to acknowledge subscription: {error}"), + None, + ) + })?; + Ok(Self { + request, + requested, + accepted, + sink, + _active_guard: active.drop_guard(), + }) + } + + /// Return the filter requested by the client. + pub fn requested(&self) -> &SubscriptionFilter { + &self.requested + } + + /// Return the subset accepted by the server. + pub fn accepted(&self) -> &SubscriptionFilter { + self.accepted.as_ref() + } + + /// Return a cloneable, filter-enforcing notification sink. + pub fn sink(&self) -> &SubscriptionSink { + &self.sink + } + + /// Wait until the subscription request is cancelled. + pub async fn cancelled(&self) { + self.request.ct.cancelled().await; + } + + /// Access the underlying request context. + pub fn request_context(&self) -> &RequestContext { + &self.request + } +} + impl> ServiceExt for S { fn serve_with_ct( self, diff --git a/crates/rmcp/src/transport/common/server_side_http.rs b/crates/rmcp/src/transport/common/server_side_http.rs index 32609bb75..4969ff793 100644 --- a/crates/rmcp/src/transport/common/server_side_http.rs +++ b/crates/rmcp/src/transport/common/server_side_http.rs @@ -6,7 +6,7 @@ use http::Response; use http_body::Body; use http_body_util::{BodyExt, Empty, Full, combinators::BoxBody}; use sse_stream::{KeepAlive, Sse, SseBody}; -use tokio_util::sync::CancellationToken; +use tokio_util::sync::{CancellationToken, DropGuard}; use super::http_header::EVENT_STREAM_MIME_TYPE; use crate::model::{ClientJsonRpcMessage, ServerJsonRpcMessage}; @@ -57,6 +57,25 @@ impl sse_stream::Timer for TokioTimer { } } +pin_project_lite::pin_project! { + struct CancelOnDropStream { + #[pin] + inner: S, + _drop_guard: DropGuard, + } +} + +impl futures::Stream for CancelOnDropStream { + type Item = S::Item; + + fn poll_next( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + self.project().inner.poll_next(cx) + } +} + #[derive(Debug, Clone, Default)] #[non_exhaustive] pub struct ServerSseMessage { @@ -108,6 +127,7 @@ pub(crate) fn sse_stream_response( ct: CancellationToken, ) -> Response> { use futures::StreamExt; + let cancelled = ct.clone(); let stream = stream .map(|message| { let mut sse = if let Some(ref msg) = message.message { @@ -126,7 +146,11 @@ pub(crate) fn sse_stream_response( Result::::Ok(sse) }) - .take_until(async move { ct.cancelled().await }); + .take_until(async move { cancelled.cancelled().await }); + let stream = CancelOnDropStream { + inner: stream, + _drop_guard: ct.drop_guard(), + }; let stream = SseBody::new(stream); let stream = match keep_alive { @@ -140,6 +164,7 @@ pub(crate) fn sse_stream_response( .status(http::StatusCode::OK) .header(http::header::CONTENT_TYPE, EVENT_STREAM_MIME_TYPE) .header(http::header::CACHE_CONTROL, "no-cache") + .header("X-Accel-Buffering", "no") .body(stream) .expect("valid response") } diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 8ba3e423c..67a34ae85 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -30,6 +30,7 @@ use crate::{ }; type BoxedSseStream = BoxStream<'static, Result>; +type SseTaskResult = (Option, Result<(), StreamableHttpError>); const SESSION_CLEANUP_TIMEOUT: Duration = Duration::from_secs(5); fn build_request_headers( @@ -678,7 +679,7 @@ impl StreamableHttpClientWorker { } fn spawn_common_stream( - streams: &mut tokio::task::JoinSet>>, + streams: &mut tokio::task::JoinSet>, client: C, session_id: Arc, config: &StreamableHttpClientTransportConfig, @@ -694,7 +695,7 @@ impl StreamableHttpClientWorker { let max_sse_event_size = config.max_sse_event_size; streams.spawn(async move { - match client + let result = match client .get_stream_with_max_sse_event_size( uri, session_id.clone(), @@ -734,7 +735,8 @@ impl StreamableHttpClientWorker { tracing::error!("fail to get common stream: {error}"); Err(error) } - } + }; + (None, result) }); } @@ -882,7 +884,10 @@ impl Worker for StreamableHttpClientWorker { )); } }; - let mut session_id: Option> = if let Some(session_id) = session_id { + let mut uses_modern_http = !is_legacy_startup; + let mut session_id: Option> = if uses_modern_http { + None + } else if let Some(session_id) = session_id { Some(session_id.into()) } else { if !self.config.allow_stateless { @@ -946,10 +951,14 @@ impl Worker for StreamableHttpClientWorker { enum Event { ClientMessage(WorkerSendRequest), ServerMessage(ServerJsonRpcMessage), - StreamResult(Result<(), StreamableHttpError>), + StreamResult { + request_id: Option, + result: Result<(), StreamableHttpError>, + }, } let mut streams = tokio::task::JoinSet::new(); let mut pending_stream_response_ids = HashSet::new(); + let mut request_stream_cancellations = HashMap::::new(); let mut awaiting_fallback_initialized = false; if let Some(session_id) = &session_id { Self::spawn_common_stream( @@ -985,7 +994,15 @@ impl Worker for StreamableHttpClientWorker { terminated_stream = streams.join_next(), if !streams.is_empty() => { match terminated_stream { Some(result) => { - Event::StreamResult(result.map_err(StreamableHttpError::TokioJoinError).and_then(std::convert::identity)) + match result { + Ok((request_id, result)) => { + Event::StreamResult { request_id, result } + } + Err(error) => Event::StreamResult { + request_id: None, + result: Err(StreamableHttpError::TokioJoinError(error)), + }, + } } None => { continue @@ -996,6 +1013,25 @@ impl Worker for StreamableHttpClientWorker { match event { Event::ClientMessage(send_request) => { let WorkerSendRequest { message, responder } = send_request; + let cancellation_request_id = match &message { + ClientJsonRpcMessage::Notification(notification) => { + match ¬ification.notification { + ClientNotification::CancelledNotification(cancelled) => { + cancelled.params.request_id.clone() + } + _ => None, + } + } + _ => None, + }; + if uses_modern_http && let Some(request_id) = cancellation_request_id { + if let Some(stream_ct) = request_stream_cancellations.remove(&request_id) { + stream_ct.cancel(); + } + pending_stream_response_ids.remove(&request_id); + let _ = responder.send(Ok(())); + continue; + } let is_fallback_initialize = saved_init_request.is_none() && matches!( &message, @@ -1016,6 +1052,7 @@ impl Worker for StreamableHttpClientWorker { && streams.is_empty(), "discover bootstrap must not create session state" ); + uses_modern_http = false; let response = self .client @@ -1222,6 +1259,7 @@ impl Worker for StreamableHttpClientWorker { Ok(()) } Ok(StreamableHttpPostResponse::Sse(stream, ..)) => { + let stream_request_id = request_id.clone(); Self::mark_stream_response_pending( &mut pending_stream_response_ids, request_id, @@ -1236,12 +1274,24 @@ impl Worker for StreamableHttpClientWorker { config.max_sse_event_size, self.config.retry_config.clone(), ); - streams.spawn(Self::execute_sse_stream( - sse_stream, - sse_worker_tx.clone(), - true, - transport_task_ct.child_token(), - )); + let stream_ct = transport_task_ct.child_token(); + if uses_modern_http + && let Some(request_id) = + stream_request_id.as_ref() + { + request_stream_cancellations.insert( + request_id.clone(), + stream_ct.clone(), + ); + } + let stream_tx = sse_worker_tx.clone(); + streams.spawn(async move { + let result = Self::execute_sse_stream( + sse_stream, stream_tx, true, stream_ct, + ) + .await; + (stream_request_id, result) + }); tracing::trace!("got new sse stream after re-init"); Ok(()) } @@ -1268,6 +1318,7 @@ impl Worker for StreamableHttpClientWorker { Ok(()) } Ok(StreamableHttpPostResponse::Sse(stream, ..)) => { + let stream_request_id = request_id.clone(); Self::mark_stream_response_pending( &mut pending_stream_response_ids, request_id, @@ -1282,12 +1333,20 @@ impl Worker for StreamableHttpClientWorker { config.max_sse_event_size, self.config.retry_config.clone(), ); - streams.spawn(Self::execute_sse_stream( - sse_stream, - sse_worker_tx.clone(), - true, - transport_task_ct.child_token(), - )); + let stream_ct = transport_task_ct.child_token(); + if uses_modern_http && let Some(request_id) = stream_request_id.as_ref() + { + request_stream_cancellations + .insert(request_id.clone(), stream_ct.clone()); + } + let stream_tx = sse_worker_tx.clone(); + streams.spawn(async move { + let result = Self::execute_sse_stream( + sse_stream, stream_tx, true, stream_ct, + ) + .await; + (stream_request_id, result) + }); tracing::trace!("got new sse stream"); Ok(()) } @@ -1312,6 +1371,11 @@ impl Worker for StreamableHttpClientWorker { let _ = responder.send(send_result); } Event::ServerMessage(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) + { + stream_ct.cancel(); + } Self::clear_stream_response_pending( &mut pending_stream_response_ids, &json_rpc_message, @@ -1322,7 +1386,26 @@ impl Worker for StreamableHttpClientWorker { break 'main_loop Err(e); } } - Event::StreamResult(result) => { + Event::StreamResult { request_id, result } => { + if let Some(request_id) = request_id { + Self::drain_queued_stream_messages( + &mut sse_worker_rx, + &mut context, + &mut pending_stream_response_ids, + ) + .await?; + request_stream_cancellations.remove(&request_id); + if pending_stream_response_ids.remove(&request_id) { + context + .send_to_handler(ServerJsonRpcMessage::error( + ErrorData::transport_closed( + "streamable HTTP response stream closed before its final response", + ), + Some(request_id), + )) + .await?; + } + } if result.is_err() { tracing::warn!( "sse client event stream terminated with error: {:?}", diff --git a/crates/rmcp/src/transport/streamable_http_server/tower.rs b/crates/rmcp/src/transport/streamable_http_server/tower.rs index 1994d554a..99b360903 100644 --- a/crates/rmcp/src/transport/streamable_http_server/tower.rs +++ b/crates/rmcp/src/transport/streamable_http_server/tower.rs @@ -22,7 +22,7 @@ use crate::{ ProtocolVersion, RequestId, ServerJsonRpcMessage, }, serve_server, - service::serve_directly, + service::{serve_directly, serve_directly_with_ct}, transport::{ OneshotTransport, TransportAdapterIdentity, common::{ @@ -242,6 +242,28 @@ fn message_has_per_request_protocol_version(message: &ClientJsonRpcMessage) -> b } } +fn headers_use_modern_http(headers: &HeaderMap) -> bool { + headers + .get(HEADER_MCP_PROTOCOL_VERSION) + .and_then(|value| value.to_str().ok()) + .is_some_and(|version| version >= ProtocolVersion::V_2026_07_28.as_str()) +} + +// Complete per-request identity metadata selects the discover lifecycle even +// when protocolVersion requests older application-level semantics. +fn message_uses_discover_lifecycle(message: &ClientJsonRpcMessage) -> bool { + matches!( + message, + ClientJsonRpcMessage::Request(request) + if !matches!(&request.request, ClientRequest::InitializeRequest(_)) + && request + .request + .get_meta() + .missing_required_keys(&ProtocolVersion::V_2026_07_28) + .is_empty() + ) +} + fn invalid_request_jsonrpc_response( id: Option, message: impl Into>, @@ -852,12 +874,14 @@ where request.request.extensions_mut().insert(parts); let (transport, mut receiver) = OneshotTransport::::new(ClientJsonRpcMessage::Request(request)); - let service = serve_directly(service, transport, peer_info); + let request_ct = self.config.cancellation_token.child_token(); + let service = + serve_directly_with_ct(service, transport, peer_info, request_ct.child_token()); tokio::spawn(async move { let _ = service.waiting().await; }); - let cancel = self.config.cancellation_token.child_token(); + let cancel = request_ct.child_token(); let first = tokio::select! { message = receiver.recv() => message, _ = cancel.cancelled() => None, @@ -869,7 +893,13 @@ where )) })?; - if self.config.json_response || jsonrpc_http_status(&first) != http::StatusCode::OK { + let terminal = matches!( + &first, + ServerJsonRpcMessage::Response(_) | ServerJsonRpcMessage::Error(_) + ); + if jsonrpc_http_status(&first) != http::StatusCode::OK + || (self.config.json_response && terminal) + { return jsonrpc_message_response(first, true); } @@ -882,7 +912,7 @@ where Ok(sse_stream_response( stream, self.config.sse_keep_alive, - self.config.cancellation_token.child_token(), + request_ct, )) } @@ -1127,15 +1157,16 @@ where return response; } let method = request.method().clone(); - let allowed_methods = match self.config.stateful_mode { + let modern_http = headers_use_modern_http(request.headers()); + let allowed_methods = match self.config.stateful_mode && !modern_http { true => "GET, POST, DELETE", false => "POST", }; let result = match (method, self.config.stateful_mode) { (Method::POST, _) => self.handle_post(request).await, // if we're not in stateful mode, we don't support GET or DELETE because there is no session - (Method::GET, true) => self.handle_get(request).await, - (Method::DELETE, true) => self.handle_delete(request).await, + (Method::GET, true) if !modern_http => self.handle_get(request).await, + (Method::DELETE, true) if !modern_http => self.handle_delete(request).await, _ => { // Handle other methods or return an error let response = Response::builder() @@ -1309,7 +1340,9 @@ where Err(response) => return Ok(response), }; - if self.config.stateful_mode { + let modern_http = + headers_use_modern_http(&part.headers) || message_uses_discover_lifecycle(&message); + if self.config.stateful_mode && !modern_http { // do we have a session id? let session_id = part .headers diff --git a/crates/rmcp/tests/test_message_schema/client_json_rpc_message_schema.json b/crates/rmcp/tests/test_message_schema/client_json_rpc_message_schema.json index 15e6e2945..922469c23 100644 --- a/crates/rmcp/tests/test_message_schema/client_json_rpc_message_schema.json +++ b/crates/rmcp/tests/test_message_schema/client_json_rpc_message_schema.json @@ -1160,20 +1160,23 @@ { "$ref": "#/definitions/Request9" }, + { + "$ref": "#/definitions/Request10" + }, { "$ref": "#/definitions/RequestOptionalParam4" }, { - "$ref": "#/definitions/Request10" + "$ref": "#/definitions/Request11" }, { "$ref": "#/definitions/RequestOptionalParam5" }, { - "$ref": "#/definitions/Request11" + "$ref": "#/definitions/Request12" }, { - "$ref": "#/definitions/Request12" + "$ref": "#/definitions/Request13" }, { "$ref": "#/definitions/CustomRequest" @@ -1567,6 +1570,22 @@ ] }, "Request10": { + "description": "Represents a JSON-RPC request with method, parameters, and extensions.\n\nThis is the core structure for all MCP requests, containing:\n- `method`: The name of the method being called\n- `params`: The parameters for the method\n- `extensions`: Additional context data (similar to HTTP headers)", + "type": "object", + "properties": { + "method": { + "$ref": "#/definitions/CallToolRequestMethod" + }, + "params": { + "$ref": "#/definitions/CallToolRequestParams" + } + }, + "required": [ + "method", + "params" + ] + }, + "Request11": { "description": "Represents a JSON-RPC request with method, parameters, and extensions.\n\nThis is the core structure for all MCP requests, containing:\n- `method`: The name of the method being called\n- `params`: The parameters for the method\n- `extensions`: Additional context data (similar to HTTP headers)", "type": "object", "properties": { @@ -1582,7 +1601,7 @@ "params" ] }, - "Request11": { + "Request12": { "description": "Represents a JSON-RPC request with method, parameters, and extensions.\n\nThis is the core structure for all MCP requests, containing:\n- `method`: The name of the method being called\n- `params`: The parameters for the method\n- `extensions`: Additional context data (similar to HTTP headers)", "type": "object", "properties": { @@ -1598,7 +1617,7 @@ "params" ] }, - "Request12": { + "Request13": { "description": "Represents a JSON-RPC request with method, parameters, and extensions.\n\nThis is the core structure for all MCP requests, containing:\n- `method`: The name of the method being called\n- `params`: The parameters for the method\n- `extensions`: Additional context data (similar to HTTP headers)", "type": "object", "properties": { @@ -1699,10 +1718,10 @@ "type": "object", "properties": { "method": { - "$ref": "#/definitions/SubscribeRequestMethod" + "$ref": "#/definitions/SubscriptionsListenRequestMethod" }, "params": { - "$ref": "#/definitions/SubscribeRequestParams" + "$ref": "#/definitions/SubscriptionsListenRequestParams" } }, "required": [ @@ -1715,10 +1734,10 @@ "type": "object", "properties": { "method": { - "$ref": "#/definitions/UnsubscribeRequestMethod" + "$ref": "#/definitions/SubscribeRequestMethod" }, "params": { - "$ref": "#/definitions/UnsubscribeRequestParams" + "$ref": "#/definitions/SubscribeRequestParams" } }, "required": [ @@ -1731,10 +1750,10 @@ "type": "object", "properties": { "method": { - "$ref": "#/definitions/CallToolRequestMethod" + "$ref": "#/definitions/UnsubscribeRequestMethod" }, "params": { - "$ref": "#/definitions/CallToolRequestParams" + "$ref": "#/definitions/UnsubscribeRequestParams" } }, "required": [ @@ -2303,6 +2322,77 @@ "uri" ] }, + "SubscriptionFilter": { + "description": "Notification categories a client opts in to on a `subscriptions/listen` stream.", + "type": "object", + "properties": { + "promptsListChanged": { + "type": "boolean" + }, + "resourceSubscriptions": { + "type": "array", + "items": { + "type": "string" + } + }, + "resourcesListChanged": { + "type": "boolean" + }, + "toolsListChanged": { + "type": "boolean" + } + } + }, + "SubscriptionsListenRequestMethod": { + "type": "string", + "format": "const", + "const": "subscriptions/listen" + }, + "SubscriptionsListenRequestParams": { + "description": "Parameters for opening a long-lived notification subscription.", + "type": "object", + "properties": { + "_meta": { + "description": "Protocol-level metadata. Required by the draft wire schema.", + "type": "object", + "properties": { + "io.modelcontextprotocol/clientCapabilities": { + "$ref": "#/definitions/ClientCapabilities" + }, + "io.modelcontextprotocol/clientInfo": { + "$ref": "#/definitions/Implementation" + }, + "io.modelcontextprotocol/logLevel": { + "$ref": "#/definitions/LoggingLevel" + }, + "io.modelcontextprotocol/protocolVersion": { + "type": "string" + }, + "progressToken": { + "$ref": "#/definitions/ProgressToken" + } + }, + "additionalProperties": true, + "required": [ + "io.modelcontextprotocol/protocolVersion", + "io.modelcontextprotocol/clientInfo", + "io.modelcontextprotocol/clientCapabilities" + ] + }, + "notifications": { + "description": "Notification categories requested for this stream.", + "allOf": [ + { + "$ref": "#/definitions/SubscriptionFilter" + } + ] + } + }, + "required": [ + "_meta", + "notifications" + ] + }, "TaskMetadata": { "description": "Metadata for augmenting a request with task execution (spec `TaskMetadata`).", "type": "object", diff --git a/crates/rmcp/tests/test_message_schema/client_json_rpc_message_schema_current.json b/crates/rmcp/tests/test_message_schema/client_json_rpc_message_schema_current.json index 15e6e2945..922469c23 100644 --- a/crates/rmcp/tests/test_message_schema/client_json_rpc_message_schema_current.json +++ b/crates/rmcp/tests/test_message_schema/client_json_rpc_message_schema_current.json @@ -1160,20 +1160,23 @@ { "$ref": "#/definitions/Request9" }, + { + "$ref": "#/definitions/Request10" + }, { "$ref": "#/definitions/RequestOptionalParam4" }, { - "$ref": "#/definitions/Request10" + "$ref": "#/definitions/Request11" }, { "$ref": "#/definitions/RequestOptionalParam5" }, { - "$ref": "#/definitions/Request11" + "$ref": "#/definitions/Request12" }, { - "$ref": "#/definitions/Request12" + "$ref": "#/definitions/Request13" }, { "$ref": "#/definitions/CustomRequest" @@ -1567,6 +1570,22 @@ ] }, "Request10": { + "description": "Represents a JSON-RPC request with method, parameters, and extensions.\n\nThis is the core structure for all MCP requests, containing:\n- `method`: The name of the method being called\n- `params`: The parameters for the method\n- `extensions`: Additional context data (similar to HTTP headers)", + "type": "object", + "properties": { + "method": { + "$ref": "#/definitions/CallToolRequestMethod" + }, + "params": { + "$ref": "#/definitions/CallToolRequestParams" + } + }, + "required": [ + "method", + "params" + ] + }, + "Request11": { "description": "Represents a JSON-RPC request with method, parameters, and extensions.\n\nThis is the core structure for all MCP requests, containing:\n- `method`: The name of the method being called\n- `params`: The parameters for the method\n- `extensions`: Additional context data (similar to HTTP headers)", "type": "object", "properties": { @@ -1582,7 +1601,7 @@ "params" ] }, - "Request11": { + "Request12": { "description": "Represents a JSON-RPC request with method, parameters, and extensions.\n\nThis is the core structure for all MCP requests, containing:\n- `method`: The name of the method being called\n- `params`: The parameters for the method\n- `extensions`: Additional context data (similar to HTTP headers)", "type": "object", "properties": { @@ -1598,7 +1617,7 @@ "params" ] }, - "Request12": { + "Request13": { "description": "Represents a JSON-RPC request with method, parameters, and extensions.\n\nThis is the core structure for all MCP requests, containing:\n- `method`: The name of the method being called\n- `params`: The parameters for the method\n- `extensions`: Additional context data (similar to HTTP headers)", "type": "object", "properties": { @@ -1699,10 +1718,10 @@ "type": "object", "properties": { "method": { - "$ref": "#/definitions/SubscribeRequestMethod" + "$ref": "#/definitions/SubscriptionsListenRequestMethod" }, "params": { - "$ref": "#/definitions/SubscribeRequestParams" + "$ref": "#/definitions/SubscriptionsListenRequestParams" } }, "required": [ @@ -1715,10 +1734,10 @@ "type": "object", "properties": { "method": { - "$ref": "#/definitions/UnsubscribeRequestMethod" + "$ref": "#/definitions/SubscribeRequestMethod" }, "params": { - "$ref": "#/definitions/UnsubscribeRequestParams" + "$ref": "#/definitions/SubscribeRequestParams" } }, "required": [ @@ -1731,10 +1750,10 @@ "type": "object", "properties": { "method": { - "$ref": "#/definitions/CallToolRequestMethod" + "$ref": "#/definitions/UnsubscribeRequestMethod" }, "params": { - "$ref": "#/definitions/CallToolRequestParams" + "$ref": "#/definitions/UnsubscribeRequestParams" } }, "required": [ @@ -2303,6 +2322,77 @@ "uri" ] }, + "SubscriptionFilter": { + "description": "Notification categories a client opts in to on a `subscriptions/listen` stream.", + "type": "object", + "properties": { + "promptsListChanged": { + "type": "boolean" + }, + "resourceSubscriptions": { + "type": "array", + "items": { + "type": "string" + } + }, + "resourcesListChanged": { + "type": "boolean" + }, + "toolsListChanged": { + "type": "boolean" + } + } + }, + "SubscriptionsListenRequestMethod": { + "type": "string", + "format": "const", + "const": "subscriptions/listen" + }, + "SubscriptionsListenRequestParams": { + "description": "Parameters for opening a long-lived notification subscription.", + "type": "object", + "properties": { + "_meta": { + "description": "Protocol-level metadata. Required by the draft wire schema.", + "type": "object", + "properties": { + "io.modelcontextprotocol/clientCapabilities": { + "$ref": "#/definitions/ClientCapabilities" + }, + "io.modelcontextprotocol/clientInfo": { + "$ref": "#/definitions/Implementation" + }, + "io.modelcontextprotocol/logLevel": { + "$ref": "#/definitions/LoggingLevel" + }, + "io.modelcontextprotocol/protocolVersion": { + "type": "string" + }, + "progressToken": { + "$ref": "#/definitions/ProgressToken" + } + }, + "additionalProperties": true, + "required": [ + "io.modelcontextprotocol/protocolVersion", + "io.modelcontextprotocol/clientInfo", + "io.modelcontextprotocol/clientCapabilities" + ] + }, + "notifications": { + "description": "Notification categories requested for this stream.", + "allOf": [ + { + "$ref": "#/definitions/SubscriptionFilter" + } + ] + } + }, + "required": [ + "_meta", + "notifications" + ] + }, "TaskMetadata": { "description": "Metadata for augmenting a request with task execution (spec `TaskMetadata`).", "type": "object", diff --git a/crates/rmcp/tests/test_message_schema/server_json_rpc_message_schema.json b/crates/rmcp/tests/test_message_schema/server_json_rpc_message_schema.json index 4cf0a0baf..520702420 100644 --- a/crates/rmcp/tests/test_message_schema/server_json_rpc_message_schema.json +++ b/crates/rmcp/tests/test_message_schema/server_json_rpc_message_schema.json @@ -1650,6 +1650,9 @@ { "$ref": "#/definitions/Notification5" }, + { + "$ref": "#/definitions/Notification6" + }, { "$ref": "#/definitions/CustomNotification" } @@ -2222,6 +2225,21 @@ ] }, "Notification5": { + "type": "object", + "properties": { + "method": { + "$ref": "#/definitions/SubscriptionsAcknowledgedNotificationMethod" + }, + "params": { + "$ref": "#/definitions/SubscriptionsAcknowledgedNotificationParams" + } + }, + "required": [ + "method", + "params" + ] + }, + "Notification6": { "type": "object", "properties": { "method": { @@ -3295,6 +3313,9 @@ { "$ref": "#/definitions/ReadResourceResult" }, + { + "$ref": "#/definitions/SubscriptionsListenResult" + }, { "$ref": "#/definitions/ListToolsResult" }, @@ -3438,6 +3459,75 @@ "format": "const", "const": "string" }, + "SubscriptionFilter": { + "description": "Notification categories a client opts in to on a `subscriptions/listen` stream.", + "type": "object", + "properties": { + "promptsListChanged": { + "type": "boolean" + }, + "resourceSubscriptions": { + "type": "array", + "items": { + "type": "string" + } + }, + "resourcesListChanged": { + "type": "boolean" + }, + "toolsListChanged": { + "type": "boolean" + } + } + }, + "SubscriptionsAcknowledgedNotificationMethod": { + "type": "string", + "format": "const", + "const": "notifications/subscriptions/acknowledged" + }, + "SubscriptionsAcknowledgedNotificationParams": { + "description": "Parameters reporting the accepted subset of a subscription filter.", + "type": "object", + "properties": { + "_meta": { + "$ref": "#/definitions/NotificationMetaObject" + }, + "notifications": { + "$ref": "#/definitions/SubscriptionFilter" + } + }, + "required": [ + "notifications" + ] + }, + "SubscriptionsListenResult": { + "description": "Final response indicating that a subscription ended gracefully.", + "type": "object", + "properties": { + "_meta": { + "$ref": "#/definitions/SubscriptionsListenResultMeta" + }, + "resultType": { + "$ref": "#/definitions/ResultType" + } + }, + "required": [ + "resultType", + "_meta" + ] + }, + "SubscriptionsListenResultMeta": { + "type": "object", + "properties": { + "io.modelcontextprotocol/subscriptionId": { + "$ref": "#/definitions/NumberOrString" + } + }, + "additionalProperties": true, + "required": [ + "io.modelcontextprotocol/subscriptionId" + ] + }, "Task": { "description": "Primary Task object that surfaces metadata during the task lifecycle.\n\nPer spec, `lastUpdatedAt` and `ttl` are required fields.\n`ttl` is nullable (`null` means unlimited retention).", "type": "object", diff --git a/crates/rmcp/tests/test_message_schema/server_json_rpc_message_schema_current.json b/crates/rmcp/tests/test_message_schema/server_json_rpc_message_schema_current.json index 4cf0a0baf..520702420 100644 --- a/crates/rmcp/tests/test_message_schema/server_json_rpc_message_schema_current.json +++ b/crates/rmcp/tests/test_message_schema/server_json_rpc_message_schema_current.json @@ -1650,6 +1650,9 @@ { "$ref": "#/definitions/Notification5" }, + { + "$ref": "#/definitions/Notification6" + }, { "$ref": "#/definitions/CustomNotification" } @@ -2222,6 +2225,21 @@ ] }, "Notification5": { + "type": "object", + "properties": { + "method": { + "$ref": "#/definitions/SubscriptionsAcknowledgedNotificationMethod" + }, + "params": { + "$ref": "#/definitions/SubscriptionsAcknowledgedNotificationParams" + } + }, + "required": [ + "method", + "params" + ] + }, + "Notification6": { "type": "object", "properties": { "method": { @@ -3295,6 +3313,9 @@ { "$ref": "#/definitions/ReadResourceResult" }, + { + "$ref": "#/definitions/SubscriptionsListenResult" + }, { "$ref": "#/definitions/ListToolsResult" }, @@ -3438,6 +3459,75 @@ "format": "const", "const": "string" }, + "SubscriptionFilter": { + "description": "Notification categories a client opts in to on a `subscriptions/listen` stream.", + "type": "object", + "properties": { + "promptsListChanged": { + "type": "boolean" + }, + "resourceSubscriptions": { + "type": "array", + "items": { + "type": "string" + } + }, + "resourcesListChanged": { + "type": "boolean" + }, + "toolsListChanged": { + "type": "boolean" + } + } + }, + "SubscriptionsAcknowledgedNotificationMethod": { + "type": "string", + "format": "const", + "const": "notifications/subscriptions/acknowledged" + }, + "SubscriptionsAcknowledgedNotificationParams": { + "description": "Parameters reporting the accepted subset of a subscription filter.", + "type": "object", + "properties": { + "_meta": { + "$ref": "#/definitions/NotificationMetaObject" + }, + "notifications": { + "$ref": "#/definitions/SubscriptionFilter" + } + }, + "required": [ + "notifications" + ] + }, + "SubscriptionsListenResult": { + "description": "Final response indicating that a subscription ended gracefully.", + "type": "object", + "properties": { + "_meta": { + "$ref": "#/definitions/SubscriptionsListenResultMeta" + }, + "resultType": { + "$ref": "#/definitions/ResultType" + } + }, + "required": [ + "resultType", + "_meta" + ] + }, + "SubscriptionsListenResultMeta": { + "type": "object", + "properties": { + "io.modelcontextprotocol/subscriptionId": { + "$ref": "#/definitions/NumberOrString" + } + }, + "additionalProperties": true, + "required": [ + "io.modelcontextprotocol/subscriptionId" + ] + }, "Task": { "description": "Primary Task object that surfaces metadata during the task lifecycle.\n\nPer spec, `lastUpdatedAt` and `ttl` are required fields.\n`ttl` is nullable (`null` means unlimited retention).", "type": "object", diff --git a/crates/rmcp/tests/test_mrtr_behavior.rs b/crates/rmcp/tests/test_mrtr_behavior.rs index b9087f706..3cb5da31b 100644 --- a/crates/rmcp/tests/test_mrtr_behavior.rs +++ b/crates/rmcp/tests/test_mrtr_behavior.rs @@ -281,9 +281,9 @@ fn client_info(protocol_version: ProtocolVersion) -> ClientInfo { .with_protocol_version(protocol_version) } -fn server_info_2026() -> ServerInfo { +fn server_info(protocol_version: ProtocolVersion) -> ServerInfo { let mut info = ServerInfo::new(ServerCapabilities::builder().enable_tools().build()); - info.protocol_version = ProtocolVersion::V_2026_07_28; + info.protocol_version = protocol_version; info } @@ -302,6 +302,7 @@ where .run_until(async move { let (server_transport, client_transport) = tokio::io::duplex(8192); let server_peer_info = client_info(client_protocol); + let client_peer_info = server_info(server_peer_info.protocol_version.clone()); let server_task = tokio::task::spawn_local(async move { let running = serve_directly::( server, @@ -315,7 +316,7 @@ where let client = serve_directly::( MrtrClient, client_transport, - Some(server_info_2026()), + Some(client_peer_info), ); let result = body(client).await; @@ -579,7 +580,7 @@ async fn request_state_codec_seals_and_verifies_through_the_loop() -> anyhow::Re let client = serve_directly::( MrtrClient, client_transport, - Some(server_info_2026()), + Some(server_info(ProtocolVersion::V_2026_07_28)), ); let result = client diff --git a/crates/rmcp/tests/test_notification.rs b/crates/rmcp/tests/test_notification.rs index 073396ee7..9aafd3ccd 100644 --- a/crates/rmcp/tests/test_notification.rs +++ b/crates/rmcp/tests/test_notification.rs @@ -1,4 +1,5 @@ #![cfg(not(feature = "local"))] +#![allow(deprecated)] use std::sync::Arc; use rmcp::{ diff --git a/crates/rmcp/tests/test_subscriptions.rs b/crates/rmcp/tests/test_subscriptions.rs new file mode 100644 index 000000000..3c19dea99 --- /dev/null +++ b/crates/rmcp/tests/test_subscriptions.rs @@ -0,0 +1,691 @@ +#![cfg(all( + not(feature = "local"), + feature = "client", + feature = "server", + feature = "transport-io" +))] + +use std::{ + num::NonZeroUsize, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, +}; + +use rmcp::{ + ClientHandler, ClientServiceExt, ServerHandler, ServiceExt, + model::{ + ClientNotification, ClientRequest, DiscoverResult, GetMeta, Implementation, + NotificationMetaObject, PromptListChangedNotification, ProtocolVersion, ServerCapabilities, + ServerInfo, ServerNotification, ServerResult, SubscriptionFilter, + SubscriptionsAcknowledgedNotification, SubscriptionsAcknowledgedNotificationParams, + SubscriptionsListenResult, + }, + service::{ + NotificationContext, RequestContext, RoleClient, RoleServer, SubscriptionContext, + SubscriptionEnd, SubscriptionSendError, SubscriptionSink, + }, +}; +use tokio::sync::{Mutex, Notify}; + +struct ToolsOnlyServer; + +#[derive(Clone)] +struct CountingClient { + tool_changes: Arc, +} + +impl ClientHandler for CountingClient { + async fn on_tool_list_changed(&self, _context: NotificationContext) { + self.tool_changes.fetch_add(1, Ordering::Relaxed); + } +} + +impl ServerHandler for ToolsOnlyServer { + fn get_info(&self) -> ServerInfo { + ServerInfo::new( + ServerCapabilities::builder() + .enable_tools() + .enable_tool_list_changed() + .build(), + ) + } + + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.supported_by(&self.get_info().capabilities)) + } + + async fn listen(&self, context: SubscriptionContext) -> Result<(), rmcp::ErrorData> { + context + .sink() + .notify_tool_list_changed() + .await + .expect("accepted tool notification"); + assert!(matches!( + context.sink().notify_prompt_list_changed().await, + Err(SubscriptionSendError::NotificationNotAccepted( + "notifications/prompts/list_changed" + )) + )); + Ok(()) + } +} + +struct ToolsAndPromptsServer; + +impl ServerHandler for ToolsAndPromptsServer { + fn get_info(&self) -> ServerInfo { + ServerInfo::new( + ServerCapabilities::builder() + .enable_tools() + .enable_tool_list_changed() + .enable_prompts() + .enable_prompts_list_changed() + .build(), + ) + } + + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.supported_by(&self.get_info().capabilities)) + } + + async fn listen(&self, context: SubscriptionContext) -> Result<(), rmcp::ErrorData> { + if context.accepted().tools_list_changed == Some(true) { + context + .sink() + .notify_tool_list_changed() + .await + .expect("send tool notification"); + } + if context.accepted().prompts_list_changed == Some(true) { + context + .sink() + .notify_prompt_list_changed() + .await + .expect("send prompt notification"); + } + Ok(()) + } +} + +struct ResourceSubscriptionServer; + +impl ServerHandler for ResourceSubscriptionServer { + fn get_info(&self) -> ServerInfo { + ServerInfo::new( + ServerCapabilities::builder() + .enable_resources() + .enable_resources_subscribe() + .build(), + ) + } + + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.supported_by(&self.get_info().capabilities)) + } + + async fn listen(&self, context: SubscriptionContext) -> Result<(), rmcp::ErrorData> { + context + .sink() + .notify_resource_updated("file:///accepted") + .await + .expect("accepted URI"); + assert!(matches!( + context + .sink() + .notify_resource_updated("file:///not-requested") + .await, + Err(SubscriptionSendError::NotificationNotAccepted( + "notifications/resources/updated" + )) + )); + Ok(()) + } +} + +struct RemoteCancellationServer; + +impl ServerHandler for RemoteCancellationServer { + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.clone()) + } + + async fn listen(&self, context: SubscriptionContext) -> Result<(), rmcp::ErrorData> { + context + .request_context() + .peer + .notify_cancelled(rmcp::model::CancelledNotificationParam::new( + Some(context.sink().id().clone()), + Some("server shutdown".to_owned()), + )) + .await + .expect("send server cancellation"); + std::future::pending().await + } +} + +struct FloodServer; + +impl ServerHandler for FloodServer { + fn get_info(&self) -> ServerInfo { + ServerInfo::new( + ServerCapabilities::builder() + .enable_tools() + .enable_tool_list_changed() + .build(), + ) + } + + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.clone()) + } + + async fn listen(&self, context: SubscriptionContext) -> Result<(), rmcp::ErrorData> { + for _ in 0..10 { + if context.sink().notify_tool_list_changed().await.is_err() { + break; + } + } + context.cancelled().await; + Ok(()) + } +} + +#[derive(Clone)] +struct ClosedSinkServer { + sink: Arc>>, +} + +struct LeakyServer; + +impl ServerHandler for LeakyServer { + fn get_info(&self) -> ServerInfo { + ServerInfo::new( + ServerCapabilities::builder() + .enable_tools() + .enable_tool_list_changed() + .build(), + ) + } + + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.supported_by(&self.get_info().capabilities)) + } + + async fn listen(&self, context: SubscriptionContext) -> Result<(), rmcp::ErrorData> { + let mut notification = + ServerNotification::PromptListChangedNotification(PromptListChangedNotification { + method: Default::default(), + extensions: Default::default(), + }); + notification + .get_meta_mut() + .set_subscription_id(context.sink().id().clone()); + context + .request_context() + .peer + .send_notification(notification) + .await + .expect("send deliberately invalid notification"); + std::future::pending().await + } +} + +struct MalformedAcknowledgmentServer { + cancelled: Arc, +} + +impl rmcp::service::Service for MalformedAcknowledgmentServer { + async fn handle_request( + &self, + request: ClientRequest, + context: RequestContext, + ) -> Result { + match request { + ClientRequest::DiscoverRequest(_) => { + Ok(ServerResult::DiscoverResult(DiscoverResult::new( + vec![ProtocolVersion::V_2026_07_28], + ServerCapabilities::builder() + .enable_tools() + .enable_tool_list_changed() + .enable_prompts() + .enable_prompts_list_changed() + .build(), + Implementation::new("malformed-ack-server", "1.0.0"), + ))) + } + ClientRequest::SubscriptionsListenRequest(_) => { + let mut acknowledgment = SubscriptionsAcknowledgedNotification::new( + SubscriptionsAcknowledgedNotificationParams::new( + SubscriptionFilter::builder().prompts_list_changed().build(), + ), + ); + let mut meta = NotificationMetaObject::new(); + meta.set_subscription_id(context.id.clone()); + acknowledgment.extensions.insert(meta); + context + .peer + .send_notification(ServerNotification::SubscriptionsAcknowledgedNotification( + acknowledgment, + )) + .await + .map_err(|error| rmcp::ErrorData::internal_error(error.to_string(), None))?; + context.ct.cancelled().await; + Ok(ServerResult::SubscriptionsListenResult( + SubscriptionsListenResult::complete(context.id), + )) + } + _ => Err(rmcp::ErrorData::invalid_request( + "unexpected test request", + None, + )), + } + } + + async fn handle_notification( + &self, + notification: ClientNotification, + _context: NotificationContext, + ) -> Result<(), rmcp::ErrorData> { + if matches!(notification, ClientNotification::CancelledNotification(_)) { + self.cancelled.notify_one(); + } + Ok(()) + } + + fn get_info(&self) -> rmcp::model::ServerInfo { + ServerInfo::default() + } +} + +impl ServerHandler for ClosedSinkServer { + fn get_info(&self) -> ServerInfo { + ServerInfo::new( + ServerCapabilities::builder() + .enable_tools() + .enable_tool_list_changed() + .build(), + ) + } + + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.supported_by(&self.get_info().capabilities)) + } + + async fn listen(&self, context: SubscriptionContext) -> Result<(), rmcp::ErrorData> { + self.sink.lock().await.replace(context.sink().clone()); + Ok(()) + } +} + +#[derive(Clone)] +struct CancellationServer { + cancelled: Arc, +} + +impl ServerHandler for CancellationServer { + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.clone()) + } + + async fn listen(&self, context: SubscriptionContext) -> Result<(), rmcp::ErrorData> { + context.cancelled().await; + self.cancelled.notify_one(); + Ok(()) + } +} + +#[derive(Clone)] +struct AbruptServer { + started: Arc, +} + +impl ServerHandler for AbruptServer { + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.clone()) + } + + async fn listen(&self, _context: SubscriptionContext) -> Result<(), rmcp::ErrorData> { + self.started.notify_one(); + std::future::pending().await + } +} + +async fn modern_client( + server: S, +) -> anyhow::Result> { + let (server_transport, client_transport) = tokio::io::duplex(16 * 1024); + tokio::spawn(async move { + let server = server.serve(server_transport).await?; + server.waiting().await?; + anyhow::Ok(()) + }); + ().serve_with_lifecycle( + client_transport, + rmcp::ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + }, + ) + .await + .map_err(Into::into) +} + +#[tokio::test] +async fn listen_exposes_acknowledged_filter_and_graceful_result() -> anyhow::Result<()> { + let client = modern_client(ToolsOnlyServer).await?; + let mut subscription = client + .listen( + SubscriptionFilter::builder() + .tools_list_changed() + .prompts_list_changed() + .build(), + ) + .await?; + + assert_eq!( + subscription.acknowledged(), + &SubscriptionFilter::builder().tools_list_changed().build() + ); + + let notification = tokio::time::timeout(Duration::from_secs(5), subscription.next()) + .await?? + .expect("tool notification"); + assert!(matches!( + notification, + ServerNotification::ToolListChangedNotification(_) + )); + assert_eq!( + notification.get_meta().subscription_id(), + Some(subscription.id().clone()) + ); + + assert!( + tokio::time::timeout(Duration::from_secs(5), subscription.next()) + .await?? + .is_none() + ); + let Some(SubscriptionEnd::Graceful(result)) = subscription.end() else { + panic!("expected graceful final result"); + }; + assert_eq!( + result.meta.subscription_id().as_ref(), + Some(subscription.id()) + ); + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn typed_subscription_notifications_do_not_reach_handler_callbacks() -> anyhow::Result<()> { + let (server_transport, client_transport) = tokio::io::duplex(16 * 1024); + tokio::spawn(async move { + let server = ToolsOnlyServer.serve(server_transport).await?; + server.waiting().await?; + anyhow::Ok(()) + }); + let tool_changes = Arc::new(AtomicUsize::new(0)); + let client = CountingClient { + tool_changes: tool_changes.clone(), + } + .serve_with_lifecycle( + client_transport, + rmcp::ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + }, + ) + .await?; + let mut subscription = client + .listen(SubscriptionFilter::builder().tools_list_changed().build()) + .await?; + + assert!(subscription.next().await?.is_some()); + assert!(subscription.next().await?.is_none()); + tokio::task::yield_now().await; + assert_eq!(tool_changes.load(Ordering::Relaxed), 0); + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn concurrent_subscriptions_are_demultiplexed_by_request_id() -> anyhow::Result<()> { + let client = modern_client(ToolsAndPromptsServer).await?; + let (tools, prompts) = tokio::join!( + client.listen(SubscriptionFilter::builder().tools_list_changed().build()), + client.listen(SubscriptionFilter::builder().prompts_list_changed().build()) + ); + let mut tools = tools?; + let mut prompts = prompts?; + + let tool_notification = tools.next().await?.expect("tool notification"); + let prompt_notification = prompts.next().await?.expect("prompt notification"); + assert!(matches!( + tool_notification, + ServerNotification::ToolListChangedNotification(_) + )); + assert!(matches!( + prompt_notification, + ServerNotification::PromptListChangedNotification(_) + )); + assert_ne!(tools.id(), prompts.id()); + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn stdio_cancellation_sends_cancelled_for_the_listen_request() -> anyhow::Result<()> { + let cancelled = Arc::new(Notify::new()); + let client = modern_client(CancellationServer { + cancelled: cancelled.clone(), + }) + .await?; + let mut subscription = client.listen(SubscriptionFilter::new()).await?; + + subscription.cancel().await?; + assert!(matches!( + subscription.end(), + Some(SubscriptionEnd::Cancelled) + )); + tokio::time::timeout(Duration::from_secs(5), cancelled.notified()).await?; + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn resource_updates_are_filtered_by_exact_uri_membership() -> anyhow::Result<()> { + let client = modern_client(ResourceSubscriptionServer).await?; + let mut subscription = client + .listen( + SubscriptionFilter::builder() + .resource_subscription("file:///accepted") + .build(), + ) + .await?; + + let notification = subscription.next().await?.expect("resource update"); + let ServerNotification::ResourceUpdatedNotification(update) = notification else { + panic!("expected resource update"); + }; + assert_eq!(update.params.uri, "file:///accepted"); + assert!(subscription.next().await?.is_none()); + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn stdio_server_cancellation_ends_the_matching_subscription() -> anyhow::Result<()> { + let client = modern_client(RemoteCancellationServer).await?; + let mut subscription = client.listen(SubscriptionFilter::new()).await?; + + assert!(subscription.next().await?.is_none()); + assert!(matches!( + subscription.end(), + Some(SubscriptionEnd::Cancelled) + )); + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn slow_consumer_reports_subscription_lag() -> anyhow::Result<()> { + let client = modern_client(FloodServer).await?; + let mut subscription = client + .listen_with_capacity( + SubscriptionFilter::builder().tools_list_changed().build(), + NonZeroUsize::MIN, + ) + .await?; + + tokio::time::sleep(Duration::from_millis(50)).await; + while subscription.next().await?.is_some() {} + assert!( + matches!( + subscription.end(), + Some(SubscriptionEnd::Lagged { capacity: 1 }) + ), + "unexpected end: {:?}", + subscription.end() + ); + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn sink_rejects_notifications_after_graceful_completion() -> anyhow::Result<()> { + let sink = Arc::new(Mutex::new(None)); + let client = modern_client(ClosedSinkServer { sink: sink.clone() }).await?; + let mut subscription = client + .listen(SubscriptionFilter::builder().tools_list_changed().build()) + .await?; + assert!(subscription.next().await?.is_none()); + + let sink = sink.lock().await.clone().expect("captured sink"); + assert!(matches!( + sink.notify_tool_list_changed().await, + Err(SubscriptionSendError::SubscriptionClosed) + )); + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn client_rejects_notifications_outside_the_acknowledged_filter() -> anyhow::Result<()> { + let client = modern_client(LeakyServer).await?; + let mut subscription = client + .listen(SubscriptionFilter::builder().tools_list_changed().build()) + .await?; + + assert!(matches!( + subscription.next().await, + Err(rmcp::ServiceError::UnexpectedResponse) + )); + assert!(matches!(subscription.end(), Some(SubscriptionEnd::Abrupt))); + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn malformed_acknowledgment_cancels_pending_listen_request() -> anyhow::Result<()> { + let (server_transport, client_transport) = tokio::io::duplex(16 * 1024); + let cancelled = Arc::new(Notify::new()); + let server_cancelled = cancelled.clone(); + tokio::spawn(async move { + let server = MalformedAcknowledgmentServer { + cancelled: server_cancelled, + } + .serve(server_transport) + .await?; + server.waiting().await?; + anyhow::Ok(()) + }); + let client = () + .serve_with_lifecycle( + client_transport, + rmcp::ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + }, + ) + .await?; + + assert!(matches!( + client + .listen(SubscriptionFilter::builder().tools_list_changed().build()) + .await, + Err(rmcp::ServiceError::UnexpectedResponse) + )); + tokio::time::timeout(Duration::from_secs(5), cancelled.notified()).await?; + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn transport_close_without_final_result_is_abrupt() -> anyhow::Result<()> { + let (server_transport, client_transport) = tokio::io::duplex(16 * 1024); + let started = Arc::new(Notify::new()); + let server_started = started.clone(); + tokio::spawn(async move { + let server = AbruptServer { + started: server_started.clone(), + } + .serve(server_transport) + .await?; + server_started.notified().await; + server.cancel().await?; + anyhow::Ok(()) + }); + let client = () + .serve_with_lifecycle( + client_transport, + rmcp::ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + }, + ) + .await?; + let mut subscription = client.listen(SubscriptionFilter::new()).await?; + + assert!( + tokio::time::timeout(Duration::from_secs(5), subscription.next()) + .await?? + .is_none() + ); + assert!(matches!(subscription.end(), Some(SubscriptionEnd::Abrupt))); + Ok(()) +} diff --git a/crates/rmcp/tests/test_subscriptions_model.rs b/crates/rmcp/tests/test_subscriptions_model.rs new file mode 100644 index 000000000..b0c8e51c3 --- /dev/null +++ b/crates/rmcp/tests/test_subscriptions_model.rs @@ -0,0 +1,237 @@ +use rmcp::model::{ + ClientJsonRpcMessage, ClientRequest, GetMeta, JsonRpcNotification, JsonRpcRequest, + NotificationMetaObject, RequestId, RequestMetaObject, ServerJsonRpcMessage, ServerNotification, + SubscriptionFilter, SubscriptionsAcknowledgedNotification, + SubscriptionsAcknowledgedNotificationParams, SubscriptionsListenRequest, + SubscriptionsListenRequestParams, SubscriptionsListenResult, SubscriptionsListenResultMeta, +}; +use serde_json::json; + +#[test] +fn subscription_filter_serializes_only_opted_in_notifications() { + let filter = SubscriptionFilter::builder() + .tools_list_changed() + .resource_subscription("file:///one") + .resource_subscription("file:///two") + .build(); + + assert_eq!( + serde_json::to_value(filter).expect("serialize filter"), + json!({ + "toolsListChanged": true, + "resourceSubscriptions": ["file:///one", "file:///two"], + }) + ); +} + +#[test] +fn subscription_filter_subset_is_order_independent_and_ignores_false_flags() { + let requested = SubscriptionFilter::builder() + .tools_list_changed() + .resource_subscriptions(["file:///one", "file:///two"]) + .build(); + let mut accepted = SubscriptionFilter::builder() + .resource_subscriptions(["file:///two", "file:///one"]) + .build(); + accepted.tools_list_changed = Some(false); + + assert!(accepted.is_subset_of(&requested)); +} + +#[test] +fn subscription_filter_omits_empty_resource_intersection() { + let requested = SubscriptionFilter::builder() + .resource_subscription("file:///requested") + .build(); + let accepted = SubscriptionFilter::builder() + .resource_subscription("file:///different") + .build(); + + assert_eq!( + serde_json::to_value(requested.intersection(&accepted)).expect("serialize intersection"), + json!({}) + ); +} + +#[test] +fn listen_request_round_trips_required_fields_and_arbitrary_metadata() { + let mut request = SubscriptionsListenRequest::new(SubscriptionsListenRequestParams::new( + SubscriptionFilter::builder().prompts_list_changed().build(), + )); + let mut meta = RequestMetaObject::new(); + meta.insert("com.example/request".into(), json!("value")); + request.extensions.insert(meta); + let message = ClientJsonRpcMessage::request( + ClientRequest::SubscriptionsListenRequest(request), + RequestId::String("subscription-1".into()), + ); + + let value = serde_json::to_value(&message).expect("serialize listen request"); + assert_eq!( + value, + json!({ + "jsonrpc": "2.0", + "id": "subscription-1", + "method": "subscriptions/listen", + "params": { + "_meta": { + "com.example/request": "value", + }, + "notifications": { + "promptsListChanged": true, + }, + }, + }) + ); + + let round_trip: ClientJsonRpcMessage = + serde_json::from_value(value).expect("deserialize listen request"); + let ClientJsonRpcMessage::Request(JsonRpcRequest { request, .. }) = round_trip else { + panic!("expected request"); + }; + let ClientRequest::SubscriptionsListenRequest(request) = request else { + panic!("expected subscriptions/listen request"); + }; + assert_eq!( + request + .extensions + .get::() + .and_then(|meta| meta.get("com.example/request")), + Some(&json!("value")) + ); +} + +#[test] +fn acknowledged_notification_round_trips_numeric_subscription_id_and_metadata() { + let mut notification = SubscriptionsAcknowledgedNotification::new( + SubscriptionsAcknowledgedNotificationParams::new( + SubscriptionFilter::builder() + .resources_list_changed() + .build(), + ), + ); + let mut meta = NotificationMetaObject::new(); + meta.set_subscription_id(RequestId::Number(7)); + meta.insert("com.example/notification".into(), json!(true)); + notification.extensions.insert(meta); + let message = ServerJsonRpcMessage::notification( + ServerNotification::SubscriptionsAcknowledgedNotification(notification), + ); + + let value = serde_json::to_value(&message).expect("serialize acknowledgment"); + assert_eq!( + value, + json!({ + "jsonrpc": "2.0", + "method": "notifications/subscriptions/acknowledged", + "params": { + "_meta": { + "io.modelcontextprotocol/subscriptionId": 7, + "com.example/notification": true, + }, + "notifications": { + "resourcesListChanged": true, + }, + }, + }) + ); + + let round_trip: ServerJsonRpcMessage = + serde_json::from_value(value).expect("deserialize acknowledgment"); + let ServerJsonRpcMessage::Notification(JsonRpcNotification { notification, .. }) = round_trip + else { + panic!("expected notification"); + }; + assert_eq!( + notification.get_meta().subscription_id(), + Some(RequestId::Number(7)) + ); + assert_eq!( + notification.get_meta().get("com.example/notification"), + Some(&json!(true)) + ); +} + +#[test] +fn listen_result_requires_matching_string_subscription_id_and_preserves_metadata() { + let mut meta = SubscriptionsListenResultMeta::new(RequestId::String("subscription-2".into())); + meta.insert("com.example/result".into(), json!({ "reason": "shutdown" })); + let result = SubscriptionsListenResult::new(meta); + + let value = serde_json::to_value(&result).expect("serialize listen result"); + assert_eq!( + value, + json!({ + "resultType": "complete", + "_meta": { + "io.modelcontextprotocol/subscriptionId": "subscription-2", + "com.example/result": { + "reason": "shutdown", + }, + }, + }) + ); + + let round_trip: SubscriptionsListenResult = + serde_json::from_value(value).expect("deserialize listen result"); + assert_eq!( + round_trip.meta.subscription_id(), + Some(RequestId::String("subscription-2".into())) + ); + assert_eq!( + round_trip.meta.get("com.example/result"), + Some(&json!({ "reason": "shutdown" })) + ); +} + +#[test] +fn listen_result_meta_returns_none_after_required_id_is_removed() { + let mut meta = SubscriptionsListenResultMeta::new(RequestId::Number(1)); + meta.remove("io.modelcontextprotocol/subscriptionId"); + + assert_eq!(meta.subscription_id(), None); +} + +#[cfg(feature = "schemars")] +#[test] +fn subscription_schemas_mark_only_draft_required_fields_as_required() { + let request_schema = + serde_json::to_value(schemars::schema_for!(SubscriptionsListenRequestParams)) + .expect("request schema"); + let filter_schema = + serde_json::to_value(schemars::schema_for!(SubscriptionFilter)).expect("filter schema"); + let acknowledgment_schema = serde_json::to_value(schemars::schema_for!( + SubscriptionsAcknowledgedNotificationParams + )) + .expect("acknowledgment schema"); + let result_schema = serde_json::to_value(schemars::schema_for!(SubscriptionsListenResult)) + .expect("result schema"); + + assert_eq!( + request_schema["required"], + json!(["_meta", "notifications"]) + ); + assert_eq!( + request_schema["properties"]["_meta"]["required"], + json!([ + "io.modelcontextprotocol/protocolVersion", + "io.modelcontextprotocol/clientInfo", + "io.modelcontextprotocol/clientCapabilities" + ]) + ); + assert!(filter_schema.get("required").is_none()); + assert_eq!( + filter_schema["properties"]["toolsListChanged"]["type"], + "boolean" + ); + assert_eq!( + filter_schema["properties"]["resourceSubscriptions"]["type"], + "array" + ); + assert_eq!(acknowledgment_schema["required"], json!(["notifications"])); + assert_eq!( + acknowledgment_schema["properties"]["_meta"]["$ref"], + "#/$defs/NotificationMetaObject" + ); + assert_eq!(result_schema["required"], json!(["resultType", "_meta"])); +} diff --git a/crates/rmcp/tests/test_subscriptions_streamable_http.rs b/crates/rmcp/tests/test_subscriptions_streamable_http.rs new file mode 100644 index 000000000..f9de65f3b --- /dev/null +++ b/crates/rmcp/tests/test_subscriptions_streamable_http.rs @@ -0,0 +1,310 @@ +#![cfg(all( + not(feature = "local"), + feature = "client", + feature = "server", + feature = "transport-streamable-http-client-reqwest", + feature = "transport-streamable-http-server" +))] + +use std::{ + borrow::Cow, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, +}; + +use rmcp::{ + ClientLifecycleMode, ClientServiceExt, ServerHandler, + model::{ + ClientInfo, ClientRequest, ListToolsRequest, ProtocolVersion, RequestMetaObject, + ServerCapabilities, ServerInfo, ServerNotification, SubscriptionFilter, + }, + service::{PeerRequestOptions, SubscriptionContext, SubscriptionEnd}, + transport::{ + StreamableHttpClientTransport, + streamable_http_client::StreamableHttpClientTransportConfig, + streamable_http_server::{ + StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager, + }, + }, +}; +use tokio::sync::Notify; +use tokio_util::sync::CancellationToken; + +#[derive(Clone)] +struct HttpSubscriptionServer { + cancelled: Arc, + started: Arc, + ending: ServerEnding, +} + +#[derive(Clone, Copy)] +enum ServerEnding { + ClientCancellation, + Graceful, + Abrupt, +} + +impl ServerHandler for HttpSubscriptionServer { + fn supported_protocol_versions(&self) -> Cow<'static, [ProtocolVersion]> { + Cow::Borrowed(&[ProtocolVersion::V_2026_07_28, ProtocolVersion::V_2025_11_25]) + } + + fn get_info(&self) -> ServerInfo { + ServerInfo::new( + ServerCapabilities::builder() + .enable_tools() + .enable_tool_list_changed() + .build(), + ) + } + + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.supported_by(&self.get_info().capabilities)) + } + + async fn listen(&self, context: SubscriptionContext) -> Result<(), rmcp::ErrorData> { + context + .sink() + .notify_tool_list_changed() + .await + .expect("send tool notification"); + self.started.notify_one(); + match self.ending { + ServerEnding::Graceful => Ok(()), + ServerEnding::ClientCancellation => { + context.cancelled().await; + self.cancelled.notify_one(); + Ok(()) + } + ServerEnding::Abrupt => std::future::pending().await, + } + } +} + +async fn spawn_server( + ending: ServerEnding, +) -> ( + String, + CancellationToken, + Arc, + Arc, + Arc, +) { + let cancellation_token = CancellationToken::new(); + let subscription_cancelled = Arc::new(Notify::new()); + let subscription_started = Arc::new(Notify::new()); + let server = HttpSubscriptionServer { + cancelled: subscription_cancelled.clone(), + started: subscription_started.clone(), + ending, + }; + let service: StreamableHttpService = + StreamableHttpService::new( + move || Ok(server.clone()), + Default::default(), + StreamableHttpServerConfig::default() + .with_stateful_mode(true) + .with_json_response(true) + .with_sse_keep_alive(Some(Duration::from_millis(50))) + .with_cancellation_token(cancellation_token.child_token()), + ); + let get_requests = Arc::new(AtomicUsize::new(0)); + let observed_get_requests = get_requests.clone(); + let router = + axum::Router::new() + .nest_service("/mcp", service) + .layer(axum::middleware::from_fn( + move |request: axum::extract::Request, next: axum::middleware::Next| { + let observed_get_requests = observed_get_requests.clone(); + async move { + if request.method() == axum::http::Method::GET { + observed_get_requests.fetch_add(1, Ordering::Relaxed); + } + next.run(request).await + } + }, + )); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("listener"); + let address = listener.local_addr().expect("listener address"); + tokio::spawn({ + let cancellation_token = cancellation_token.clone(); + async move { + let _ = axum::serve(listener, router) + .with_graceful_shutdown(async move { cancellation_token.cancelled_owned().await }) + .await; + } + }); + ( + format!("http://{address}/mcp"), + cancellation_token, + subscription_cancelled, + subscription_started, + get_requests, + ) +} + +#[tokio::test] +async fn modern_http_listen_uses_post_stream_and_cancels_by_closing_it() -> anyhow::Result<()> { + let (url, server_ct, subscription_cancelled, _, get_requests) = + spawn_server(ServerEnding::ClientCancellation).await; + let transport = StreamableHttpClientTransport::from_config( + StreamableHttpClientTransportConfig::with_uri(url.clone()), + ); + let client = ClientInfo::default() + .serve_with_lifecycle( + transport, + ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + }, + ) + .await?; + assert_eq!(get_requests.load(Ordering::Relaxed), 0); + let mut subscription = client + .listen(SubscriptionFilter::builder().tools_list_changed().build()) + .await?; + + assert!(matches!( + subscription.next().await?.expect("tool notification"), + ServerNotification::ToolListChangedNotification(_) + )); + let mut older_version = RequestMetaObject::new(); + older_version.set_protocol_version(ProtocolVersion::V_2025_11_25); + client + .send_request_with_option( + ClientRequest::ListToolsRequest(ListToolsRequest { + method: Default::default(), + params: None, + extensions: Default::default(), + }), + PeerRequestOptions::no_options().with_meta(older_version), + ) + .await? + .await_response() + .await?; + subscription.cancel().await?; + tokio::time::timeout(Duration::from_secs(5), subscription_cancelled.notified()).await?; + + client.cancel().await?; + server_ct.cancel(); + Ok(()) +} + +#[tokio::test] +async fn modern_http_graceful_close_returns_final_listen_result() -> anyhow::Result<()> { + let (url, server_ct, _, _, _) = spawn_server(ServerEnding::Graceful).await; + let transport = StreamableHttpClientTransport::from_config( + StreamableHttpClientTransportConfig::with_uri(url), + ); + let client = ClientInfo::default() + .serve_with_lifecycle( + transport, + ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + }, + ) + .await?; + let mut subscription = client + .listen(SubscriptionFilter::builder().tools_list_changed().build()) + .await?; + + assert!(subscription.next().await?.is_some()); + assert!(subscription.next().await?.is_none()); + assert!(matches!( + subscription.end(), + Some(SubscriptionEnd::Graceful(_)) + )); + + client.cancel().await?; + server_ct.cancel(); + Ok(()) +} + +#[tokio::test] +async fn modern_http_stream_close_without_result_is_abrupt() -> anyhow::Result<()> { + let (url, server_ct, _, subscription_started, _) = spawn_server(ServerEnding::Abrupt).await; + let transport = StreamableHttpClientTransport::from_config( + StreamableHttpClientTransportConfig::with_uri(url), + ); + let client = ClientInfo::default() + .serve_with_lifecycle( + transport, + ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + }, + ) + .await?; + let mut subscription = client + .listen(SubscriptionFilter::builder().tools_list_changed().build()) + .await?; + assert!(subscription.next().await?.is_some()); + subscription_started.notified().await; + + server_ct.cancel(); + assert!( + tokio::time::timeout(Duration::from_secs(5), subscription.next()) + .await?? + .is_none() + ); + assert!(matches!(subscription.end(), Some(SubscriptionEnd::Abrupt))); + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn modern_http_lifecycle_stays_sessionless_for_older_application_version() +-> anyhow::Result<()> { + let (url, server_ct, _, _, get_requests) = spawn_server(ServerEnding::ClientCancellation).await; + let transport = StreamableHttpClientTransport::from_config( + StreamableHttpClientTransportConfig::with_uri(url), + ); + let client = ClientInfo::default() + .serve_with_lifecycle( + transport, + ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2025_11_25], + }, + ) + .await?; + + client.list_tools(None).await?; + assert_eq!(get_requests.load(Ordering::Relaxed), 0); + + client.cancel().await?; + server_ct.cancel(); + Ok(()) +} + +#[tokio::test] +async fn modern_http_get_and_delete_are_method_not_allowed_in_stateful_mode() { + let (url, server_ct, _, _, _) = spawn_server(ServerEnding::ClientCancellation).await; + let client = reqwest::Client::new(); + + for method in [reqwest::Method::GET, reqwest::Method::DELETE] { + let response = client + .request(method, &url) + .header("Accept", "text/event-stream") + .header("MCP-Protocol-Version", "2026-07-28") + .send() + .await + .expect("request"); + assert_eq!(response.status(), reqwest::StatusCode::METHOD_NOT_ALLOWED); + assert_eq!( + response + .headers() + .get(reqwest::header::ALLOW) + .and_then(|value| value.to_str().ok()), + Some("POST") + ); + } + + server_ct.cancel(); +} diff --git a/examples/clients/Cargo.toml b/examples/clients/Cargo.toml index 416057ac8..419a176f9 100644 --- a/examples/clients/Cargo.toml +++ b/examples/clients/Cargo.toml @@ -65,3 +65,7 @@ path = "src/auth/client_credentials.rs" [[example]] name = "clients_task_stdio" path = "src/task_stdio.rs" + +[[example]] +name = "clients_subscriptions_streamhttp" +path = "src/subscriptions_streamhttp.rs" diff --git a/examples/clients/README.md b/examples/clients/README.md index 419cdac22..f082a4926 100644 --- a/examples/clients/README.md +++ b/examples/clients/README.md @@ -16,6 +16,13 @@ A client that communicates with a Git-related MCP server using standard input/ou A client that communicates with an MCP server using HTTP streaming transport. - Connects to an MCP server running at `http://localhost:8000` + +### Modern Subscription Client (`subscriptions_streamhttp.rs`) + +Uses modern discovery and `subscriptions/listen`, prints the accepted filter, +and consumes tagged notifications until graceful closure or cancellation. + +- Run with `cargo run -p mcp-client-examples --example clients_subscriptions_streamhttp` - Retrieves server information and list of available tools - Calls a tool named "increment" diff --git a/examples/clients/src/subscriptions_streamhttp.rs b/examples/clients/src/subscriptions_streamhttp.rs new file mode 100644 index 000000000..7ab612bb9 --- /dev/null +++ b/examples/clients/src/subscriptions_streamhttp.rs @@ -0,0 +1,49 @@ +use rmcp::{ + ClientLifecycleMode, ClientServiceExt, + model::{ClientInfo, ProtocolVersion, SubscriptionFilter}, + transport::StreamableHttpClientTransport, +}; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()), + ) + .init(); + + let transport = StreamableHttpClientTransport::from_uri("http://127.0.0.1:8000/mcp"); + let client = ClientInfo::default() + .serve_with_lifecycle( + transport, + ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + }, + ) + .await?; + let mut subscription = client + .listen(SubscriptionFilter::builder().tools_list_changed().build()) + .await?; + + println!("accepted filter: {:?}", subscription.acknowledged()); + loop { + tokio::select! { + result = subscription.next() => { + match result? { + Some(notification) => println!("notification: {notification:?}"), + None => { + println!("subscription ended: {:?}", subscription.end()); + break; + } + } + } + _ = tokio::signal::ctrl_c() => { + subscription.cancel().await?; + break; + } + } + } + + client.cancel().await?; + Ok(()) +} diff --git a/examples/servers/Cargo.toml b/examples/servers/Cargo.toml index f189c9e7f..bc983a154 100644 --- a/examples/servers/Cargo.toml +++ b/examples/servers/Cargo.toml @@ -118,3 +118,7 @@ path = "src/task_stdio.rs" [[example]] name = "servers_mrtr" path = "src/mrtr.rs" + +[[example]] +name = "servers_subscriptions_streamhttp" +path = "src/subscriptions_streamhttp.rs" diff --git a/examples/servers/README.md b/examples/servers/README.md index a6f0dcf76..5fa5011ec 100644 --- a/examples/servers/README.md +++ b/examples/servers/README.md @@ -27,6 +27,13 @@ A server using streamable HTTP transport for MCP communication, with axum. - Provides counter tools via HTTP streaming - Demonstrates streamable HTTP transport configuration +### Modern Subscription Server (`subscriptions_streamhttp.rs`) + +A stateless `2026-07-28` server that opens `subscriptions/listen` response +streams, acknowledges the accepted filter, and emits tagged tool-list changes. + +- Run with `cargo run -p mcp-server-examples --example servers_subscriptions_streamhttp` + ### Counter Streamable HTTP Server with Hyper (`counter_hyper_streamable_http.rs`) A server using streamable HTTP transport for MCP communication, with hyper. diff --git a/examples/servers/src/subscriptions_streamhttp.rs b/examples/servers/src/subscriptions_streamhttp.rs new file mode 100644 index 000000000..015ae3cbe --- /dev/null +++ b/examples/servers/src/subscriptions_streamhttp.rs @@ -0,0 +1,83 @@ +use std::{borrow::Cow, time::Duration}; + +use rmcp::{ + ErrorData, ServerHandler, + model::{ProtocolVersion, ServerCapabilities, ServerInfo, SubscriptionFilter}, + service::SubscriptionContext, + transport::streamable_http_server::{ + StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager, + }, +}; +use tokio_util::sync::CancellationToken; + +#[derive(Clone)] +struct SubscriptionServer; + +impl ServerHandler for SubscriptionServer { + fn supported_protocol_versions(&self) -> Cow<'static, [ProtocolVersion]> { + Cow::Borrowed(&[ProtocolVersion::V_2026_07_28]) + } + + fn get_info(&self) -> ServerInfo { + ServerInfo::new( + ServerCapabilities::builder() + .enable_tools() + .enable_tool_list_changed() + .build(), + ) + } + + fn accepted_subscription_filter( + &self, + requested: &SubscriptionFilter, + ) -> Option { + Some(requested.supported_by(&self.get_info().capabilities)) + } + + async fn listen(&self, context: SubscriptionContext) -> Result<(), ErrorData> { + loop { + tokio::select! { + () = context.cancelled() => return Ok(()), + () = tokio::time::sleep(Duration::from_secs(2)) => { + context + .sink() + .notify_tool_list_changed() + .await + .map_err(|error| { + ErrorData::internal_error(error.to_string(), None) + })?; + } + } + } + } +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()), + ) + .init(); + + let cancellation_token = CancellationToken::new(); + let service: StreamableHttpService = + StreamableHttpService::new( + || Ok(SubscriptionServer), + Default::default(), + StreamableHttpServerConfig::default() + .with_stateful_mode(false) + .with_sse_keep_alive(Some(Duration::from_secs(10))) + .with_cancellation_token(cancellation_token.child_token()), + ); + let router = axum::Router::new().nest_service("/mcp", service); + let listener = tokio::net::TcpListener::bind("127.0.0.1:8000").await?; + + axum::serve(listener, router) + .with_graceful_shutdown(async move { + let _ = tokio::signal::ctrl_c().await; + cancellation_token.cancel(); + }) + .await?; + Ok(()) +}