Fix compilation errors from rmcp 1.6.0 upgrade (CVE-2026-42559)

- Implement StreamableHttpClient for McpHttpClient wrapper (reqwest 0.12 compat)
- Remove legacy SSE transport references (SseClientTransport removed in rmcp 1.6)
- Use builder patterns for non-exhaustive structs (CallToolRequestParams, etc.)
- Add wildcard arms for non-exhaustive enum matches (RmcpError, ServiceError)
- Fix register_client call signature (added scopes parameter)
- Update type aliases for deprecated types (CallToolRequestParam -> Params)

Co-Authored-By: Oz <oz-agent@warp.dev>

Author: acarl005

Cargo.lock

@@ -293,9 +293,10 @@ aws-types = "1"
 
 rmcp = { workspace = true, features = [
     "auth",
-    "transport-streamable-http-client-reqwest",
+    "transport-streamable-http-client",
     "transport-child-process",
 ] }
+sse-stream = "0.2"
 notify-debouncer-full.workspace = true
 rquickjs = { workspace = true, optional = true }
 rustls.workspace = true

app/Cargo.toml

@@ -141,10 +141,10 @@ impl CallMCPToolExecutor {
             ActionExecution::new_async(
                 async move {
                     reconnecting_peer
-                        .call_tool(rmcp::model::CallToolRequestParam {
-                            name: name_owned_inner.into(),
-                            arguments: Some(arguments),
-                        })
+                        .call_tool(
+                            rmcp::model::CallToolRequestParams::new(name_owned_inner)
+                                .with_arguments(arguments),
+                        )
                         .await
                 },
                 move |res, ctx| handle_call_tool_result(res, server_output_id, name_clone, ctx),

app/src/ai/blocklist/action_model/execute/call_mcp_tool.rs

@@ -128,7 +128,7 @@ impl ReadMCPResourceExecutor {
             ActionExecution::new_async(
                 async move {
                     reconnecting_peer
-                        .read_resource(rmcp::model::ReadResourceRequestParam { uri })
+                        .read_resource(rmcp::model::ReadResourceRequestParams::new(uri))
                         .await
                 },
                 |res, _ctx| handle_read_resource_result(res),

app/src/ai/blocklist/action_model/execute/read_mcp_resource.rs

@@ -1,8 +1,25 @@
-use std::collections::HashMap;
+use std::{borrow::Cow, collections::HashMap, sync::Arc};
 
-use reqwest::header::HeaderMap;
+use futures::{StreamExt, stream::BoxStream};
+use http::{HeaderName, HeaderValue, header::WWW_AUTHENTICATE};
+use reqwest::header::{HeaderMap, ACCEPT};
+use rmcp::{
+    model::{ClientJsonRpcMessage, JsonRpcMessage, ServerJsonRpcMessage},
+    transport::{
+        common::http_header::{
+            EVENT_STREAM_MIME_TYPE, HEADER_LAST_EVENT_ID, HEADER_SESSION_ID, JSON_MIME_TYPE,
+        },
+        streamable_http_client::*,
+    },
+};
+use sse_stream::{Sse, SseStream};
 
-type ReqwestHttpTransport = rmcp::transport::StreamableHttpClientTransport<reqwest::Client>;
+/// Newtype wrapper around `reqwest::Client` so we can implement the foreign
+/// `StreamableHttpClient` trait (orphan rule).
+#[derive(Clone, Debug, Default)]
+pub struct McpHttpClient(pub reqwest::Client);
+
+pub type ReqwestHttpTransport = rmcp::transport::StreamableHttpClientTransport<McpHttpClient>;
 
 /// Builds a `HeaderMap` from a `HashMap<String, String>` of user-provided headers.
 ///
@@ -27,3 +44,247 @@ pub fn build_client_with_headers(
             ))
         })
 }
