From d7428f7568ba23cd452504f2a4cb19a34145e6ba Mon Sep 17 00:00:00 2001 From: hugocasa Date: Mon, 12 Oct 2026 09:57:37 +0200 Subject: [PATCH] fix: drop the url from mcp client errors before rmcp logs them Co-Authored-By: Claude Opus 5.5 --- backend/Cargo.lock | 1 + backend/windmill-mcp/Cargo.toml | 1 + backend/windmill-mcp/src/client/mod.rs | 40 ++----- backend/windmill-mcp/src/client/redacting.rs | 115 +++++++++++++++++++ 4 files changed, 125 insertions(+), 32 deletions(-) create mode 100644 backend/windmill-mcp/src/client/redacting.rs diff --git a/backend/Cargo.lock b/backend/Cargo.lock index aa3d601492..657417ac4d 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -16517,6 +16517,7 @@ dependencies = [ "serde", "serde_json", "sqlx", + "sse-stream", "tokio", "tokio-util", "tracing", diff --git a/backend/windmill-mcp/Cargo.toml b/backend/windmill-mcp/Cargo.toml index 8ac23e885d..2c07b92d44 100644 --- a/backend/windmill-mcp/Cargo.toml +++ b/backend/windmill-mcp/Cargo.toml @@ -22,6 +22,7 @@ serde.workspace = true serde_json.workspace = true tracing.workspace = true rmcp.workspace = true +sse-stream = "0.2" sqlx = { workspace = true, optional = true } chrono = { workspace = true, optional = true } async-trait = { workspace = true, optional = true } diff --git a/backend/windmill-mcp/src/client/mod.rs b/backend/windmill-mcp/src/client/mod.rs index 6f2c37f1b6..df4e9ded54 100644 --- a/backend/windmill-mcp/src/client/mod.rs +++ b/backend/windmill-mcp/src/client/mod.rs @@ -3,11 +3,13 @@ //! This module provides functionality for connecting to external MCP servers //! and executing tools on them. +mod redacting; mod types; pub use types::{McpResource, McpToolSource}; use anyhow::{Context, Result}; +use redacting::UrlRedactingClient; use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; use rmcp::{ model::{ @@ -29,20 +31,6 @@ pub struct McpClient { client: RunningService, /// Cached list of available tools from the server available_tools: Vec, - server_url: ServerUrl, -} - -/// reqwest names the URL it failed on in its errors, and the path or query of -/// an MCP server URL can carry the server's key: errors keep the origin only. -struct ServerUrl { - full: String, - origin: String, -} - -impl ServerUrl { - fn redact(&self, e: anyhow::Error) -> anyhow::Error { - anyhow::anyhow!("{}", format!("{e:#}").replace(&self.full, &self.origin)) - } } impl McpClient { @@ -66,16 +54,6 @@ impl McpClient { ) })?; - let mut parsed_url = - reqwest::Url::parse(&resource.url).context("Invalid MCP server URL")?; - // reqwest moves userinfo to a header, so its errors print the URL without it. - let _ = parsed_url.set_username(""); - let _ = parsed_url.set_password(None); - let server_url = ServerUrl { - full: parsed_url.to_string(), - origin: parsed_url.origin().ascii_serialization(), - }; - // Build custom reqwest client with headers if provided let mut headers = HeaderMap::new(); if let Some(token) = token { @@ -121,7 +99,8 @@ impl McpClient { // Create the HTTP transport with custom client let config = StreamableHttpClientTransportConfig::with_uri(resource.url.as_str()); - let transport = StreamableHttpClientTransport::with_client(reqwest_client, config); + let transport = + StreamableHttpClientTransport::with_client(UrlRedactingClient(reqwest_client), config); // Set up client info let client_info = ClientInfo::new( @@ -134,18 +113,16 @@ impl McpClient { let client = client_info .serve(transport) .await - .context("Failed to connect to MCP server") - .map_err(|e| server_url.redact(e))?; + .context("Failed to connect to MCP server")?; // Immediately fetch available tools let available_tools = client .list_tools(Default::default()) .await - .context("Failed to list tools from MCP server") - .map_err(|e| server_url.redact(e))? + .context("Failed to list tools from MCP server")? .tools; - Ok(Self { client, available_tools, server_url }) + Ok(Self { client, available_tools }) } /// Get the list of available tools from the MCP server @@ -168,8 +145,7 @@ impl McpClient { .client .call_tool(params) .await - .context(format!("Failed to call MCP tool: {}", name)) - .map_err(|e| self.server_url.redact(e))?; + .context(format!("Failed to call MCP tool: {}", name))?; // Convert the result to a JSON value // MCP tools return ToolResult which contains content array diff --git a/backend/windmill-mcp/src/client/redacting.rs b/backend/windmill-mcp/src/client/redacting.rs new file mode 100644 index 0000000000..1beffefaf3 --- /dev/null +++ b/backend/windmill-mcp/src/client/redacting.rs @@ -0,0 +1,115 @@ +use std::{collections::HashMap, sync::Arc}; + +use futures::stream::BoxStream; +use reqwest::header::{HeaderName, HeaderValue}; +use rmcp::{ + model::ClientJsonRpcMessage, + transport::streamable_http_client::{ + SseError, StreamableHttpClient, StreamableHttpError, StreamableHttpPostResponse, + }, +}; +use sse_stream::Sse; + +type Error = StreamableHttpError; + +/// reqwest names the URL it failed on in its errors, and an MCP server URL can +/// carry the server's key. rmcp logs those errors itself as well as returning +/// them, so the URL is dropped here, before either can happen. +#[derive(Clone)] +pub(super) struct UrlRedactingClient(pub(super) reqwest::Client); + +fn without_url(e: Error) -> Error { + match e { + StreamableHttpError::Client(e) => StreamableHttpError::Client(e.without_url()), + e => e, + } +} + +impl StreamableHttpClient for UrlRedactingClient { + type Error = reqwest::Error; + + async fn post_message( + &self, + uri: Arc, + message: ClientJsonRpcMessage, + session_id: Option>, + auth_header: Option, + custom_headers: HashMap, + ) -> Result { + self.0 + .post_message(uri, message, session_id, auth_header, custom_headers) + .await + .map_err(without_url) + } + + async fn post_message_with_max_sse_event_size( + &self, + uri: Arc, + message: ClientJsonRpcMessage, + session_id: Option>, + auth_header: Option, + custom_headers: HashMap, + max_sse_event_size: usize, + ) -> Result { + self.0 + .post_message_with_max_sse_event_size( + uri, + message, + session_id, + auth_header, + custom_headers, + max_sse_event_size, + ) + .await + .map_err(without_url) + } + + async fn delete_session( + &self, + uri: Arc, + session_id: Arc, + auth_header: Option, + custom_headers: HashMap, + ) -> Result<(), Error> { + self.0 + .delete_session(uri, session_id, auth_header, custom_headers) + .await + .map_err(without_url) + } + + async fn get_stream( + &self, + uri: Arc, + session_id: Option>, + last_event_id: Option, + auth_header: Option, + custom_headers: HashMap, + ) -> Result>, Error> { + self.0 + .get_stream(uri, session_id, last_event_id, auth_header, custom_headers) + .await + .map_err(without_url) + } + + async fn get_stream_with_max_sse_event_size( + &self, + uri: Arc, + session_id: Option>, + last_event_id: Option, + auth_header: Option, + custom_headers: HashMap, + max_sse_event_size: usize, + ) -> Result>, Error> { + self.0 + .get_stream_with_max_sse_event_size( + uri, + session_id, + last_event_id, + auth_header, + custom_headers, + max_sse_event_size, + ) + .await + .map_err(without_url) + } +}