feat: route SEP-2260 associated server requests to the originating SSE stream

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"

crates/rmcp/src/service.rs

@@ -149,6 +149,29 @@ pub trait ServiceRole: std::fmt::Debug + Send + Sync + 'static + Copy + Clone {
     ) -> impl Future<Output = ()> + MaybeSendFuture {
         async {}
     }
+
+    #[doc(hidden)]
+    fn enforce_request_association(
+        _request: &Self::Req,
+        _peer_info: Option<&Self::PeerInfo>,
+        _in_request_handler_scope: bool,
+    ) -> 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(
@@ -159,6 +182,33 @@ 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()
+}
+
+/// 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.
+///
+/// # 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);
+
 pub type TxJsonRpcMessage<R> =
     JsonRpcMessage<<R as ServiceRole>::Req, <R as ServiceRole>::Resp, <R as ServiceRole>::Not>;
 pub type RxJsonRpcMessage<R> = JsonRpcMessage<
@@ -725,6 +775,16 @@ impl<R: ServiceRole> Peer<R> {
         options: PeerRequestOptions,
         subscription_sender: Option<SubscriptionChannel<R::PeerNot>>,
     ) -> Result<RequestHandle<R>, ServiceError> {
+        R::enforce_request_association(
+            &request,
+            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() {
@@ -1389,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();
@@ -1409,9 +1487,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) => {
@@ -1608,3 +1687,50 @@ 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<RequestId>) -> <RoleServer as ServiceRole>::Req {
+        // peer_info None keeps enforcement non-strict; only the sink message matters.
+        let (peer, mut rx) =
+            Peer::<RoleServer>::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::<OriginatingRequestId>()
+            .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::<OriginatingRequestId>().is_none());
+    }
+}

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<Self>, notification: &Self::PeerNot) {
         match notification {
             ServerNotification::ResourceUpdatedNotification(notification) => {

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.
@@ -744,6 +769,10 @@ impl Peer<RoleServer> {
         }
     }
 
+    /// # SEP-2260: request association
+    ///
+    /// From protocol version `2026-07-28` this must be issued while handling a
+    /// 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"
@@ -778,16 +807,32 @@ impl Peer<RoleServer> {
         }
     }
     method!(
+        /// # SEP-2260: request association
+        ///
+        /// From protocol version `2026-07-28` this must be issued while handling a
+        /// 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"
         )]
         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 must be issued while handling a
+        /// client request; see [`OriginatingRequestId`].
+        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 must be issued while handling a
+        /// client request; see [`OriginatingRequestId`].
+        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));
@@ -1014,6 +1059,11 @@ impl Peer<RoleServer> {
     /// # Ok(())
     /// # }
     /// ```
+    ///
+    /// # SEP-2260: request association
+    ///
+    /// From protocol version `2026-07-28` this must be issued while handling a
+    /// client request; see [`OriginatingRequestId`].
     #[cfg(all(feature = "schemars", feature = "elicitation"))]
     pub async fn elicit<T>(&self, message: impl Into<String>) -> Result<Option<T>, ElicitationError>
     where
@@ -1075,6 +1125,11 @@ impl Peer<RoleServer> {
     /// # Ok(())
     /// # }
     /// ```
+    ///
+    /// # SEP-2260: request association
+    ///
+    /// From protocol version `2026-07-28` this must be issued while handling a
+    /// client request; see [`OriginatingRequestId`].
     #[cfg(all(feature = "schemars", feature = "elicitation"))]
     pub async fn elicit_with_timeout<T>(
         &self,
@@ -1170,6 +1225,11 @@ impl Peer<RoleServer> {
     ///  Ok(())
     /// }
     /// ```
+    ///
+    /// # SEP-2260: request association
+    ///
+    /// From protocol version `2026-07-28` this must be issued while handling a
+    /// client request; see [`OriginatingRequestId`].
     #[cfg(feature = "elicitation")]
     pub async fn elicit_url(
         &self,
@@ -1221,6 +1281,11 @@ impl Peer<RoleServer> {
     ///  Ok(())
     /// }
     /// ```
+    ///
+    /// # SEP-2260: request association
+    ///
+    /// From protocol version `2026-07-28` this must be issued while handling a
+    /// client request; see [`OriginatingRequestId`].
     #[cfg(feature = "elicitation")]
     pub async fn elicit_url_with_timeout(
         &self,