+
+/// Reserved headers that must not be overridden by custom headers.
+/// Matches the validation logic in rmcp's `validate_custom_header`.
+const RESERVED_HEADERS: &[&str] = &[
+    "accept",
+    "content-type",
+    "mcp-session-id",
+    "last-event-id",
+    "authorization",
+    "host",
+    "origin",
+];
+
+/// Applies custom headers to a request builder, rejecting reserved headers.
+fn apply_custom_headers(
+    mut builder: reqwest::RequestBuilder,
+    custom_headers: HashMap<HeaderName, HeaderValue>,
+) -> Result<reqwest::RequestBuilder, StreamableHttpError<reqwest::Error>> {
+    for (name, value) in custom_headers {
+        let name_lower = name.as_str().to_lowercase();
+        if RESERVED_HEADERS.contains(&name_lower.as_str()) {
+            return Err(StreamableHttpError::ReservedHeaderConflict(name.to_string()));
+        }
+        builder = builder.header(name, value);
+    }
+    Ok(builder)
+}
+
+/// Extracts the scope value from a WWW-Authenticate header.
+fn extract_scope(header: &str) -> Option<String> {
+    header.split(',').find_map(|part| {
+        let part = part.trim();
+        if let Some(rest) = part.strip_prefix("scope=") {
+            Some(rest.trim_matches('"').to_string())
+        } else {
+            None
+        }
+    })
+}
+
+/// Attempts to parse `body` as a JSON-RPC error message.
+fn parse_json_rpc_error(body: &str) -> Option<ServerJsonRpcMessage> {
+    match serde_json::from_str::<ServerJsonRpcMessage>(body) {
+        Ok(message @ JsonRpcMessage::Error(_)) => Some(message),
+        _ => None,
+    }
+}
+
+/// Implement `StreamableHttpClient` for our newtype wrapper around reqwest 0.12's `Client`.
+///
+/// rmcp 1.6.0 ships its own impl for reqwest 0.13, but warp uses reqwest 0.12.
+/// This provides the equivalent implementation against the 0.12 API.
+impl StreamableHttpClient for McpHttpClient {
+    type Error = reqwest::Error;
+
+    async fn get_stream(
+        &self,
+        uri: Arc<str>,
+        session_id: Arc<str>,
+        last_event_id: Option<String>,
+        auth_token: Option<String>,
+        custom_headers: HashMap<HeaderName, HeaderValue>,
+    ) -> Result<BoxStream<'static, Result<Sse, SseError>>, StreamableHttpError<Self::Error>> {
+        let mut request_builder = self.0
+            .get(uri.as_ref())
+            .header(ACCEPT, [EVENT_STREAM_MIME_TYPE, JSON_MIME_TYPE].join(", "))
+            .header(HEADER_SESSION_ID, session_id.as_ref());
+        if let Some(last_event_id) = last_event_id {
+            request_builder = request_builder.header(HEADER_LAST_EVENT_ID, last_event_id);
+        }
+        if let Some(auth_header) = auth_token {
+            request_builder = request_builder.bearer_auth(auth_header);
+        }
+        request_builder = apply_custom_headers(request_builder, custom_headers)?;
+        let response = request_builder
+            .send()
+            .await
+            .map_err(StreamableHttpError::Client)?;
+        if response.status() == reqwest::StatusCode::METHOD_NOT_ALLOWED {
+            return Err(StreamableHttpError::ServerDoesNotSupportSse);
+        }
+        let response = response
+            .error_for_status()
+            .map_err(StreamableHttpError::Client)?;
+        match response.headers().get(reqwest::header::CONTENT_TYPE) {
+            Some(ct) => {
+                if !ct.as_bytes().starts_with(EVENT_STREAM_MIME_TYPE.as_bytes())
+                    && !ct.as_bytes().starts_with(JSON_MIME_TYPE.as_bytes())
+                {
+                    return Err(StreamableHttpError::UnexpectedContentType(Some(
+                        String::from_utf8_lossy(ct.as_bytes()).to_string(),
+                    )));
+                }
+            }
+            None => {
+                return Err(StreamableHttpError::UnexpectedContentType(None));
+            }
+        }
+        let event_stream = SseStream::from_byte_stream(response.bytes_stream()).boxed();
+        Ok(event_stream)
+    }
+
+    async fn delete_session(
+        &self,
+        uri: Arc<str>,
+        session: Arc<str>,
+        auth_token: Option<String>,
+        custom_headers: HashMap<HeaderName, HeaderValue>,
+    ) -> Result<(), StreamableHttpError<Self::Error>> {
+        let mut request_builder = self.0.delete(uri.as_ref());
+        if let Some(auth_header) = auth_token {
+            request_builder = request_builder.bearer_auth(auth_header);
+        }
+        request_builder = request_builder.header(HEADER_SESSION_ID, session.as_ref());
+        request_builder = apply_custom_headers(request_builder, custom_headers)?;
+        let response = request_builder
+            .send()
+            .await
+            .map_err(StreamableHttpError::Client)?;
+        if response.status() == reqwest::StatusCode::METHOD_NOT_ALLOWED {
+            tracing::debug!("this server doesn't support deleting session");
+            return Ok(());
+        }
+        let _response = response
+            .error_for_status()
+            .map_err(StreamableHttpError::Client)?;
+        Ok(())
+    }
+
+    async fn post_message(
+        &self,
+        uri: Arc<str>,
+        message: ClientJsonRpcMessage,
+        session_id: Option<Arc<str>>,
+        auth_token: Option<String>,
+        custom_headers: HashMap<HeaderName, HeaderValue>,
+    ) -> Result<StreamableHttpPostResponse, StreamableHttpError<Self::Error>> {
+        let mut request = self.0
+            .post(uri.as_ref())
+            .header(ACCEPT, [EVENT_STREAM_MIME_TYPE, JSON_MIME_TYPE].join(", "));
+        if let Some(auth_header) = auth_token {
+            request = request.bearer_auth(auth_header);
+        }
+        request = apply_custom_headers(request, custom_headers)?;
+        let session_was_attached = session_id.is_some();
+        if let Some(session_id) = session_id {
+            request = request.header(HEADER_SESSION_ID, session_id.as_ref());
+        }
+        let response = request
+            .json(&message)
+            .send()
+            .await
+            .map_err(StreamableHttpError::Client)?;
+        if response.status() == reqwest::StatusCode::UNAUTHORIZED {
+            if let Some(header) = response.headers().get(WWW_AUTHENTICATE) {
+                let header = header
+                    .to_str()
+                    .map_err(|_| {
+                        StreamableHttpError::UnexpectedServerResponse(Cow::from(
+                            "invalid www-authenticate header value",
+                        ))
+                    })?
+                    .to_string();
+                return Err(StreamableHttpError::AuthRequired(
+                    AuthRequiredError::new(header),
+                ));
+            }
+        }
+        if response.status() == reqwest::StatusCode::FORBIDDEN {
+            if let Some(header) = response.headers().get(WWW_AUTHENTICATE) {
+                let header_str = header.to_str().map_err(|_| {
+                    StreamableHttpError::UnexpectedServerResponse(Cow::from(
+                        "invalid www-authenticate header value",
+                    ))
+                })?;
+                return Err(StreamableHttpError::InsufficientScope(
+                    InsufficientScopeError::new(header_str.to_string(), extract_scope(header_str)),
+                ));
+            }
+        }
+        let status = response.status();
+        if matches!(
+            status,
+            reqwest::StatusCode::ACCEPTED | reqwest::StatusCode::NO_CONTENT
+        ) {
+            return Ok(StreamableHttpPostResponse::Accepted);
+        }
+        if status == reqwest::StatusCode::NOT_FOUND && session_was_attached {
+            return Err(StreamableHttpError::SessionExpired);
+        }
+        let content_type = response
+            .headers()
+            .get(reqwest::header::CONTENT_TYPE)
+            .map(|ct| String::from_utf8_lossy(ct.as_bytes()).to_string());
+        let session_id = response
+            .headers()
+            .get(HEADER_SESSION_ID)
+            .and_then(|v| v.to_str().ok())
+            .map(|s| s.to_string());
+        if !status.is_success() {
+            let body = response
+                .text()
+                .await
+                .unwrap_or_else(|_| "<failed to read response body>".to_owned());
+            if content_type
+                .as_deref()
+                .is_some_and(|ct| ct.as_bytes().starts_with(JSON_MIME_TYPE.as_bytes()))
+            {
+                match parse_json_rpc_error(&body) {
+                    Some(message) => {
+                        return Ok(StreamableHttpPostResponse::Json(message, session_id));
+                    }
+                    None => tracing::warn!(
+                        "HTTP {status}: could not parse JSON body as a JSON-RPC error"
+                    ),
+                }
+            }
+            return Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned(
+                format!("HTTP {status}: {body}"),
+            )));
+        }
+        match content_type.as_deref() {
+            Some(ct) if ct.as_bytes().starts_with(EVENT_STREAM_MIME_TYPE.as_bytes()) => {
+                let event_stream = SseStream::from_byte_stream(response.bytes_stream()).boxed();
+                Ok(StreamableHttpPostResponse::Sse(event_stream, session_id))
+            }
+            Some(ct) if ct.as_bytes().starts_with(JSON_MIME_TYPE.as_bytes()) => {
+                match response.json::<ServerJsonRpcMessage>().await {
+                    Ok(message) => Ok(StreamableHttpPostResponse::Json(message, session_id)),
+                    Err(e) => {
+                        tracing::warn!(
+                            "could not parse JSON response as ServerJsonRpcMessage, treating as accepted: {e}"
+                        );
+                        Ok(StreamableHttpPostResponse::Accepted)
+                    }
+                }
+            }
+            _ => {
+                tracing::error!("unexpected content type: {:?}", content_type);
+                Err(StreamableHttpError::UnexpectedContentType(content_type))
+            }
+        }
+    }
+}

app/src/ai/mcp/http_client.rs

@@ -120,7 +120,7 @@ impl ReconnectingPeer {
     /// Calls a tool on the MCP server.
     pub async fn call_tool(
         &self,
-        params: rmcp::model::CallToolRequestParam,
+        params: rmcp::model::CallToolRequestParams,
     ) -> Result<rmcp::model::CallToolResult, rmcp::ServiceError> {
         self.with_reconnect_retry(params, |peer, p| async move { peer.call_tool(p).await })
             .await
@@ -129,7 +129,7 @@ impl ReconnectingPeer {
     /// Reads a resource from the MCP server.
     pub async fn read_resource(
         &self,
-        params: rmcp::model::ReadResourceRequestParam,
+        params: rmcp::model::ReadResourceRequestParams,
     ) -> Result<rmcp::model::ReadResourceResult, rmcp::ServiceError> {
         self.with_reconnect_retry(params, |peer, p| async move { peer.read_resource(p).await })
             .await