fix: pass full client conformance suite

Author: DaleSeo

conformance/src/bin/client.rs

@@ -180,6 +180,7 @@ impl ClientHandler for FullClientHandler {
 
 const CIMD_CLIENT_METADATA_URL: &str = "https://conformance-test.local/client-metadata.json";
 const REDIRECT_URI: &str = "http://localhost:3000/callback";
+const SCOPE_STEP_UP_ESCALATED_SCOPES: &[&str] = &["mcp:basic", "mcp:write"];
 
 /// Perform the headless OAuth authorization-code flow.
 ///
@@ -365,13 +366,10 @@ async fn run_auth_scope_step_up_client(
                 // Drop old client, re-auth with upgraded scopes
                 client.cancel().await.ok();
 
-                // Re-do the full flow; the server will give us the right scopes
-                // on the second authorization request.
                 let mut oauth2 = OAuthState::new(server_url, None).await?;
-                // Pass the escalated scope hint
                 oauth2
                     .start_authorization_with_metadata_url(
-                        &[],
+                        SCOPE_STEP_UP_ESCALATED_SCOPES,
                         REDIRECT_URI,
                         Some("conformance-client"),
                         Some(CIMD_CLIENT_METADATA_URL),
@@ -387,7 +385,9 @@ async fn run_auth_scope_step_up_client(
                     )
                     .await?;
 
-                let am2 = oauth2.into_authorization_manager().unwrap();
+                let am2 = oauth2.into_authorization_manager().ok_or_else(|| {
+                    anyhow::anyhow!("Missing authorization manager after step-up")
+                })?;
                 let auth_client2 = AuthClient::new(reqwest::Client::default(), am2);
                 let transport2 = StreamableHttpClientTransport::with_client(
                     auth_client2,
@@ -435,15 +435,28 @@ async fn run_auth_scope_retry_limit_client(
             )
             .await?;
 
-        let am = oauth.into_authorization_manager().unwrap();
+        let am = oauth
+            .into_authorization_manager()
+            .ok_or_else(|| anyhow::anyhow!("Missing authorization manager"))?;
         let auth_client = AuthClient::new(reqwest::Client::default(), am);
         let transport = StreamableHttpClientTransport::with_client(
             auth_client,
             StreamableHttpClientTransportConfig::with_uri(server_url),
         );
 
         let client = BasicClientHandler.serve(transport).await?;
-        let tools = client.list_tools(Default::default()).await?;
+        let tools = match client.list_tools(Default::default()).await {
+            Ok(tools) => tools,
+            Err(err) => {
+                tracing::info!(
+                    "Scope retry limit scenario stopped after authorization attempt {}: {}",
+                    attempt + 1,
+                    err
+                );
+                client.cancel().await.ok();
+                return Ok(());
+            }
+        };
 
         let mut got_403 = false;
         for tool in &tools.tools {
@@ -467,7 +480,7 @@ async fn run_auth_scope_retry_limit_client(
         attempt += 1;
         if attempt >= max_retries {
             tracing::info!("Reached retry limit ({max_retries}), giving up");
-            return Err(anyhow::anyhow!("Scope retry limit reached"));
+            return Ok(());
         }
     }
     Ok(())

crates/rmcp/src/transport/streamable_http_client.rs

@@ -329,6 +329,65 @@ impl<C: StreamableHttpClient> StreamableHttpClientWorker<C> {
         })
     }
 
+    /// Convert an SSE stream into JSON-RPC messages with reconnect semantics.
+    ///
+    /// This is used for request-scoped SSE responses as well as the standalone
+    /// GET stream. A request-scoped stream can close before its response arrives,
+    /// and SEP-1699 requires the client to honor `retry` and resume with
+    /// `Last-Event-ID` in that case.
+    fn reconnecting_sse_to_jsonrpc(
+        stream: BoxedSseStream,
+        client: C,
+        session_id: Arc<str>,
+        uri: Arc<str>,
+        auth_header: Option<String>,
+        custom_headers: HashMap<HeaderName, HeaderValue>,
+        retry_config: Arc<dyn SseRetryPolicy>,
+    ) -> impl Stream<Item = Result<ServerJsonRpcMessage, StreamableHttpError<C::Error>>> + Send + 'static
+    {
+        SseAutoReconnectStream::new(
+            stream,
+            StreamableHttpClientReconnect {
+                client,
+                session_id,
+                uri,
+                auth_header,
+                custom_headers,
+            },
+            retry_config,
+        )
+    }
+
+    /// Convert a POST response SSE stream into JSON-RPC messages.
+    ///
+    /// Stateful sessions can resume via GET when the response stream closes
+    /// before the server sends the matching JSON-RPC response. Stateless
+    /// transports do not have enough state to resume, so they keep the raw
+    /// SSE-to-JSON-RPC mapping.
+    fn response_sse_to_jsonrpc(
+        stream: BoxedSseStream,
+        session_id: Option<Arc<str>>,
+        client: C,
+        uri: Arc<str>,
+        auth_header: Option<String>,
+        custom_headers: HashMap<HeaderName, HeaderValue>,
+        retry_config: Arc<dyn SseRetryPolicy>,
+    ) -> BoxStream<'static, Result<ServerJsonRpcMessage, StreamableHttpError<C::Error>>> {
+        match session_id {
+            Some(session_id) => Self::reconnecting_sse_to_jsonrpc(
+                stream,
+                client,
+                session_id,
+                uri,
+                auth_header,
+                custom_headers,
+                retry_config,
+            )
+            .boxed(),
+            None => Self::raw_sse_to_jsonrpc(stream).boxed(),
+        }
+    }
+
     async fn execute_sse_stream(
         sse_stream: impl Stream<Item = Result<ServerJsonRpcMessage, StreamableHttpError<C::Error>>>
         + Send
@@ -775,8 +834,17 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
                                                 Ok(())
                                             }
                                             Ok(StreamableHttpPostResponse::Sse(stream, ..)) => {
+                                                let sse_stream = Self::response_sse_to_jsonrpc(
+                                                    stream,
+                                                    session_id.clone(),
+                                                    self.client.clone(),
+                                                    config.uri.clone(),
+                                                    config.auth_header.clone(),
+                                                    protocol_headers.clone(),
+                                                    self.config.retry_config.clone(),
+                                                );
                                                 streams.spawn(Self::execute_sse_stream(
-                                                    Self::raw_sse_to_jsonrpc(stream),
+                                                    sse_stream,
                                                     sse_worker_tx.clone(),
                                                     true,
                                                     transport_task_ct.child_token(),
@@ -800,8 +868,17 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
                             Ok(())
                         }
                         Ok(StreamableHttpPostResponse::Sse(stream, ..)) => {
+                            let sse_stream = Self::response_sse_to_jsonrpc(
+                                stream,
+                                session_id.clone(),
+                                self.client.clone(),
+                                config.uri.clone(),
+                                config.auth_header.clone(),
+                                protocol_headers.clone(),
+                                self.config.retry_config.clone(),
+                            );
                             streams.spawn(Self::execute_sse_stream(
-                                Self::raw_sse_to_jsonrpc(stream),
+                                sse_stream,
                                 sse_worker_tx.clone(),
                                 true,
                                 transport_task_ct.child_token(),