mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-20 00:03:21 +00:00
fix: resolve variables in the url and headers of an mcp resource
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
bc02fc0748
commit
ded49c51f8
+16
@@ -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(())
|
||||
}
|
||||
|
||||
@@ -33,6 +33,22 @@ async fn with_deadline<T>(
|
||||
})?
|
||||
}
|
||||
|
||||
/// The MCP resource is read raw so that `token` stays a variable path, which
|
||||
/// leaves a `$var:` in its `url` or header values unresolved. A scoped token
|
||||
/// resolves it as the token rather than as its user, as resource interpolation does.
|
||||
async fn resolve_var(
|
||||
authed: &ApiAuthed,
|
||||
db_authed: &DbWithOptAuthed<'_, ApiAuthed>,
|
||||
w_id: &str,
|
||||
value: String,
|
||||
) -> Result<String> {
|
||||
let Some(var_path) = value.strip_prefix("$var:") else {
|
||||
return Ok(value);
|
||||
};
|
||||
check_scopes(authed, || format!("variables:read:{}", var_path))?;
|
||||
get_value_internal(db_authed, w_id, var_path, false).await
|
||||
}
|
||||
|
||||
/// Connect to the MCP server described by the `mcp` resource at `path`.
|
||||
///
|
||||
/// The caller is responsible for the scope check; everything else (resource
|
||||
@@ -65,8 +81,16 @@ async fn connect_mcp_client(
|
||||
let resource_value = not_found_if_none(resource_value_o, "Resource", path)?
|
||||
.ok_or_else(|| Error::BadRequest(format!("Empty resource value for {}", path)))?;
|
||||
|
||||
let mcp_resource = serde_json::from_str::<windmill_mcp::McpResource>(resource_value.0.get())
|
||||
.map_err(|e| Error::BadRequest(format!("Failed to parse MCP resource: {}", e)))?;
|
||||
let mut mcp_resource =
|
||||
serde_json::from_str::<windmill_mcp::McpResource>(resource_value.0.get())
|
||||
.map_err(|e| Error::BadRequest(format!("Failed to parse MCP resource: {}", e)))?;
|
||||
|
||||
let db_authed = DbWithOptAuthed::from_authed(authed, db.clone(), Some(user_db.clone()));
|
||||
let url = std::mem::take(&mut mcp_resource.url);
|
||||
mcp_resource.url = resolve_var(authed, &db_authed, w_id, url).await?;
|
||||
for value in mcp_resource.headers.iter_mut().flat_map(|h| h.values_mut()) {
|
||||
*value = resolve_var(authed, &db_authed, w_id, std::mem::take(value)).await?;
|
||||
}
|
||||
|
||||
#[cfg(feature = "oauth2")]
|
||||
{
|
||||
@@ -119,7 +143,6 @@ async fn connect_mcp_client(
|
||||
if token_var_path.trim().is_empty() {
|
||||
None
|
||||
} else {
|
||||
let db_authed = DbWithOptAuthed::from_authed(authed, db.clone(), Some(user_db.clone()));
|
||||
Some(get_value_internal(&db_authed, w_id, token_var_path, false).await?)
|
||||
}
|
||||
} else {
|
||||
|
||||
@@ -29,6 +29,20 @@ 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 {
|
||||
@@ -52,6 +66,12 @@ impl McpClient {
|
||||
)
|
||||
})?;
|
||||
|
||||
let parsed_url = reqwest::Url::parse(&resource.url).context("Invalid MCP server URL")?;
|
||||
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 {
|
||||
@@ -109,16 +129,18 @@ impl McpClient {
|
||||
let client = client_info
|
||||
.serve(transport)
|
||||
.await
|
||||
.context("Failed to connect to MCP server")?;
|
||||
.context("Failed to connect to MCP server")
|
||||
.map_err(|e| server_url.redact(e))?;
|
||||
|
||||
// Immediately fetch available tools
|
||||
let available_tools = client
|
||||
.list_tools(Default::default())
|
||||
.await
|
||||
.context("Failed to list tools from MCP server")?
|
||||
.context("Failed to list tools from MCP server")
|
||||
.map_err(|e| server_url.redact(e))?
|
||||
.tools;
|
||||
|
||||
Ok(Self { client, available_tools })
|
||||
Ok(Self { client, available_tools, server_url })
|
||||
}
|
||||
|
||||
/// Get the list of available tools from the MCP server
|
||||
@@ -141,7 +163,8 @@ impl McpClient {
|
||||
.client
|
||||
.call_tool(params)
|
||||
.await
|
||||
.context(format!("Failed to call MCP tool: {}", name))?;
|
||||
.context(format!("Failed to call MCP tool: {}", name))
|
||||
.map_err(|e| self.server_url.redact(e))?;
|
||||
|
||||
// Convert the result to a JSON value
|
||||
// MCP tools return ToolResult which contains content array
|
||||
|
||||
@@ -549,6 +549,26 @@ async fn refresh_token_if_expired(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The MCP resource is read raw so that `token` stays a variable path, which
|
||||
/// leaves a `$var:` in its `url` or header values unresolved: resolve it through
|
||||
/// the job's permissioned variable path, like the token.
|
||||
#[cfg(feature = "mcp")]
|
||||
async fn resolve_mcp_var(
|
||||
client: &AuthedClient,
|
||||
value: String,
|
||||
resource_name: &str,
|
||||
) -> Result<String, Error> {
|
||||
let Some(var_path) = value.strip_prefix("$var:") else {
|
||||
return Ok(value);
|
||||
};
|
||||
client.get_variable_value(var_path).await.map_err(|e| {
|
||||
Error::internal_err(format!(
|
||||
"Failed to resolve variable {} for MCP resource {}: {}",
|
||||
var_path, resource_name, e
|
||||
))
|
||||
})
|
||||
}
|
||||
|
||||
/// Load tools from MCP servers and return both the clients and tools
|
||||
/// Returns a map of resource name -> client, and a vector of tools
|
||||
#[cfg(feature = "mcp")]
|
||||
@@ -570,7 +590,7 @@ pub async fn load_mcp_tools(
|
||||
// is not allowed to read (resources:read:{path}). Reading through the raw
|
||||
// db pool here would bypass the authorization enforced by the regular MCP
|
||||
// tools API (get_mcp_tools) and act as a confused deputy.
|
||||
let mcp_resource = client
|
||||
let mut mcp_resource = client
|
||||
.get_resource_value::<McpResource>(path)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
@@ -582,6 +602,12 @@ pub async fn load_mcp_tools(
|
||||
|
||||
let resource_name = mcp_resource.name.clone();
|
||||
|
||||
let url = std::mem::take(&mut mcp_resource.url);
|
||||
mcp_resource.url = resolve_mcp_var(client, url, &resource_name).await?;
|
||||
for value in mcp_resource.headers.iter_mut().flat_map(|h| h.values_mut()) {
|
||||
*value = resolve_mcp_var(client, std::mem::take(value), &resource_name).await?;
|
||||
}
|
||||
|
||||
// Resolve the token through the job's permissioned (RLS + audit) path so
|
||||
// the AI agent cannot exfiltrate a secret its identity is not allowed to
|
||||
// read by pointing an MCP resource's token at it.
|
||||
|
||||
Reference in New Issue
Block a user