mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-13 00:03:00 +00:00
fix: drop the url from mcp client errors before rmcp logs them
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
6d0fd3be83
commit
d7428f7568
Generated
+1
@@ -16517,6 +16517,7 @@ dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
"sqlx",
|
||||
"sse-stream",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"tracing",
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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<RoleClient, InitializeRequestParams>,
|
||||
/// Cached list of available tools from the server
|
||||
available_tools: Vec<McpTool>,
|
||||
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
|
||||
|
||||
@@ -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::Error>;
|
||||
|
||||
/// 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<str>,
|
||||
message: ClientJsonRpcMessage,
|
||||
session_id: Option<Arc<str>>,
|
||||
auth_header: Option<String>,
|
||||
custom_headers: HashMap<HeaderName, HeaderValue>,
|
||||
) -> Result<StreamableHttpPostResponse, Error> {
|
||||
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<str>,
|
||||
message: ClientJsonRpcMessage,
|
||||
session_id: Option<Arc<str>>,
|
||||
auth_header: Option<String>,
|
||||
custom_headers: HashMap<HeaderName, HeaderValue>,
|
||||
max_sse_event_size: usize,
|
||||
) -> Result<StreamableHttpPostResponse, Error> {
|
||||
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<str>,
|
||||
session_id: Arc<str>,
|
||||
auth_header: Option<String>,
|
||||
custom_headers: HashMap<HeaderName, HeaderValue>,
|
||||
) -> 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<str>,
|
||||
session_id: Option<Arc<str>>,
|
||||
last_event_id: Option<String>,
|
||||
auth_header: Option<String>,
|
||||
custom_headers: HashMap<HeaderName, HeaderValue>,
|
||||
) -> Result<BoxStream<'static, Result<Sse, SseError>>, 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<str>,
|
||||
session_id: Option<Arc<str>>,
|
||||
last_event_id: Option<String>,
|
||||
auth_header: Option<String>,
|
||||
custom_headers: HashMap<HeaderName, HeaderValue>,
|
||||
max_sse_event_size: usize,
|
||||
) -> Result<BoxStream<'static, Result<Sse, SseError>>, 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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user