From 003314a787e4c56809073b587d0c8c21893f0bfc Mon Sep 17 00:00:00 2001 From: Alex Hancock Date: Wed, 22 Jul 2026 14:55:31 -0400 Subject: [PATCH 01/12] feat: implement SEP-2260 require server requests to associate with client requests --- crates/rmcp/src/service.rs | 27 ++- crates/rmcp/src/service/server.rs | 25 +++ crates/rmcp/src/task_manager.rs | 80 ++++++++- .../test_sep_2260_request_association.rs | 159 ++++++++++++++++++ 4 files changed, 288 insertions(+), 3 deletions(-) create mode 100644 crates/rmcp/tests/test_sep_2260_request_association.rs diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 18da427d..391ca5d6 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -149,6 +149,15 @@ pub trait ServiceRole: std::fmt::Debug + Send + Sync + 'static + Copy + Clone { ) -> impl Future + MaybeSendFuture { async {} } + + #[doc(hidden)] + fn enforce_request_association( + _request: &Self::Req, + _peer_info: Option<&Self::PeerInfo>, + _in_request_handler_scope: bool, + ) -> Result<(), ServiceError> { + Ok(()) + } } pub(crate) fn uses_legacy_lifecycle( @@ -159,6 +168,14 @@ pub(crate) fn uses_legacy_lifecycle( && protocol_version.is_none_or(|version| version < &ProtocolVersion::V_2026_07_28) } +tokio::task_local! { + pub(crate) static ORIGINATING_REQUEST: RequestId; +} + +pub(crate) fn in_request_handler_scope() -> bool { + ORIGINATING_REQUEST.try_with(|_| ()).is_ok() +} + pub type TxJsonRpcMessage = JsonRpcMessage<::Req, ::Resp, ::Not>; pub type RxJsonRpcMessage = JsonRpcMessage< @@ -725,6 +742,11 @@ impl Peer { options: PeerRequestOptions, subscription_sender: Option>, ) -> Result, ServiceError> { + R::enforce_request_association( + &request, + self.peer_info().as_deref(), + in_request_handler_scope(), + )?; let id = self.request_id_provider.next_request_id(); let progress_token = self.progress_token_provider.next_progress_token(); if let Some(metadata) = self.client_request_metadata.get() { @@ -1409,9 +1431,10 @@ where extensions, }; let current_span = tracing::Span::current(); + let handler_id = id.clone(); spawn_service_task(async move { - let result = service - .handle_request(request, context) + let result = ORIGINATING_REQUEST + .scope(handler_id, service.handle_request(request, context)) .await; let response = match result { Ok(result) => { diff --git a/crates/rmcp/src/service/server.rs b/crates/rmcp/src/service/server.rs index f4529908..6d2a0d6b 100644 --- a/crates/rmcp/src/service/server.rs +++ b/crates/rmcp/src/service/server.rs @@ -49,6 +49,31 @@ impl ServiceRole for RoleServer { _ => None, } } + + fn enforce_request_association( + request: &Self::Req, + peer_info: Option<&Self::PeerInfo>, + in_request_handler_scope: bool, + ) -> Result<(), ServiceError> { + let restricted = matches!( + request, + ServerRequest::CreateMessageRequest(_) + | ServerRequest::ListRootsRequest(_) + | ServerRequest::ElicitRequest(_) + ); + if !restricted { + return Ok(()); + } + let strict = + peer_info.is_some_and(|info| info.protocol_version >= ProtocolVersion::V_2026_07_28); + if strict && !in_request_handler_scope { + return Err(ServiceError::McpError(ErrorData::invalid_request( + "SEP-2260: server-to-client requests must be associated with an originating client request", + None, + ))); + } + Ok(()) + } } /// It represents the error that may occur when serving the server. diff --git a/crates/rmcp/src/task_manager.rs b/crates/rmcp/src/task_manager.rs index df1c389a..0e8339c0 100644 --- a/crates/rmcp/src/task_manager.rs +++ b/crates/rmcp/src/task_manager.rs @@ -349,8 +349,11 @@ impl TaskManager { let future = make_future(context); let inner = self.inner.clone(); let id_for_task = task_id.clone(); + let originating_request = crate::service::ORIGINATING_REQUEST + .try_with(|id| id.clone()) + .ok(); let handle = tokio::spawn(async move { - let result = future.await; + let result = run_task_operation(originating_request, future).await; let mut inner = inner.lock().expect("task manager lock poisoned"); if let Some(entry) = inner.tasks.get_mut(&id_for_task) { if entry.terminal.is_none() { @@ -538,6 +541,16 @@ fn unknown_task(task_id: &str) -> McpError { McpError::invalid_params(format!("unknown task: {task_id}"), None) } +async fn run_task_operation( + originating_request: Option, + future: TaskFuture, +) -> Result { + match originating_request { + Some(id) => crate::service::ORIGINATING_REQUEST.scope(id, future).await, + None => future.await, + } +} + fn result_to_object(result: &CallToolResult) -> JsonObject { match serde_json::to_value(result) { Ok(serde_json::Value::Object(map)) => map, @@ -917,4 +930,69 @@ mod tests { } panic!("task did not complete after input response"); } + + #[tokio::test] + async fn task_operation_reestablishes_request_association_scope() { + use crate::{ + model::RequestId, + service::{ORIGINATING_REQUEST, in_request_handler_scope}, + }; + + let manager = TaskManager::new(); + let observed = Arc::new(Mutex::new(None::)); + let observed_in_task = observed.clone(); + + ORIGINATING_REQUEST + .scope(RequestId::Number(7), async { + manager.spawn(TaskOptions::default(), move |_ctx| { + let observed_in_task = observed_in_task.clone(); + Box::pin(async move { + *observed_in_task.lock().unwrap() = Some(in_request_handler_scope()); + Ok(ok_result("done")) + }) + }) + }) + .await; + + for _ in 0..100 { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + if let Some(scoped) = *observed.lock().unwrap() { + assert!( + scoped, + "task operation must run inside the originating request's association scope" + ); + return; + } + } + panic!("task operation did not run"); + } + + #[tokio::test] + async fn task_operation_without_originating_request_is_unscoped() { + use crate::service::in_request_handler_scope; + + let manager = TaskManager::new(); + let observed = Arc::new(Mutex::new(None::)); + let observed_in_task = observed.clone(); + + manager.spawn(TaskOptions::default(), move |_ctx| { + let observed_in_task = observed_in_task.clone(); + Box::pin(async move { + *observed_in_task.lock().unwrap() = Some(in_request_handler_scope()); + Ok(ok_result("done")) + }) + }); + + for _ in 0..100 { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + if let Some(scoped) = *observed.lock().unwrap() { + assert!( + !scoped, + "task operation started without an originating request must remain unscoped" + ); + return; + } + } + panic!("task operation did not run"); + } } diff --git a/crates/rmcp/tests/test_sep_2260_request_association.rs b/crates/rmcp/tests/test_sep_2260_request_association.rs new file mode 100644 index 00000000..de25b8a4 --- /dev/null +++ b/crates/rmcp/tests/test_sep_2260_request_association.rs @@ -0,0 +1,159 @@ +#![cfg(all(feature = "server", feature = "client", not(feature = "local")))] +#![allow(deprecated)] + +use std::sync::{Arc, Mutex}; + +use rmcp::{ + ClientHandler, RoleClient, RoleServer, ServerHandler, ServiceError, ServiceExt, + model::{ + CallToolRequestParams, CallToolResponse, CallToolResult, ClientInfo, ContentBlock, + CreateMessageRequest, CreateMessageRequestParams, CreateMessageResult, ProtocolVersion, + SamplingMessage, ServerCapabilities, ServerInfo, ServerRequest, + }, + service::RequestContext, +}; +use tokio::sync::oneshot; + +#[derive(Clone)] +struct SamplingServer { + outside: Arc>>>>, +} + +impl ServerHandler for SamplingServer { + fn get_info(&self) -> ServerInfo { + ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) + } + + async fn call_tool( + &self, + request: CallToolRequestParams, + context: RequestContext, + ) -> Result { + let peer = context.peer.clone(); + let slot = self.outside.clone(); + + let use_generic = request.name == "sample_generic"; + tokio::spawn(async move { + let outside = if use_generic { + peer.send_request(ServerRequest::CreateMessageRequest( + CreateMessageRequest::new(CreateMessageRequestParams::new( + vec![SamplingMessage::user_text("standalone-generic")], + 16, + )), + )) + .await + .map(|_| ()) + } else { + peer.create_message(CreateMessageRequestParams::new( + vec![SamplingMessage::user_text("standalone")], + 16, + )) + .await + .map(|_| ()) + }; + if let Some(tx) = slot.lock().unwrap().take() { + let _ = tx.send(outside); + } + }); + + let nested = context + .peer + .create_message(CreateMessageRequestParams::new( + vec![SamplingMessage::user_text("nested")], + 16, + )) + .await; + nested.map_err(|e| rmcp::ErrorData::internal_error(e.to_string(), None))?; + Ok(CallToolResult::success(vec![ContentBlock::text("ok")]).into()) + } +} + +#[derive(Clone)] +struct SamplingClient; + +impl ClientHandler for SamplingClient { + async fn create_message( + &self, + _params: CreateMessageRequestParams, + _context: RequestContext, + ) -> Result { + Ok(CreateMessageResult::new( + SamplingMessage::assistant_text("pong"), + "test-model".to_string(), + ) + .with_stop_reason(CreateMessageResult::STOP_REASON_END_TURN)) + } + + fn get_info(&self) -> ClientInfo { + let mut info = ClientInfo::default(); + info.protocol_version = ProtocolVersion::V_2026_07_28; + info + } +} + +#[tokio::test] +async fn nested_sampling_allowed_standalone_rejected() -> anyhow::Result<()> { + let (server_transport, client_transport) = tokio::io::duplex(4096); + let (tx, rx) = oneshot::channel(); + let server = SamplingServer { + outside: Arc::new(Mutex::new(Some(tx))), + }; + let server_handle = tokio::spawn(async move { + let running = server.serve(server_transport).await?; + running.waiting().await?; + anyhow::Ok(()) + }); + + let client = SamplingClient.serve(client_transport).await?; + + let result = client + .peer() + .call_tool(CallToolRequestParams::new("sample")) + .await?; + assert_eq!( + result.content.first().unwrap().as_text().unwrap().text, + "ok" + ); + + let outside = rx.await?; + assert!(matches!(outside, Err(ServiceError::McpError(_)))); + + client.cancel().await?; + let _ = server_handle.await?; + Ok(()) +} + +#[tokio::test] +async fn generic_send_request_bypass_rejected() -> anyhow::Result<()> { + let (server_transport, client_transport) = tokio::io::duplex(4096); + let (tx, rx) = oneshot::channel(); + let server = SamplingServer { + outside: Arc::new(Mutex::new(Some(tx))), + }; + let server_handle = tokio::spawn(async move { + let running = server.serve(server_transport).await?; + running.waiting().await?; + anyhow::Ok(()) + }); + + let client = SamplingClient.serve(client_transport).await?; + + let result = client + .peer() + .call_tool(CallToolRequestParams::new("sample_generic")) + .await?; + assert_eq!( + result.content.first().unwrap().as_text().unwrap().text, + "ok" + ); + + let outside = rx.await?; + assert!( + matches!(outside, Err(ServiceError::McpError(_))), + "generic send_request must not bypass SEP-2260 enforcement" + ); + + client.cancel().await?; + let _ = server_handle.await?; + Ok(()) +} From 425625029fd8a3e2dbdd7fc203c8bbbc83c1dddd Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Wed, 22 Jul 2026 17:55:59 -0400 Subject: [PATCH 02/12] feat(service): attach originating request id to in-handler outbound requests (SEP-2260) Reuses the ORIGINATING_REQUEST task-local from #1027; the marker rides the request's non-serialized Extensions so the streamable HTTP server can route associated requests to the originating SSE stream. --- crates/rmcp/src/service.rs | 70 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 70 insertions(+) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 391ca5d6..df5cd3ba 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -176,6 +176,23 @@ pub(crate) fn in_request_handler_scope() -> bool { ORIGINATING_REQUEST.try_with(|_| ()).is_ok() } +/// Marker stored in an outbound request's [`Extensions`](crate::model::Extensions) +/// identifying the in-flight peer request it was issued from (SEP-2260). +/// +/// Attached automatically whenever a request is sent from within a request +/// handler. The streamable HTTP server uses it to deliver server-initiated +/// requests on the originating client request's SSE stream. +/// +/// # In-memory only +/// +/// `Extensions` are never serialized, so this marker does not appear on the +/// wire (SEP-2260 defines no wire field). Session managers that serialize +/// messages between processes lose it; such requests fall back to the +/// standalone stream (logged as a warning). +#[derive(Debug, Clone, PartialEq, Eq)] +#[expect(clippy::exhaustive_structs, reason = "intentionally exhaustive")] +pub struct OriginatingRequestId(pub RequestId); + pub type TxJsonRpcMessage = JsonRpcMessage<::Req, ::Resp, ::Not>; pub type RxJsonRpcMessage = JsonRpcMessage< @@ -747,6 +764,11 @@ impl Peer { self.peer_info().as_deref(), in_request_handler_scope(), )?; + if let Ok(originating) = ORIGINATING_REQUEST.try_with(|id| id.clone()) { + request + .extensions_mut() + .insert(OriginatingRequestId(originating)); + } let id = self.request_id_provider.next_request_id(); let progress_token = self.progress_token_provider.next_progress_token(); if let Some(metadata) = self.client_request_metadata.get() { @@ -1631,3 +1653,51 @@ where dg: ct.drop_guard(), } } + +#[cfg(all(test, feature = "server"))] +mod sep2260_marker_tests { + use std::sync::Arc; + + use super::*; + use crate::model::{PingRequest, RequestId, ServerRequest}; + + fn ping() -> ServerRequest { + ServerRequest::PingRequest(PingRequest { + method: Default::default(), + extensions: Default::default(), + }) + } + + async fn send_and_capture(scope: Option) -> ::Req { + // peer_info None => enforce_request_association is non-strict, so the + // send is accepted regardless of scope; we only inspect the sink message. + let (peer, mut rx) = + Peer::::new(Arc::new(AtomicU32RequestIdProvider::default()), None); + let send = peer.send_request_with_option(ping(), PeerRequestOptions::no_options()); + let _handle = match scope { + Some(id) => ORIGINATING_REQUEST.scope(id, send).await.unwrap(), + None => send.await.unwrap(), + }; + let PeerSinkMessage::Request { request, .. } = rx.recv().await.expect("sink message") + else { + panic!("expected a request sink message"); + }; + request + } + + #[tokio::test] + async fn outbound_request_carries_originating_id_when_in_scope() { + let request = send_and_capture(Some(RequestId::Number(7))).await; + let marker = request + .extensions() + .get::() + .expect("marker attached"); + assert_eq!(marker.0, RequestId::Number(7)); + } + + #[tokio::test] + async fn outbound_request_has_no_marker_outside_scope() { + let request = send_and_capture(None).await; + assert!(request.extensions().get::().is_none()); + } +} From 4736211e15d710e4ed904ccdf9e0bab55ef8ab30 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Wed, 22 Jul 2026 18:10:49 -0400 Subject: [PATCH 03/12] feat(transport): route associated server requests to originating SSE stream (SEP-2260) --- .../streamable_http_server/session.rs | 12 ++ .../streamable_http_server/session/local.rs | 103 +++++++++++++++++- 2 files changed, 114 insertions(+), 1 deletion(-) diff --git a/crates/rmcp/src/transport/streamable_http_server/session.rs b/crates/rmcp/src/transport/streamable_http_server/session.rs index 28bd98bc..da32e155 100644 --- a/crates/rmcp/src/transport/streamable_http_server/session.rs +++ b/crates/rmcp/src/transport/streamable_http_server/session.rs @@ -77,6 +77,18 @@ pub enum RestoreOutcome { /// trait for every HTTP request that carries (or should carry) a session ID. /// /// See the [module-level docs](self) for background on sessions. +/// +/// # SEP-2260 request association +/// +/// Server-initiated requests issued while handling a client request carry an +/// [`OriginatingRequestId`](crate::service::OriginatingRequestId) marker in +/// their non-serialized `Extensions`. Implementations that keep messages +/// in-memory (like the bundled local session manager) deliver such requests on +/// the originating request's SSE stream, per SEP-2260. Implementations that +/// serialize messages between processes lose the marker; those requests then +/// fall back to the standalone stream, which violates SEP-2260 for protocol +/// version 2026-07-28+ clients — distributed session managers need their own +/// association mechanism. pub trait SessionManager: Send + Sync + 'static { type Error: std::error::Error + Send + 'static; type Transport: crate::transport::Transport; diff --git a/crates/rmcp/src/transport/streamable_http_server/session/local.rs b/crates/rmcp/src/transport/streamable_http_server/session/local.rs index ca88c088..3a04fd20 100644 --- a/crates/rmcp/src/transport/streamable_http_server/session/local.rs +++ b/crates/rmcp/src/transport/streamable_http_server/session/local.rs @@ -527,7 +527,42 @@ impl LocalSessionWorker { } fn resolve_outbound_channel(&self, message: &ServerJsonRpcMessage) -> OutboundChannel { match &message { - ServerJsonRpcMessage::Request(_) => OutboundChannel::Common, + // SEP-2260: server-initiated requests issued while handling a + // client request carry an OriginatingRequestId marker and are + // delivered on that request's SSE stream, never the standalone + // GET stream. Unmarked requests (sent outside a handler on + // legacy protocol versions) keep using the common stream. If the + // marker is present but the originating request already completed + // (e.g. a task operation outliving its originating request), fall + // back to the common stream loudly — never a closed channel. + ServerJsonRpcMessage::Request(json_rpc_request) => { + use crate::model::GetExtensions; + let originating = json_rpc_request + .request + .extensions() + .get::(); + match originating { + Some(originating) => match self + .resource_router + .get(&ResourceKey::McpRequestId(originating.0.clone())) + { + Some(id) => OutboundChannel::RequestWise { + id: *id, + close: false, + }, + None => { + tracing::warn!( + originating_request_id = %originating.0, + "associated server request could not be routed to its \ + originating stream (request completed or association \ + lost); falling back to standalone stream" + ); + OutboundChannel::Common + } + }, + None => OutboundChannel::Common, + } + } ServerJsonRpcMessage::Notification(JsonRpcNotification { notification: ServerNotification::ProgressNotification(Notification { @@ -1281,3 +1316,69 @@ fn create_local_session_with_event_store( }; (handle, session_worker) } + +#[cfg(test)] +mod sep2260_routing_tests { + use super::*; + use crate::service::OriginatingRequestId; + + fn roots_request(originating: Option) -> ServerJsonRpcMessage { + #[allow(deprecated)] + let mut request = crate::model::ListRootsRequest { + method: Default::default(), + extensions: Default::default(), + }; + if let Some(id) = originating { + request.extensions.insert(OriginatingRequestId(id)); + } + ServerJsonRpcMessage::request( + crate::model::ServerRequest::ListRootsRequest(request), + RequestId::Number(1000), + ) + } + + #[tokio::test] + async fn associated_server_request_routes_to_originating_stream() { + let (_handle, mut worker) = create_local_session("test-session", SessionConfig::default()); + let receiver = worker.establish_request_wise_channel().await.unwrap(); + let http_request_id = receiver.http_request_id.unwrap(); + let originating_id = RequestId::Number(7); + worker.register_resource( + ResourceKey::McpRequestId(originating_id.clone()), + http_request_id, + ); + + let channel = worker.resolve_outbound_channel(&roots_request(Some(originating_id))); + assert!( + matches!(channel, OutboundChannel::RequestWise { id, close: false } if id == http_request_id) + ); + } + + #[tokio::test] + async fn unassociated_server_request_routes_to_common_stream() { + let (_handle, mut worker) = create_local_session("test-session", SessionConfig::default()); + let _receiver = worker.establish_request_wise_channel().await.unwrap(); + let channel = worker.resolve_outbound_channel(&roots_request(None)); + assert!(matches!(channel, OutboundChannel::Common)); + } + + #[tokio::test] + async fn associated_request_for_completed_request_falls_back_to_common() { + // Completion race / re-scoped task operations: the originating + // request's resources were unregistered before the associated request + // reached the worker. It must fall back to the common stream (with a + // warning), never be dropped into a closed request-wise channel. + let (_handle, mut worker) = create_local_session("test-session", SessionConfig::default()); + let receiver = worker.establish_request_wise_channel().await.unwrap(); + let http_request_id = receiver.http_request_id.unwrap(); + let originating_id = RequestId::Number(7); + worker.register_resource( + ResourceKey::McpRequestId(originating_id.clone()), + http_request_id, + ); + worker.unregister_resource(&ResourceKey::McpRequestId(originating_id.clone())); + + let channel = worker.resolve_outbound_channel(&roots_request(Some(originating_id))); + assert!(matches!(channel, OutboundChannel::Common)); + } +} From e0f0f11c828798ba039830a35b4944effe28b072 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Wed, 22 Jul 2026 18:26:06 -0400 Subject: [PATCH 04/12] test: end-to-end SEP-2260 stream routing over streamable HTTP --- .../tests/test_sep_2260_stream_routing.rs | 195 ++++++++++++++++++ 1 file changed, 195 insertions(+) create mode 100644 crates/rmcp/tests/test_sep_2260_stream_routing.rs diff --git a/crates/rmcp/tests/test_sep_2260_stream_routing.rs b/crates/rmcp/tests/test_sep_2260_stream_routing.rs new file mode 100644 index 00000000..51af4e41 --- /dev/null +++ b/crates/rmcp/tests/test_sep_2260_stream_routing.rs @@ -0,0 +1,195 @@ +//! SEP-2260 end-to-end: a server→client request issued while handling a +//! client request is delivered on the originating POST's SSE stream, and the +//! standalone GET stream never carries it. +#![cfg(all( + not(feature = "local"), + feature = "server", + feature = "elicitation", + feature = "transport-streamable-http-server" +))] + +use std::{collections::BTreeMap, sync::Arc, time::Duration}; + +use futures::StreamExt; +use rmcp::{ + ErrorData as McpError, RoleServer, ServerHandler, + model::{ + CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock, ElicitRequestParams, + ElicitationSchema, ListToolsResult, PaginatedRequestParams, ServerCapabilities, ServerInfo, + Tool, + }, + service::RequestContext, + transport::streamable_http_server::{ + StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager, + }, +}; +use serde_json::json; +use tokio_util::sync::CancellationToken; + +#[derive(Clone)] +struct ElicitingServer; + +impl ServerHandler for ElicitingServer { + fn get_info(&self) -> ServerInfo { + ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) + } + + async fn list_tools( + &self, + _request: Option, + _context: RequestContext, + ) -> Result { + Ok(ListToolsResult { + tools: vec![Tool::new( + "ask", + "asks the user", + serde_json::from_value::( + json!({"type": "object", "properties": {}}), + ) + .unwrap(), + )], + ..Default::default() + }) + } + + async fn call_tool( + &self, + _request: CallToolRequestParams, + context: RequestContext, + ) -> Result { + // Blocks awaiting the client's answer; the test only needs the elicit + // request to be EMITTED on the right stream, so it never answers and + // tears the server down at the end instead. + let _ = context + .peer + .create_elicitation(ElicitRequestParams::FormElicitationParams { + meta: None, + message: "need input".to_string(), + requested_schema: ElicitationSchema::new(BTreeMap::new()), + }) + .await; + Ok(CallToolResult::success(vec![ContentBlock::text("done")]).into()) + } +} + +async fn start_server(ct: CancellationToken) -> String { + let service = StreamableHttpService::new( + move || Ok(ElicitingServer), + Arc::new(LocalSessionManager::default()), + StreamableHttpServerConfig::default().with_cancellation_token(ct.child_token()), + ); + let router = axum::Router::new().nest_service("/mcp", service); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!( + "http://127.0.0.1:{}/mcp", + listener.local_addr().unwrap().port() + ); + let ct = ct.clone(); + tokio::spawn(async move { + axum::serve(listener, router) + .with_graceful_shutdown(async move { ct.cancelled().await }) + .await + .unwrap(); + }); + tokio::time::sleep(Duration::from_millis(100)).await; + url +} + +/// Read an SSE byte stream until `needle` appears or timeout. +async fn sse_contains(resp: reqwest::Response, needle: &str, timeout: Duration) -> bool { + let mut stream = resp.bytes_stream(); + tokio::time::timeout(timeout, async { + let mut buffer = String::new(); + while let Some(Ok(chunk)) = stream.next().await { + buffer.push_str(&String::from_utf8_lossy(&chunk)); + if buffer.contains(needle) { + return true; + } + } + false + }) + .await + .unwrap_or(false) +} + +#[tokio::test] +async fn elicitation_rides_originating_post_stream_not_standalone_get() { + let ct = CancellationToken::new(); + let url = start_server(ct.clone()).await; + let client = reqwest::Client::new(); + + // Initialize with 2025-11-25: the latest version with sessions. Per + // SEP-2567 the server serves 2026-07-28 statelessly (no session, no + // standalone GET stream), so the SEP-2260 stream-routing scenario only + // exists on session-carrying versions. + let resp = client + .post(&url) + .header("Accept", "text/event-stream, application/json") + .header("Content-Type", "application/json") + .json(&json!({ + "jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": { + "protocolVersion": "2025-11-25", + "capabilities": { "elicitation": {} }, + "clientInfo": { "name": "test-client", "version": "1.0.0" } + } + })) + .send() + .await + .unwrap(); + assert!(resp.status().is_success()); + let session_id = resp + .headers() + .get("Mcp-Session-Id") + .expect("session id") + .to_str() + .unwrap() + .to_string(); + + client + .post(&url) + .header("Accept", "text/event-stream, application/json") + .header("Content-Type", "application/json") + .header("Mcp-Session-Id", &session_id) + .json(&json!({"jsonrpc": "2.0", "method": "notifications/initialized"})) + .send() + .await + .unwrap(); + + // Standalone GET stream — must NEVER carry the elicitation request. + let get_stream = client + .get(&url) + .header("Accept", "text/event-stream") + .header("Mcp-Session-Id", &session_id) + .send() + .await + .unwrap(); + assert_eq!(get_stream.status(), 200); + + // tools/call — the response is an SSE stream; the in-handler elicitation + // request must appear on THIS stream while the handler awaits the answer. + let post_stream = client + .post(&url) + .header("Accept", "text/event-stream, application/json") + .header("Content-Type", "application/json") + .header("Mcp-Session-Id", &session_id) + .json(&json!({ + "jsonrpc": "2.0", "id": 2, "method": "tools/call", + "params": { "name": "ask", "arguments": {} } + })) + .send() + .await + .unwrap(); + assert!(post_stream.status().is_success()); + + assert!( + sse_contains(post_stream, "elicitation/create", Duration::from_secs(5)).await, + "elicitation request must be delivered on the originating POST SSE stream" + ); + assert!( + !sse_contains(get_stream, "elicitation/create", Duration::from_millis(500)).await, + "standalone GET stream must not carry the elicitation request" + ); + + ct.cancel(); +} From 98a51acb5bb9d366951c7173b24542e4fc3fc399 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Wed, 22 Jul 2026 18:35:12 -0400 Subject: [PATCH 05/12] test: register SEP-2260 routing test with required-features; drop dead list_tools --- crates/rmcp/Cargo.toml | 5 ++++ .../tests/test_sep_2260_stream_routing.rs | 28 ++----------------- 2 files changed, 7 insertions(+), 26 deletions(-) diff --git a/crates/rmcp/Cargo.toml b/crates/rmcp/Cargo.toml index 752b5c7e..73768278 100644 --- a/crates/rmcp/Cargo.toml +++ b/crates/rmcp/Cargo.toml @@ -445,3 +445,8 @@ required-features = [ "reqwest", ] path = "tests/test_streamable_http_disconnect_cancel.rs" + +[[test]] +name = "test_sep_2260_stream_routing" +required-features = ["server", "elicitation", "transport-streamable-http-server", "reqwest"] +path = "tests/test_sep_2260_stream_routing.rs" diff --git a/crates/rmcp/tests/test_sep_2260_stream_routing.rs b/crates/rmcp/tests/test_sep_2260_stream_routing.rs index 51af4e41..88c1b231 100644 --- a/crates/rmcp/tests/test_sep_2260_stream_routing.rs +++ b/crates/rmcp/tests/test_sep_2260_stream_routing.rs @@ -1,12 +1,7 @@ //! SEP-2260 end-to-end: a server→client request issued while handling a //! client request is delivered on the originating POST's SSE stream, and the //! standalone GET stream never carries it. -#![cfg(all( - not(feature = "local"), - feature = "server", - feature = "elicitation", - feature = "transport-streamable-http-server" -))] +#![cfg(not(feature = "local"))] use std::{collections::BTreeMap, sync::Arc, time::Duration}; @@ -15,8 +10,7 @@ use rmcp::{ ErrorData as McpError, RoleServer, ServerHandler, model::{ CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock, ElicitRequestParams, - ElicitationSchema, ListToolsResult, PaginatedRequestParams, ServerCapabilities, ServerInfo, - Tool, + ElicitationSchema, ServerCapabilities, ServerInfo, }, service::RequestContext, transport::streamable_http_server::{ @@ -34,24 +28,6 @@ impl ServerHandler for ElicitingServer { ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) } - async fn list_tools( - &self, - _request: Option, - _context: RequestContext, - ) -> Result { - Ok(ListToolsResult { - tools: vec![Tool::new( - "ask", - "asks the user", - serde_json::from_value::( - json!({"type": "object", "properties": {}}), - ) - .unwrap(), - )], - ..Default::default() - }) - } - async fn call_tool( &self, _request: CallToolRequestParams, From 7618d2b9ae2efb52c4b29b0c53d09b9653139a3c Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Wed, 22 Jul 2026 18:39:10 -0400 Subject: [PATCH 06/12] docs: document SEP-2260 association requirement and stream routing --- crates/rmcp/src/service/server.rs | 100 +++++++++++++++++++++++++++++- 1 file changed, 98 insertions(+), 2 deletions(-) diff --git a/crates/rmcp/src/service/server.rs b/crates/rmcp/src/service/server.rs index 6d2a0d6b..646af1fd 100644 --- a/crates/rmcp/src/service/server.rs +++ b/crates/rmcp/src/service/server.rs @@ -769,6 +769,17 @@ impl Peer { } } + /// # SEP-2260: request association + /// + /// From protocol version `2026-07-28`, this request must be issued while + /// handling a client request — call it on the [`Peer`] from a + /// [`RequestContext`], from the handler's + /// own task. Calling it outside a handler returns an + /// `invalid_request` error. Note that task-local association does not + /// cross `tokio::spawn`: a task spawned inside a handler is treated as + /// outside the handler — use the task manager for long-running work. + /// Requests issued in-handler are delivered on the originating request's + /// SSE stream in the streamable HTTP transport. #[deprecated( since = "1.8.0", note = "Sampling is deprecated by SEP-2577 and will be removed in a future release. See https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2577" @@ -803,6 +814,17 @@ impl Peer { } } method!( + /// # SEP-2260: request association + /// + /// From protocol version `2026-07-28`, this request must be issued while + /// handling a client request — call it on the [`Peer`] from a + /// [`RequestContext`], from the handler's + /// own task. Calling it outside a handler returns an + /// `invalid_request` error. Note that task-local association does not + /// cross `tokio::spawn`: a task spawned inside a handler is treated as + /// outside the handler — use the task manager for long-running work. + /// Requests issued in-handler are delivered on the originating request's + /// SSE stream in the streamable HTTP transport. #[deprecated( since = "1.8.0", note = "Roots is deprecated by SEP-2577 and will be removed in a future release. See https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2577" @@ -810,9 +832,35 @@ impl Peer { peer_req list_roots ListRootsRequest() => ListRootsResult ); #[cfg(feature = "elicitation")] - method!(peer_req create_elicitation ElicitRequest(ElicitRequestParams) => ElicitResult); + method!( + /// # SEP-2260: request association + /// + /// From protocol version `2026-07-28`, this request must be issued while + /// handling a client request — call it on the [`Peer`] from a + /// [`RequestContext`], from the handler's + /// own task. Calling it outside a handler returns an + /// `invalid_request` error. Note that task-local association does not + /// cross `tokio::spawn`: a task spawned inside a handler is treated as + /// outside the handler — use the task manager for long-running work. + /// Requests issued in-handler are delivered on the originating request's + /// SSE stream in the streamable HTTP transport. + peer_req create_elicitation ElicitRequest(ElicitRequestParams) => ElicitResult + ); #[cfg(feature = "elicitation")] - method!(peer_req_with_timeout create_elicitation_with_timeout ElicitRequest(ElicitRequestParams) => ElicitResult); + method!( + /// # SEP-2260: request association + /// + /// From protocol version `2026-07-28`, this request must be issued while + /// handling a client request — call it on the [`Peer`] from a + /// [`RequestContext`], from the handler's + /// own task. Calling it outside a handler returns an + /// `invalid_request` error. Note that task-local association does not + /// cross `tokio::spawn`: a task spawned inside a handler is treated as + /// outside the handler — use the task manager for long-running work. + /// Requests issued in-handler are delivered on the originating request's + /// SSE stream in the streamable HTTP transport. + peer_req_with_timeout create_elicitation_with_timeout ElicitRequest(ElicitRequestParams) => ElicitResult + ); method!(peer_not notify_cancelled CancelledNotification(CancelledNotificationParam)); method!(peer_not notify_progress ProgressNotification(ProgressNotificationParam)); @@ -1039,6 +1087,18 @@ impl Peer { /// # Ok(()) /// # } /// ``` + /// + /// # SEP-2260: request association + /// + /// From protocol version `2026-07-28`, this request must be issued while + /// handling a client request — call it on the [`Peer`] from a + /// [`RequestContext`], from the handler's + /// own task. Calling it outside a handler returns an + /// `invalid_request` error. Note that task-local association does not + /// cross `tokio::spawn`: a task spawned inside a handler is treated as + /// outside the handler — use the task manager for long-running work. + /// Requests issued in-handler are delivered on the originating request's + /// SSE stream in the streamable HTTP transport. #[cfg(all(feature = "schemars", feature = "elicitation"))] pub async fn elicit(&self, message: impl Into) -> Result, ElicitationError> where @@ -1100,6 +1160,18 @@ impl Peer { /// # Ok(()) /// # } /// ``` + /// + /// # SEP-2260: request association + /// + /// From protocol version `2026-07-28`, this request must be issued while + /// handling a client request — call it on the [`Peer`] from a + /// [`RequestContext`], from the handler's + /// own task. Calling it outside a handler returns an + /// `invalid_request` error. Note that task-local association does not + /// cross `tokio::spawn`: a task spawned inside a handler is treated as + /// outside the handler — use the task manager for long-running work. + /// Requests issued in-handler are delivered on the originating request's + /// SSE stream in the streamable HTTP transport. #[cfg(all(feature = "schemars", feature = "elicitation"))] pub async fn elicit_with_timeout( &self, @@ -1195,6 +1267,18 @@ impl Peer { /// Ok(()) /// } /// ``` + /// + /// # SEP-2260: request association + /// + /// From protocol version `2026-07-28`, this request must be issued while + /// handling a client request — call it on the [`Peer`] from a + /// [`RequestContext`], from the handler's + /// own task. Calling it outside a handler returns an + /// `invalid_request` error. Note that task-local association does not + /// cross `tokio::spawn`: a task spawned inside a handler is treated as + /// outside the handler — use the task manager for long-running work. + /// Requests issued in-handler are delivered on the originating request's + /// SSE stream in the streamable HTTP transport. #[cfg(feature = "elicitation")] pub async fn elicit_url( &self, @@ -1246,6 +1330,18 @@ impl Peer { /// Ok(()) /// } /// ``` + /// + /// # SEP-2260: request association + /// + /// From protocol version `2026-07-28`, this request must be issued while + /// handling a client request — call it on the [`Peer`] from a + /// [`RequestContext`], from the handler's + /// own task. Calling it outside a handler returns an + /// `invalid_request` error. Note that task-local association does not + /// cross `tokio::spawn`: a task spawned inside a handler is treated as + /// outside the handler — use the task manager for long-running work. + /// Requests issued in-handler are delivered on the originating request's + /// SSE stream in the streamable HTTP transport. #[cfg(feature = "elicitation")] pub async fn elicit_url_with_timeout( &self, From f2bae31a8c5fc2ac48bf4f4aa017e0a6cdf99942 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Wed, 22 Jul 2026 18:39:46 -0400 Subject: [PATCH 07/12] docs: fix redundant explicit link on OriginatingRequestId --- crates/rmcp/src/service.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index df5cd3ba..648568b6 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -176,7 +176,7 @@ pub(crate) fn in_request_handler_scope() -> bool { ORIGINATING_REQUEST.try_with(|_| ()).is_ok() } -/// Marker stored in an outbound request's [`Extensions`](crate::model::Extensions) +/// Marker stored in an outbound request's [`Extensions`] /// identifying the in-flight peer request it was issued from (SEP-2260). /// /// Attached automatically whenever a request is sent from within a request From 3eced5ef679cf6f31c52b17a1ab0670eac2dff78 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Thu, 23 Jul 2026 02:03:52 -0400 Subject: [PATCH 08/12] docs: note marker is role-agnostic; add reason to deprecated expect Review follow-ups: the OriginatingRequestId rustdoc now states the marker is attached for both roles with only the server transport reading it, and the test's deprecation suppression carries a reason per house style. --- crates/rmcp/src/service.rs | 5 +++-- .../src/transport/streamable_http_server/session/local.rs | 5 ++++- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 648568b6..61a8610e 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -180,8 +180,9 @@ pub(crate) fn in_request_handler_scope() -> bool { /// identifying the in-flight peer request it was issued from (SEP-2260). /// /// Attached automatically whenever a request is sent from within a request -/// handler. The streamable HTTP server uses it to deliver server-initiated -/// requests on the originating client request's SSE stream. +/// handler, for both roles; currently only the streamable HTTP server reads +/// it, to deliver server-initiated requests on the originating client +/// request's SSE stream. /// /// # In-memory only /// diff --git a/crates/rmcp/src/transport/streamable_http_server/session/local.rs b/crates/rmcp/src/transport/streamable_http_server/session/local.rs index 3a04fd20..dd1aca74 100644 --- a/crates/rmcp/src/transport/streamable_http_server/session/local.rs +++ b/crates/rmcp/src/transport/streamable_http_server/session/local.rs @@ -1323,7 +1323,10 @@ mod sep2260_routing_tests { use crate::service::OriginatingRequestId; fn roots_request(originating: Option) -> ServerJsonRpcMessage { - #[allow(deprecated)] + #[expect( + deprecated, + reason = "roots is SEP-2577-deprecated; any restricted request works here" + )] let mut request = crate::model::ListRootsRequest { method: Default::default(), extensions: Default::default(), From c6f7aeadaabdc0c46764b26ca92c86510bf5b5ce Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Thu, 23 Jul 2026 02:27:44 -0400 Subject: [PATCH 09/12] docs: condense inline comments to repo convention Trims multi-line inline comments added in this branch down to the terse one-to-two-line style used elsewhere in the codebase, keeping only the non-obvious rationale (spec references, fallback invariant, version choice). --- crates/rmcp/src/service.rs | 3 +-- .../streamable_http_server/session/local.rs | 16 ++++------------ .../rmcp/tests/test_sep_2260_stream_routing.rs | 18 ++++++------------ 3 files changed, 11 insertions(+), 26 deletions(-) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 61a8610e..d05af0a7 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -1670,8 +1670,7 @@ mod sep2260_marker_tests { } async fn send_and_capture(scope: Option) -> ::Req { - // peer_info None => enforce_request_association is non-strict, so the - // send is accepted regardless of scope; we only inspect the sink message. + // peer_info None keeps enforcement non-strict; only the sink message matters. let (peer, mut rx) = Peer::::new(Arc::new(AtomicU32RequestIdProvider::default()), None); let send = peer.send_request_with_option(ping(), PeerRequestOptions::no_options()); diff --git a/crates/rmcp/src/transport/streamable_http_server/session/local.rs b/crates/rmcp/src/transport/streamable_http_server/session/local.rs index dd1aca74..dc8a70b8 100644 --- a/crates/rmcp/src/transport/streamable_http_server/session/local.rs +++ b/crates/rmcp/src/transport/streamable_http_server/session/local.rs @@ -527,14 +527,8 @@ impl LocalSessionWorker { } fn resolve_outbound_channel(&self, message: &ServerJsonRpcMessage) -> OutboundChannel { match &message { - // SEP-2260: server-initiated requests issued while handling a - // client request carry an OriginatingRequestId marker and are - // delivered on that request's SSE stream, never the standalone - // GET stream. Unmarked requests (sent outside a handler on - // legacy protocol versions) keep using the common stream. If the - // marker is present but the originating request already completed - // (e.g. a task operation outliving its originating request), fall - // back to the common stream loudly — never a closed channel. + // SEP-2260: requests carrying an OriginatingRequestId marker ride the + // originating request's SSE stream, never the standalone GET stream. ServerJsonRpcMessage::Request(json_rpc_request) => { use crate::model::GetExtensions; let originating = json_rpc_request @@ -1367,10 +1361,8 @@ mod sep2260_routing_tests { #[tokio::test] async fn associated_request_for_completed_request_falls_back_to_common() { - // Completion race / re-scoped task operations: the originating - // request's resources were unregistered before the associated request - // reached the worker. It must fall back to the common stream (with a - // warning), never be dropped into a closed request-wise channel. + // Originating request already unregistered (completion race / re-scoped + // task op): must fall back to Common, never a closed request-wise channel. let (_handle, mut worker) = create_local_session("test-session", SessionConfig::default()); let receiver = worker.establish_request_wise_channel().await.unwrap(); let http_request_id = receiver.http_request_id.unwrap(); diff --git a/crates/rmcp/tests/test_sep_2260_stream_routing.rs b/crates/rmcp/tests/test_sep_2260_stream_routing.rs index 88c1b231..824ef0c7 100644 --- a/crates/rmcp/tests/test_sep_2260_stream_routing.rs +++ b/crates/rmcp/tests/test_sep_2260_stream_routing.rs @@ -1,6 +1,5 @@ -//! SEP-2260 end-to-end: a server→client request issued while handling a -//! client request is delivered on the originating POST's SSE stream, and the -//! standalone GET stream never carries it. +//! SEP-2260 end-to-end: in-handler server→client requests ride the originating +//! POST's SSE stream, never the standalone GET stream. #![cfg(not(feature = "local"))] use std::{collections::BTreeMap, sync::Arc, time::Duration}; @@ -33,9 +32,7 @@ impl ServerHandler for ElicitingServer { _request: CallToolRequestParams, context: RequestContext, ) -> Result { - // Blocks awaiting the client's answer; the test only needs the elicit - // request to be EMITTED on the right stream, so it never answers and - // tears the server down at the end instead. + // Never answered: the test only checks the request is emitted on the right stream. let _ = context .peer .create_elicitation(ElicitRequestParams::FormElicitationParams { @@ -94,10 +91,8 @@ async fn elicitation_rides_originating_post_stream_not_standalone_get() { let url = start_server(ct.clone()).await; let client = reqwest::Client::new(); - // Initialize with 2025-11-25: the latest version with sessions. Per - // SEP-2567 the server serves 2026-07-28 statelessly (no session, no - // standalone GET stream), so the SEP-2260 stream-routing scenario only - // exists on session-carrying versions. + // 2025-11-25 is the latest session-carrying version; SEP-2567 serves + // 2026-07-28+ statelessly, with no standalone GET stream to test against. let resp = client .post(&url) .header("Accept", "text/event-stream, application/json") @@ -142,8 +137,7 @@ async fn elicitation_rides_originating_post_stream_not_standalone_get() { .unwrap(); assert_eq!(get_stream.status(), 200); - // tools/call — the response is an SSE stream; the in-handler elicitation - // request must appear on THIS stream while the handler awaits the answer. + // The in-handler elicitation must appear on this tools/call SSE stream. let post_stream = client .post(&url) .header("Accept", "text/event-stream, application/json") From 5a5de82fdb878db9100eb553d8e1986c00ee9e5f Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Thu, 23 Jul 2026 02:35:02 -0400 Subject: [PATCH 10/12] docs: condense SEP-2260 rustdoc sections Tightens the duplicated per-method association section from eleven lines to five, and trims the marker and SessionManager docs to the same information in fewer words. --- crates/rmcp/src/service.rs | 21 ++-- crates/rmcp/src/service/server.rs | 104 ++++++------------ .../streamable_http_server/session.rs | 12 +- 3 files changed, 44 insertions(+), 93 deletions(-) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index d05af0a7..104efeec 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -176,20 +176,13 @@ pub(crate) fn in_request_handler_scope() -> bool { ORIGINATING_REQUEST.try_with(|_| ()).is_ok() } -/// Marker stored in an outbound request's [`Extensions`] -/// identifying the in-flight peer request it was issued from (SEP-2260). -/// -/// Attached automatically whenever a request is sent from within a request -/// handler, for both roles; currently only the streamable HTTP server reads -/// it, to deliver server-initiated requests on the originating client -/// request's SSE stream. -/// -/// # In-memory only -/// -/// `Extensions` are never serialized, so this marker does not appear on the -/// wire (SEP-2260 defines no wire field). Session managers that serialize -/// messages between processes lose it; such requests fall back to the -/// standalone stream (logged as a warning). +/// Marker in an outbound request's non-serialized [`Extensions`] identifying +/// the in-flight peer request it was issued from (SEP-2260). Attached for both +/// roles whenever a request is sent from within a request handler; the +/// streamable HTTP server reads it to deliver server-initiated requests on the +/// originating request's SSE stream. Never on the wire (SEP-2260 defines no +/// wire field), so session managers that serialize messages between processes +/// lose it and such requests fall back to the standalone stream with a warning. #[derive(Debug, Clone, PartialEq, Eq)] #[expect(clippy::exhaustive_structs, reason = "intentionally exhaustive")] pub struct OriginatingRequestId(pub RequestId); diff --git a/crates/rmcp/src/service/server.rs b/crates/rmcp/src/service/server.rs index 646af1fd..927e87c3 100644 --- a/crates/rmcp/src/service/server.rs +++ b/crates/rmcp/src/service/server.rs @@ -771,15 +771,10 @@ impl Peer { /// # SEP-2260: request association /// - /// From protocol version `2026-07-28`, this request must be issued while - /// handling a client request — call it on the [`Peer`] from a - /// [`RequestContext`], from the handler's - /// own task. Calling it outside a handler returns an - /// `invalid_request` error. Note that task-local association does not - /// cross `tokio::spawn`: a task spawned inside a handler is treated as - /// outside the handler — use the task manager for long-running work. - /// Requests issued in-handler are delivered on the originating request's - /// SSE stream in the streamable HTTP transport. + /// From protocol version `2026-07-28` this must be issued while handling a + /// client request; outside a handler it returns an `invalid_request` error. + /// The association does not cross `tokio::spawn`, so use the task manager + /// for long-running work. #[deprecated( since = "1.8.0", note = "Sampling is deprecated by SEP-2577 and will be removed in a future release. See https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2577" @@ -816,15 +811,10 @@ impl Peer { method!( /// # SEP-2260: request association /// - /// From protocol version `2026-07-28`, this request must be issued while - /// handling a client request — call it on the [`Peer`] from a - /// [`RequestContext`], from the handler's - /// own task. Calling it outside a handler returns an - /// `invalid_request` error. Note that task-local association does not - /// cross `tokio::spawn`: a task spawned inside a handler is treated as - /// outside the handler — use the task manager for long-running work. - /// Requests issued in-handler are delivered on the originating request's - /// SSE stream in the streamable HTTP transport. + /// From protocol version `2026-07-28` this must be issued while handling a + /// client request; outside a handler it returns an `invalid_request` error. + /// The association does not cross `tokio::spawn`, so use the task manager + /// for long-running work. #[deprecated( since = "1.8.0", note = "Roots is deprecated by SEP-2577 and will be removed in a future release. See https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2577" @@ -835,30 +825,20 @@ impl Peer { method!( /// # SEP-2260: request association /// - /// From protocol version `2026-07-28`, this request must be issued while - /// handling a client request — call it on the [`Peer`] from a - /// [`RequestContext`], from the handler's - /// own task. Calling it outside a handler returns an - /// `invalid_request` error. Note that task-local association does not - /// cross `tokio::spawn`: a task spawned inside a handler is treated as - /// outside the handler — use the task manager for long-running work. - /// Requests issued in-handler are delivered on the originating request's - /// SSE stream in the streamable HTTP transport. + /// From protocol version `2026-07-28` this must be issued while handling a + /// client request; outside a handler it returns an `invalid_request` error. + /// The association does not cross `tokio::spawn`, so use the task manager + /// for long-running work. peer_req create_elicitation ElicitRequest(ElicitRequestParams) => ElicitResult ); #[cfg(feature = "elicitation")] method!( /// # SEP-2260: request association /// - /// From protocol version `2026-07-28`, this request must be issued while - /// handling a client request — call it on the [`Peer`] from a - /// [`RequestContext`], from the handler's - /// own task. Calling it outside a handler returns an - /// `invalid_request` error. Note that task-local association does not - /// cross `tokio::spawn`: a task spawned inside a handler is treated as - /// outside the handler — use the task manager for long-running work. - /// Requests issued in-handler are delivered on the originating request's - /// SSE stream in the streamable HTTP transport. + /// From protocol version `2026-07-28` this must be issued while handling a + /// client request; outside a handler it returns an `invalid_request` error. + /// The association does not cross `tokio::spawn`, so use the task manager + /// for long-running work. peer_req_with_timeout create_elicitation_with_timeout ElicitRequest(ElicitRequestParams) => ElicitResult ); @@ -1090,15 +1070,10 @@ impl Peer { /// /// # SEP-2260: request association /// - /// From protocol version `2026-07-28`, this request must be issued while - /// handling a client request — call it on the [`Peer`] from a - /// [`RequestContext`], from the handler's - /// own task. Calling it outside a handler returns an - /// `invalid_request` error. Note that task-local association does not - /// cross `tokio::spawn`: a task spawned inside a handler is treated as - /// outside the handler — use the task manager for long-running work. - /// Requests issued in-handler are delivered on the originating request's - /// SSE stream in the streamable HTTP transport. + /// From protocol version `2026-07-28` this must be issued while handling a + /// client request; outside a handler it returns an `invalid_request` error. + /// The association does not cross `tokio::spawn`, so use the task manager + /// for long-running work. #[cfg(all(feature = "schemars", feature = "elicitation"))] pub async fn elicit(&self, message: impl Into) -> Result, ElicitationError> where @@ -1163,15 +1138,10 @@ impl Peer { /// /// # SEP-2260: request association /// - /// From protocol version `2026-07-28`, this request must be issued while - /// handling a client request — call it on the [`Peer`] from a - /// [`RequestContext`], from the handler's - /// own task. Calling it outside a handler returns an - /// `invalid_request` error. Note that task-local association does not - /// cross `tokio::spawn`: a task spawned inside a handler is treated as - /// outside the handler — use the task manager for long-running work. - /// Requests issued in-handler are delivered on the originating request's - /// SSE stream in the streamable HTTP transport. + /// From protocol version `2026-07-28` this must be issued while handling a + /// client request; outside a handler it returns an `invalid_request` error. + /// The association does not cross `tokio::spawn`, so use the task manager + /// for long-running work. #[cfg(all(feature = "schemars", feature = "elicitation"))] pub async fn elicit_with_timeout( &self, @@ -1270,15 +1240,10 @@ impl Peer { /// /// # SEP-2260: request association /// - /// From protocol version `2026-07-28`, this request must be issued while - /// handling a client request — call it on the [`Peer`] from a - /// [`RequestContext`], from the handler's - /// own task. Calling it outside a handler returns an - /// `invalid_request` error. Note that task-local association does not - /// cross `tokio::spawn`: a task spawned inside a handler is treated as - /// outside the handler — use the task manager for long-running work. - /// Requests issued in-handler are delivered on the originating request's - /// SSE stream in the streamable HTTP transport. + /// From protocol version `2026-07-28` this must be issued while handling a + /// client request; outside a handler it returns an `invalid_request` error. + /// The association does not cross `tokio::spawn`, so use the task manager + /// for long-running work. #[cfg(feature = "elicitation")] pub async fn elicit_url( &self, @@ -1333,15 +1298,10 @@ impl Peer { /// /// # SEP-2260: request association /// - /// From protocol version `2026-07-28`, this request must be issued while - /// handling a client request — call it on the [`Peer`] from a - /// [`RequestContext`], from the handler's - /// own task. Calling it outside a handler returns an - /// `invalid_request` error. Note that task-local association does not - /// cross `tokio::spawn`: a task spawned inside a handler is treated as - /// outside the handler — use the task manager for long-running work. - /// Requests issued in-handler are delivered on the originating request's - /// SSE stream in the streamable HTTP transport. + /// From protocol version `2026-07-28` this must be issued while handling a + /// client request; outside a handler it returns an `invalid_request` error. + /// The association does not cross `tokio::spawn`, so use the task manager + /// for long-running work. #[cfg(feature = "elicitation")] pub async fn elicit_url_with_timeout( &self, diff --git a/crates/rmcp/src/transport/streamable_http_server/session.rs b/crates/rmcp/src/transport/streamable_http_server/session.rs index da32e155..ea2e85b2 100644 --- a/crates/rmcp/src/transport/streamable_http_server/session.rs +++ b/crates/rmcp/src/transport/streamable_http_server/session.rs @@ -82,13 +82,11 @@ pub enum RestoreOutcome { /// /// Server-initiated requests issued while handling a client request carry an /// [`OriginatingRequestId`](crate::service::OriginatingRequestId) marker in -/// their non-serialized `Extensions`. Implementations that keep messages -/// in-memory (like the bundled local session manager) deliver such requests on -/// the originating request's SSE stream, per SEP-2260. Implementations that -/// serialize messages between processes lose the marker; those requests then -/// fall back to the standalone stream, which violates SEP-2260 for protocol -/// version 2026-07-28+ clients — distributed session managers need their own -/// association mechanism. +/// their non-serialized `Extensions`; the bundled local session manager uses +/// it to deliver such requests on the originating request's SSE stream. +/// Implementations that serialize messages between processes lose the marker +/// and fall back to the standalone stream, violating SEP-2260 for 2026-07-28+ +/// clients; they need their own association mechanism. pub trait SessionManager: Send + Sync + 'static { type Error: std::error::Error + Send + 'static; type Transport: crate::transport::Transport; From cdce2a057fb9a9f5e56769f0e7f235d7c4d4d4bf Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Thu, 23 Jul 2026 03:20:26 -0400 Subject: [PATCH 11/12] docs: single-source SEP-2260 caller requirements on OriginatingRequestId --- crates/rmcp/src/service.rs | 8 ++++++++ crates/rmcp/src/service/server.rs | 32 ++++++++----------------------- 2 files changed, 16 insertions(+), 24 deletions(-) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 104efeec..246e8e13 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -183,6 +183,14 @@ pub(crate) fn in_request_handler_scope() -> bool { /// originating request's SSE stream. Never on the wire (SEP-2260 defines no /// wire field), so session managers that serialize messages between processes /// lose it and such requests fall back to the standalone stream with a warning. +/// +/// # Caller requirements +/// +/// From protocol version `2026-07-28`, server-to-client sampling, roots, and +/// elicitation requests must be issued while handling a client request; +/// outside a handler they return an `invalid_request` error. The association +/// is task-local and does not cross `tokio::spawn`, so use the task manager +/// for long-running work. #[derive(Debug, Clone, PartialEq, Eq)] #[expect(clippy::exhaustive_structs, reason = "intentionally exhaustive")] pub struct OriginatingRequestId(pub RequestId); diff --git a/crates/rmcp/src/service/server.rs b/crates/rmcp/src/service/server.rs index 927e87c3..2b84a943 100644 --- a/crates/rmcp/src/service/server.rs +++ b/crates/rmcp/src/service/server.rs @@ -772,9 +772,7 @@ impl Peer { /// # SEP-2260: request association /// /// From protocol version `2026-07-28` this must be issued while handling a - /// client request; outside a handler it returns an `invalid_request` error. - /// The association does not cross `tokio::spawn`, so use the task manager - /// for long-running work. + /// client request; see [`OriginatingRequestId`]. #[deprecated( since = "1.8.0", note = "Sampling is deprecated by SEP-2577 and will be removed in a future release. See https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2577" @@ -812,9 +810,7 @@ impl Peer { /// # SEP-2260: request association /// /// From protocol version `2026-07-28` this must be issued while handling a - /// client request; outside a handler it returns an `invalid_request` error. - /// The association does not cross `tokio::spawn`, so use the task manager - /// for long-running work. + /// client request; see [`OriginatingRequestId`]. #[deprecated( since = "1.8.0", note = "Roots is deprecated by SEP-2577 and will be removed in a future release. See https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2577" @@ -826,9 +822,7 @@ impl Peer { /// # SEP-2260: request association /// /// From protocol version `2026-07-28` this must be issued while handling a - /// client request; outside a handler it returns an `invalid_request` error. - /// The association does not cross `tokio::spawn`, so use the task manager - /// for long-running work. + /// client request; see [`OriginatingRequestId`]. peer_req create_elicitation ElicitRequest(ElicitRequestParams) => ElicitResult ); #[cfg(feature = "elicitation")] @@ -836,9 +830,7 @@ impl Peer { /// # SEP-2260: request association /// /// From protocol version `2026-07-28` this must be issued while handling a - /// client request; outside a handler it returns an `invalid_request` error. - /// The association does not cross `tokio::spawn`, so use the task manager - /// for long-running work. + /// client request; see [`OriginatingRequestId`]. peer_req_with_timeout create_elicitation_with_timeout ElicitRequest(ElicitRequestParams) => ElicitResult ); @@ -1071,9 +1063,7 @@ impl Peer { /// # SEP-2260: request association /// /// From protocol version `2026-07-28` this must be issued while handling a - /// client request; outside a handler it returns an `invalid_request` error. - /// The association does not cross `tokio::spawn`, so use the task manager - /// for long-running work. + /// client request; see [`OriginatingRequestId`]. #[cfg(all(feature = "schemars", feature = "elicitation"))] pub async fn elicit(&self, message: impl Into) -> Result, ElicitationError> where @@ -1139,9 +1129,7 @@ impl Peer { /// # SEP-2260: request association /// /// From protocol version `2026-07-28` this must be issued while handling a - /// client request; outside a handler it returns an `invalid_request` error. - /// The association does not cross `tokio::spawn`, so use the task manager - /// for long-running work. + /// client request; see [`OriginatingRequestId`]. #[cfg(all(feature = "schemars", feature = "elicitation"))] pub async fn elicit_with_timeout( &self, @@ -1241,9 +1229,7 @@ impl Peer { /// # SEP-2260: request association /// /// From protocol version `2026-07-28` this must be issued while handling a - /// client request; outside a handler it returns an `invalid_request` error. - /// The association does not cross `tokio::spawn`, so use the task manager - /// for long-running work. + /// client request; see [`OriginatingRequestId`]. #[cfg(feature = "elicitation")] pub async fn elicit_url( &self, @@ -1299,9 +1285,7 @@ impl Peer { /// # SEP-2260: request association /// /// From protocol version `2026-07-28` this must be issued while handling a - /// client request; outside a handler it returns an `invalid_request` error. - /// The association does not cross `tokio::spawn`, so use the task manager - /// for long-running work. + /// client request; see [`OriginatingRequestId`]. #[cfg(feature = "elicitation")] pub async fn elicit_url_with_timeout( &self, From 4039fd9b7d050318d76bfabcd5d5df4eecb9790a Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Thu, 23 Jul 2026 11:37:03 -0400 Subject: [PATCH 12/12] feat(service): reject unassociated server-to-client requests on the client (SEP-2260) Implements the client receive-side of SEP-2260: restricted server requests (sampling/createMessage, roots/list, elicitation/create) received while the client has no outbound request in flight are answered with -32602 invalid params instead of being dispatched to the handler. Gated on negotiated protocol >= 2026-07-28; ping is exempt. With a request in flight we cannot tell which one the server request belongs to (SEP-2260 defines no wire field), so this is a deliberate under-approximation of the spec's SHOULD; exact stream-based enforcement for streamable HTTP needs receive-side provenance plumbing and is left as a follow-up. --- crates/rmcp/src/service.rs | 32 +++++ crates/rmcp/src/service/client.rs | 29 +++++ .../test_sep_2260_request_association.rs | 116 +++++++++++++++++- 3 files changed, 176 insertions(+), 1 deletion(-) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 246e8e13..217464ff 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -158,6 +158,20 @@ pub trait ServiceRole: std::fmt::Debug + Send + Sync + 'static + Copy + Clone { ) -> Result<(), ServiceError> { Ok(()) } + + /// Receive-side counterpart of [`Self::enforce_request_association`]: + /// SEP-2260 says clients receiving a server-to-client request with no + /// associated outbound request should reject it with invalid params. An + /// error return is sent back to the peer instead of dispatching to the + /// handler. + #[doc(hidden)] + fn enforce_peer_request_association( + _peer_request: &Self::PeerReq, + _peer_info: Option<&Self::PeerInfo>, + _has_pending_outbound_request: bool, + ) -> Result<(), McpError> { + Ok(()) + } } pub(crate) fn uses_legacy_lifecycle( @@ -1435,6 +1449,24 @@ where .. })) => { tracing::debug!(%id, ?request, "received request"); + if let Err(error) = R::enforce_peer_request_association( + &request, + peer.peer_info().as_deref(), + !local_responder_pool.is_empty(), + ) { + tracing::warn!(%id, message = %error.message, "rejected peer request"); + // send directly: the sink proxy path would drop the + // error since the request was never registered in + // local_ct_pool + let send = transport.send(JsonRpcMessage::error(error, Some(id))); + let current_span = tracing::Span::current(); + response_send_tasks.spawn(async move { + if let Err(error) = send.await { + tracing::error!(%error, "fail to send rejection error"); + } + }.instrument(current_span)); + continue; + } { let service = shared_service.clone(); let sink = sink_proxy_tx.clone(); diff --git a/crates/rmcp/src/service/client.rs b/crates/rmcp/src/service/client.rs index 166207b7..40d410d5 100644 --- a/crates/rmcp/src/service/client.rs +++ b/crates/rmcp/src/service/client.rs @@ -213,6 +213,35 @@ impl ServiceRole for RoleClient { } } + // SEP-2260: with no outbound request in flight there is nothing the + // server request could be associated with, so reject it. With one in + // flight we cannot tell which request it belongs to (no wire field), so + // we accept — an under-approximation of the spec's SHOULD. + fn enforce_peer_request_association( + peer_request: &Self::PeerReq, + peer_info: Option<&Self::PeerInfo>, + has_pending_outbound_request: bool, + ) -> Result<(), ErrorData> { + let restricted = matches!( + peer_request, + ServerRequest::CreateMessageRequest(_) + | ServerRequest::ListRootsRequest(_) + | ServerRequest::ElicitRequest(_) + ); + if !restricted { + return Ok(()); + } + let strict = + peer_info.is_some_and(|info| info.protocol_version >= ProtocolVersion::V_2026_07_28); + if strict && !has_pending_outbound_request { + return Err(ErrorData::invalid_params( + "SEP-2260: server-to-client requests must be associated with an in-flight client request", + None, + )); + } + Ok(()) + } + async fn invalidate_response_cache(peer: &Peer, notification: &Self::PeerNot) { match notification { ServerNotification::ResourceUpdatedNotification(notification) => { diff --git a/crates/rmcp/tests/test_sep_2260_request_association.rs b/crates/rmcp/tests/test_sep_2260_request_association.rs index de25b8a4..d4e20e1e 100644 --- a/crates/rmcp/tests/test_sep_2260_request_association.rs +++ b/crates/rmcp/tests/test_sep_2260_request_association.rs @@ -12,7 +12,11 @@ use rmcp::{ }, service::RequestContext, }; -use tokio::sync::oneshot; +use serde_json::{Value, json}; +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader, DuplexStream, Lines, ReadHalf, WriteHalf}, + sync::oneshot, +}; #[derive(Clone)] struct SamplingServer { @@ -157,3 +161,113 @@ async fn generic_send_request_bypass_rejected() -> anyhow::Result<()> { let _ = server_handle.await?; Ok(()) } + +// A compliant rmcp server cannot produce an unassociated server-to-client +// request at >= 2026-07-28 (send-side enforcement blocks it), so the client's +// receive-side enforcement is exercised with a raw JSON-RPC server. +type RawServer = ( + Lines>>, + WriteHalf, +); + +async fn raw_initialize(io: DuplexStream, protocol_version: &str) -> anyhow::Result { + let (read, mut write) = tokio::io::split(io); + let mut lines = BufReader::new(read).lines(); + let init: Value = serde_json::from_str(&lines.next_line().await?.expect("initialize request"))?; + assert_eq!(init["method"], "initialize"); + let response = json!({ + "jsonrpc": "2.0", + "id": init["id"], + "result": { + "protocolVersion": protocol_version, + "capabilities": {}, + "serverInfo": { "name": "raw-server", "version": "0.0.0" } + } + }); + write.write_all(format!("{response}\n").as_bytes()).await?; + let initialized: Value = + serde_json::from_str(&lines.next_line().await?.expect("initialized notification"))?; + assert_eq!(initialized["method"], "notifications/initialized"); + Ok((lines, write)) +} + +async fn raw_request(server: &mut RawServer, request: Value) -> anyhow::Result { + let (lines, write) = server; + write.write_all(format!("{request}\n").as_bytes()).await?; + Ok(serde_json::from_str( + &lines.next_line().await?.expect("response"), + )?) +} + +fn raw_sampling_request(id: u32) -> Value { + json!({ + "jsonrpc": "2.0", + "id": id, + "method": "sampling/createMessage", + "params": { + "messages": [{ "role": "user", "content": { "type": "text", "text": "hi" } }], + "maxTokens": 16 + } + }) +} + +#[tokio::test] +async fn unassociated_server_request_rejected_with_invalid_params() -> anyhow::Result<()> { + let (client_io, server_io) = tokio::io::duplex(4096); + let raw = tokio::spawn(async move { + let mut server = raw_initialize(server_io, "2026-07-28").await?; + raw_request(&mut server, raw_sampling_request(100)).await + }); + + let client = SamplingClient.serve(client_io).await?; + let response = raw.await??; + assert_eq!( + response["error"]["code"], -32602, + "SEP-2260: unassociated server-to-client request must be rejected with invalid params, got {response}" + ); + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn unassociated_server_request_allowed_on_legacy_protocol() -> anyhow::Result<()> { + let (client_io, server_io) = tokio::io::duplex(4096); + let raw = tokio::spawn(async move { + let mut server = raw_initialize(server_io, "2025-11-25").await?; + raw_request(&mut server, raw_sampling_request(100)).await + }); + + let client = SamplingClient.serve(client_io).await?; + let response = raw.await??; + assert_eq!( + response["result"]["model"], "test-model", + "pre-2026-07-28 peers keep the permissive behavior, got {response}" + ); + + client.cancel().await?; + Ok(()) +} + +#[tokio::test] +async fn unassociated_ping_allowed() -> anyhow::Result<()> { + let (client_io, server_io) = tokio::io::duplex(4096); + let raw = tokio::spawn(async move { + let mut server = raw_initialize(server_io, "2026-07-28").await?; + raw_request( + &mut server, + json!({ "jsonrpc": "2.0", "id": 101, "method": "ping" }), + ) + .await + }); + + let client = SamplingClient.serve(client_io).await?; + let response = raw.await??; + assert!( + response.get("error").is_none(), + "SEP-2260 excepts ping from request association, got {response}" + ); + + client.cancel().await?; + Ok(()) +}