fix: resolve variables in the url and headers of an mcp resource (#11680)

* fix(cli): count superadmins as able to preserve ownership on push Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011BzU9zd53JsTShCsbKzYzB * fix: resolve variables in the url and headers of an mcp resource Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: keep resolved mcp secrets out of header warnings and userinfo urls Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: drop the url from mcp client errors before rmcp logs them Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: run rmcp's reqwest client without a log subscriber Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: scrub the request target from mcp error bodies Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: drop server error bodies from mcp transport errors Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: fail closed when an mcp error message has no status prefix Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-10-13 16:02:59 +00:00 · 2026-10-13 13:52:55 +00:00
parent f629df3382
commit 4bc2394322
8 changed files with 228 additions and 6 deletions
@@ -16523,6 +16523,7 @@ dependencies = [
 "serde",
 "serde_json",
 "sqlx",
+ "sse-stream",
 "tokio",
 "tokio-util",
 "tracing",
@@ -23,3 +23,19 @@ VALUES (
    '{}',
    'test-user'
 );
+
+-- MCP resource whose URL lives in a variable. The host only appears in the
+-- variable, so an error naming it proves the reference was resolved.
+INSERT INTO variable (workspace_id, path, value, is_secret, description, extra_perms)
+VALUES ('test-workspace', 'u/test-user/mcp_url', 'https://mcp-from-var.invalid.windmill.test/key', false, '', '{}');
+
+INSERT INTO resource (workspace_id, path, value, description, resource_type, extra_perms, created_by)
+VALUES (
+    'test-workspace',
+    'u/test-user/var_mcp',
+    '{"name": "var_mcp", "url": "$var:u/test-user/mcp_url"}',
+    'MCP resource with its URL in a variable',
+    'mcp',
+    '{}',
+    'test-user'
+);
@@ -23,9 +23,9 @@ use windmill_common::client::AuthedClient;
 use windmill_test_utils::*;
 use windmill_worker::{load_mcp_tools, McpResourceConfig};

-fn config() -> Vec<McpResourceConfig> {
+fn config(resource_path: &str) -> Vec<McpResourceConfig> {
    vec![McpResourceConfig {
-        resource_path: "$res:f/private/admin_mcp".to_string(),
+        resource_path: resource_path.to_string(),
        include_tools: None,
        exclude_tools: None,
    }]
@@ -34,7 +34,16 @@ fn config() -> Vec<McpResourceConfig> {
 // The Ok variant ((HashMap<_, Arc<McpClient>>, Vec<Tool>)) does not implement
 // Debug, so `expect_err` is unavailable — extract the error explicitly.
 async fn load_err(db: &Pool<Postgres>, client: &AuthedClient, ctx: &str) -> String {
-    match load_mcp_tools(db, "test-workspace", config(), client).await {
+    load_err_for(db, client, "$res:f/private/admin_mcp", ctx).await
+}
+
+async fn load_err_for(
+    db: &Pool<Postgres>,
+    client: &AuthedClient,
+    resource_path: &str,
+    ctx: &str,
+) -> String {
+    match load_mcp_tools(db, "test-workspace", config(resource_path), client).await {
        Ok(_) => panic!("{ctx}"),
        Err(e) => e.to_string(),
    }
@@ -100,3 +109,31 @@ async fn test_mcp_resource_not_loaded_without_authorization(

    Ok(())
 }
+
+/// A `$var:` in the resource's `url` is resolved before connecting, rather than
+/// being handed to the client as a literal (and rejected as a relative URL).
+#[sqlx::test(fixtures("base", "mcp_resource_authz"))]
+async fn test_mcp_resource_url_variable_is_resolved(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let server = ApiServer::start(db.clone()).await?;
+    let client = AuthedClient::new(
+        format!("http://localhost:{}", server.addr.port()),
+        "test-workspace".to_string(),
+        "SECRET_TOKEN".to_string(),
+        None,
+    );
+    let msg = load_err_for(
+        &db,
+        &client,
+        "$res:u/test-user/var_mcp",
+        "the variable's URL is non-resolvable, so the connect step must fail",
+    )
+    .await;
+    assert!(
+        msg.contains("mcp-from-var.invalid.windmill.test"),
+        "the URL should come from the variable: {msg}"
+    );
+
+    Ok(())
+}
@@ -125,8 +125,10 @@ async fn connect_mcp_client(
        None
    };

-    // Header secrets go through the same permissioned path as the token.
-    for value in mcp_resource.headers.iter_mut().flatten().map(|(_, v)| v) {
+    // The URL and header secrets go through the same permissioned path as the token.
+    for value in std::iter::once(&mut mcp_resource.url)
+        .chain(mcp_resource.headers.iter_mut().flatten().map(|(_, v)| v))
+    {
        if let Some(var_path) = value.strip_prefix("$var:") {
            check_scopes(authed, || format!("variables:read:{}", var_path))?;
            let db_authed = DbWithOptAuthed::from_authed(authed, db.clone(), Some(user_db.clone()));
@@ -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::{
@@ -97,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(
@@ -0,0 +1,151 @@
+use std::{borrow::Cow, collections::HashMap, future::Future, 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;
+use tracing::{instrument::WithSubscriber, Dispatch};
+
+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 as well as returning them, so
+/// the URL is dropped before the transport sees one, and rmcp's reqwest client,
+/// which logs the errors it swallows, runs without a subscriber.
+#[derive(Clone)]
+pub(super) struct UrlRedactingClient(pub(super) reqwest::Client);
+
+async fn redacted<T>(request: impl Future<Output = Result<T, Error>>) -> Result<T, Error> {
+    match request.with_subscriber(Dispatch::none()).await {
+        Err(StreamableHttpError::Client(e)) => Err(StreamableHttpError::Client(e.without_url())),
+        // An owned message is `HTTP <status>: <body>`, and the body is the server's
+        // to write: a common one echoes the request target (`Cannot POST /mcp/<key>`).
+        Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned(message))) => {
+            let status = message.split_once(": ").map_or("", |(status, _)| status);
+            let status = status.to_string();
+            Err(StreamableHttpError::UnexpectedServerResponse(status.into()))
+        }
+        r => r,
+    }
+}
+
+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> {
+        redacted(
+            self.0
+                .post_message(uri, message, session_id, auth_header, custom_headers),
+        )
+        .await
+    }
+
+    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> {
+        redacted(self.0.post_message_with_max_sse_event_size(
+            uri,
+            message,
+            session_id,
+            auth_header,
+            custom_headers,
+            max_sse_event_size,
+        ))
+        .await
+    }
+
+    async fn delete_session(
+        &self,
+        uri: Arc<str>,
+        session_id: Arc<str>,
+        auth_header: Option<String>,
+        custom_headers: HashMap<HeaderName, HeaderValue>,
+    ) -> Result<(), Error> {
+        redacted(
+            self.0
+                .delete_session(uri, session_id, auth_header, custom_headers),
+        )
+        .await
+    }
+
+    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> {
+        redacted(
+            self.0
+                .get_stream(uri, session_id, last_event_id, auth_header, custom_headers),
+        )
+        .await
+    }
+
+    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> {
+        redacted(self.0.get_stream_with_max_sse_event_size(
+            uri,
+            session_id,
+            last_event_id,
+            auth_header,
+            custom_headers,
+            max_sse_event_size,
+        ))
+        .await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn server_error_body_is_dropped() {
+        for (message, expected) in [
+            (
+                "HTTP 404 Not Found: Cannot POST /mcp/KEY",
+                "HTTP 404 Not Found",
+            ),
+            ("Cannot POST /mcp/KEY", ""),
+        ] {
+            let err = redacted(async {
+                Err::<(), Error>(StreamableHttpError::UnexpectedServerResponse(
+                    message.to_string().into(),
+                ))
+            })
+            .await
+            .unwrap_err();
+            let StreamableHttpError::UnexpectedServerResponse(got) = err else {
+                panic!("unexpected variant: {err}");
+            };
+            assert_eq!(got, expected);
+        }
+    }
+}
@@ -615,6 +615,17 @@ pub async fn load_mcp_tools(
            None
        };

+        // A URL can carry the server's key, so it goes through the same
+        // permissioned path as the token.
+        if let Some(var_path) = mcp_resource.url.strip_prefix("$var:") {
+            mcp_resource.url = client.get_variable_value(var_path).await.map_err(|e| {
+                Error::internal_err(format!(
+                    "Failed to resolve variable {} for the url of MCP resource {}: {}",
+                    var_path, resource_name, e
+                ))
+            })?;
+        }
+
        // Header secrets go through the same permissioned path as the token.
        for (name, value) in mcp_resource.headers.iter_mut().flatten() {
            if let Some(var_path) = value.strip_prefix("$var:") {