feat: route SEP-2260 associated server requests to the originating SSE stream (#1029)

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

* 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.

* feat(transport): route associated server requests to originating SSE stream (SEP-2260)

* test: end-to-end SEP-2260 stream routing over streamable HTTP

* test: register SEP-2260 routing test with required-features; drop dead list_tools

* docs: document SEP-2260 association requirement and stream routing

* docs: fix redundant explicit link on OriginatingRequestId

* 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.

* 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).

* 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.

* docs: single-source SEP-2260 caller requirements on OriginatingRequestId

* 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.

---------

Co-authored-by: Alex Hancock <alexhancock@block.xyz>

crates/rmcp/tests/test_sep_2260_request_association.rs

@@ -0,0 +1,273 @@
+#![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 serde_json::{Value, json};
+use tokio::{
+    io::{AsyncBufReadExt, AsyncWriteExt, BufReader, DuplexStream, Lines, ReadHalf, WriteHalf},
+    sync::oneshot,
+};
+
+#[derive(Clone)]
+struct SamplingServer {
+    outside: Arc<Mutex<Option<oneshot::Sender<Result<(), ServiceError>>>>>,
+}
+
+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<RoleServer>,
+    ) -> Result<CallToolResponse, rmcp::ErrorData> {
+        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<RoleClient>,
+    ) -> Result<CreateMessageResult, rmcp::ErrorData> {
+        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(())
+}
+
+// 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(())
+}

crates/rmcp/tests/test_sep_2260_stream_routing.rs

@@ -0,0 +1,165 @@
+//! 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};
+
+use futures::StreamExt;
+use rmcp::{
+    ErrorData as McpError, RoleServer, ServerHandler,
+    model::{
+        CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock, ElicitRequestParams,
+        ElicitationSchema, ServerCapabilities, ServerInfo,
+    },
+    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 call_tool(
+        &self,
+        _request: CallToolRequestParams,
+        context: RequestContext<RoleServer>,
+    ) -> Result<CallToolResponse, McpError> {
+        // Never answered: the test only checks the request is emitted on the right stream.
+        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();
+
+    // 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")
+        .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);
+
+    // 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")
+        .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();
+}