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>
chore(deps): bump the codeql-action group with 4 updates (#1277 )
2026-09-18 13:47:35 -04:00 · 2026-09-18 13:47:13 -04:00 · 2026-09-18 13:46:56 -04:00 · 2026-09-18 13:46:37 -04:00 · 2026-09-15 11:37:39 -04:00 · 2026-09-15 11:18:33 -04:00
78 changed files with 2580 additions and 442 deletions
--- a/.github/workflows/ci.yml
+++ b/.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
--- a/.github/workflows/codeql.yml
+++ b/.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
--- a/.github/workflows/scorecard.yml
+++ b/.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
--- a/Cargo.toml
+++ b/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-macros = { version = "3.3.0", path = "./crates/rmcp-macros" }
+rmcp = { version = "3.4.0", path = "./crates/rmcp" }
+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/"
--- a/README.md
+++ b/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 {
-        ServerInfo::new(
+    fn get_info(&self) -> ServerConfig {
+        ServerConfig::new(
            ServerCapabilities::builder()
                .enable_resources()
                .build(),
@ -582,8 +582,8 @@ impl MyServer {

 #[prompt_handler]
 impl ServerHandler for MyServer {
-    fn get_info(&self) -> ServerInfo {
-        ServerInfo::new(ServerCapabilities::builder().enable_prompts().build())
+    fn get_info(&self) -> ServerConfig {
+        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 {
-        ServerInfo::new(
+    fn get_info(&self) -> ServerConfig {
+        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 {
-        ServerInfo::new(
+    fn get_info(&self) -> ServerConfig {
+        ServerConfig::new(
            ServerCapabilities::builder()
                .enable_completions()
                .enable_prompts()
@ -1253,8 +1253,8 @@ use rmcp::{
 };

 impl ServerHandler for MyServer {
-    fn get_info(&self) -> ServerInfo {
-        ServerInfo::new(
+    fn get_info(&self) -> ServerConfig {
+        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 {
-    ServerInfo::new(
+fn get_info(&self) -> ServerConfig {
+    ServerConfig::new(
        ServerCapabilities::builder()
            .enable_tools()
            .enable_prompts()
--- a/conformance/src/bin/client.rs
+++ b/conformance/src/bin/client.rs
@ -74,8 +74,8 @@ impl ClientHandler for BasicClientHandler {}
 struct ElicitationDefaultsClientHandler;

 impl ClientHandler for ElicitationDefaultsClientHandler {
-    fn get_info(&self) -> ClientInfo {
-        let mut info = ClientInfo::default();
+    fn get_info(&self) -> ClientConfig {
+        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 {
-        let mut info = ClientInfo::default();
+    fn get_info(&self) -> ClientConfig {
+        let mut info = ClientConfig::default();
        info.capabilities.elicitation = Some(
            ElicitationCapability::new()
                .with_form(FormElicitationCapability::new().with_schema_validation(true)),
--- a/conformance/src/bin/server.rs
+++ b/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 {
-        ServerInfo::new(
+    fn get_info(&self) -> ServerConfig {
+        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(),
--- a/crates/rmcp-macros/CHANGELOG.md
+++ b/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
--- a/crates/rmcp-macros/src/lib.rs
+++ b/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 {
-///         ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
+///     fn get_info(&self) -> ServerConfig {
+///         ServerConfig::new(ServerCapabilities::builder().enable_tools().build())
 ///     }
 /// }
 /// ```
--- a/crates/rmcp-macros/src/tool_handler.rs
+++ b/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 {
-            rmcp::model::ServerInfo::new(
+        fn get_info(&self) -> rmcp::model::ServerConfig {
+            rmcp::model::ServerConfig::new(
                rmcp::model::ServerCapabilities::builder()
                    #(#capability_calls)*
                    .build()
--- a/crates/rmcp/CHANGELOG.md
+++ b/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
--- a/crates/rmcp/Cargo.toml
+++ b/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"
--- a/crates/rmcp/src/handler/client.rs
+++ b/crates/rmcp/src/handler/client.rs
@ -274,8 +274,8 @@ macro_rules! client_handler_methods {
            std::future::ready(())
        }

-        fn get_info(&self) -> ClientInfo {
-            ClientInfo::default()
+        fn get_info(&self) -> ClientConfig {
+            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.
-impl ClientHandler for ClientInfo {
-    fn get_info(&self) -> ClientInfo {
+/// Do nothing, with a specific client config.
+impl ClientHandler for ClientConfig {
+    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()
            }
        }
--- a/crates/rmcp/src/handler/server.rs
+++ b/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 {
-        ///         ServerInfo::default()
+        ///     fn get_info(&self) -> ServerConfig {
+        ///         ServerConfig::default()
        ///     }
        ///
        ///     async fn initialize(
@ -600,8 +600,8 @@ macro_rules! server_handler_methods {
            std::future::ready(())
        }

-        fn get_info(&self) -> ServerInfo {
-            ServerInfo::default()
+        fn get_info(&self) -> ServerConfig {
+            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()
            }

--- a/crates/rmcp/src/model.rs
+++ b/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(),
--- a/crates/rmcp/src/service.rs
+++ b/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() {
-                                if R::IS_CLIENT {
-                                    if let Some(responder) =
-                                        local_responder_pool.remove(request_id)
+                            // Modern servers cancel listen requests; legacy peers cancel
+                            // requests they originated, even when both directions share an ID.
+                            if R::peer_cancels_subscriptions(&peer) {
+                                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;
--- a/crates/rmcp/src/service/client.rs
+++ b/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,
-        InputRequiredResult, InputResponses, JsonRpcResponse, ListPromptsRequest,
+        GetTaskResult, InitializeRequest, InitializeRequestParams, InitializedNotification,
+        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};
--- a/crates/rmcp/src/service/server.rs
+++ b/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,
-        ListRootsResult, LoggingMessageNotification, LoggingMessageNotificationParam,
-        ProgressNotification, ProgressNotificationParam, PromptListChangedNotification,
-        ProtocolVersion, ResourceListChangedNotification, ResourceUpdatedNotification,
-        ResourceUpdatedNotificationParam, ServerInfo, ServerNotification, ServerRequest,
-        ServerResult, SubscriptionFilter, SubscriptionsAcknowledgedNotification,
-        SubscriptionsAcknowledgedNotificationParams, ToolListChangedNotification,
+        CreateMessageRequestParams, CreateMessageResult, EmptyResult, ErrorData,
+        InitializeRequestParams, ListRootsRequest, ListRootsResult, LoggingMessageNotification,
+        LoggingMessageNotificationParam, ProgressNotification, ProgressNotificationParam,
+        PromptListChangedNotification, ProtocolVersion, ResourceListChangedNotification,
+        ResourceUpdatedNotification, ResourceUpdatedNotificationParam, ServerConfig,
+        ServerNotification, ServerRequest, ServerResult, SubscriptionFilter,
+        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 PeerInfo = ClientInfo;
+    type Info = ServerConfig;
+    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 {
-                ct: ct.child_token(),
-                id: id.clone(),
-                meta: std::mem::take(request.get_meta_mut()),
-                extensions: std::mem::take(request.extensions_mut()),
-                peer: peer.clone(),
-            };
-            let response = match service.handle_request(request, context).await {
-                Ok(result) => ServerJsonRpcMessage::response(result, id),
-                Err(error) => ServerJsonRpcMessage::error(error, Some(id)),
-            };
-            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));
+            // Dispatch the request from inside the service loop rather than
+            // inline: its handler may send notifications through `peer`, which
+            // only complete once the loop drains `peer_rx`.
+            return Ok(serve_inner(
+                service,
+                transport,
+                peer,
+                peer_rx,
+                VecDeque::from([ClientJsonRpcMessage::request(request, id)]),
+                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 {
--- a/crates/rmcp/src/transport/auth.rs
+++ b/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)
-            .await
+        self.authorization_metadata_from_resource_metadata(
+            &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 {
-            candidates.extend(list);
+        for candidate in resource_metadata.authorization_servers.unwrap_or_default() {
+            push_candidate(candidate);
        }

-        for candidate in candidates {
-            let candidate = candidate.trim();
-            if candidate.is_empty() {
-                continue;
-            }
+        if candidates.is_empty() {
+            debug!(
+                "protected resource metadata at {resource_metadata_url} names no authorization server"
+            );
+            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> {
-        if let Some(resource_metadata_url) =
-            self.probe_resource_metadata_url(&self.base_url).await?
-        {
-            return Ok(Some(resource_metadata_url));
+    /// Look for the protected resource metadata document, reading it where it is
+    /// found so that a candidate answering with something else only costs that
+    /// candidate.
+    async fn discover_resource_metadata(
+        &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();
-            discovery_url.set_query(None);
-            discovery_url.set_fragment(None);
-            discovery_url.set_path(&candidate_path);
-            if let Some(resource_metadata_url) =
-                self.probe_resource_metadata_url(&discovery_url).await?
-            {
-                return Ok(Some(resource_metadata_url));
+            let mut candidate_url = self.base_url.clone();
+            candidate_url.set_query(None);
+            candidate_url.set_fragment(None);
+            candidate_url.set_path(&candidate_path);
+
+            if !requested.insert(candidate_url.clone()) {
+                continue;
+            }
+
+            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
-    /// 200 (the url itself is the metadata document) or from a 401's
-    /// WWW-Authenticate header value.
+    /// Probe the resource itself, looking only for a `WWW-Authenticate` challenge
+    /// that carries a `resource_metadata` pointer.
+    ///
+    /// 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() {
-            StatusCode::OK => Ok(Some(url.clone())),
-            StatusCode::UNAUTHORIZED => Ok(self
+        if response.status() == StatusCode::UNAUTHORIZED {
+            return Ok(self
                .extract_resource_metadata_url_from_www_authenticate(&response)
-                .await),
-            status => {
-                debug!("resource metadata probe returned unexpected status: {status}");
-                Ok(None)
-            }
+                .await);
        }
+
+        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,
--- a/crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs
+++ b/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
-                // malformed (e.g. a 200 response to a notification that lacks
-                // an `id` field), treat it as accepted rather than failing.
-                match response.json::<ServerJsonRpcMessage>().await {
-                    Ok(message) => Ok(StreamableHttpPostResponse::Json(message, session_id)),
-                    Err(e) => {
+                // A notification/response/error POST does not await a JSON-RPC
+                // reply. Treat an unusable JSON body as Accepted. A request
+                // still needs a reply; the same body is an error so the worker
+                // does not wait on SSE forever.
+                let body = response.bytes().await?;
+                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),
+                    ))),
                }
            }
            _ => {
--- a/Show more
+++ b/Show more