feat: Implement SEP-2260 require server requests to associate with client requests

Author: alexhancock

crates/rmcp/src/service.rs

@@ -136,6 +136,24 @@ pub trait ServiceRole: std::fmt::Debug + Send + Sync + 'static + Copy + Clone {
     fn peer_cancelled_params(_notification: &Self::PeerNot) -> Option<&CancelledNotificationParam> {
         None
     }
+
+    /// SEP-2260 hook: enforce that outbound requests which MUST be associated
+    /// with an originating peer request are only sent from within a request
+    /// handler scope.
+    ///
+    /// This is invoked at the single transport chokepoint
+    /// ([`Peer::send_request_with_option_and_subscription`]) so the guarantee
+    /// holds regardless of which public sending API is used (including the
+    /// generic [`Peer::send_request`]). The default is a no-op; only
+    /// [`RoleServer`] restricts anything.
+    #[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(
@@ -146,6 +164,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<R> =
     JsonRpcMessage<<R as ServiceRole>::Req, <R as ServiceRole>::Resp, <R as ServiceRole>::Not>;
 pub type RxJsonRpcMessage<R> = JsonRpcMessage<
@@ -706,6 +732,15 @@ impl<R: ServiceRole> Peer<R> {
         options: PeerRequestOptions,
         subscription_sender: Option<SubscriptionChannel<R::PeerNot>>,
     ) -> Result<RequestHandle<R>, ServiceError> {
+        // SEP-2260: enforce request-association at the single transport
+        // chokepoint. This covers every public sending API (typed convenience
+        // methods and the generic `send_request`/`send_cancellable_request`),
+        // so the guarantee cannot be bypassed.
+        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() {
@@ -1379,9 +1414,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) => {

crates/rmcp/src/service/server.rs

@@ -49,6 +49,40 @@ impl ServiceRole for RoleServer {
             _ => None,
         }
     }
+
+    /// SEP-2260: `sampling/createMessage`, `roots/list`, and
+    /// `elicitation/create` MUST be associated with an originating
+    /// client-to-server request. We enforce this for peers negotiating
+    /// protocol `2026-07-28` or newer by requiring an active request-handler
+    /// scope (see [`crate::service::ORIGINATING_REQUEST`]).
+    ///
+    /// `ping` is explicitly excepted, and `CustomRequest` is left to the
+    /// implementor. Older peers are not enforced, preserving backward
+    /// compatibility.
+    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.

crates/rmcp/src/task_manager.rs

@@ -349,8 +349,18 @@ impl TaskManager {
         let future = make_future(context);
         let inner = self.inner.clone();
         let id_for_task = task_id.clone();
+        // SEP-2260: a task operation runs on a detached `tokio::spawn`, which
+        // does not inherit the caller's task-local `ORIGINATING_REQUEST`
+        // scope. Capture the originating client request id (if we are inside a
+        // request handler, which is the normal case for `tools/call`) and
+        // re-establish it inside the spawned task. Without this, legitimate
+        // nested sampling/elicitation/roots requests surfaced mid-task would
+        // be wrongly rejected as unassociated.
+        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 +548,24 @@ fn unknown_task(task_id: &str) -> McpError {
     McpError::invalid_params(format!("unknown task: {task_id}"), None)
 }
 
+/// Run a task operation future, re-establishing the originating request's
+/// SEP-2260 association scope if one was captured at spawn time.
+///
+/// When the operation was started from within a request handler (the normal
+/// `tools/call` path), `originating_request` carries that request's id and the
+/// future runs inside an `ORIGINATING_REQUEST` scope so nested
+/// sampling/elicitation/roots requests are recognized as associated. When
+/// there is no originating request, the future runs unscoped.
+async fn run_task_operation(
+    originating_request: Option<crate::model::RequestId>,
+    future: TaskFuture,
+) -> Result<CallToolResult, TaskExit> {
+    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 +945,77 @@ mod tests {
         }
         panic!("task did not complete after input response");
     }
+
+    // SEP-2260 (GAP 2): a task operation runs on a detached `tokio::spawn`,
+    // which does not inherit the caller's `ORIGINATING_REQUEST` task-local.
+    // Verify the originating request's association scope is re-established
+    // inside the spawned task so nested sampling/elicitation/roots requests
+    // are recognized as associated.
+    #[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::<bool>));
+        let observed_in_task = observed.clone();
+
+        // Spawn from within a request-handler scope (as `tools/call` does).
+        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");
+    }
+
+    // SEP-2260 (GAP 2): a task spawned outside any request-handler scope must
+    // not fabricate an association scope.
+    #[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::<bool>));
+        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");
+    }
 }