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.

Author: gocamille

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();

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/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<BufReader<ReadHalf<DuplexStream>>>,
+    WriteHalf<DuplexStream>,
+);
+
+async fn raw_initialize(io: DuplexStream, protocol_version: &str) -> anyhow::Result<RawServer> {
+    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<Value> {
+    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(())
+}