chore(deps): update rstest requirement from 0.26.1 to 0.27.0 (#1276)

Updates the requirements on [rstest](https://github.com/la10736/rstest) to permit the latest version.
- [Release notes](https://github.com/la10736/rstest/releases)
- [Changelog](https://github.com/la10736/rstest/blob/master/CHANGELOG.md)
- [Commits](https://github.com/la10736/rstest/compare/v0.26.1...v0.27.0)

---
updated-dependencies:
- dependency-name: rstest
  dependency-version: 0.27.0
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.github/workflows/ci.yml

@@ -85,7 +85,7 @@ jobs:
       - uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2
 
       - name: Install cargo-semver-checks
-        uses: taiki-e/install-action@7b8d4719ee4aaa279bdf55df38dacb9ebfe12a6c # v2.87.6
+        uses: taiki-e/install-action@3f74d7c16a4242f1c95561e98edc25d36adb4375 # v2.87.12
         with:
           tool: cargo-semver-checks
 
@@ -139,7 +139,7 @@ jobs:
       - uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2
 
       - name: Install cargo-public-api
-        uses: taiki-e/install-action@7b8d4719ee4aaa279bdf55df38dacb9ebfe12a6c # v2.87.6
+        uses: taiki-e/install-action@3f74d7c16a4242f1c95561e98edc25d36adb4375 # v2.87.12
         with:
           tool: cargo-public-api
 
@@ -251,7 +251,7 @@ jobs:
           node-version: '22'
           
       - name: Install uv
-        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+        uses: astral-sh/setup-uv@bec219d24cd3e171d82865faccec33120bb574f4 # v10.1.0
         
       - name: Install Rust
         uses: dtolnay/rust-toolchain@4360b52568e2003a75bf9bc1d59f33a8e3fc893c # stable
@@ -280,7 +280,7 @@ jobs:
           node-version: '22'
 
       - name: Install uv
-        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+        uses: astral-sh/setup-uv@bec219d24cd3e171d82865faccec33120bb574f4 # v10.1.0
 
       - name: Install Rust
         uses: dtolnay/rust-toolchain@4360b52568e2003a75bf9bc1d59f33a8e3fc893c # stable
@@ -316,7 +316,7 @@ jobs:
           node-version: '22'
           
       - name: Install uv
-        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+        uses: astral-sh/setup-uv@bec219d24cd3e171d82865faccec33120bb574f4 # v10.1.0
         
       - name: Install Rust
         uses: dtolnay/rust-toolchain@4360b52568e2003a75bf9bc1d59f33a8e3fc893c # stable
@@ -351,7 +351,7 @@ jobs:
           node-version: '22'
 
       - name: Install uv
-        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+        uses: astral-sh/setup-uv@bec219d24cd3e171d82865faccec33120bb574f4 # v10.1.0
 
       - name: Install Rust
         uses: dtolnay/rust-toolchain@4360b52568e2003a75bf9bc1d59f33a8e3fc893c # stable

.github/workflows/codeql.yml

@@ -27,13 +27,13 @@ jobs:
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
 
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@cdf488f595d80d6e07e03d4674febd5ab45fa938 # v4.37.9
+        uses: github/codeql-action/init@b96794f015dfd88f77b49b1c93e0fa7110f94c63 # v4.38.0
         with:
           languages: ${{ matrix.language }}
           config-file: ./.github/codeql/codeql-config.yml
 
       - name: Autobuild
-        uses: github/codeql-action/autobuild@cdf488f595d80d6e07e03d4674febd5ab45fa938 # v4.37.9
+        uses: github/codeql-action/autobuild@b96794f015dfd88f77b49b1c93e0fa7110f94c63 # v4.38.0
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@cdf488f595d80d6e07e03d4674febd5ab45fa938 # v4.37.9
+        uses: github/codeql-action/analyze@b96794f015dfd88f77b49b1c93e0fa7110f94c63 # v4.38.0

.github/workflows/scorecard.yml

@@ -0,0 +1,48 @@
+# OpenSSF Scorecard supply-chain security analysis.
+name: Scorecard supply-chain security
+
+on:
+  branch_protection_rule:
+  schedule:
+    - cron: '5 23 * * 5'
+  push:
+    branches: [main]
+  pull_request:
+  workflow_dispatch:
+
+permissions: read-all
+
+jobs:
+  analysis:
+    name: Scorecard analysis
+    runs-on: ubuntu-latest
+    permissions:
+      security-events: write
+      id-token: write
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+
+      - name: Run analysis
+        uses: ossf/scorecard-action@4eaacf0543bb3f2c246792bd56e8cdeffafb205a # v2.4.3
+        with:
+          results_file: results.sarif
+          results_format: sarif
+          # Publishing only succeeds on the default branch. PR and manual runs
+          # still validate the analysis and produce SARIF without publishing.
+          publish_results: ${{ github.event_name != 'pull_request' && github.ref == format('refs/heads/{0}', github.event.repository.default_branch) }}
+
+      - name: Upload SARIF artifact
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: scorecard-results
+          path: results.sarif
+          retention-days: 5
+
+      - name: Upload to code scanning
+        uses: github/codeql-action/upload-sarif@b96794f015dfd88f77b49b1c93e0fa7110f94c63 # v4.38.0
+        with:
+          sarif_file: results.sarif

Cargo.toml

@@ -4,13 +4,13 @@ default-members = ["crates/rmcp", "crates/rmcp-macros"]
 resolver = "2"
 
 [workspace.dependencies]
-rmcp = { version = "3.3.0", path = "./crates/rmcp" }
+rmcp = { version = "3.4.0", path = "./crates/rmcp" }
-rmcp-macros = { version = "3.3.0", path = "./crates/rmcp-macros" }
+rmcp-macros = { version = "3.4.0", path = "./crates/rmcp-macros" }
 
 [workspace.package]
 edition = "2024"
 rust-version = "1.88"
-version = "3.3.0"
+version = "3.4.0"
 authors = ["4t145 <u4t145@163.com>"]
 license = "Apache-2.0"
 repository = "https://github.com/modelcontextprotocol/rust-sdk/"

README.md

@@ -99,10 +99,10 @@ Use [`ClientServiceExt::serve_with_lifecycle`](crates/rmcp/src/service/client.rs
 select another lifecycle explicitly:
 
 ```rust, ignore
-use rmcp::{ClientInfo, ClientLifecycleMode, ClientServiceExt, ProtocolVersion};
+use rmcp::{ClientLifecycleMode, ClientServiceExt, ProtocolVersion, model::ClientConfig};
 
 // Start directly with server/discover and include client metadata on every request.
-let client = ClientInfo::default()
+let client = ClientConfig::default()
     .serve_with_lifecycle(
         transport,
         ClientLifecycleMode::Discover {
@@ -113,7 +113,7 @@ let client = ClientInfo::default()
 
 // Or probe the discover lifecycle and fall back when a legacy server reports
 // that server/discover is not implemented or does not respond within 10 seconds.
-let client = ClientInfo::default()
+let client = ClientConfig::default()
     .serve_with_lifecycle(
         transport,
         ClientLifecycleMode::Auto {
@@ -369,8 +369,8 @@ use serde_json::json;
 struct MyServer;
 
 impl ServerHandler for MyServer {
-    fn get_info(&self) -> ServerInfo {
+    fn get_info(&self) -> ServerConfig {
-        ServerInfo::new(
+        ServerConfig::new(
             ServerCapabilities::builder()
                 .enable_resources()
                 .build(),
@@ -582,8 +582,8 @@ impl MyServer {
 
 #[prompt_handler]
 impl ServerHandler for MyServer {
-    fn get_info(&self) -> ServerInfo {
+    fn get_info(&self) -> ServerConfig {
-        ServerInfo::new(ServerCapabilities::builder().enable_prompts().build())
+        ServerConfig::new(ServerCapabilities::builder().enable_prompts().build())
     }
 }
 ```
@@ -934,8 +934,8 @@ Enable the logging capability, handle level changes from the client, and send lo
 use rmcp::{ServerHandler, model::*, service::RequestContext};
 
 impl ServerHandler for MyServer {
-    fn get_info(&self) -> ServerInfo {
+    fn get_info(&self) -> ServerConfig {
-        ServerInfo::new(
+        ServerConfig::new(
             ServerCapabilities::builder()
                 .enable_logging()
                 .build(),
@@ -1007,8 +1007,8 @@ Enable the completions capability and implement the `complete()` handler. Use `r
 use rmcp::{ErrorData as McpError, ServerHandler, model::*, service::RequestContext, RoleServer};
 
 impl ServerHandler for MyServer {
-    fn get_info(&self) -> ServerInfo {
+    fn get_info(&self) -> ServerConfig {
-        ServerInfo::new(
+        ServerConfig::new(
             ServerCapabilities::builder()
                 .enable_completions()
                 .enable_prompts()
@@ -1253,8 +1253,8 @@ use rmcp::{
 };
 
 impl ServerHandler for MyServer {
-    fn get_info(&self) -> ServerInfo {
+    fn get_info(&self) -> ServerConfig {
-        ServerInfo::new(
+        ServerConfig::new(
             ServerCapabilities::builder()
                 .enable_tools()
                 .enable_tool_list_changed()
@@ -1584,7 +1584,7 @@ use rmcp::transport::StreamableHttpClientTransport;
 
 // Defaults are stateless-friendly.
 let transport = StreamableHttpClientTransport::from_uri("http://localhost:8000/mcp");
-let client = ClientInfo::default().serve(transport).await?;
+let client = ClientConfig::default().serve(transport).await?;
 ```
 
 **Example:** [`examples/servers/src/counter_streamhttp.rs`](examples/servers/src/counter_streamhttp.rs) (server), [`examples/clients/src/streamable_http.rs`](examples/clients/src/streamable_http.rs) (client)
@@ -1638,7 +1638,7 @@ server example). The client transport connects with a single URI:
 use rmcp::transport::StreamableHttpClientTransport;
 
 let transport = StreamableHttpClientTransport::from_uri("http://localhost:8000/mcp");
-let client = ClientInfo::default().serve(transport).await?;
+let client = ClientConfig::default().serve(transport).await?;
 ```
 
 The client allows up to 16 ordinary http POSTs at once. Configure this with
@@ -1742,10 +1742,10 @@ knows what the other supports. Declare yours with the `ServerCapabilities`
 builder in `get_info()`:
 
 ```rust,ignore
-use rmcp::model::{ServerCapabilities, ServerInfo};
+use rmcp::model::{ServerCapabilities, ServerConfig};
 
-fn get_info(&self) -> ServerInfo {
+fn get_info(&self) -> ServerConfig {
-    ServerInfo::new(
+    ServerConfig::new(
         ServerCapabilities::builder()
             .enable_tools()
             .enable_prompts()

conformance/src/bin/client.rs

@@ -74,8 +74,8 @@ impl ClientHandler for BasicClientHandler {}
 struct ElicitationDefaultsClientHandler;
 
 impl ClientHandler for ElicitationDefaultsClientHandler {
-    fn get_info(&self) -> ClientInfo {
+    fn get_info(&self) -> ClientConfig {
-        let mut info = ClientInfo::default();
+        let mut info = ClientConfig::default();
         info.capabilities.elicitation = Some(
             ElicitationCapability::new()
                 .with_form(FormElicitationCapability::new().with_schema_validation(true)),
@@ -163,8 +163,8 @@ impl ClientHandler for ElicitationDefaultsClientHandler {
 struct FullClientHandler;
 
 impl ClientHandler for FullClientHandler {
-    fn get_info(&self) -> ClientInfo {
+    fn get_info(&self) -> ClientConfig {
-        let mut info = ClientInfo::default();
+        let mut info = ClientConfig::default();
         info.capabilities.elicitation = Some(
             ElicitationCapability::new()
                 .with_form(FormElicitationCapability::new().with_schema_validation(true)),

conformance/src/bin/server.rs

@@ -704,8 +704,8 @@ impl ServerHandler for ConformanceServer {
         (name == "test_custom_header").then(custom_header_tool)
     }
 
-    fn get_info(&self) -> ServerInfo {
+    fn get_info(&self) -> ServerConfig {
-        ServerInfo::new(
+        ServerConfig::new(
             ServerCapabilities::builder()
                 .enable_prompts()
                 .enable_prompts_list_changed()
@@ -1744,7 +1744,9 @@ async fn main() -> anyhow::Result<()> {
     tracing::info!("Starting conformance server on {}", bind_addr);
 
     let server = ConformanceServer::new();
-    let config = StreamableHttpServerConfig::default();
+    let config = StreamableHttpServerConfig::default()
+        .with_allowed_origins([format!("http://{bind_addr}")])
+        .enforce_origin_validation();
     let service = StreamableHttpService::new(
         move || Ok(server.clone()),
         LocalSessionManager::default().into(),

crates/rmcp-macros/CHANGELOG.md

@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [3.4.0](https://github.com/modelcontextprotocol/rust-sdk/compare/rmcp-macros-v3.3.0...rmcp-macros-v3.4.0) - 2026-09-15
+
+### Added
+
+- *(model)* add ServerConfig and ClientConfig ([#1266](https://github.com/modelcontextprotocol/rust-sdk/pull/1266))
+
+### Fixed
+
+- *(model)* deprecate ServerInfo and ClientInfo aliases ([#1156](https://github.com/modelcontextprotocol/rust-sdk/pull/1156))
+
 ## [3.3.0](https://github.com/modelcontextprotocol/rust-sdk/compare/rmcp-macros-v3.2.0...rmcp-macros-v3.3.0) - 2026-09-10
 
 ### Added

crates/rmcp-macros/src/lib.rs

@@ -217,8 +217,8 @@ pub fn tool_router(attr: TokenStream, input: TokenStream) -> TokenStream {
 /// ```rust,ignore
 /// #[tool_handler]
 /// impl ServerHandler for MyToolHandler {
-///     fn get_info(&self) -> ServerInfo {
+///     fn get_info(&self) -> ServerConfig {
-///         ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
+///         ServerConfig::new(ServerCapabilities::builder().enable_tools().build())
 ///     }
 /// }
 /// ```

crates/rmcp-macros/src/tool_handler.rs

@@ -116,7 +116,7 @@ pub(crate) enum CallerCapability {
     Prompts,
 }
 
-/// Build a `get_info()` method that returns `ServerInfo` with the appropriate capabilities.
+/// Build a `get_info()` method that returns `ServerConfig` with the appropriate capabilities.
 ///
 /// The caller declares its own capability via `caller`. Sibling handler attributes
 /// (`prompt_handler`, `tool_handler`) are detected automatically
@@ -157,8 +157,8 @@ pub(crate) fn build_get_info(
     }
 
     syn::parse2::<ImplItem>(quote! {
-        fn get_info(&self) -> rmcp::model::ServerInfo {
+        fn get_info(&self) -> rmcp::model::ServerConfig {
-            rmcp::model::ServerInfo::new(
+            rmcp::model::ServerConfig::new(
                 rmcp::model::ServerCapabilities::builder()
                     #(#capability_calls)*
                     .build()

crates/rmcp/CHANGELOG.md

@@ -7,6 +7,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [3.4.0](https://github.com/modelcontextprotocol/rust-sdk/compare/rmcp-v3.3.0...rmcp-v3.4.0) - 2026-09-15
+
+### Added
+
+- *(model)* add ServerConfig and ClientConfig ([#1266](https://github.com/modelcontextprotocol/rust-sdk/pull/1266))
+
+### Fixed
+
+- *(rmcp)* use ServerConfig in cancellation test ([#1273](https://github.com/modelcontextprotocol/rust-sdk/pull/1273))
+- *(rmcp)* route peer cancellation by lifecycle ([#1262](https://github.com/modelcontextprotocol/rust-sdk/pull/1262))
+- run first pre-init request in service loop ([#1263](https://github.com/modelcontextprotocol/rust-sdk/pull/1263))
+- *(auth)* let url origin decide prm discovery ([#1264](https://github.com/modelcontextprotocol/rust-sdk/pull/1264))
+- *(model)* deprecate ServerInfo and ClientInfo aliases ([#1156](https://github.com/modelcontextprotocol/rust-sdk/pull/1156))
+- *(auth)* ignore non-metadata JSON when probing for protected resource metadata ([#1204](https://github.com/modelcontextprotocol/rust-sdk/pull/1204))
+- do not treat malformed JSON 200 as Accepted for requests ([#1208](https://github.com/modelcontextprotocol/rust-sdk/pull/1208))
+- *(streamable-http-server)* map handler-generated HeaderMismatch to HTTP 400 ([#1259](https://github.com/modelcontextprotocol/rust-sdk/pull/1259))
+- *(http)* enforce Origin validation semantics ([#1192](https://github.com/modelcontextprotocol/rust-sdk/pull/1192))
+
 ## [3.3.0](https://github.com/modelcontextprotocol/rust-sdk/compare/rmcp-v3.2.0...rmcp-v3.3.0) - 2026-09-10
 
 ### Added

crates/rmcp/Cargo.toml

@@ -215,7 +215,7 @@ tracing-subscriber = { version = "0.3", features = [
   "fmt",
 ] }
 async-trait = "0.1"
-rstest = "0.26.1"
+rstest = "0.27.0"
 [[test]]
 name = "test_tool_macros"
 required-features = ["server", "client"]
@@ -348,6 +348,15 @@ name = "test_streamable_http_4xx_error_body"
 required-features = ["transport-streamable-http-client", "transport-streamable-http-client-reqwest"]
 path = "tests/test_streamable_http_4xx_error_body.rs"
 
+[[test]]
+name = "test_streamable_http_malformed_json_request"
+required-features = [
+  "client",
+  "transport-streamable-http-client",
+  "transport-streamable-http-client-reqwest",
+]
+path = "tests/test_streamable_http_malformed_json_request.rs"
+
 
 [[test]]
 name = "test_custom_request"

crates/rmcp/src/handler/client.rs

@@ -274,8 +274,8 @@ macro_rules! client_handler_methods {
             std::future::ready(())
         }
 
-        fn get_info(&self) -> ClientInfo {
+        fn get_info(&self) -> ClientConfig {
-            ClientInfo::default()
+            ClientConfig::default()
         }
     };
 }
@@ -292,12 +292,12 @@ pub trait ClientHandler: Sized + 'static {
     client_handler_methods!();
 }
 
-/// Do nothing, with default client info.
+/// Do nothing, with the default client config.
 impl ClientHandler for () {}
 
-/// Do nothing, with a specific client info.
+/// Do nothing, with a specific client config.
-impl ClientHandler for ClientInfo {
+impl ClientHandler for ClientConfig {
-    fn get_info(&self) -> ClientInfo {
+    fn get_info(&self) -> ClientConfig {
         self.clone()
     }
 }
@@ -422,7 +422,7 @@ macro_rules! impl_client_handler_for_wrapper {
                 (**self).on_custom_notification(notification, context)
             }
 
-            fn get_info(&self) -> ClientInfo {
+            fn get_info(&self) -> ClientConfig {
                 (**self).get_info()
             }
         }

crates/rmcp/src/handler/server.rs

@@ -334,15 +334,15 @@ macro_rules! server_handler_methods {
         /// ```
         /// use rmcp::{
         ///     ErrorData as McpError, RoleServer, ServerHandler,
-        ///     model::{InitializeRequestParams, InitializeResult, ServerInfo},
+        ///     model::{InitializeRequestParams, InitializeResult, ServerConfig},
         ///     service::RequestContext,
         /// };
         ///
         /// struct MyServer;
         ///
         /// impl ServerHandler for MyServer {
-        ///     fn get_info(&self) -> ServerInfo {
+        ///     fn get_info(&self) -> ServerConfig {
-        ///         ServerInfo::default()
+        ///         ServerConfig::default()
         ///     }
         ///
         ///     async fn initialize(
@@ -600,8 +600,8 @@ macro_rules! server_handler_methods {
             std::future::ready(())
         }
 
-        fn get_info(&self) -> ServerInfo {
+        fn get_info(&self) -> ServerConfig {
-            ServerInfo::default()
+            ServerConfig::default()
         }
 
         /// SEP-2663 `tasks/get`: return the current [`DetailedTask`] state.
@@ -839,7 +839,7 @@ macro_rules! impl_server_handler_for_wrapper {
                 (**self).on_custom_notification(notification, context)
             }
 
-            fn get_info(&self) -> ServerInfo {
+            fn get_info(&self) -> ServerConfig {
                 (**self).get_info()
             }
 

crates/rmcp/src/model.rs

@@ -1119,7 +1119,58 @@ impl InitializeResult {
     }
 }
 
+/// A server's own configuration: the protocol version, capabilities,
+/// implementation identity, and instructions it advertises to clients.
+///
+/// This is what `ServerHandler::get_info` returns. It is an alias for
+/// [`InitializeResult`] because the same value is sent as the `initialize`
+/// response on the wire.
+///
+/// # Examples
+///
+/// ```
+/// use rmcp::model::{ServerCapabilities, ServerConfig};
+///
+/// let config = ServerConfig::new(ServerCapabilities::builder().enable_tools().build())
+///     .with_instructions("Call `add` to sum two numbers.");
+/// assert!(config.capabilities.tools.is_some());
+/// ```
+pub type ServerConfig = InitializeResult;
+
+/// A client's own configuration: the protocol version, capabilities, and
+/// implementation identity it advertises to servers.
+///
+/// This is what `ClientHandler::get_info` returns. It is an alias for
+/// [`InitializeRequestParams`] because the same value is sent as the
+/// `initialize` request on the wire.
+///
+/// # Examples
+///
+/// ```
+/// use rmcp::model::{ClientCapabilities, ClientConfig, Implementation};
+///
+/// let config = ClientConfig::new(
+///     ClientCapabilities::builder().enable_elicitation().build(),
+///     Implementation::new("my-client", "1.0.0"),
+/// );
+/// assert!(config.capabilities.elicitation.is_some());
+/// ```
+pub type ClientConfig = InitializeRequestParams;
+
+/// Deprecated alias for [`ServerConfig`].
+///
+/// The name collides with the protocol's `serverInfo` field, which is only the
+/// [`Implementation`] identity, so `server_info.server_info` was easy to
+/// misread (#1082).
+#[deprecated(note = "use `ServerConfig` instead")]
 pub type ServerInfo = InitializeResult;
+
+/// Deprecated alias for [`ClientConfig`].
+///
+/// The name collides with the protocol's `clientInfo` field, which is only the
+/// [`Implementation`] identity, so `client_info.client_info` was easy to
+/// misread (#1082).
+#[deprecated(note = "use `ClientConfig` instead")]
 pub type ClientInfo = InitializeRequestParams;
 
 /// Information negotiated about a server peer.
@@ -1279,18 +1330,18 @@ impl DiscoverResult {
         self
     }
 
-    /// Create a discovery result from the server's initialization information.
+    /// Create a discovery result from the server's configuration.
     pub fn from_server_info(
         supported_versions: Vec<ProtocolVersion>,
-        server_info: ServerInfo,
+        server_config: ServerConfig,
     ) -> Self {
-        let ServerInfo {
+        let ServerConfig {
             capabilities,
             server_info,
             instructions,
             meta,
             ..
-        } = server_info;
+        } = server_config;
         let mut result = Self {
             result_type: ResultType::COMPLETE,
             supported_versions,
@@ -1332,9 +1383,9 @@ impl ServerPeerInfo {
 }
 
 #[allow(clippy::derivable_impls)]
-impl Default for ServerInfo {
+impl Default for InitializeResult {
     fn default() -> Self {
-        ServerInfo {
+        InitializeResult {
             protocol_version: ProtocolVersion::default(),
             capabilities: ServerCapabilities::default(),
             server_info: Implementation::from_build_env(),
@@ -1345,9 +1396,9 @@ impl Default for ServerInfo {
 }
 
 #[allow(clippy::derivable_impls)]
-impl Default for ClientInfo {
+impl Default for InitializeRequestParams {
     fn default() -> Self {
-        ClientInfo {
+        InitializeRequestParams {
             meta: None,
             protocol_version: ProtocolVersion::default(),
             capabilities: ClientCapabilities::default(),

crates/rmcp/src/service.rs

@@ -136,6 +136,14 @@ pub trait ServiceRole: std::fmt::Debug + Send + Sync + 'static + Copy + Clone {
     fn peer_cancelled_params(_notification: &Self::PeerNot) -> Option<&CancelledNotificationParam> {
         None
     }
+    #[doc(hidden)]
+    fn peer_cancels_subscriptions(_peer: &Peer<Self>) -> bool {
+        false
+    }
+    #[doc(hidden)]
+    fn is_subscription_request(_request: &Self::Req) -> bool {
+        false
+    }
     /// Invalidate any response cache affected by an inbound peer notification.
     ///
     /// The serve loop calls this for every notification *before* subscription
@@ -522,6 +530,10 @@ impl ProgressNotificationToken for ServerNotification {
 }
 
 type Responder<T> = tokio::sync::oneshot::Sender<T>;
+struct PendingResponse<T> {
+    responder: Responder<T>,
+    is_subscription: bool,
+}
 type ProgressTimeoutWatchers = Arc<tokio::sync::RwLock<HashMap<ProgressToken, mpsc::Sender<()>>>>;
 type SubscriptionChannel<N> = (mpsc::Sender<N>, usize);
 type SubscriptionChannelMap<N> = HashMap<RequestId, SubscriptionChannel<N>>;
@@ -1298,7 +1310,14 @@ where
 {
     let (peer, peer_rx) = Peer::new(Arc::new(AtomicU32RequestIdProvider::default()), peer_info);
     R::configure_direct_peer(&peer, &service.get_info());
-    serve_inner(service, transport.into_transport(), peer, peer_rx, ct)
+    serve_inner(
+        service,
+        transport.into_transport(),
+        peer,
+        peer_rx,
+        VecDeque::new(),
+        ct,
+    )
 }
 
 /// Spawn a task that may hold `!Send` state when the `local` feature is active.
@@ -1323,12 +1342,19 @@ where
     tokio::task::spawn_local(future)
 }
 
+/// Run the service loop over `transport`.
+///
+/// `initial_messages` are messages the caller already read from `transport`
+/// (e.g. the first request of an `initialize`-less session). They are
+/// dispatched by the loop, in order, before anything else is read from the
+/// transport, so that their handlers run with the loop draining `peer_rx`.
 #[instrument(skip_all)]
 fn serve_inner<R, S, T>(
     service: S,
     transport: T,
     peer: Peer<R>,
     mut peer_rx: tokio::sync::mpsc::Receiver<PeerSinkMessage<R>>,
+    initial_messages: VecDeque<RxJsonRpcMessage<R>>,
     ct: CancellationToken,
 ) -> RunningService<R, S>
 where
@@ -1348,7 +1374,7 @@ where
     }
 
     let mut local_responder_pool =
-        HashMap::<RequestId, Responder<Result<R::PeerResp, ServiceError>>>::new();
+        HashMap::<RequestId, PendingResponse<Result<R::PeerResp, ServiceError>>>::new();
     let mut local_ct_pool = HashMap::<RequestId, CancellationToken>::new();
     let shared_service = Arc::new(service);
     // for return
@@ -1361,7 +1387,7 @@ where
     let current_span = tracing::Span::current();
     let handle = spawn_service_task(async move {
         let mut transport = transport.into_transport();
-        let mut batch_messages = VecDeque::<RxJsonRpcMessage<R>>::new();
+        let mut batch_messages = initial_messages;
         let mut send_task_set = tokio::task::JoinSet::<SendTaskResult>::new();
         let mut response_send_tasks = tokio::task::JoinSet::<()>::new();
         #[derive(Debug)]
@@ -1445,7 +1471,7 @@ where
                 Event::SendTaskResult(SendTaskResult::Request { id, result }) => {
                     if let Err(e) = result
                         && let Some(responder) = local_responder_pool.remove(&id) {
-                            let _ = responder.send(Err(ServiceError::TransportSend(e)));
+                            let _ = responder.responder.send(Err(ServiceError::TransportSend(e)));
                         }
                 }
                 Event::SendTaskResult(SendTaskResult::Notification {
@@ -1463,7 +1489,7 @@ where
                         && let Some(request_id) = &param.request_id
                             && let Some(responder) = local_responder_pool.remove(request_id) {
                                 tracing::info!(id = %request_id, reason = param.reason, "cancelled");
-                                let _response_result = responder.send(Err(ServiceError::Cancelled {
+                                let _response_result = responder.responder.send(Err(ServiceError::Cancelled {
                                     reason: param.reason.clone(),
                                 }));
                             }
@@ -1500,7 +1526,10 @@ where
                     id,
                     responder,
                 }) => {
-                    local_responder_pool.insert(id.clone(), responder);
+                    local_responder_pool.insert(id.clone(), PendingResponse {
+                        responder,
+                        is_subscription: R::is_subscription_request(&request),
+                    });
                     let send = transport.send(JsonRpcMessage::request(request, id.clone()));
                     {
                         let id = id.clone();
@@ -1601,31 +1630,38 @@ where
                 })) => {
                     tracing::info!(?notification, "received notification");
                     R::invalidate_response_cache(&peer, &notification).await;
-                    let cancellation_request_id =
+                    let subscription_id =
                         if let Some(cancelled) = R::peer_cancelled_params(&notification) {
                             let request_id = cancelled.request_id.clone();
-                            if let Some(request_id) = request_id.as_ref() {
+                            // Modern servers cancel listen requests; legacy peers cancel
-                                if R::IS_CLIENT {
+                            // requests they originated, even when both directions share an ID.
-                                    if let Some(responder) =
+                            if R::peer_cancels_subscriptions(&peer) {
-                                        local_responder_pool.remove(request_id)
+                                if let Some(request_id) = request_id.as_ref() {
+                                    if local_responder_pool.get(request_id)
+                                        .is_some_and(|pending| pending.is_subscription)
+                                        && let Some(pending) = local_responder_pool.remove(request_id)
                                     {
-                                        let _ = responder.send(Err(ServiceError::Cancelled {
+                                        let _ = pending.responder.send(Err(ServiceError::Cancelled {
                                             reason: cancelled.reason.clone(),
                                         }));
+                                    } else {
+                                        tracing::debug!(%request_id, "ignoring cancellation of unknown subscription");
+                                        continue;
                                     }
-                                } else if let Some(ct) = local_ct_pool.remove(request_id) {
+                                }
+                                request_id
+                            } else {
+                                if let Some(request_id) = request_id.as_ref()
+                                    && let Some(ct) = local_ct_pool.remove(request_id)
+                                {
                                     tracing::info!(id = %request_id, reason = cancelled.reason, "cancelled");
                                     ct.cancel();
                                 }
+                                None
                             }
-                            request_id
                         } else {
-                            None
+                            notification.get_meta().subscription_id()
                         };
-                    let subscription_id = notification
-                        .get_meta()
-                        .subscription_id()
-                        .or(cancellation_request_id);
                     if let Some(subscription_id) = subscription_id
                         && let Some((sender, capacity)) =
                             peer.subscription_sender(&subscription_id)
@@ -1642,7 +1678,7 @@ where
                                     && let Some(responder) =
                                         local_responder_pool.remove(&subscription_id)
                                 {
-                                    let _ = responder
+                                    let _ = responder.responder
                                         .send(Err(ServiceError::SubscriptionLagged { capacity }));
                                 }
                                 peer.unregister_subscription(&subscription_id);
@@ -1694,7 +1730,7 @@ where
                     if let Some(responder) =
                         remove_pending_request(&mut local_responder_pool, &id)
                     {
-                        let response_result = responder.send(Ok(result));
+                        let response_result = responder.responder.send(Ok(result));
                         if let Err(_error) = response_result {
                             tracing::warn!(%id, "Error sending response");
                         }
@@ -1715,7 +1751,7 @@ where
                         } else {
                             ServiceError::McpError(error)
                         };
-                        let _response_result = responder.send(Err(service_error));
+                        let _response_result = responder.responder.send(Err(service_error));
                         if let Err(_error) = _response_result {
                             tracing::warn!(%id, "Error sending response");
                         }
@@ -1749,6 +1785,17 @@ where
                 // Then drain any handler responses still in the channel
                 // (handlers that finished after the loop broke).
                 while let Some(m) = sink_proxy_rx.recv().await {
+                    if let Some(id) = match &m {
+                        JsonRpcMessage::Response(response) => Some(&response.id),
+                        JsonRpcMessage::Error(error) => error.id.as_ref(),
+                        _ => None,
+                    } {
+                        let Some(ct) = local_ct_pool.remove(id) else {
+                            tracing::debug!(%id, "dropping response for cancelled request");
+                            continue;
+                        };
+                        ct.cancel();
+                    }
                     if let Err(error) = transport.send(m).await {
                         tracing::error!(%error, "failed to send pending response during drain");
                         break;

crates/rmcp/src/service/client.rs

@@ -13,13 +13,13 @@ use crate::{
     model::{
         ArgumentInfo, CacheScope, CallToolRequest, CallToolRequestParams, CallToolResponse,
         CallToolResult, CancelTaskParams, CancelTaskRequest, CancelledNotification,
-        CancelledNotificationParam, ClientInfo, ClientJsonRpcMessage, ClientNotification,
+        CancelledNotificationParam, ClientConfig, ClientJsonRpcMessage, ClientNotification,
         ClientRequest, ClientResult, CompleteRequest, CompleteRequestParams, CompleteResult,
         CompletionContext, CompletionInfo, DEFAULT_MRTR_MAX_ROUNDS, DiscoverRequest,
         DiscoverRequestParams, DiscoverResult, ErrorData, GetExtensions, GetMeta, GetPromptRequest,
         GetPromptRequestParams, GetPromptResponse, GetPromptResult, GetTaskParams, GetTaskRequest,
-        GetTaskResult, InitializeRequest, InitializedNotification, InputRequest,
+        GetTaskResult, InitializeRequest, InitializeRequestParams, InitializedNotification,
-        InputRequiredResult, InputResponses, JsonRpcResponse, ListPromptsRequest,
+        InputRequest, InputRequiredResult, InputResponses, JsonRpcResponse, ListPromptsRequest,
         ListPromptsResult, ListResourceTemplatesRequest, ListResourceTemplatesResult,
         ListResourcesRequest, ListResourcesResult, ListToolsRequest, ListToolsResult,
         NumberOrString, PaginatedRequestParams, ProgressNotification, ProgressNotificationParam,
@@ -266,7 +266,7 @@ impl ServiceRole for RoleClient {
     type PeerReq = ServerRequest;
     type PeerResp = ServerResult;
     type PeerNot = ServerNotification;
-    type Info = ClientInfo;
+    type Info = ClientConfig;
     type PeerInfo = ServerPeerInfo;
     type InitializeError = ClientInitializeError;
     const IS_CLIENT: bool = true;
@@ -292,6 +292,20 @@ impl ServiceRole for RoleClient {
         }
     }
 
+    fn peer_cancels_subscriptions(peer: &Peer<Self>) -> bool {
+        // Discovery keeps modern lifecycle semantics even with an older application version.
+        !super::uses_legacy_lifecycle(
+            peer.peer_info()
+                .as_deref()
+                .map(|info| &info.protocol_version),
+            peer.client_request_metadata.get().is_some(),
+        )
+    }
+
+    fn is_subscription_request(request: &Self::Req) -> bool {
+        matches!(request, ClientRequest::SubscriptionsListenRequest(_))
+    }
+
     // SEP-2260: reject restricted server requests that arrived unassociated
     // with any in-flight outbound request. Without stream separation
     // (`Unknown`) the coarse in-flight check under-approximates the SHOULD.
@@ -827,7 +841,14 @@ where
             }
         }
     }
-    Ok(serve_inner(service, transport, peer, peer_rx, ct))
+    Ok(serve_inner(
+        service,
+        transport,
+        peer,
+        peer_rx,
+        VecDeque::new(),
+        ct,
+    ))
 }
 
 /// Modern-era JSON-RPC error codes a server can return from `server/discover`
@@ -870,7 +891,7 @@ async fn legacy_startup<S, T>(
     transport: &mut T,
     id_provider: &Arc<AtomicU32RequestIdProvider>,
     peer: &Peer<RoleClient>,
-    client_info: ClientInfo,
+    client_info: InitializeRequestParams,
 ) -> Result<(), ClientInitializeError>
 where
     S: Service<RoleClient>,
@@ -919,7 +940,7 @@ async fn discover_startup<S, T>(
     transport: &mut T,
     id_provider: &Arc<AtomicU32RequestIdProvider>,
     peer: &Peer<RoleClient>,
-    client_info: &ClientInfo,
+    client_info: &InitializeRequestParams,
     preferred_versions: Vec<ProtocolVersion>,
 ) -> Result<DiscoverOutcome, ClientInitializeError>
 where
@@ -2193,6 +2214,232 @@ where
 mod tests {
     use super::*;
 
+    #[tokio::test]
+    async fn server_cancellation_retires_subscription_responder() {
+        tokio::task::LocalSet::new()
+            .run_until(async {
+                for (discover, version) in [
+                    (false, ProtocolVersion::V_2026_07_28),
+                    (true, ProtocolVersion::V_2026_07_28),
+                    (true, ProtocolVersion::V_2025_11_25),
+                ] {
+                    tokio::time::timeout(
+                        Duration::from_secs(5),
+                        check_subscription_cancellation(discover, version),
+                    )
+                    .await
+                    .expect("subscription cancellation timed out");
+                }
+            })
+            .await;
+    }
+
+    async fn check_subscription_cancellation(discover: bool, version: ProtocolVersion) {
+        use crate::model::{
+            GetMeta, PingRequest, ServerInfo, SubscriptionsAcknowledgedNotification,
+            SubscriptionsAcknowledgedNotificationParams,
+        };
+
+        let (server_transport, client_transport) = tokio::io::duplex(4096);
+        let mut server =
+            crate::transport::IntoTransport::<RoleServer, _, _>::into_transport(server_transport);
+        let info = ServerInfo {
+            protocol_version: version.clone(),
+            ..Default::default()
+        };
+        let client = if discover {
+            let (client, ()) = tokio::join!(
+                serve_client_with_lifecycle(
+                    (),
+                    client_transport,
+                    ClientLifecycleMode::Discover {
+                        preferred_versions: vec![version.clone()]
+                    }
+                ),
+                async {
+                    let Some(ClientJsonRpcMessage::Request(request)) = server.receive().await
+                    else {
+                        panic!("expected discover request");
+                    };
+                    assert!(matches!(request.request, ClientRequest::DiscoverRequest(_)));
+                    server
+                        .send(ServerJsonRpcMessage::response(
+                            ServerResult::DiscoverResult(DiscoverResult::new(
+                                vec![version],
+                                info.capabilities,
+                            )),
+                            request.id,
+                        ))
+                        .await
+                        .unwrap();
+                }
+            );
+            client.unwrap()
+        } else {
+            super::super::serve_directly::<RoleClient, _, _, _, _>(
+                (),
+                client_transport,
+                Some(info.into()),
+            )
+        };
+        let filter = SubscriptionFilter::default();
+        let (subscription, ()) = tokio::join!(client.listen(filter.clone()), async {
+            let Some(ClientJsonRpcMessage::Request(request)) = server.receive().await else {
+                panic!("expected listen request");
+            };
+            assert!(matches!(
+                request.request,
+                ClientRequest::SubscriptionsListenRequest(_)
+            ));
+            let mut ack: ServerNotification = SubscriptionsAcknowledgedNotification::new(
+                SubscriptionsAcknowledgedNotificationParams::new(filter),
+            )
+            .into();
+            ack.get_meta_mut().set_subscription_id(request.id);
+            server
+                .send(ServerJsonRpcMessage::notification(ack))
+                .await
+                .unwrap();
+        });
+        let mut subscription = subscription.unwrap();
+        let id = subscription.id().clone();
+        let mut ordinary = client
+            .send_cancellable_request(
+                PingRequest {
+                    method: Default::default(),
+                    extensions: Default::default(),
+                }
+                .into(),
+                PeerRequestOptions::no_options(),
+            )
+            .await
+            .unwrap();
+        assert!(matches!(
+            server.receive().await,
+            Some(ClientJsonRpcMessage::Request(_))
+        ));
+        for unknown in [
+            ordinary.id.clone(),
+            RequestId::String(id.to_string().into()),
+            RequestId::String("unknown".into()),
+        ] {
+            server
+                .send(ServerJsonRpcMessage::notification(
+                    CancelledNotification::new(CancelledNotificationParam::new(
+                        Some(unknown),
+                        None,
+                    ))
+                    .into(),
+                ))
+                .await
+                .unwrap();
+        }
+        server
+            .send(ServerJsonRpcMessage::request(
+                PingRequest {
+                    method: Default::default(),
+                    extensions: Default::default(),
+                }
+                .into(),
+                RequestId::String("before-cancellation".into()),
+            ))
+            .await
+            .unwrap();
+        assert!(matches!(
+            server.receive().await,
+            Some(ClientJsonRpcMessage::Response(_))
+        ));
+        assert!(matches!(
+            subscription.request.as_mut().unwrap().rx.try_recv(),
+            Err(tokio::sync::oneshot::error::TryRecvError::Empty)
+        ));
+        assert!(matches!(
+            ordinary.rx.try_recv(),
+            Err(tokio::sync::oneshot::error::TryRecvError::Empty)
+        ));
+        assert!(matches!(
+            subscription.notifications.try_recv(),
+            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
+        ));
+        server
+            .send(ServerJsonRpcMessage::notification(
+                CancelledNotification::new(CancelledNotificationParam::new(
+                    Some(id.clone()),
+                    Some("subscription ended".to_owned()),
+                ))
+                .into(),
+            ))
+            .await
+            .unwrap();
+        // Ordered input makes the ping response a barrier for cancellation handling.
+        server
+            .send(ServerJsonRpcMessage::request(
+                PingRequest {
+                    method: Default::default(),
+                    extensions: Default::default(),
+                }
+                .into(),
+                RequestId::String("after-cancellation".into()),
+            ))
+            .await
+            .unwrap();
+        assert!(matches!(
+            server.receive().await,
+            Some(ClientJsonRpcMessage::Response(_))
+        ));
+        assert!(
+            matches!(
+                subscription.request.as_mut().unwrap().rx.try_recv(),
+                Ok(Err(ServiceError::Cancelled { .. }))
+            ),
+            "the outbound responder must retire without a final response or disconnect"
+        );
+        assert!(subscription.next().await.unwrap().is_none());
+        assert!(matches!(
+            subscription.end(),
+            Some(SubscriptionEnd::Cancelled)
+        ));
+        assert!(client.peer().subscription_sender(&id).is_none());
+        server
+            .send(ServerJsonRpcMessage::response(
+                ServerResult::empty(()),
+                ordinary.id.clone(),
+            ))
+            .await
+            .unwrap();
+        assert!(ordinary.await_response().await.is_ok());
+
+        // Raw listen requests also retire, even without a registered notification channel.
+        let raw = client
+            .send_cancellable_request(
+                ClientRequest::SubscriptionsListenRequest(SubscriptionsListenRequest::new(
+                    SubscriptionsListenRequestParams::new(SubscriptionFilter::default()),
+                )),
+                PeerRequestOptions::no_options(),
+            )
+            .await
+            .unwrap();
+        assert!(matches!(
+            server.receive().await,
+            Some(ClientJsonRpcMessage::Request(_))
+        ));
+        server
+            .send(ServerJsonRpcMessage::notification(
+                CancelledNotification::new(CancelledNotificationParam::new(
+                    Some(raw.id.clone()),
+                    None,
+                ))
+                .into(),
+            ))
+            .await
+            .unwrap();
+        assert!(matches!(
+            raw.await_response().await,
+            Err(ServiceError::Cancelled { .. })
+        ));
+        client.cancel().await.unwrap();
+    }
+
     #[tokio::test]
     async fn auto_startup_falls_back_when_discover_is_ignored() {
         use crate::model::{InitializeResult, ServerCapabilities};

crates/rmcp/src/service/server.rs

@@ -13,15 +13,16 @@ use super::*;
 use crate::model::{ElicitRequest, ElicitRequestParams, ElicitResult, ElicitationAction};
 use crate::{
     model::{
-        CancelledNotification, CancelledNotificationParam, ClientInfo, ClientJsonRpcMessage,
+        CancelledNotification, CancelledNotificationParam, ClientJsonRpcMessage,
         ClientNotification, ClientRequest, ClientResult, CreateMessageRequest,
-        CreateMessageRequestParams, CreateMessageResult, EmptyResult, ErrorData, ListRootsRequest,
+        CreateMessageRequestParams, CreateMessageResult, EmptyResult, ErrorData,
-        ListRootsResult, LoggingMessageNotification, LoggingMessageNotificationParam,
+        InitializeRequestParams, ListRootsRequest, ListRootsResult, LoggingMessageNotification,
-        ProgressNotification, ProgressNotificationParam, PromptListChangedNotification,
+        LoggingMessageNotificationParam, ProgressNotification, ProgressNotificationParam,
-        ProtocolVersion, ResourceListChangedNotification, ResourceUpdatedNotification,
+        PromptListChangedNotification, ProtocolVersion, ResourceListChangedNotification,
-        ResourceUpdatedNotificationParam, ServerInfo, ServerNotification, ServerRequest,
+        ResourceUpdatedNotification, ResourceUpdatedNotificationParam, ServerConfig,
-        ServerResult, SubscriptionFilter, SubscriptionsAcknowledgedNotification,
+        ServerNotification, ServerRequest, ServerResult, SubscriptionFilter,
-        SubscriptionsAcknowledgedNotificationParams, ToolListChangedNotification,
+        SubscriptionsAcknowledgedNotification, SubscriptionsAcknowledgedNotificationParams,
+        ToolListChangedNotification,
     },
     transport::DynamicTransportError,
 };
@@ -37,8 +38,8 @@ impl ServiceRole for RoleServer {
     type PeerReq = ClientRequest;
     type PeerResp = ClientResult;
     type PeerNot = ClientNotification;
-    type Info = ServerInfo;
+    type Info = ServerConfig;
-    type PeerInfo = ClientInfo;
+    type PeerInfo = InitializeRequestParams;
 
     type InitializeError = ServerInitializeError;
     const IS_CLIENT: bool = false;
@@ -593,7 +594,7 @@ where
 
     let initialize_request = match request {
         ClientRequest::InitializeRequest(request) => request,
-        mut request => {
+        request => {
             let missing_metadata = request
                 .get_meta()
                 .missing_required_keys(&ProtocolVersion::V_2026_07_28);
@@ -616,21 +617,17 @@ where
             }
             let (peer, peer_rx) = Peer::new(id_provider, None);
             peer.require_request_metadata();
-            let context = RequestContext {
+            // Dispatch the request from inside the service loop rather than
-                ct: ct.child_token(),
+            // inline: its handler may send notifications through `peer`, which
-                id: id.clone(),
+            // only complete once the loop drains `peer_rx`.
-                meta: std::mem::take(request.get_meta_mut()),
+            return Ok(serve_inner(
-                extensions: std::mem::take(request.extensions_mut()),
+                service,
-                peer: peer.clone(),
+                transport,
-            };
+                peer,
-            let response = match service.handle_request(request, context).await {
+                peer_rx,
-                Ok(result) => ServerJsonRpcMessage::response(result, id),
+                VecDeque::from([ClientJsonRpcMessage::request(request, id)]),
-                Err(error) => ServerJsonRpcMessage::error(error, Some(id)),
+                ct,
-            };
+            ));
-            transport.send(response).await.map_err(|error| {
-                ServerInitializeError::transport::<T>(error, "sending negotiated request response")
-            })?;
-            return Ok(serve_inner(service, transport, peer, peer_rx, ct));
         }
     };
     let requested_protocol_version = initialize_request.params.protocol_version.clone();
@@ -680,7 +677,14 @@ where
     // Streamable HTTP has no ordering guarantee between POSTs, and the MCP spec uses
     // SHOULD NOT (not MUST NOT) for pre-initialized messages, so any request arriving
     // before initialized is processed normally.
-    Ok(serve_inner(service, transport, peer, peer_rx, ct))
+    Ok(serve_inner(
+        service,
+        transport,
+        peer,
+        peer_rx,
+        VecDeque::new(),
+        ct,
+    ))
 }
 
 macro_rules! method {

crates/rmcp/src/transport/auth.rs

@@ -1,5 +1,5 @@
 use std::{
-    collections::HashMap,
+    collections::{HashMap, HashSet},
     future::Future,
     net::{IpAddr, Ipv4Addr, Ipv6Addr},
     pin::Pin,
@@ -685,6 +685,19 @@ struct ResourceServerMetadata {
     scopes_supported: Option<Vec<String>>,
 }
 
+/// How a url that may hold protected resource metadata was arrived at, which
+/// decides what a document that is not metadata means there.
+#[derive(Debug, Clone, Copy)]
+enum ResourceMetadataUrlOrigin {
+    /// The resource itself named this url in the `resource_metadata` parameter
+    /// of its `WWW-Authenticate` challenge.
+    Advertised,
+    /// The url was derived from the base url on the chance that the document is
+    /// published there, or named by a challenge on such a url. Either way the
+    /// run only guessed its way here.
+    WellKnownGuess,
+}
+
 /// Parameters extracted from WWW-Authenticate header
 #[derive(Debug, Clone, Default)]
 #[non_exhaustive]
@@ -2525,26 +2538,50 @@ impl AuthorizationManager {
     async fn discover_oauth_server_via_resource_metadata(
         &self,
     ) -> Result<Option<AuthorizationMetadata>, AuthError> {
-        let Some(resource_metadata_url) = self.discover_resource_metadata_url().await? else {
+        let Some((resource_metadata_url, resource_metadata)) =
+            self.discover_resource_metadata().await?
+        else {
             return Ok(None);
         };
-        self.discover_oauth_server_from_resource_metadata_url(&resource_metadata_url)
+        self.authorization_metadata_from_resource_metadata(
-            .await
+            &resource_metadata_url,
+            resource_metadata,
+        )
+        .await
     }
 
+    /// Read protected resource metadata from the url a `WWW-Authenticate`
+    /// challenge advertised.
     async fn discover_oauth_server_from_resource_metadata_url(
         &self,
         resource_metadata_url: &Url,
     ) -> Result<Option<AuthorizationMetadata>, AuthError> {
         let Some(resource_metadata) = self
-            .fetch_resource_metadata_from_url(resource_metadata_url)
+            .fetch_resource_metadata_from_url(
+                resource_metadata_url,
+                ResourceMetadataUrlOrigin::Advertised,
+            )
             .await?
         else {
             return Ok(None);
         };
 
-        self.validate_resource_metadata_resource(&resource_metadata)?;
+        self.authorization_metadata_from_resource_metadata(resource_metadata_url, resource_metadata)
+            .await
+    }
 
+    /// Walk the authorization servers a protected resource metadata document
+    /// names, keeping the first one that answers with usable metadata. A document
+    /// that names servers and none of them answers is an error, not a reason to
+    /// go looking elsewhere; a document that names none returns `Ok(None)`.
+    ///
+    /// The document arrives here through `read_resource_metadata`, which is where
+    /// it is decided to be this resource's metadata at all.
+    async fn authorization_metadata_from_resource_metadata(
+        &self,
+        resource_metadata_url: &Url,
+        resource_metadata: ResourceServerMetadata,
+    ) -> Result<Option<AuthorizationMetadata>, AuthError> {
         self.discovered_resource
             .write()
             .await
@@ -2557,21 +2594,32 @@ impl AuthorizationManager {
             *self.resource_scopes.write().await = scopes;
         }
 
-        let mut candidates = Vec::new();
+        // A server naming the same authorization server in both the singular draft
+        // field and the list would otherwise have each of that server's well-known
+        // forms requested twice.
+        let mut candidates: Vec<String> = Vec::new();
+        let mut push_candidate = |candidate: String| {
+            let candidate = candidate.trim();
+            if !candidate.is_empty() && !candidates.iter().any(|kept| kept == candidate) {
+                candidates.push(candidate.to_string());
+            }
+        };
 
         if let Some(single) = resource_metadata.authorization_server {
-            candidates.push(single);
+            push_candidate(single);
         }
-        if let Some(list) = resource_metadata.authorization_servers {
+        for candidate in resource_metadata.authorization_servers.unwrap_or_default() {
-            candidates.extend(list);
+            push_candidate(candidate);
         }
 
-        for candidate in candidates {
+        if candidates.is_empty() {
-            let candidate = candidate.trim();
+            debug!(
-            if candidate.is_empty() {
+                "protected resource metadata at {resource_metadata_url} names no authorization server"
-                continue;
+            );
-            }
+            return Ok(None);
+        }
 
+        for candidate in &candidates {
             let candidate_url = match Url::parse(candidate) {
                 Ok(url) => url,
                 Err(_) => match resource_metadata_url.join(candidate) {
@@ -2608,7 +2656,12 @@ impl AuthorizationManager {
             }
         }
 
-        Ok(None)
+        // Falling back to the base url would send the user somewhere the resource
+        // never named.
+        Err(AuthError::MetadataError(format!(
+            "protected resource metadata at {resource_metadata_url} names authorization servers {}, but none published usable metadata",
+            candidates.join(", ")
+        )))
     }
 
     fn validate_resource_metadata_resource(
@@ -2668,52 +2721,115 @@ impl AuthorizationManager {
                 || expected_path.as_bytes().get(actual_path.len()) == Some(&b'/'))
     }
 
-    async fn discover_resource_metadata_url(&self) -> Result<Option<Url>, AuthError> {
+    /// Look for the protected resource metadata document, reading it where it is
-        if let Some(resource_metadata_url) =
+    /// found so that a candidate answering with something else only costs that
-            self.probe_resource_metadata_url(&self.base_url).await?
+    /// candidate.
-        {
+    async fn discover_resource_metadata(
-            return Ok(Some(resource_metadata_url));
+        &self,
+    ) -> Result<Option<(Url, ResourceServerMetadata)>, AuthError> {
+        // A url the resource points at can also be one of the candidates below.
+        let mut requested = HashSet::new();
+
+        if let Some(advertised_url) = self.probe_resource_endpoint_for_challenge().await? {
+            requested.insert(advertised_url.clone());
+            if let Some(metadata) = self
+                .fetch_resource_metadata_from_url(
+                    &advertised_url,
+                    ResourceMetadataUrlOrigin::Advertised,
+                )
+                .await?
+            {
+                return Ok(Some((advertised_url, metadata)));
+            }
+            // Nothing was published there. The candidates below are reached from the
+            // base url rather than from that pointer, so they are still worth trying.
         }
 
-        // If the primary URL doesn't use WWW-Authenticate, try oauth-protected-resource discovery.
+        // The other place the document can be is the well-known location.
         // https://www.rfc-editor.org/rfc/rfc9728.html#name-obtaining-protected-resourc
         for candidate_path in
             Self::well_known_paths(self.base_url.path(), "oauth-protected-resource")
         {
-            let mut discovery_url = self.base_url.clone();
+            let mut candidate_url = self.base_url.clone();
-            discovery_url.set_query(None);
+            candidate_url.set_query(None);
-            discovery_url.set_fragment(None);
+            candidate_url.set_fragment(None);
-            discovery_url.set_path(&candidate_path);
+            candidate_url.set_path(&candidate_path);
-            if let Some(resource_metadata_url) =
+
-                self.probe_resource_metadata_url(&discovery_url).await?
+            if !requested.insert(candidate_url.clone()) {
-            {
+                continue;
-                return Ok(Some(resource_metadata_url));
+            }
+
+            let response = self
+                .discovery_get(&candidate_url)
+                .await
+                .map_err(|error| Self::discovery_failed(&candidate_url, error))?;
+
+            match response.status() {
+                // The candidate url is the document itself, so read the body here
+                // instead of requesting the same url again.
+                StatusCode::OK => {
+                    if let Some(metadata) = self.read_resource_metadata(
+                        &candidate_url,
+                        response.body(),
+                        ResourceMetadataUrlOrigin::WellKnownGuess,
+                    )? {
+                        return Ok(Some((candidate_url, metadata)));
+                    }
+                }
+                StatusCode::UNAUTHORIZED => {
+                    let Some(pointer_url) = self
+                        .extract_resource_metadata_url_from_www_authenticate(&response)
+                        .await
+                    else {
+                        continue;
+                    };
+                    if !requested.insert(pointer_url.clone()) {
+                        continue;
+                    }
+                    // Found on a url this run guessed at, so a wrong answer only rules
+                    // out the pointer.
+                    if let Some(metadata) = self
+                        .fetch_resource_metadata_from_url(
+                            &pointer_url,
+                            ResourceMetadataUrlOrigin::WellKnownGuess,
+                        )
+                        .await?
+                    {
+                        return Ok(Some((pointer_url, metadata)));
+                    }
+                }
+                status => debug!("resource metadata probe returned unexpected status: {status}"),
             }
         }
 
         Ok(None)
     }
 
-    /// Probe `url` with a GET, extracting the resource metadata url from a
+    /// Probe the resource itself, looking only for a `WWW-Authenticate` challenge
-    /// 200 (the url itself is the metadata document) or from a 401's
+    /// that carries a `resource_metadata` pointer.
-    /// WWW-Authenticate header value.
+    ///
+    /// A 200 here says nothing about metadata. RFC 9728 publishes the document at
+    /// the well-known URI and advertises it through the challenge parameter, so the
+    /// resource answering its own GET is not the document and must not end
+    /// discovery before the well-known candidates are tried.
     /// https://www.rfc-editor.org/rfc/rfc9728.html#name-use-of-www-authenticate-for
-    async fn probe_resource_metadata_url(&self, url: &Url) -> Result<Option<Url>, AuthError> {
+    async fn probe_resource_endpoint_for_challenge(&self) -> Result<Option<Url>, AuthError> {
         let response = self
-            .discovery_get(url)
+            .discovery_get(&self.base_url)
             .await
-            .map_err(|error| Self::discovery_failed(url, error))?;
+            .map_err(|error| Self::discovery_failed(&self.base_url, error))?;
 
-        match response.status() {
+        if response.status() == StatusCode::UNAUTHORIZED {
-            StatusCode::OK => Ok(Some(url.clone())),
+            return Ok(self
-            StatusCode::UNAUTHORIZED => Ok(self
                 .extract_resource_metadata_url_from_www_authenticate(&response)
-                .await),
+                .await);
-            status => {
-                debug!("resource metadata probe returned unexpected status: {status}");
-                Ok(None)
-            }
         }
+
+        debug!(
+            "resource endpoint probe returned {}, no WWW-Authenticate pointer to follow",
+            response.status()
+        );
+        Ok(None)
     }
 
     async fn extract_resource_metadata_url_from_www_authenticate(
@@ -2739,6 +2855,7 @@ impl AuthorizationManager {
     async fn fetch_resource_metadata_from_url(
         &self,
         resource_metadata_url: &Url,
+        origin: ResourceMetadataUrlOrigin,
     ) -> Result<Option<ResourceServerMetadata>, AuthError> {
         debug!(
             "resource metadata discovery url: {:?}",
@@ -2757,13 +2874,69 @@ impl AuthorizationManager {
             return Ok(None);
         }
 
-        let metadata = match serde_json::from_slice::<ResourceServerMetadata>(response.body()) {
+        self.read_resource_metadata(resource_metadata_url, response.body(), origin)
+    }
+
+    /// Read a response body as this resource's protected resource metadata.
+    ///
+    /// A body that is not that document rules out the url it came from, and where
+    /// that url came from decides whether ruling it out leaves anything to try.
+    fn read_resource_metadata(
+        &self,
+        resource_metadata_url: &Url,
+        body: &[u8],
+        origin: ResourceMetadataUrlOrigin,
+    ) -> Result<Option<ResourceServerMetadata>, AuthError> {
+        let metadata = match serde_json::from_slice::<ResourceServerMetadata>(body) {
             Ok(metadata) => metadata,
             Err(e) => {
                 debug!("failed to parse resource metadata as JSON: {}", e);
                 return Ok(None);
             }
         };
+
+        // Every field of `ResourceServerMetadata` is optional, so an unrelated JSON
+        // object deserializes into an all-`None` value and then fails validation
+        // fatally. RFC 9728 requires `resource`, and MCP requires an authorization
+        // server reference, so a document carrying neither is not a protected
+        // resource metadata document.
+        if metadata.resource.is_none()
+            && metadata.authorization_server.is_none()
+            && metadata.authorization_servers.is_none()
+        {
+            return match origin {
+                // The server named this url, so there is nothing better to move on
+                // to: the alternatives all drop the resource binding the document
+                // was supposed to carry. Report it instead.
+                ResourceMetadataUrlOrigin::Advertised => Err(AuthError::MetadataError(format!(
+                    "the server advertised {resource_metadata_url} as protected resource metadata, but the document carries neither `resource` nor an authorization server reference"
+                ))),
+                // Nothing advertised this url, so its answer only rules out this
+                // candidate.
+                ResourceMetadataUrlOrigin::WellKnownGuess => {
+                    debug!(
+                        "response at {resource_metadata_url} is not a protected resource metadata document"
+                    );
+                    Ok(None)
+                }
+            };
+        }
+
+        // Carrying those fields only makes the body a metadata document; validation
+        // is what makes it this resource's. Both answers rule out the url the same
+        // way, so both are read the same way.
+        if let Err(error) = self.validate_resource_metadata_resource(&metadata) {
+            return match origin {
+                ResourceMetadataUrlOrigin::Advertised => Err(error),
+                ResourceMetadataUrlOrigin::WellKnownGuess => {
+                    debug!(
+                        "document at {resource_metadata_url} is not this resource's metadata: {error}"
+                    );
+                    Ok(None)
+                }
+            };
+        }
+
         Ok(Some(metadata))
     }
 
@@ -4441,13 +4614,6 @@ mod tests {
                     "authorization_servers": ["https://auth.example.com/tenant1"]
                 }),
             ),
-            http_response(
-                200,
-                serde_json::json!({
-                    "resource": "https://mcp.example.com/",
-                    "authorization_servers": ["https://auth.example.com/tenant1"]
-                }),
-            ),
             http_response(
                 200,
                 serde_json::json!({
@@ -4482,7 +4648,6 @@ mod tests {
                 vec![
                     "https://mcp.example.com/",
                     "https://mcp.example.com/.well-known/oauth-protected-resource",
-                    "https://mcp.example.com/.well-known/oauth-protected-resource",
                     "https://auth.example.com/.well-known/oauth-authorization-server/tenant1",
                 ],
             )
@@ -4500,13 +4665,6 @@ mod tests {
                     "authorization_servers": ["https://auth.example.com/tenant1/"]
                 }),
             ),
-            http_response(
-                200,
-                serde_json::json!({
-                    "resource": "https://mcp.example.com/",
-                    "authorization_servers": ["https://auth.example.com/tenant1/"]
-                }),
-            ),
             http_response(
                 200,
                 serde_json::json!({
@@ -4970,6 +5128,579 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn resolve_metadata_reaches_the_well_known_document_past_a_non_metadata_base_url() {
+        let client = RecordingOAuthHttpClient::with_responses(vec![
+            // the MCP endpoint answers GET with a health payload, not metadata
+            http_response(
+                200,
+                serde_json::json!({"status": "healthy", "message": "MCP server is running"}),
+            ),
+            // the well-known candidate carries the real document
+            http_response(
+                200,
+                serde_json::json!({
+                    "resource": "https://mcp.example.com/",
+                    "authorization_servers": ["https://auth.example.com"]
+                }),
+            ),
+            http_response(
+                200,
+                serde_json::json!({
+                    "issuer": "https://auth.example.com",
+                    "authorization_endpoint": "https://auth.example.com/authorize",
+                    "token_endpoint": "https://auth.example.com/token"
+                }),
+            ),
+        ]);
+        let recorder = client.clone();
+        let manager = AuthorizationManager::new_with_oauth_http_client(
+            "https://mcp.example.com/",
+            Arc::new(client),
+        )
+        .await
+        .unwrap();
+
+        let resolution = manager.resolve_metadata().await.unwrap();
+
+        assert_eq!(
+            (
+                resolution.source,
+                resolution.metadata.token_endpoint.as_str(),
+            ),
+            (
+                AuthorizationMetadataSource::ProtectedResourceMetadata,
+                "https://auth.example.com/token",
+            )
+        );
+        assert!(
+            recorder.requests().iter().any(|request| {
+                request.uri == "https://mcp.example.com/.well-known/oauth-protected-resource"
+            }),
+            "the well-known candidate was never probed: {:?}",
+            recorder.requests()
+        );
+    }
+
+    #[tokio::test]
+    async fn resolve_metadata_ignores_a_well_known_url_that_is_not_a_metadata_document() {
+        let health = || {
+            http_response(
+                200,
+                serde_json::json!({"status": "healthy", "message": "MCP server is running"}),
+            )
+        };
+        let client = RecordingOAuthHttpClient::with_responses(vec![
+            // the MCP endpoint answers GET with a health payload, not metadata
+            health(),
+            // so does the well-known candidate
+            health(),
+            http_response(
+                200,
+                serde_json::json!({
+                    "issuer": "https://mcp.example.com",
+                    "authorization_endpoint": "https://mcp.example.com/oauth/authorize",
+                    "token_endpoint": "https://mcp.example.com/oauth/token"
+                }),
+            ),
+        ]);
+        let manager = AuthorizationManager::new_with_oauth_http_client(
+            "https://mcp.example.com/",
+            Arc::new(client),
+        )
+        .await
+        .unwrap();
+
+        let resolution = manager.resolve_metadata().await.unwrap();
+
+        assert_eq!(
+            (
+                resolution.source,
+                resolution.metadata.token_endpoint.as_str(),
+            ),
+            (
+                AuthorizationMetadataSource::AuthorizationServerMetadata,
+                "https://mcp.example.com/oauth/token",
+            )
+        );
+    }
+
+    #[tokio::test]
+    async fn resolve_metadata_tries_the_next_well_known_candidate_past_a_non_metadata_document() {
+        let client = RecordingOAuthHttpClient::with_responses(vec![
+            // the MCP endpoint answers GET with a health payload, not metadata
+            http_response(
+                200,
+                serde_json::json!({"status": "healthy", "message": "MCP server is running"}),
+            ),
+            // so does the first well-known candidate
+            http_response(
+                200,
+                serde_json::json!({"status": "healthy", "message": "MCP server is running"}),
+            ),
+            // the second candidate carries the real document
+            http_response(
+                200,
+                serde_json::json!({
+                    "resource": "https://mcp.example.com/mcp",
+                    "authorization_servers": ["https://auth.example.com"]
+                }),
+            ),
+            http_response(
+                200,
+                serde_json::json!({
+                    "issuer": "https://auth.example.com",
+                    "authorization_endpoint": "https://auth.example.com/authorize",
+                    "token_endpoint": "https://auth.example.com/token"
+                }),
+            ),
+        ]);
+        let recorder = client.clone();
+        let manager = AuthorizationManager::new_with_oauth_http_client(
+            "https://mcp.example.com/mcp",
+            Arc::new(client),
+        )
+        .await
+        .unwrap();
+
+        let resolution = manager.resolve_metadata().await.unwrap();
+
+        assert_eq!(
+            (
+                resolution.source,
+                resolution.metadata.token_endpoint.as_str(),
+            ),
+            (
+                AuthorizationMetadataSource::ProtectedResourceMetadata,
+                "https://auth.example.com/token",
+            )
+        );
+        let document_requests = recorder
+            .requests()
+            .iter()
+            .filter(|request| {
+                request.uri == "https://mcp.example.com/mcp/.well-known/oauth-protected-resource"
+            })
+            .count();
+        assert_eq!(
+            document_requests,
+            1,
+            "the candidate holding the document should be requested exactly once: {:?}",
+            recorder.requests()
+        );
+    }
+
+    #[tokio::test]
+    async fn resolve_metadata_from_challenge_reports_an_advertised_url_without_metadata() {
+        let mut responses = vec![http_response(200, serde_json::json!({}))];
+        // enough responses for the fallback to reach the legacy endpoints, so that
+        // treating the document as a soft failure would resolve rather than error
+        responses.extend(std::iter::repeat_with(|| empty_response(404)).take(8));
+        let client = RecordingOAuthHttpClient::with_responses(responses);
+        let manager = AuthorizationManager::new_with_oauth_http_client(
+            "https://mcp.example.com/mcp",
+            Arc::new(client),
+        )
+        .await
+        .unwrap();
+
+        let error = manager
+            .resolve_metadata_from_challenge(Some(
+                r#"Bearer resource_metadata="https://mcp.example.com/.well-known/oauth-protected-resource""#,
+            ))
+            .await
+            .unwrap_err();
+
+        assert_eq!(
+            error.to_string(),
+            "Metadata error: the server advertised https://mcp.example.com/.well-known/oauth-protected-resource as protected resource metadata, but the document carries neither `resource` nor an authorization server reference"
+        );
+    }
+
+    #[tokio::test]
+    async fn resolve_metadata_tries_the_next_well_known_candidate_past_an_unusable_document() {
+        let client = RecordingOAuthHttpClient::with_responses(vec![
+            // the MCP endpoint answers GET with a health payload, not metadata
+            http_response(
+                200,
+                serde_json::json!({"status": "healthy", "message": "MCP server is running"}),
+            ),
+            // a catch-all handler answers the first candidate with its own error
+            // shape, which carries a `resource` that only validation rejects
+            http_response(
+                200,
+                serde_json::json!({
+                    "error": "not_found",
+                    "resource": "/.well-known/oauth-protected-resource"
+                }),
+            ),
+            // the second candidate carries the real document
+            http_response(
+                200,
+                serde_json::json!({
+                    "resource": "https://mcp.example.com/mcp",
+                    "authorization_servers": ["https://auth.example.com"]
+                }),
+            ),
+            http_response(
+                200,
+                serde_json::json!({
+                    "issuer": "https://auth.example.com",
+                    "authorization_endpoint": "https://auth.example.com/authorize",
+                    "token_endpoint": "https://auth.example.com/token"
+                }),
+            ),
+        ]);
+        let recorder = client.clone();
+        let manager = AuthorizationManager::new_with_oauth_http_client(
+            "https://mcp.example.com/mcp",
+            Arc::new(client),
+        )
+        .await
+        .unwrap();
+
+        let resolution = manager.resolve_metadata().await.unwrap();
+
+        assert_eq!(
+            (
+                resolution.source,
+                resolution.metadata.token_endpoint.as_str(),
+            ),
+            (
+                AuthorizationMetadataSource::ProtectedResourceMetadata,
+                "https://auth.example.com/token",
+            )
+        );
+        assert!(
+            recorder.requests().iter().any(|request| {
+                request.uri == "https://mcp.example.com/mcp/.well-known/oauth-protected-resource"
+            }),
+            "the candidate after the rejected one was never probed: {:?}",
+            recorder.requests()
+        );
+    }
+
+    #[tokio::test]
+    async fn resolve_metadata_probes_the_candidates_past_an_advertised_url_that_is_not_served() {
+        let challenge = oauth2::http::Response::builder()
+            .status(401)
+            .header(
+                "www-authenticate",
+                r#"Bearer resource_metadata="https://mcp.example.com/prm""#,
+            )
+            .body(Vec::new())
+            .unwrap();
+        let client = RecordingOAuthHttpClient::with_responses(vec![
+            challenge,
+            // the advertised url is not where the document is served
+            empty_response(404),
+            // neither is the first candidate
+            empty_response(404),
+            // the second candidate carries the document
+            http_response(
+                200,
+                serde_json::json!({
+                    "resource": "https://mcp.example.com/mcp",
+                    "authorization_servers": ["https://auth.example.com"]
+                }),
+            ),
+            http_response(
+                200,
+                serde_json::json!({
+                    "issuer": "https://auth.example.com",
+                    "authorization_endpoint": "https://auth.example.com/authorize",
+                    "token_endpoint": "https://auth.example.com/token"
+                }),
+            ),
+        ]);
+        let recorder = client.clone();
+        let manager = AuthorizationManager::new_with_oauth_http_client(
+            "https://mcp.example.com/mcp",
+            Arc::new(client),
+        )
+        .await
+        .unwrap();
+
+        let resolution = manager.resolve_metadata().await.unwrap();
+
+        assert_eq!(
+            (
+                resolution.source,
+                resolution.metadata.token_endpoint.as_str(),
+            ),
+            (
+                AuthorizationMetadataSource::ProtectedResourceMetadata,
+                "https://auth.example.com/token",
+            )
+        );
+        assert!(
+            recorder.requests().iter().any(|request| {
+                request.uri == "https://mcp.example.com/mcp/.well-known/oauth-protected-resource"
+            }),
+            "the candidates were skipped after the advertised url answered 404: {:?}",
+            recorder.requests()
+        );
+    }
+
+    #[tokio::test]
+    async fn resolve_metadata_requests_an_advertised_url_that_is_also_a_candidate_once() {
+        let challenge = oauth2::http::Response::builder()
+            .status(401)
+            .header(
+                "www-authenticate",
+                r#"Bearer resource_metadata="https://mcp.example.com/.well-known/oauth-protected-resource""#,
+            )
+            .body(Vec::new())
+            .unwrap();
+        let mut responses = vec![
+            // the MCP endpoint answers GET with a health payload, not metadata
+            http_response(
+                200,
+                serde_json::json!({"status": "healthy", "message": "MCP server is running"}),
+            ),
+            // the first candidate points at the last candidate of the same run
+            challenge,
+        ];
+        // the document is served nowhere, so the run walks every candidate and
+        // settles on the legacy endpoints
+        responses.extend(std::iter::repeat_with(|| empty_response(404)).take(10));
+        let client = RecordingOAuthHttpClient::with_responses(responses);
+        let recorder = client.clone();
+        let manager = AuthorizationManager::new_with_oauth_http_client(
+            "https://mcp.example.com/mcp",
+            Arc::new(client),
+        )
+        .await
+        .unwrap();
+
+        let resolution = manager.resolve_metadata().await.unwrap();
+
+        assert_eq!(
+            resolution.source,
+            AuthorizationMetadataSource::LegacyEndpointFallback
+        );
+        let advertised_requests = recorder
+            .requests()
+            .iter()
+            .filter(|request| {
+                request.uri == "https://mcp.example.com/.well-known/oauth-protected-resource"
+            })
+            .count();
+        assert_eq!(
+            advertised_requests,
+            1,
+            "the url the challenge named was requested again as a candidate: {:?}",
+            recorder.requests()
+        );
+    }
+
+    #[tokio::test]
+    async fn resolve_metadata_requests_an_authorization_server_named_twice_once() {
+        let mut responses = vec![
+            // the MCP endpoint answers GET with a health payload, not metadata
+            http_response(
+                200,
+                serde_json::json!({"status": "healthy", "message": "MCP server is running"}),
+            ),
+            // the document names the same authorization server in both the singular
+            // draft field and the list
+            http_response(
+                200,
+                serde_json::json!({
+                    "resource": "https://mcp.example.com/",
+                    "authorization_server": "https://auth.example.com",
+                    "authorization_servers": ["https://auth.example.com"]
+                }),
+            ),
+        ];
+        // the authorization server publishes no metadata; more 404s than the walk
+        // can use, so that requesting it twice would show up in the sequence
+        responses.extend(std::iter::repeat_with(|| empty_response(404)).take(10));
+        let client = RecordingOAuthHttpClient::with_responses(responses);
+        let recorder = client.clone();
+        let manager = AuthorizationManager::new_with_oauth_http_client(
+            "https://mcp.example.com/",
+            Arc::new(client),
+        )
+        .await
+        .unwrap();
+
+        manager.resolve_metadata().await.unwrap_err();
+
+        let requests = recorder.requests();
+        let requested: Vec<_> = requests
+            .iter()
+            .map(|request| request.uri.as_str())
+            .collect();
+        assert_eq!(
+            requested,
+            [
+                "https://mcp.example.com/",
+                "https://mcp.example.com/.well-known/oauth-protected-resource",
+                "https://auth.example.com/.well-known/oauth-authorization-server",
+                "https://auth.example.com/.well-known/openid-configuration",
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn resolve_metadata_reports_named_authorization_servers_that_publish_no_metadata() {
+        let mut responses = vec![
+            // the MCP endpoint answers GET with a health payload, not metadata
+            http_response(
+                200,
+                serde_json::json!({"status": "healthy", "message": "MCP server is running"}),
+            ),
+            // the document names two authorization servers
+            http_response(
+                200,
+                serde_json::json!({
+                    "resource": "https://mcp.example.com/",
+                    "authorization_servers": ["https://auth.example.com", "https://sso.example.com"]
+                }),
+            ),
+        ];
+        // neither publishes metadata; the surplus 404s would let the run reach the
+        // resource host's own well-known urls and the legacy endpoints if it kept going
+        responses.extend(std::iter::repeat_with(|| empty_response(404)).take(10));
+        let client = RecordingOAuthHttpClient::with_responses(responses);
+        let recorder = client.clone();
+        let manager = AuthorizationManager::new_with_oauth_http_client(
+            "https://mcp.example.com/",
+            Arc::new(client),
+        )
+        .await
+        .unwrap();
+
+        let error = manager.resolve_metadata().await.unwrap_err();
+
+        assert_eq!(
+            error.to_string(),
+            "Metadata error: protected resource metadata at https://mcp.example.com/.well-known/oauth-protected-resource names authorization servers https://auth.example.com, https://sso.example.com, but none published usable metadata"
+        );
+        let requests = recorder.requests();
+        let requested: Vec<_> = requests
+            .iter()
+            .map(|request| request.uri.as_str())
+            .collect();
+        assert_eq!(
+            requested,
+            [
+                "https://mcp.example.com/",
+                "https://mcp.example.com/.well-known/oauth-protected-resource",
+                "https://auth.example.com/.well-known/oauth-authorization-server",
+                "https://auth.example.com/.well-known/openid-configuration",
+                "https://sso.example.com/.well-known/oauth-authorization-server",
+                "https://sso.example.com/.well-known/openid-configuration",
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn resolve_metadata_keeps_falling_back_when_the_document_names_no_authorization_server() {
+        let mut responses = vec![
+            // the MCP endpoint answers GET with a health payload, not metadata
+            http_response(
+                200,
+                serde_json::json!({"status": "healthy", "message": "MCP server is running"}),
+            ),
+            // the document says nothing about where to authorize
+            http_response(
+                200,
+                serde_json::json!({"resource": "https://mcp.example.com/"}),
+            ),
+        ];
+        responses.extend(std::iter::repeat_with(|| empty_response(404)).take(10));
+        let client = RecordingOAuthHttpClient::with_responses(responses);
+        let manager = AuthorizationManager::new_with_oauth_http_client(
+            "https://mcp.example.com/",
+            Arc::new(client),
+        )
+        .await
+        .unwrap();
+
+        let resolution = manager.resolve_metadata().await.unwrap();
+
+        assert_eq!(
+            resolution.source,
+            AuthorizationMetadataSource::LegacyEndpointFallback
+        );
+    }
+
+    #[tokio::test]
+    async fn resolve_metadata_tries_the_next_candidate_past_a_pointer_found_while_guessing() {
+        let challenge = oauth2::http::Response::builder()
+            .status(401)
+            .header(
+                "www-authenticate",
+                r#"Bearer resource_metadata="https://mcp.example.com/prm""#,
+            )
+            .body(Vec::new())
+            .unwrap();
+        let client = RecordingOAuthHttpClient::with_responses(vec![
+            // the MCP endpoint answers GET with a health payload, not metadata
+            http_response(
+                200,
+                serde_json::json!({"status": "healthy", "message": "MCP server is running"}),
+            ),
+            // auth sits in front of the first candidate and its challenge names a url
+            challenge,
+            // which serves something that is not the document
+            http_response(200, serde_json::json!({})),
+            // the second candidate carries the real document
+            http_response(
+                200,
+                serde_json::json!({
+                    "resource": "https://mcp.example.com/mcp",
+                    "authorization_servers": ["https://auth.example.com"]
+                }),
+            ),
+            http_response(
+                200,
+                serde_json::json!({
+                    "issuer": "https://auth.example.com",
+                    "authorization_endpoint": "https://auth.example.com/authorize",
+                    "token_endpoint": "https://auth.example.com/token"
+                }),
+            ),
+        ]);
+        let recorder = client.clone();
+        let manager = AuthorizationManager::new_with_oauth_http_client(
+            "https://mcp.example.com/mcp",
+            Arc::new(client),
+        )
+        .await
+        .unwrap();
+
+        let resolution = manager.resolve_metadata().await.unwrap();
+
+        assert_eq!(
+            (
+                resolution.source,
+                resolution.metadata.token_endpoint.as_str(),
+            ),
+            (
+                AuthorizationMetadataSource::ProtectedResourceMetadata,
+                "https://auth.example.com/token",
+            )
+        );
+        let requests = recorder.requests();
+        let requested: Vec<_> = requests
+            .iter()
+            .map(|request| request.uri.as_str())
+            .collect();
+        assert_eq!(
+            requested,
+            [
+                "https://mcp.example.com/mcp",
+                "https://mcp.example.com/.well-known/oauth-protected-resource/mcp",
+                "https://mcp.example.com/prm",
+                "https://mcp.example.com/mcp/.well-known/oauth-protected-resource",
+                "https://auth.example.com/.well-known/oauth-authorization-server",
+            ]
+        );
+    }
+
     #[rstest]
     #[case::protected_resource_metadata(
         AuthorizationMetadataSource::ProtectedResourceMetadata,

crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs

@@ -308,17 +308,29 @@ impl StreamableHttpClient for reqwest::Client {
                 Ok(StreamableHttpPostResponse::Sse(event_stream, session_id))
             }
             Some(ct) if ct.as_bytes().starts_with(JSON_MIME_TYPE.as_bytes()) => {
-                // Try to parse as a valid JSON-RPC message. If the body is
+                // A notification/response/error POST does not await a JSON-RPC
-                // malformed (e.g. a 200 response to a notification that lacks
+                // reply. Treat an unusable JSON body as Accepted. A request
-                // an `id` field), treat it as accepted rather than failing.
+                // still needs a reply; the same body is an error so the worker
-                match response.json::<ServerJsonRpcMessage>().await {
+                // does not wait on SSE forever.
-                    Ok(message) => Ok(StreamableHttpPostResponse::Json(message, session_id)),
+                let body = response.bytes().await?;
-                    Err(e) => {
+                match serde_json::from_slice::<ServerJsonRpcMessage>(&body) {
+                    Ok(parsed) => Ok(StreamableHttpPostResponse::Json(parsed, session_id)),
+                    Err(e)
+                        if matches!(
+                            message,
+                            ClientJsonRpcMessage::Notification(_)
+                                | ClientJsonRpcMessage::Response(_)
+                                | ClientJsonRpcMessage::Error(_)
+                        ) =>
+                    {
                         tracing::warn!(
                             "could not parse JSON response as ServerJsonRpcMessage, treating as accepted: {e}"
                         );
                         Ok(StreamableHttpPostResponse::Accepted)
                     }
+                    Err(e) => Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned(
+                        json_rpc_parse_error_message(&e, &body),
+                    ))),
                 }
             }
             _ => {