From 437bad4cb7d6bfac136b1b41971edba66cd93eb9 Mon Sep 17 00:00:00 2001 From: centdix <40307056+centdix@users.noreply.github.com> Date: Fri, 16 Jan 2026 21:55:14 +0100 Subject: [PATCH] chore: move mcp logic to windmill-mcp (#7584) * draft * clean up mcp logic * cleaning * cleaning * better code * error logging * cleaning --- backend/Cargo.lock | 6 + .../generate_mcp_tools.py | 28 +- backend/windmill-api/src/lib.rs | 5 +- .../{tools => }/auto_generated_endpoints.rs | 16 +- backend/windmill-api/src/mcp/core.rs | 448 ++++++++++++++ backend/windmill-api/src/mcp/mod.rs | 10 +- backend/windmill-api/src/mcp/server.rs | 561 ------------------ .../src/mcp/tools/endpoint_tools.rs | 307 ---------- .../windmill-api/src/mcp/tools/flow_tools.rs | 40 -- .../windmill-api/src/mcp/tools/hub_tools.rs | 43 -- backend/windmill-api/src/mcp/tools/mod.rs | 10 - .../src/mcp/tools/script_tools.rs | 40 -- backend/windmill-api/src/mcp/utils.rs | 414 +++++++++++++ .../windmill-api/src/mcp/utils/database.rs | 243 -------- backend/windmill-api/src/mcp/utils/mod.rs | 10 - backend/windmill-api/src/mcp/utils/schema.rs | 143 ----- backend/windmill-mcp/Cargo.toml | 8 +- backend/windmill-mcp/src/client/mod.rs | 209 +++++++ backend/windmill-mcp/src/client/types.rs | 32 + backend/windmill-mcp/src/common/mod.rs | 16 + backend/windmill-mcp/src/common/schema.rs | 41 ++ .../src/common/scope.rs} | 48 +- .../src/common}/transform.rs | 39 +- .../src/common/types.rs} | 37 +- backend/windmill-mcp/src/lib.rs | 275 +-------- backend/windmill-mcp/src/server/backend.rs | 159 +++++ backend/windmill-mcp/src/server/endpoints.rs | 103 ++++ backend/windmill-mcp/src/server/mod.rs | 32 + backend/windmill-mcp/src/server/runner.rs | 373 ++++++++++++ backend/windmill-mcp/src/server/tools.rs | 184 ++++++ 30 files changed, 2161 insertions(+), 1719 deletions(-) rename backend/windmill-api/src/mcp/{tools => }/auto_generated_endpoints.rs (99%) create mode 100644 backend/windmill-api/src/mcp/core.rs delete mode 100644 backend/windmill-api/src/mcp/server.rs delete mode 100644 backend/windmill-api/src/mcp/tools/endpoint_tools.rs delete mode 100644 backend/windmill-api/src/mcp/tools/flow_tools.rs delete mode 100644 backend/windmill-api/src/mcp/tools/hub_tools.rs delete mode 100644 backend/windmill-api/src/mcp/tools/mod.rs delete mode 100644 backend/windmill-api/src/mcp/tools/script_tools.rs create mode 100644 backend/windmill-api/src/mcp/utils.rs delete mode 100644 backend/windmill-api/src/mcp/utils/database.rs delete mode 100644 backend/windmill-api/src/mcp/utils/mod.rs delete mode 100644 backend/windmill-api/src/mcp/utils/schema.rs create mode 100644 backend/windmill-mcp/src/client/mod.rs create mode 100644 backend/windmill-mcp/src/client/types.rs create mode 100644 backend/windmill-mcp/src/common/mod.rs create mode 100644 backend/windmill-mcp/src/common/schema.rs rename backend/{windmill-api/src/mcp/utils/scope_matcher.rs => windmill-mcp/src/common/scope.rs} (86%) rename backend/{windmill-api/src/mcp/utils => windmill-mcp/src/common}/transform.rs (78%) rename backend/{windmill-api/src/mcp/utils/models.rs => windmill-mcp/src/common/types.rs} (68%) create mode 100644 backend/windmill-mcp/src/server/backend.rs create mode 100644 backend/windmill-mcp/src/server/endpoints.rs create mode 100644 backend/windmill-mcp/src/server/mod.rs create mode 100644 backend/windmill-mcp/src/server/runner.rs create mode 100644 backend/windmill-mcp/src/server/tools.rs diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 7455c4e5a6..af26999421 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -15624,11 +15624,17 @@ name = "windmill-mcp" version = "1.608.0" dependencies = [ "anyhow", + "async-trait", + "futures", + "http 1.4.0", "oauth2", "reqwest 0.12.28", "rmcp", "serde", "serde_json", + "sqlx", + "tokio", + "tokio-util", "tracing", "windmill-common", ] diff --git a/backend/generate_mcp_endpoints_tools/generate_mcp_tools.py b/backend/generate_mcp_endpoints_tools/generate_mcp_tools.py index ae1e27a288..154e3e4376 100644 --- a/backend/generate_mcp_endpoints_tools/generate_mcp_tools.py +++ b/backend/generate_mcp_endpoints_tools/generate_mcp_tools.py @@ -13,21 +13,7 @@ from typing import Dict, List, Any, Optional IMPORTS = """ use std::borrow::Cow; -use serde::{Deserialize, Serialize}; -""" - -ENDPOINT_STRUCT = """ -#[derive(Debug, Clone, Deserialize, Serialize)] -pub struct EndpointTool { - pub name: Cow<'static, str>, - pub description: Cow<'static, str>, - pub instructions: Cow<'static, str>, - pub path: Cow<'static, str>, - pub method: Cow<'static, str>, - pub path_params_schema: Option, - pub query_params_schema: Option, - pub body_schema: Option, -} +use windmill_mcp::server::EndpointTool; """ def load_openapi_spec(file_path: str) -> Dict[str, Any]: @@ -341,13 +327,11 @@ export const mcpEndpointTools: EndpointTool[] = [ def generate_rust_code(tools: List[Dict[str, Any]], spec: Dict[str, Any], base_path: str = "") -> str: """Generate the complete Rust code with MCP tools.""" if not tools: - return """// No MCP tools found in the OpenAPI specification - + return f"""// No MCP tools found in the OpenAPI specification {IMPORTS} -{ENDPOINT_STRUCT} -pub fn all_tools() -> Vec { +pub fn all_tools() -> Vec {{ vec![] -} +}} """ tool_definitions = [] @@ -386,9 +370,7 @@ pub fn all_tools() -> Vec { rust_code = f"""// Auto-generated MCP tools from OpenAPI specification // This file is generated by generate_mcp_tools.py - DO NOT EDIT MANUALLY - {IMPORTS} -{ENDPOINT_STRUCT} pub fn all_tools() -> Vec {{ vec![ {tool_definitions_str} @@ -405,7 +387,7 @@ def main(): project_dir = backend_dir.parent openapi_file = backend_dir / "windmill-api" / "openapi.yaml" - rust_output_file = backend_dir / "windmill-api" / "src" / "mcp" / "tools" / "auto_generated_endpoints.rs" + rust_output_file = backend_dir / "windmill-api" / "src" / "mcp" / "auto_generated_endpoints.rs" ts_output_file = project_dir / "frontend" / "src" / "lib" / "mcpEndpointTools.ts" if not openapi_file.exists(): diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index a46f37a811..df00619425 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -355,7 +355,7 @@ pub async fn run_server( { let smtp_server = Arc::new(SmtpServer { db: db.clone(), - user_db: user_db, + user_db: user_db.clone(), auth_cache: auth_cache.clone(), base_internal_url: _base_internal_url.clone(), }); @@ -409,7 +409,8 @@ pub async fn run_server( let (mcp_router, mcp_cancellation_token) = { #[cfg(feature = "mcp")] if server_mode || mcp_mode { - let (mcp_router, mcp_cancellation_token) = setup_mcp_server().await?; + let (mcp_router, mcp_cancellation_token) = + setup_mcp_server(db.clone(), user_db).await?; let mcp_middleware = axum::middleware::from_fn(extract_and_store_workspace_id); ( mcp_router.layer(mcp_middleware), diff --git a/backend/windmill-api/src/mcp/tools/auto_generated_endpoints.rs b/backend/windmill-api/src/mcp/auto_generated_endpoints.rs similarity index 99% rename from backend/windmill-api/src/mcp/tools/auto_generated_endpoints.rs rename to backend/windmill-api/src/mcp/auto_generated_endpoints.rs index d5425dcd7f..91a07e3f7a 100644 --- a/backend/windmill-api/src/mcp/tools/auto_generated_endpoints.rs +++ b/backend/windmill-api/src/mcp/auto_generated_endpoints.rs @@ -1,22 +1,8 @@ // Auto-generated MCP tools from OpenAPI specification // This file is generated by generate_mcp_tools.py - DO NOT EDIT MANUALLY - use std::borrow::Cow; -use serde::{Deserialize, Serialize}; - - -#[derive(Debug, Clone, Deserialize, Serialize)] -pub struct EndpointTool { - pub name: Cow<'static, str>, - pub description: Cow<'static, str>, - pub instructions: Cow<'static, str>, - pub path: Cow<'static, str>, - pub method: Cow<'static, str>, - pub path_params_schema: Option, - pub query_params_schema: Option, - pub body_schema: Option, -} +use windmill_mcp::server::EndpointTool; pub fn all_tools() -> Vec { vec![ diff --git a/backend/windmill-api/src/mcp/core.rs b/backend/windmill-api/src/mcp/core.rs new file mode 100644 index 0000000000..6b5917398e --- /dev/null +++ b/backend/windmill-api/src/mcp/core.rs @@ -0,0 +1,448 @@ +//! Windmill MCP Backend implementation +//! +//! This module provides the concrete implementation of the McpBackend trait +//! for the Windmill platform. + +use async_trait::async_trait; +use serde_json::Value; +use std::collections::HashMap; +use windmill_common::{db::UserDB, utils::StripPath, DB}; +use windmill_mcp::common::transform::apply_key_transformation; +use windmill_mcp::common::types::{ + FlowInfo, HubScriptInfo, ResourceInfo, ResourceType, SchemaType, ScriptInfo, +}; +use windmill_mcp::server::{BackendResult, EndpointTool, ErrorData, McpAuth, McpBackend}; + +use crate::db::ApiAuthed; +use crate::jobs::{ + run_wait_result_flow_by_path_internal, run_wait_result_script_by_path_internal, RunJobQuery, +}; + +use super::auto_generated_endpoints::all_tools; +use super::utils::{ + build_query_string, build_request_body, create_http_request, get_hub_script_schema, + get_item_schema, get_items, get_resources, get_resources_types, get_scripts_from_hub, + parse_response_body, prepare_push_args, substitute_path_params, +}; + +use std::sync::Arc; +use std::time::Duration; + +use tokio_util::sync::CancellationToken; +use windmill_mcp::server::{ + LocalSessionManager, Runner, StreamableHttpServerConfig, StreamableHttpService, +}; +use windmill_mcp::WorkspaceId; + +use axum::{ + extract::Path, http::Request, middleware::Next, response::Response, routing::get, Json, Router, +}; +use windmill_common::error::JsonResult; + +/// Implement McpAuth for ApiAuthed +impl McpAuth for ApiAuthed { + fn username(&self) -> &str { + &self.username + } + + fn email(&self) -> &str { + &self.email + } + + fn is_admin(&self) -> bool { + self.is_admin + } + + fn is_operator(&self) -> bool { + self.is_operator + } + + fn groups(&self) -> &[String] { + &self.groups + } + + fn folders(&self) -> &[(String, bool, bool)] { + &self.folders + } + + fn scopes(&self) -> Option<&[String]> { + self.scopes.as_deref() + } +} + +/// Windmill's MCP backend implementation +#[derive(Clone)] +pub struct WindmillBackend { + pub db: DB, + pub user_db: UserDB, +} + +impl WindmillBackend { + pub fn new(db: DB, user_db: UserDB) -> Self { + Self { db, user_db } + } +} + +#[async_trait] +impl McpBackend for WindmillBackend { + type Auth = ApiAuthed; + + async fn list_scripts( + &self, + auth: &ApiAuthed, + workspace_id: &str, + favorites_only: bool, + ) -> BackendResult> { + let scope_type = if favorites_only { "favorites" } else { "all" }; + get_items::(&self.user_db, auth, workspace_id, scope_type, "script") + .await + .map_err(|e| ErrorData::internal_error(e.message, None)) + } + + async fn list_flows( + &self, + auth: &ApiAuthed, + workspace_id: &str, + favorites_only: bool, + ) -> BackendResult> { + let scope_type = if favorites_only { "favorites" } else { "all" }; + get_items::(&self.user_db, auth, workspace_id, scope_type, "flow") + .await + .map_err(|e| ErrorData::internal_error(e.message, None)) + } + + async fn list_resource_types( + &self, + auth: &ApiAuthed, + workspace_id: &str, + ) -> BackendResult> { + get_resources_types(&self.user_db, auth, workspace_id) + .await + .map_err(|e| ErrorData::internal_error(e.message, None)) + } + + async fn list_resources( + &self, + auth: &ApiAuthed, + workspace_id: &str, + resource_type: &str, + ) -> BackendResult> { + get_resources(&self.user_db, auth, workspace_id, resource_type) + .await + .map_err(|e| ErrorData::internal_error(e.message, None)) + } + + async fn list_hub_scripts( + &self, + app_filter: Option<&str>, + ) -> BackendResult> { + get_scripts_from_hub(&self.db, app_filter) + .await + .map_err(|e| ErrorData::internal_error(e.message, None)) + } + + async fn get_item_schema( + &self, + auth: &ApiAuthed, + workspace_id: &str, + path: &str, + item_type: &str, + ) -> BackendResult> { + let schema = get_item_schema(path, &self.user_db, auth, workspace_id, item_type) + .await + .map_err(|e| ErrorData::internal_error(e.message, None))?; + + if let Some(ref s) = schema { + match serde_json::from_str::(s.0.get()) { + Ok(val) => Ok(Some(val)), + Err(e) => { + tracing::warn!("Failed to parse schema: {}", e); + Ok(None) + } + } + } else { + Ok(None) + } + } + + async fn get_hub_script_schema(&self, path: &str) -> BackendResult> { + let schema = get_hub_script_schema(path, &self.db) + .await + .map_err(|e| ErrorData::internal_error(e.message, None))?; + + if let Some(ref s) = schema { + match serde_json::from_str::(s.0.get()) { + Ok(val) => Ok(Some(val)), + Err(e) => { + tracing::warn!("Failed to parse hub schema: {}", e); + Ok(None) + } + } + } else { + Ok(None) + } + } + + fn transform_schema_for_resources( + &self, + schema: &SchemaType, + resources_cache: &HashMap>, + resources_types: &[ResourceType], + ) -> SchemaType { + let mut schema_obj = schema.clone(); + + // Replace invalid char in property key with underscore + let replacements: Vec<(String, String, Value)> = schema_obj + .properties + .iter() + .filter_map(|(key, value)| { + if key.chars().any(|c| !c.is_alphanumeric() && c != '_') { + let new_key = apply_key_transformation(key); + Some((key.clone(), new_key, value.clone())) + } else { + None + } + }) + .collect(); + + for (old_key, new_key, value) in replacements { + schema_obj.properties.remove(&old_key); + schema_obj.properties.insert(new_key, value); + } + + for (_key, prop_value) in schema_obj.properties.iter_mut() { + if let Value::Object(prop_map) = prop_value { + if let Some(format_value) = prop_map.get("format") { + if let Value::String(format_str) = format_value { + if format_str.starts_with("resource-") { + let resource_type_key = + format_str.split("-").last().unwrap_or_default().to_string(); + let resource_type = resources_types + .iter() + .find(|rt| rt.name == resource_type_key); + let resource_type_obj = resource_type.cloned(); + + if let Some(resource_cache) = resources_cache.get(&resource_type_key) { + let resources_count = resource_cache.len(); + let description = match resource_type_obj { + Some(resource_type_obj) => format!( + "This is a resource named `{}` with the following description: `{}`.\\nThe path of the resource should be used to specify the resource.\\n{}", + resource_type_obj.name, + resource_type_obj.description.as_deref().unwrap_or("No description"), + if resources_count == 0 { + "This resource does not have any available instances, you should create one from your windmill workspace." + } else if resources_count > 1 { + "This resource has multiple available instances, you should precisely select the one you want to use." + } else { + "There is 1 resource available." + } + ), + None => "An object parameter.".to_string(), + }; + prop_map.insert( + "type".to_string(), + Value::String("string".to_string()), + ); + prop_map + .insert("description".to_string(), Value::String(description)); + if resources_count > 0 { + let resources_description = resource_cache + .iter() + .map(|resource| { + format!( + "{}: $res:{}", + resource + .description + .as_deref() + .unwrap_or("No title"), + resource.path + ) + }) + .collect::>() + .join("\\n"); + + prop_map.insert( + "description".to_string(), + Value::String(format!( + "{}\\nHere are the available resources, in the format title:path. Title can be empty. Path should be used to specify the resource:\\n{}", + prop_map.get("description").unwrap_or(&Value::String("No description".to_string())), + resources_description + )), + ); + } + } + } + } + } + } + } + + schema_obj + } + + async fn run_script( + &self, + auth: &ApiAuthed, + workspace_id: &str, + path: &str, + args: Value, + ) -> BackendResult { + let push_args = prepare_push_args(args); + + let result = run_wait_result_script_by_path_internal( + self.db.clone(), + RunJobQuery::default(), + StripPath(path.to_string()), + auth.clone(), + self.user_db.clone(), + workspace_id.to_string(), + push_args, + ) + .await + .map_err(|e| ErrorData::internal_error(e.to_string(), None))?; + + parse_response_body(result).await + } + + async fn run_flow( + &self, + auth: &ApiAuthed, + workspace_id: &str, + path: &str, + args: Value, + ) -> BackendResult { + let push_args = prepare_push_args(args); + + let result = run_wait_result_flow_by_path_internal( + self.db.clone(), + RunJobQuery::default(), + StripPath(path.to_string()), + auth.clone(), + self.user_db.clone(), + push_args, + workspace_id.to_string(), + ) + .await + .map_err(|e| ErrorData::internal_error(e.to_string(), None))?; + + parse_response_body(result).await + } + + async fn call_endpoint( + &self, + auth: &ApiAuthed, + workspace_id: &str, + endpoint_tool: &EndpointTool, + args: Value, + ) -> BackendResult { + let args_map = match &args { + Value::Object(map) => map, + _ => { + return Err(ErrorData::invalid_params( + "Arguments must be an object", + None, + )); + } + }; + + // Build URL with path substitutions + let path_template = substitute_path_params( + &endpoint_tool.path, + workspace_id, + args_map, + &endpoint_tool.path_params_schema, + )?; + let query_string = build_query_string(args_map, &endpoint_tool.query_params_schema); + let full_url = format!( + "{}/api{}{}", + windmill_common::BASE_INTERNAL_URL.as_str(), + path_template, + query_string + ); + + // Prepare request body + let body_json = + build_request_body(&endpoint_tool.method, args_map, &endpoint_tool.body_schema); + + // Create and execute request + let response = create_http_request( + &endpoint_tool.method, + &full_url, + workspace_id, + auth, + body_json, + ) + .await?; + + let status = response.status(); + let response_text = response.text().await.map_err(|e| { + ErrorData::internal_error(format!("Failed to read response text: {}", e), None) + })?; + + if status.is_success() { + Ok(serde_json::from_str(&response_text) + .unwrap_or_else(|_| Value::String(response_text))) + } else { + Err(ErrorData::internal_error( + format!( + "HTTP {} {}: {}", + status.as_u16(), + status.canonical_reason().unwrap_or(""), + response_text + ), + None, + )) + } + } + + fn all_endpoint_tools(&self) -> Vec { + all_tools() + } +} + +/// Extract workspace ID from path and store it in request extensions +pub async fn extract_and_store_workspace_id( + Path(params): Path, + mut request: Request, + next: Next, +) -> Response { + let workspace_id = params; + request.extensions_mut().insert(WorkspaceId(workspace_id)); + next.run(request).await +} + +/// Setup the MCP server with HTTP transport +pub async fn setup_mcp_server( + db: DB, + user_db: UserDB, +) -> anyhow::Result<(Router, CancellationToken)> { + let cancellation_token = CancellationToken::new(); + let session_manager = Arc::new(LocalSessionManager::default()); + + let backend = WindmillBackend::new(db, user_db); + let runner = Runner::new(backend); + + let service_config = StreamableHttpServerConfig { + sse_keep_alive: Some(Duration::from_secs(15)), + stateful_mode: false, + cancellation_token: cancellation_token.clone(), + sse_retry: Some(Duration::from_secs(15)), + }; + + let service = + StreamableHttpService::new(move || Ok(runner.clone()), session_manager, service_config); + + let router = Router::new().nest_service("/", service); + Ok((router, cancellation_token)) +} + +/// HTTP handler to list MCP tools as JSON +async fn list_mcp_tools_handler() -> JsonResult> { + let endpoint_tools = all_tools(); + Ok(Json(endpoint_tools)) +} + +/// Creates a router service for listing MCP tools +pub fn list_tools_service() -> Router { + Router::new().route("/", get(list_mcp_tools_handler)) +} diff --git a/backend/windmill-api/src/mcp/mod.rs b/backend/windmill-api/src/mcp/mod.rs index e1cd855547..b7853fd7de 100644 --- a/backend/windmill-api/src/mcp/mod.rs +++ b/backend/windmill-api/src/mcp/mod.rs @@ -3,9 +3,9 @@ //! This module provides the MCP server implementation that exposes Windmill scripts, //! flows, and API endpoints as MCP tools for AI assistants to interact with. -pub mod server; -pub mod tools; -pub mod utils; +mod auto_generated_endpoints; +mod core; +mod utils; -// Re-export main components -pub use server::{extract_and_store_workspace_id, list_tools_service, setup_mcp_server}; +// Re-export only what's needed externally +pub use core::{extract_and_store_workspace_id, list_tools_service, setup_mcp_server}; diff --git a/backend/windmill-api/src/mcp/server.rs b/backend/windmill-api/src/mcp/server.rs deleted file mode 100644 index ba38de4b1c..0000000000 --- a/backend/windmill-api/src/mcp/server.rs +++ /dev/null @@ -1,561 +0,0 @@ -//! MCP Server implementation -//! -//! Contains the core MCP server handler that implements the Model Context Protocol -//! specification. This is a thin orchestration layer that delegates to the appropriate -//! modules for tool management, database operations, and schema transformation. - -use std::collections::HashMap; -use std::sync::Arc; -use std::{borrow::Cow, time::Duration}; - -use axum::body::to_bytes; -use serde_json::Value; -use tokio::try_join; -use tokio_util::sync::CancellationToken; -use windmill_common::db::UserDB; -use windmill_common::worker::to_raw_value; -use windmill_common::{utils::StripPath, DB}; -use windmill_mcp::server::{ - Annotated, CallToolRequestParam, CallToolResult, Content, ErrorData, Implementation, - InitializeRequestParam, InitializeResult, ListPromptsResult, ListResourceTemplatesResult, - ListResourcesResult, ListToolsResult, LocalSessionManager, PaginatedRequestParam, - ProtocolVersion, RawContent, RawTextContent, RequestContext, RoleServer, ServerCapabilities, - ServerHandler, ServerInfo, StreamableHttpServerConfig, StreamableHttpService, Tool, - ToolAnnotations, -}; - -use crate::db::ApiAuthed; -use crate::jobs::{ - run_wait_result_flow_by_path_internal, run_wait_result_script_by_path_internal, RunJobQuery, -}; - -use super::tools::endpoint_tools::{ - all_endpoint_tools, call_endpoint_tool, endpoint_tools_to_mcp_tools, EndpointTool, -}; -use super::utils::{ - database::{ - check_scopes, get_hub_script_schema, get_item_schema, get_items, get_resources_types, - get_scripts_from_hub, - }, - models::{ - FlowInfo, ResourceInfo, ResourceType, SchemaType, ScriptInfo, ToolableItem, WorkspaceId, - }, - schema::transform_schema_for_resources, - scope_matcher::{is_resource_allowed, parse_mcp_scopes}, - transform::{reverse_transform, reverse_transform_key}, -}; - -use axum::{ - extract::Path, http::Request, middleware::Next, response::Response, routing::get, Json, Router, -}; -use windmill_common::error::JsonResult; - -/// MCP Server Runner - implements the core MCP protocol handlers -#[derive(Clone)] -pub struct Runner {} - -impl Runner { - pub fn new() -> Self { - Self {} - } - - /// Creates a Tool from a ToolableItem - async fn create_tool_from_item( - item: &T, - user_db: &UserDB, - authed: &ApiAuthed, - workspace_id: &str, - resources_cache: &mut HashMap>, - resources_types: &Vec, - ) -> Result { - let is_hub = item.is_hub(); - let path = item.get_path_or_id(); - let item_type = item.item_type(); - let description = format!( - "This is a {} named `{}` with the following description: `{}`.{}", - item_type, - item.get_summary(), - item.get_description(), - if is_hub { - format!( - " It is a tool used for the following app: {}", - item.get_integration_type() - .unwrap_or("No integration type".to_string()) - ) - } else { - "".to_string() - } - ); - let schema_obj = transform_schema_for_resources( - &item.get_schema(), - user_db, - authed, - &workspace_id, - resources_cache, - &resources_types, - ) - .await?; - let input_schema_map = match serde_json::to_value(schema_obj) { - Ok(Value::Object(map)) => map, - Ok(_) => { - tracing::warn!("Schema object for tool '{}' did not serialize to a JSON object, using empty schema.", path); - serde_json::Map::new() - } - Err(e) => { - tracing::error!( - "Failed to serialize schema object for tool '{}': {}. Using empty schema.", - path, - e - ); - serde_json::Map::new() - } - }; - - Ok(Tool { - name: Cow::Owned(path), - description: Some(Cow::Owned(description)), - input_schema: Arc::new(input_schema_map), - title: Some(item.get_summary().to_string()), - output_schema: None, - icons: None, - annotations: Some(ToolAnnotations { - title: Some(item.get_summary().to_string()), - read_only_hint: Some(false), // Can modify environment - destructive_hint: Some(true), // Can potentially be destructive - idempotent_hint: Some(false), // Are not guaranteed to be idempotent - open_world_hint: Some(true), // Can interact with external services - }), - meta: None, - }) - } -} - -impl ServerHandler for Runner { - /// Handles the `CallTool` request from the MCP client - async fn call_tool( - &self, - request: CallToolRequestParam, - context: RequestContext, - ) -> Result { - let http_parts = context - .extensions - .get::() - .ok_or_else(|| { - tracing::error!("http::request::Parts not found"); - ErrorData::internal_error("http::request::Parts not found", None) - })?; - - let authed = http_parts.extensions.get::().ok_or_else(|| { - tracing::error!("ApiAuthed Axum extension not found"); - ErrorData::internal_error("ApiAuthed Axum extension not found", None) - })?; - - check_scopes(authed)?; - - // Parse MCP scopes for authorization - let scopes = authed.scopes.as_ref().map(|s| s.as_slice()).unwrap_or(&[]); - let scope_config = parse_mcp_scopes(scopes)?; - - if request.name.ends_with("_TRUNC") { - return Ok(CallToolResult::error( - vec![ - Annotated::new( - RawContent::Text(RawTextContent { - text: - "Tool path is too long. Consider shortening it to make it compatible with MCP." - .to_string(), - meta: None, - }), - None - ), - ] - )); - } - - let db = http_parts.extensions.get::().ok_or_else(|| { - tracing::error!("DB Axum extension not found"); - ErrorData::internal_error("DB Axum extension not found", None) - })?; - - let user_db = http_parts.extensions.get::().ok_or_else(|| { - tracing::error!("UserDB Axum extension not found"); - ErrorData::internal_error("UserDB Axum extension not found", None) - })?; - - let args = request.arguments.map(Value::Object).ok_or_else(|| { - ErrorData::invalid_params( - "Missing arguments for tool", - Some(request.name.clone().into()), - ) - })?; - - let workspace_id = http_parts - .extensions - .get::() - .ok_or_else(|| { - tracing::error!("WorkspaceId not found"); - ErrorData::internal_error("WorkspaceId not found", None) - }) - .map(|w_id| w_id.0.clone())?; - - // Check if this is a generated endpoint tool - let endpoint_tools = all_endpoint_tools(); - for endpoint_tool in endpoint_tools { - if endpoint_tool.name.as_ref() == request.name { - // Validate endpoint scope - if scope_config.granular - && !is_resource_allowed(&endpoint_tool.name, &scope_config.endpoints) - { - return Err(ErrorData::internal_error( - format!( - "Access denied: endpoint '{}' not in token scope", - endpoint_tool.name - ), - None, - )); - } - - // This is an endpoint tool, forward to the actual HTTP endpoint - let result = - call_endpoint_tool(&endpoint_tool, args.clone(), &workspace_id, &authed) - .await?; - return Ok(CallToolResult::success(vec![Content::text( - serde_json::to_string_pretty(&result).unwrap_or_else(|_| "{}".to_string()), - )])); - } - } - - // Continue with script/flow logic - let (tool_type, path, is_hub) = reverse_transform(&request.name).map_err(|e| { - ErrorData::internal_error(format!("Failed to reverse transform path: {}", e), None) - })?; - - // Validate script/flow scope - if !is_hub && scope_config.granular { - if tool_type == "script" && !is_resource_allowed(&path, &scope_config.scripts) { - return Err(ErrorData::internal_error( - format!("Access denied: script '{}' not in token scope", path), - None, - )); - } else if tool_type == "flow" && !is_resource_allowed(&path, &scope_config.flows) { - return Err(ErrorData::internal_error( - format!("Access denied: flow '{}' not in token scope", path), - None, - )); - } - } - - let item_schema = if is_hub { - get_hub_script_schema(&format!("hub/{}", path), db).await? - } else { - get_item_schema(&path, user_db, authed, &workspace_id, &tool_type).await? - }; - - let schema_obj = if let Some(ref s) = item_schema { - match serde_json::from_str::(s.0.get()) { - Ok(val) => Some(val), - Err(e) => { - tracing::warn!("Failed to parse schema: {}", e); - None - } - } - } else { - None - }; - - let push_args = if let Value::Object(map) = args.clone() { - let mut args_hash = HashMap::new(); - for (k, v) in map { - // need to transform back the key without invalid characters to the original key - let original_key = reverse_transform_key(&k, &schema_obj); - args_hash.insert(original_key, to_raw_value(&v)); - } - windmill_queue::PushArgsOwned { extra: None, args: args_hash } - } else { - windmill_queue::PushArgsOwned::default() - }; - let script_or_flow_path = if is_hub { - StripPath(format!("hub/{}", path)) - } else { - StripPath(path) - }; - let run_query = RunJobQuery::default(); - - let result = if tool_type == "script" { - run_wait_result_script_by_path_internal( - db.clone(), - run_query, - script_or_flow_path, - authed.clone(), - user_db.clone(), - workspace_id.clone(), - push_args, - ) - .await - } else { - run_wait_result_flow_by_path_internal( - db.clone(), - run_query, - script_or_flow_path, - authed.clone(), - user_db.clone(), - push_args, - workspace_id.clone(), - ) - .await - }; - - match result { - Ok(response) => { - let body_bytes = to_bytes(response.into_body(), usize::MAX) - .await - .map_err(|e| { - ErrorData::internal_error( - format!("Failed to read response body: {}", e), - None, - ) - })?; - let body_str = String::from_utf8(body_bytes.to_vec()).map_err(|e| { - ErrorData::internal_error( - format!("Failed to decode response body: {}", e), - None, - ) - })?; - Ok(CallToolResult::success(vec![Content::text(body_str)])) - } - Err(e) => Err(ErrorData::internal_error( - format!("Failed to run script: {}", e), - None, - )), - } - } - - /// Fetches available tools (scripts, flows, hub scripts) based on the user's scope - async fn list_tools( - &self, - _request: Option, - mut _context: RequestContext, - ) -> Result { - let http_parts = _context - .extensions - .get::() - .ok_or_else(|| { - tracing::error!("http::request::Parts not found"); - ErrorData::internal_error("http::request::Parts not found", None) - })?; - - let authed = http_parts.extensions.get::().ok_or_else(|| { - tracing::error!("ApiAuthed Axum extension not found"); - ErrorData::internal_error("ApiAuthed Axum extension not found", None) - })?; - - check_scopes(authed)?; - - let db = http_parts.extensions.get::().ok_or_else(|| { - tracing::error!("DB Axum extension not found"); - ErrorData::internal_error("DB Axum extension not found", None) - })?; - - let user_db = http_parts.extensions.get::().ok_or_else(|| { - tracing::error!("UserDB Axum extension not found"); - ErrorData::internal_error("UserDB Axum extension not found", None) - })?; - - let workspace_id = http_parts - .extensions - .get::() - .ok_or_else(|| { - tracing::error!("WorkspaceId not found"); - ErrorData::internal_error("WorkspaceId not found", None) - }) - .map(|w_id| w_id.0.clone())?; - - // Parse MCP scopes to determine what to expose - let scopes = authed.scopes.as_ref().map(|s| s.as_slice()).unwrap_or(&[]); - let scope_config = parse_mcp_scopes(scopes)?; - - let scope_type = if scope_config.favorites { - "favorites" - } else { - // Fetch all items if either all or granular scope set (we filter later for granular scopes) - "all" - }; - - let scripts_fn = - get_items::(user_db, authed, &workspace_id, scope_type, "script"); - let flows_fn = get_items::(user_db, authed, &workspace_id, scope_type, "flow"); - let resources_types_fn = get_resources_types(user_db, authed, &workspace_id); - let hub_scripts_fn = get_scripts_from_hub(db, scope_config.hub_apps.as_deref()); - let (scripts, flows, resources_types, hub_scripts) = if scope_config.hub_apps.is_some() { - let (scripts, flows, resources_types, hub_scripts) = - try_join!(scripts_fn, flows_fn, resources_types_fn, hub_scripts_fn)?; - (scripts, flows, resources_types, hub_scripts) - } else { - let (scripts, flows, resources_types) = - try_join!(scripts_fn, flows_fn, resources_types_fn)?; - (scripts, flows, resources_types, vec![]) - }; - - let mut resources_cache: HashMap> = HashMap::new(); - let mut tools: Vec = Vec::new(); - - // Filter and add scripts based on scope - for script in scripts { - // For granular scopes, filter by path - if scope_config.granular && !is_resource_allowed(&script.path, &scope_config.scripts) { - continue; - } - - tools.push( - Runner::create_tool_from_item( - &script, - user_db, - authed, - &workspace_id, - &mut resources_cache, - &resources_types, - ) - .await?, - ); - } - - // Filter and add flows based on scope - for flow in flows { - // For granular scopes, filter by path - if scope_config.granular && !is_resource_allowed(&flow.path, &scope_config.flows) { - continue; - } - - tools.push( - Runner::create_tool_from_item( - &flow, - user_db, - authed, - &workspace_id, - &mut resources_cache, - &resources_types, - ) - .await?, - ); - } - - for hub_script in hub_scripts { - tools.push( - Runner::create_tool_from_item( - &hub_script, - user_db, - authed, - &workspace_id, - &mut resources_cache, - &resources_types, - ) - .await?, - ); - } - - // Add endpoint tools from the generated MCP tools, filtered by scope - let endpoint_tools = all_endpoint_tools(); - for endpoint_tool in endpoint_tools { - // For granular scopes, filter by endpoint name - if scope_config.granular - && !is_resource_allowed(&endpoint_tool.name, &scope_config.endpoints) - { - continue; - } - - tools.push( - endpoint_tools_to_mcp_tools(vec![endpoint_tool]) - .into_iter() - .next() - .unwrap(), - ); - } - - Ok(ListToolsResult { tools, next_cursor: None, meta: None }) - } - - fn get_info(&self) -> ServerInfo { - ServerInfo { - protocol_version: ProtocolVersion::default(), - capabilities: ServerCapabilities::builder() - .enable_tools() - .build(), - server_info: Implementation::from_build_env(), - instructions: Some("This server provides a list of scripts and flows the user can run on Windmill. Each flow and script is a tool callable with their respective arguments.".to_string()), - } - } - - async fn initialize( - &self, - _request: InitializeRequestParam, - _context: RequestContext, - ) -> Result { - Ok(self.get_info()) - } - - async fn list_resources( - &self, - _request: Option, - _context: RequestContext, - ) -> Result { - Ok(ListResourcesResult { resources: vec![], next_cursor: None, meta: None }) - } - - async fn list_prompts( - &self, - _request: Option, - _context: RequestContext, - ) -> Result { - Ok(ListPromptsResult::default()) - } - - async fn list_resource_templates( - &self, - _request: Option, - _context: RequestContext, - ) -> Result { - Ok(ListResourceTemplatesResult::default()) - } -} - -/// Extract workspace ID from path and store it in request extensions -pub async fn extract_and_store_workspace_id( - Path(params): Path, - mut request: Request, - next: Next, -) -> Response { - let workspace_id = params; - request.extensions_mut().insert(WorkspaceId(workspace_id)); - next.run(request).await -} - -/// Setup the MCP server with HTTP transport -pub async fn setup_mcp_server() -> anyhow::Result<(Router, CancellationToken)> { - let cancellation_token = CancellationToken::new(); - let session_manager = Arc::new(LocalSessionManager::default()); - let service_config = StreamableHttpServerConfig { - sse_keep_alive: Some(Duration::from_secs(15)), - stateful_mode: false, - cancellation_token: cancellation_token.clone(), - sse_retry: Some(Duration::from_secs(15)), - }; - let service = StreamableHttpService::new( - || Ok(Runner::new()), - session_manager.clone(), - service_config, - ); - - let router = axum::Router::new().nest_service("/", service); - Ok((router, cancellation_token)) -} - -/// HTTP handler to list MCP tools as JSON -async fn list_mcp_tools_handler() -> JsonResult> { - let endpoint_tools = all_endpoint_tools(); - Ok(Json(endpoint_tools)) -} - -/// Creates a router service for listing MCP tools -pub fn list_tools_service() -> Router { - Router::new().route("/", get(list_mcp_tools_handler)) -} diff --git a/backend/windmill-api/src/mcp/tools/endpoint_tools.rs b/backend/windmill-api/src/mcp/tools/endpoint_tools.rs deleted file mode 100644 index 372310b463..0000000000 --- a/backend/windmill-api/src/mcp/tools/endpoint_tools.rs +++ /dev/null @@ -1,307 +0,0 @@ -//! Endpoint tools for MCP server -//! -//! Contains the auto-generated endpoint tools and utilities for converting -//! them to MCP tools and handling HTTP calls to Windmill API endpoints. - -use crate::db::ApiAuthed; -use std::sync::Arc; -use windmill_common::db::Authed; -use windmill_common::{auth::create_jwt_token, BASE_INTERNAL_URL}; -use windmill_mcp::server::{ErrorData, Tool, ToolAnnotations}; - -// Import the auto-generated tools -use super::auto_generated_endpoints; -pub use auto_generated_endpoints::{all_tools, EndpointTool}; - -/// Get all available endpoint tools -pub fn all_endpoint_tools() -> Vec { - all_tools() -} - -/// Convert endpoint tools to MCP tools -pub fn endpoint_tools_to_mcp_tools(endpoint_tools: Vec) -> Vec { - endpoint_tools - .into_iter() - .map(|tool| endpoint_tool_to_mcp_tool(&tool)) - .collect() -} - -/// Convert a single endpoint tool to MCP tool -pub fn endpoint_tool_to_mcp_tool(tool: &EndpointTool) -> Tool { - let mut combined_properties = serde_json::Map::new(); - let mut combined_required = Vec::new(); - - // Combine all parameter schemas - let schemas = [ - &tool.path_params_schema, - &tool.query_params_schema, - &tool.body_schema, - ]; - - for schema in schemas.iter().filter_map(|s| s.as_ref()) { - merge_schema_into(&mut combined_properties, &mut combined_required, schema); - } - - let combined_schema = serde_json::json!({ - "type": "object", - "properties": combined_properties, - "required": combined_required - }); - - let description = format!("{}. {}", tool.description, tool.instructions); - - // Create annotations based on HTTP method and endpoint characteristics - let annotations = create_endpoint_annotations(tool); - - Tool { - name: tool.name.clone(), - description: Some(description.into()), - input_schema: Arc::new(combined_schema.as_object().unwrap().clone()), - title: Some(tool.name.to_string()), - output_schema: None, - icons: None, - annotations: Some(annotations), - meta: None, - } -} - -/// Create appropriate annotations for endpoint tools based on HTTP method -fn create_endpoint_annotations(tool: &EndpointTool) -> ToolAnnotations { - let method = tool.method.as_ref(); - - // Determine characteristics based on HTTP method - let (read_only, destructive, idempotent, open_world) = match method { - "GET" => (true, false, true, true), // Read-only, safe, idempotent - "POST" => (false, true, false, true), // Can modify, potentially destructive, not idempotent - "PUT" => (false, false, true, true), // Can modify, typically idempotent updates - "DELETE" => (false, true, true, true), // Destructive but idempotent - "PATCH" => (false, false, false, true), // Partial updates, not guaranteed idempotent - _ => (false, true, false, true), // Default: assume can modify and be destructive - }; - - ToolAnnotations { - title: Some(format!("{} {}", method, tool.path)), - read_only_hint: Some(read_only), - destructive_hint: Some(destructive), - idempotent_hint: Some(idempotent), - open_world_hint: Some(open_world), - } -} - -/// Merge schema into combined properties and required fields -fn merge_schema_into( - combined_properties: &mut serde_json::Map, - combined_required: &mut Vec, - schema: &serde_json::Value, -) { - if let Some(props) = schema.get("properties").and_then(|p| p.as_object()) { - for (key, value) in props { - combined_properties.insert(key.clone(), value.clone()); - } - } - - if let Some(required) = schema.get("required").and_then(|r| r.as_array()) { - for req in required.iter().filter_map(|r| r.as_str()) { - combined_required.push(req.to_string()); - } - } -} - -/// Call an endpoint tool by making HTTP request to Windmill API -pub async fn call_endpoint_tool( - tool: &EndpointTool, - args: serde_json::Value, - workspace_id: &str, - api_authed: &ApiAuthed, -) -> Result { - let args_map = match &args { - serde_json::Value::Object(map) => map, - _ => { - return Err(ErrorData::invalid_params( - "Arguments must be an object", - Some(tool.name.clone().into()), - )) - } - }; - - // Build URL with path substitutions - let path_template = - substitute_path_params(&tool.path, workspace_id, args_map, &tool.path_params_schema)?; - let query_string = build_query_string(args_map, &tool.query_params_schema); - let full_url = format!( - "{}/api{}{}", - BASE_INTERNAL_URL.as_str(), - path_template, - query_string - ); - - // Prepare request body - let body_json = build_request_body(&tool.method, args_map, &tool.body_schema); - - // Create and execute request - let response = - create_http_request(&tool.method, &full_url, workspace_id, api_authed, body_json).await?; - - let status = response.status(); - let response_text = response.text().await.map_err(|e| { - ErrorData::internal_error(format!("Failed to read response text: {}", e), None) - })?; - - if status.is_success() { - Ok(serde_json::from_str(&response_text) - .unwrap_or_else(|_| serde_json::Value::String(response_text))) - } else { - Err(ErrorData::internal_error( - format!( - "HTTP {} {}: {}", - status.as_u16(), - status.canonical_reason().unwrap_or(""), - response_text - ), - None, - )) - } -} - -/// Substitute path parameters in the URL template -fn substitute_path_params( - path: &str, - workspace_id: &str, - args_map: &serde_json::Map, - path_schema: &Option, -) -> Result { - let mut path_template = path.replace("{workspace}", workspace_id); - - if let Some(schema) = path_schema { - if let Some(props) = schema.get("properties").and_then(|p| p.as_object()) { - for (param_name, _) in props { - let placeholder = format!("{{{}}}", param_name); - match args_map.get(param_name) { - Some(param_value) => { - if let Some(str_val) = param_value.as_str() { - path_template = path_template.replace(&placeholder, str_val); - } - } - None => { - tracing::warn!("Missing required path parameter: {}", param_name); - return Err(ErrorData::invalid_params( - format!("Missing required path parameter: {}", param_name), - None, - )); - } - } - } - } - } - - Ok(path_template) -} - -/// Build query string from arguments -fn build_query_string( - args_map: &serde_json::Map, - query_schema: &Option, -) -> String { - let Some(schema) = query_schema else { - return String::new(); - }; - let Some(props) = schema.get("properties").and_then(|p| p.as_object()) else { - return String::new(); - }; - - let query_params: Vec = props - .keys() - .filter_map(|param_name| { - args_map - .get(param_name) - .filter(|v| !v.is_null()) - .map(|value| { - let value_str = value.to_string(); - let str_val = value_str.trim_matches('"'); - format!( - "{}={}", - urlencoding::encode(param_name), - urlencoding::encode(str_val) - ) - }) - }) - .collect(); - - if query_params.is_empty() { - String::new() - } else { - format!("?{}", query_params.join("&")) - } -} - -/// Build request body from arguments -fn build_request_body( - method: &str, - args_map: &serde_json::Map, - body_schema: &Option, -) -> Option { - if method == "GET" { - return None; - } - - let schema = body_schema.as_ref()?; - let props = schema.get("properties")?.as_object()?; - - let body_map: serde_json::Map = props - .keys() - .filter_map(|param_name| { - args_map - .get(param_name) - .map(|value| (param_name.clone(), value.clone())) - }) - .collect(); - - if body_map.is_empty() { - None - } else { - Some(serde_json::Value::Object(body_map)) - } -} - -/// Create HTTP request with authentication -async fn create_http_request( - method: &str, - url: &str, - workspace_id: &str, - api_authed: &ApiAuthed, - body_json: Option, -) -> Result { - let client = &crate::HTTP_CLIENT; - let mut request_builder = match method { - "GET" => client.get(url), - "POST" => client.post(url), - "PUT" => client.put(url), - "DELETE" => client.delete(url), - "PATCH" => client.patch(url), - _ => { - return Err(ErrorData::invalid_params( - format!("Unsupported HTTP method: {}", method), - None, - )) - } - }; - - // Add authorization header - let authed = Authed::from(api_authed.clone()); - let token = create_jwt_token(authed, workspace_id, 3600, None, None, None, None) - .await - .map_err(|e| ErrorData::internal_error(e.to_string(), None))?; - request_builder = request_builder.header("Authorization", format!("Bearer {}", token)); - - // Add body if present - if let Some(body) = body_json { - request_builder = request_builder - .header("Content-Type", "application/json") - .json(&body); - } - - request_builder - .send() - .await - .map_err(|e| ErrorData::internal_error(format!("Failed to execute request: {}", e), None)) -} diff --git a/backend/windmill-api/src/mcp/tools/flow_tools.rs b/backend/windmill-api/src/mcp/tools/flow_tools.rs deleted file mode 100644 index 03b47472bf..0000000000 --- a/backend/windmill-api/src/mcp/tools/flow_tools.rs +++ /dev/null @@ -1,40 +0,0 @@ -//! Flow tools for MCP server -//! -//! Contains functionality for converting Windmill flows into MCP tools. - -use super::super::utils::{ - models::{FlowInfo, ToolableItem, SchemaType}, - schema::convert_schema_to_schema_type, - transform::transform_path, -}; - -/// Implementation of ToolableItem for FlowInfo -impl ToolableItem for FlowInfo { - fn get_path_or_id(&self) -> String { - transform_path(&self.path, "flow") - } - - fn get_summary(&self) -> &str { - self.summary.as_deref().unwrap_or("No summary") - } - - fn get_description(&self) -> &str { - self.description.as_deref().unwrap_or("No description") - } - - fn get_schema(&self) -> SchemaType { - convert_schema_to_schema_type(self.schema.clone()) - } - - fn is_hub(&self) -> bool { - false - } - - fn item_type(&self) -> &'static str { - "flow" - } - - fn get_integration_type(&self) -> Option { - None - } -} \ No newline at end of file diff --git a/backend/windmill-api/src/mcp/tools/hub_tools.rs b/backend/windmill-api/src/mcp/tools/hub_tools.rs deleted file mode 100644 index b81b973b30..0000000000 --- a/backend/windmill-api/src/mcp/tools/hub_tools.rs +++ /dev/null @@ -1,43 +0,0 @@ -//! Hub tools for MCP server -//! -//! Contains functionality for integrating Windmill Hub scripts as MCP tools. - -use super::super::utils::{ - models::{HubScriptInfo, ToolableItem, SchemaType}, -}; - -/// Implementation of ToolableItem for HubScriptInfo -impl ToolableItem for HubScriptInfo { - fn get_path_or_id(&self) -> String { - let id = self.version_id; - let summary = self.summary.as_deref().unwrap_or("No summary"); - format!("hs-{}-{}", id, summary.replace(" ", "_")) - } - - fn get_summary(&self) -> &str { - self.summary.as_deref().unwrap_or("No summary") - } - - fn get_description(&self) -> &str { - self.description.as_deref().unwrap_or("No description") - } - - fn get_schema(&self) -> SchemaType { - match serde_json::from_value::(self.schema.clone().unwrap_or_default()) { - Ok(schema_type) => schema_type, - Err(_) => SchemaType::default(), - } - } - - fn is_hub(&self) -> bool { - true - } - - fn item_type(&self) -> &'static str { - "script" - } - - fn get_integration_type(&self) -> Option { - self.app.clone() - } -} \ No newline at end of file diff --git a/backend/windmill-api/src/mcp/tools/mod.rs b/backend/windmill-api/src/mcp/tools/mod.rs deleted file mode 100644 index 1488b22d8a..0000000000 --- a/backend/windmill-api/src/mcp/tools/mod.rs +++ /dev/null @@ -1,10 +0,0 @@ -//! Tool management for MCP server -//! -//! This module handles the conversion of Windmill scripts, flows, and endpoints -//! into MCP tools that can be used by AI assistants. - -pub mod script_tools; -pub mod flow_tools; -pub mod hub_tools; -pub mod endpoint_tools; -pub mod auto_generated_endpoints; \ No newline at end of file diff --git a/backend/windmill-api/src/mcp/tools/script_tools.rs b/backend/windmill-api/src/mcp/tools/script_tools.rs deleted file mode 100644 index 716fae3a87..0000000000 --- a/backend/windmill-api/src/mcp/tools/script_tools.rs +++ /dev/null @@ -1,40 +0,0 @@ -//! Script tools for MCP server -//! -//! Contains functionality for converting Windmill scripts into MCP tools. - -use super::super::utils::{ - models::{ScriptInfo, ToolableItem, SchemaType}, - schema::convert_schema_to_schema_type, - transform::transform_path, -}; - -/// Implementation of ToolableItem for ScriptInfo -impl ToolableItem for ScriptInfo { - fn get_path_or_id(&self) -> String { - transform_path(&self.path, "script") - } - - fn get_summary(&self) -> &str { - self.summary.as_deref().unwrap_or("No summary") - } - - fn get_description(&self) -> &str { - self.description.as_deref().unwrap_or("No description") - } - - fn get_schema(&self) -> SchemaType { - convert_schema_to_schema_type(self.schema.clone()) - } - - fn is_hub(&self) -> bool { - false - } - - fn item_type(&self) -> &'static str { - "script" - } - - fn get_integration_type(&self) -> Option { - None - } -} \ No newline at end of file diff --git a/backend/windmill-api/src/mcp/utils.rs b/backend/windmill-api/src/mcp/utils.rs new file mode 100644 index 0000000000..a2513ef5f7 --- /dev/null +++ b/backend/windmill-api/src/mcp/utils.rs @@ -0,0 +1,414 @@ +//! Utility functions for MCP server +//! +//! Contains database query functions and HTTP request helpers +//! used by the MCP server implementation. + +use std::collections::HashMap; + +use axum::body::{to_bytes, Body}; +use axum::response::Response; +use serde_json::Value; +use sql_builder::prelude::*; +use windmill_common::auth::create_jwt_token; +use windmill_common::db::{Authed, UserDB}; +use windmill_common::scripts::{get_full_hub_script_by_path, Schema}; +use windmill_common::utils::{query_elems_from_hub, StripPath}; +use windmill_common::worker::to_raw_value; +use windmill_common::{DB, HUB_BASE_URL}; +use windmill_mcp::server::{BackendResult, ErrorData}; +use windmill_mcp::{HubResponse, HubScriptInfo, ItemSchema, ResourceInfo, ResourceType}; + +use crate::db::ApiAuthed; +use crate::HTTP_CLIENT; + +// items max limit +const ITEMS_FETCH_MAX_LIMIT: usize = 100; + +// ============================================================================ +// Database utilities +// ============================================================================ + +/// Get the schema for a specific item (script or flow) +pub async fn get_item_schema( + path: &str, + user_db: &UserDB, + authed: &ApiAuthed, + workspace_id: &str, + item_type: &str, +) -> Result, ErrorData> { + let mut sqlb = SqlBuilder::select_from(&format!("{} as o", item_type)); + sqlb.fields(&["o.schema"]); + sqlb.and_where("o.path = ?".bind(&path)); + sqlb.and_where("o.workspace_id = ?".bind(&workspace_id)); + sqlb.and_where("o.archived = false"); + sqlb.and_where("o.draft_only IS NOT TRUE"); + let sql = sqlb.sql().map_err(|e| { + tracing::error!("failed to build sql: {}", e); + ErrorData::internal_error(format!("failed to build sql: {}", e), None) + })?; + let mut tx = user_db.clone().begin(authed).await.map_err(|e| { + tracing::error!("failed to begin transaction: {}", e); + ErrorData::internal_error(format!("failed to begin transaction: {}", e), None) + })?; + let item = sqlx::query_as::<_, ItemSchema>(&sql) + .fetch_one(&mut *tx) + .await + .map_err(|e| { + tracing::error!("failed to fetch item schema: {}", e); + ErrorData::internal_error(format!("failed to fetch item schema: {}", e), None) + })?; + tx.commit().await.map_err(|e| { + tracing::error!("failed to commit transaction: {}", e); + ErrorData::internal_error(format!("failed to commit transaction: {}", e), None) + })?; + Ok(item.schema) +} + +/// Get all resource types from the database +pub async fn get_resources_types( + user_db: &UserDB, + authed: &ApiAuthed, + workspace_id: &str, +) -> Result, ErrorData> { + let mut sqlb = SqlBuilder::select_from("resource_type as o"); + sqlb.fields(&["o.name", "o.description"]); + sqlb.and_where("o.workspace_id = ?".bind(&workspace_id)); + let sql = sqlb.sql().map_err(|e| { + tracing::error!("failed to build sql: {}", e); + ErrorData::internal_error(format!("failed to build sql: {}", e), None) + })?; + let mut tx = user_db.clone().begin(authed).await.map_err(|e| { + tracing::error!("failed to begin transaction: {}", e); + ErrorData::internal_error(format!("failed to begin transaction: {}", e), None) + })?; + let rows = sqlx::query_as::<_, ResourceType>(&sql) + .fetch_all(&mut *tx) + .await + .map_err(|e| { + tracing::error!("failed to fetch resource types: {}", e); + ErrorData::internal_error(format!("failed to fetch resource types: {}", e), None) + })?; + tx.commit().await.map_err(|e| { + tracing::error!("failed to commit transaction: {}", e); + ErrorData::internal_error(format!("failed to commit transaction: {}", e), None) + })?; + Ok(rows) +} + +/// Get resources by type from the database +pub async fn get_resources( + user_db: &UserDB, + authed: &ApiAuthed, + workspace_id: &str, + resource_type: &str, +) -> Result, ErrorData> { + let mut sqlb = SqlBuilder::select_from("resource as o"); + sqlb.fields(&["o.path", "o.description", "o.resource_type"]); + sqlb.and_where("o.workspace_id = ?".bind(&workspace_id)); + sqlb.and_where("o.resource_type = ?".bind(&resource_type)); + let sql = sqlb.sql().map_err(|e| { + tracing::error!("failed to build sql: {}", e); + ErrorData::internal_error(format!("failed to build sql: {}", e), None) + })?; + let mut tx = user_db.clone().begin(authed).await.map_err(|e| { + tracing::error!("failed to begin transaction: {}", e); + ErrorData::internal_error(format!("failed to begin transaction: {}", e), None) + })?; + let rows = sqlx::query_as::<_, ResourceInfo>(&sql) + .fetch_all(&mut *tx) + .await + .map_err(|e| { + tracing::error!("failed to fetch resources: {}", e); + ErrorData::internal_error(format!("failed to fetch resources: {}", e), None) + })?; + tx.commit().await.map_err(|e| { + tracing::error!("failed to commit transaction: {}", e); + ErrorData::internal_error(format!("failed to commit transaction: {}", e), None) + })?; + + Ok(rows) +} + +/// Generic function to get items (scripts or flows) from the database +pub async fn get_items sqlx::FromRow<'a, sqlx::postgres::PgRow> + Send + Unpin>( + user_db: &UserDB, + authed: &ApiAuthed, + workspace_id: &str, + scope_type: &str, + item_type: &str, +) -> Result, ErrorData> { + let mut sqlb = SqlBuilder::select_from(&format!("{} as o", item_type)); + let fields = vec!["o.path", "o.summary", "o.description", "o.schema"]; + sqlb.fields(&fields); + if scope_type == "favorites" { + sqlb.join("favorite") + .on("favorite.favorite_kind = ? AND favorite.workspace_id = o.workspace_id AND favorite.path = o.path AND favorite.usr = ?".bind(&item_type) + .bind(&authed.username)); + } + sqlb.and_where("o.workspace_id = ?".bind(&workspace_id)) + .and_where("o.archived = false") + .and_where("o.draft_only IS NOT TRUE"); + + if item_type == "script" { + sqlb.and_where("(o.no_main_func IS NOT TRUE OR o.no_main_func IS NULL)"); + } + + sqlb.order_by( + if item_type == "flow" { + "o.edited_at" + } else { + "o.created_at" + }, + false, + ) + .limit(ITEMS_FETCH_MAX_LIMIT); + let sql = sqlb.sql().map_err(|e| { + tracing::error!("failed to build sql: {}", e); + ErrorData::internal_error(format!("failed to build sql: {}", e), None) + })?; + let mut tx = user_db.clone().begin(authed).await.map_err(|e| { + tracing::error!("failed to begin transaction: {}", e); + ErrorData::internal_error(format!("failed to begin transaction: {}", e), None) + })?; + let rows = sqlx::query_as::<_, T>(&sql) + .fetch_all(&mut *tx) + .await + .map_err(|e| { + tracing::error!("failed to fetch {}: {}", item_type, e); + ErrorData::internal_error(format!("failed to fetch {}: {}", item_type, e), None) + })?; + tx.commit().await.map_err(|e| { + tracing::error!("failed to commit transaction: {}", e); + ErrorData::internal_error(format!("failed to commit transaction: {}", e), None) + })?; + Ok(rows) +} + +/// Get scripts from the Hub +pub async fn get_scripts_from_hub( + db: &DB, + scope_integrations: Option<&str>, +) -> Result, ErrorData> { + let query_params = Some(vec![ + ("limit", ITEMS_FETCH_MAX_LIMIT.to_string()), + ("with_schema", "true".to_string()), + ("apps", scope_integrations.unwrap_or("").to_string()), + ]); + let url = format!("{}/scripts/top", *HUB_BASE_URL.read().await); + let (_status_code, _headers, response) = + query_elems_from_hub(&HTTP_CLIENT, &url, query_params, &db) + .await + .map_err(|e| { + tracing::error!("Failed to get items from hub: {}", e); + ErrorData::internal_error(format!("Failed to get items from hub: {}", e), None) + })?; + + use axum::body::to_bytes; + let body_bytes = to_bytes(response, usize::MAX).await.map_err(|e| { + tracing::error!("Failed to read response body: {}", e); + ErrorData::internal_error(format!("Failed to read response body: {}", e), None) + })?; + let body_str = String::from_utf8(body_bytes.to_vec()).map_err(|e| { + tracing::error!("Failed to decode response body: {}", e); + ErrorData::internal_error(format!("Failed to decode response body: {}", e), None) + })?; + let hub_response: HubResponse = serde_json::from_str(&body_str).map_err(|e| { + tracing::error!("Failed to parse hub response: {}", e); + ErrorData::internal_error(format!("Failed to parse hub response: {}", e), None) + })?; + + Ok(hub_response.asks) +} + +/// Get the schema for a Hub script +pub async fn get_hub_script_schema(path: &str, db: &DB) -> Result, ErrorData> { + let strip_path = StripPath(path.to_string()); + let res = get_full_hub_script_by_path(strip_path, &HTTP_CLIENT, Some(db)) + .await + .map_err(|e| { + tracing::error!("Failed to get hub script: {}", e); + ErrorData::internal_error(format!("Failed to get hub script: {}", e), None) + })?; + match serde_json::from_str::(res.schema.get()) { + Ok(schema) => Ok(Some(schema)), + Err(e) => { + tracing::warn!("Failed to convert schema: {}", e); + Ok(None) + } + } +} + +// ============================================================================ +// HTTP request utilities for endpoint tools +// ============================================================================ + +/// Substitute path parameters in the URL template +pub fn substitute_path_params( + path: &str, + workspace_id: &str, + args_map: &serde_json::Map, + path_schema: &Option, +) -> BackendResult { + let mut path_template = path.replace("{workspace}", workspace_id); + + if let Some(schema) = path_schema { + if let Some(props) = schema.get("properties").and_then(|p| p.as_object()) { + for (param_name, _) in props { + let placeholder = format!("{{{}}}", param_name); + match args_map.get(param_name) { + Some(param_value) => { + if let Some(str_val) = param_value.as_str() { + path_template = path_template.replace(&placeholder, str_val); + } + } + None => { + tracing::warn!("Missing required path parameter: {}", param_name); + return Err(ErrorData::invalid_params( + format!("Missing required path parameter: {}", param_name), + None, + )); + } + } + } + } + } + + Ok(path_template) +} + +/// Build query string from arguments +pub fn build_query_string( + args_map: &serde_json::Map, + query_schema: &Option, +) -> String { + let Some(schema) = query_schema else { + return String::new(); + }; + let Some(props) = schema.get("properties").and_then(|p| p.as_object()) else { + return String::new(); + }; + + let query_params: Vec = props + .keys() + .filter_map(|param_name| { + args_map + .get(param_name) + .filter(|v| !v.is_null()) + .map(|value| { + let value_str = value.to_string(); + let str_val = value_str.trim_matches('"'); + format!( + "{}={}", + urlencoding::encode(param_name), + urlencoding::encode(str_val) + ) + }) + }) + .collect(); + + if query_params.is_empty() { + String::new() + } else { + format!("?{}", query_params.join("&")) + } +} + +/// Build request body from arguments +pub fn build_request_body( + method: &str, + args_map: &serde_json::Map, + body_schema: &Option, +) -> Option { + if method == "GET" { + return None; + } + + let schema = body_schema.as_ref()?; + let props = schema.get("properties")?.as_object()?; + + let body_map: serde_json::Map = props + .keys() + .filter_map(|param_name| { + args_map + .get(param_name) + .map(|value| (param_name.clone(), value.clone())) + }) + .collect(); + + if body_map.is_empty() { + None + } else { + Some(Value::Object(body_map)) + } +} + +/// Create HTTP request with authentication +pub async fn create_http_request( + method: &str, + url: &str, + workspace_id: &str, + api_authed: &ApiAuthed, + body_json: Option, +) -> BackendResult { + let client = &HTTP_CLIENT; + let mut request_builder = match method { + "GET" => client.get(url), + "POST" => client.post(url), + "PUT" => client.put(url), + "DELETE" => client.delete(url), + "PATCH" => client.patch(url), + _ => { + return Err(ErrorData::invalid_params( + format!("Unsupported HTTP method: {}", method), + None, + )); + } + }; + + // Add authorization header + let authed = Authed::from(api_authed.clone()); + let token = create_jwt_token(authed, workspace_id, 3600, None, None, None, None) + .await + .map_err(|e| ErrorData::internal_error(e.to_string(), None))?; + request_builder = request_builder.header("Authorization", format!("Bearer {}", token)); + + // Add body if present + if let Some(body) = body_json { + request_builder = request_builder + .header("Content-Type", "application/json") + .json(&body); + } + + request_builder + .send() + .await + .map_err(|e| ErrorData::internal_error(format!("Failed to execute request: {}", e), None)) +} + +/// Convert a JSON Value into PushArgsOwned for job execution +pub fn prepare_push_args(args: Value) -> windmill_queue::PushArgsOwned { + if let Value::Object(map) = args { + let mut args_hash = HashMap::new(); + for (k, v) in map { + args_hash.insert(k, to_raw_value(&v)); + } + windmill_queue::PushArgsOwned { extra: None, args: args_hash } + } else { + windmill_queue::PushArgsOwned::default() + } +} + +/// Parse an HTTP response body into a JSON Value +pub async fn parse_response_body(response: Response) -> BackendResult { + let body_bytes = to_bytes(response.into_body(), usize::MAX) + .await + .map_err(|e| { + ErrorData::internal_error(format!("Failed to read response body: {}", e), None) + })?; + + let body_str = String::from_utf8(body_bytes.to_vec()).map_err(|e| { + ErrorData::internal_error(format!("Failed to decode response body: {}", e), None) + })?; + + Ok(serde_json::from_str(&body_str).unwrap_or_else(|_| Value::String(body_str))) +} diff --git a/backend/windmill-api/src/mcp/utils/database.rs b/backend/windmill-api/src/mcp/utils/database.rs deleted file mode 100644 index 8453177ac0..0000000000 --- a/backend/windmill-api/src/mcp/utils/database.rs +++ /dev/null @@ -1,243 +0,0 @@ -//! Database operations for MCP server -//! -//! Contains all database query functions and database-related utilities -//! used by the MCP server implementation. - -use windmill_mcp::server::ErrorData; -use sql_builder::prelude::*; -use windmill_common::db::UserDB; -use windmill_common::scripts::{get_full_hub_script_by_path, Schema}; -use windmill_common::utils::{query_elems_from_hub, StripPath}; -use windmill_common::{DB, HUB_BASE_URL}; - -use super::models::*; -use crate::db::ApiAuthed; -use crate::HTTP_CLIENT; - -/// Check if the user has proper MCP scopes -pub fn check_scopes(authed: &ApiAuthed) -> Result<(), ErrorData> { - let scopes = authed.scopes.as_ref(); - if scopes.is_none() - || scopes - .unwrap() - .iter() - .all(|scope| !scope.starts_with("mcp:")) - { - tracing::error!("Unauthorized: missing mcp scope"); - return Err(ErrorData::internal_error( - "Unauthorized: missing mcp scope".to_string(), - None, - )); - } - Ok(()) -} - -/// Get the schema for a specific item (script or flow) -pub async fn get_item_schema( - path: &str, - user_db: &UserDB, - authed: &ApiAuthed, - workspace_id: &str, - item_type: &str, -) -> Result, ErrorData> { - let mut sqlb = SqlBuilder::select_from(&format!("{} as o", item_type)); - sqlb.fields(&["o.schema"]); - sqlb.and_where("o.path = ?".bind(&path)); - sqlb.and_where("o.workspace_id = ?".bind(&workspace_id)); - sqlb.and_where("o.archived = false"); - sqlb.and_where("o.draft_only IS NOT TRUE"); - let sql = sqlb.sql().map_err(|_e| { - tracing::error!("failed to build sql: {}", _e); - ErrorData::internal_error("failed to build sql", None) - })?; - let mut tx = user_db - .clone() - .begin(authed) - .await - .map_err(|_e| ErrorData::internal_error("failed to begin transaction", None))?; - let item = sqlx::query_as::<_, ItemSchema>(&sql) - .fetch_one(&mut *tx) - .await - .map_err(|_e| { - tracing::error!("failed to fetch item schema: {}", _e); - ErrorData::internal_error("failed to fetch item schema", None) - })?; - tx.commit() - .await - .map_err(|_e| ErrorData::internal_error("failed to commit transaction", None))?; - Ok(item.schema) -} - -/// Get all resource types from the database -pub async fn get_resources_types( - user_db: &UserDB, - authed: &ApiAuthed, - workspace_id: &str, -) -> Result, ErrorData> { - let mut sqlb = SqlBuilder::select_from("resource_type as o"); - sqlb.fields(&["o.name", "o.description"]); - sqlb.and_where("o.workspace_id = ?".bind(&workspace_id)); - let sql = sqlb.sql().map_err(|_e| { - tracing::error!("failed to build sql: {}", _e); - ErrorData::internal_error("failed to build sql", None) - })?; - let mut tx = user_db - .clone() - .begin(authed) - .await - .map_err(|_e| ErrorData::internal_error("failed to begin transaction", None))?; - let rows = sqlx::query_as::<_, ResourceType>(&sql) - .fetch_all(&mut *tx) - .await - .map_err(|_e| { - tracing::error!("Failed to fetch resource types: {}", _e); - ErrorData::internal_error("failed to fetch resource types", None) - })?; - tx.commit() - .await - .map_err(|_e| ErrorData::internal_error("failed to commit transaction", None))?; - Ok(rows) -} - -/// Get resources by type from the database -pub async fn get_resources( - user_db: &UserDB, - authed: &ApiAuthed, - workspace_id: &str, - resource_type: &str, -) -> Result, ErrorData> { - let mut sqlb = SqlBuilder::select_from("resource as o"); - sqlb.fields(&["o.path", "o.description", "o.resource_type"]); - sqlb.and_where("o.workspace_id = ?".bind(&workspace_id)); - sqlb.and_where("o.resource_type = ?".bind(&resource_type)); - let sql = sqlb.sql().map_err(|_e| { - tracing::error!("failed to build sql: {}", _e); - ErrorData::internal_error("failed to build sql", None) - })?; - let mut tx = user_db - .clone() - .begin(authed) - .await - .map_err(|_e| ErrorData::internal_error("failed to begin transaction", None))?; - let rows = sqlx::query_as::<_, ResourceInfo>(&sql) - .fetch_all(&mut *tx) - .await - .map_err(|_e| { - tracing::error!("Failed to fetch resources: {}", _e); - ErrorData::internal_error("failed to fetch resources", None) - })?; - tx.commit() - .await - .map_err(|_e| ErrorData::internal_error("failed to commit transaction", None))?; - - Ok(rows) -} - -/// Generic function to get items (scripts or flows) from the database -pub async fn get_items sqlx::FromRow<'a, sqlx::postgres::PgRow> + Send + Unpin>( - user_db: &UserDB, - authed: &ApiAuthed, - workspace_id: &str, - scope_type: &str, - item_type: &str, -) -> Result, ErrorData> { - let mut sqlb = SqlBuilder::select_from(&format!("{} as o", item_type)); - let fields = vec!["o.path", "o.summary", "o.description", "o.schema"]; - sqlb.fields(&fields); - if scope_type == "favorites" { - sqlb.join("favorite") - .on("favorite.favorite_kind = ? AND favorite.workspace_id = o.workspace_id AND favorite.path = o.path AND favorite.usr = ?".bind(&item_type) - .bind(&authed.username)); - } - sqlb.and_where("o.workspace_id = ?".bind(&workspace_id)) - .and_where("o.archived = false") - .and_where("o.draft_only IS NOT TRUE"); - - if item_type == "script" { - sqlb.and_where("(o.no_main_func IS NOT TRUE OR o.no_main_func IS NULL)"); - } - - sqlb.order_by( - if item_type == "flow" { - "o.edited_at" - } else { - "o.created_at" - }, - false, - ) - .limit(100); - let sql = sqlb.sql().map_err(|_e| { - tracing::error!("failed to build sql: {}", _e); - ErrorData::internal_error("failed to build sql", None) - })?; - let mut tx = user_db - .clone() - .begin(authed) - .await - .map_err(|_e| ErrorData::internal_error("failed to begin transaction", None))?; - let rows = sqlx::query_as::<_, T>(&sql) - .fetch_all(&mut *tx) - .await - .map_err(|_e| { - tracing::error!("Failed to fetch {}: {}", item_type, _e); - ErrorData::internal_error(format!("failed to fetch {}", item_type), None) - })?; - tx.commit() - .await - .map_err(|_e| ErrorData::internal_error("failed to commit transaction", None))?; - Ok(rows) -} - -/// Get scripts from the Hub -pub async fn get_scripts_from_hub( - db: &DB, - scope_integrations: Option<&str>, -) -> Result, ErrorData> { - let query_params = Some(vec![ - ("limit", "100".to_string()), - ("with_schema", "true".to_string()), - ("apps", scope_integrations.unwrap_or("").to_string()), - ]); - let url = format!("{}/scripts/top", *HUB_BASE_URL.read().await); - let (_status_code, _headers, response) = - query_elems_from_hub(&HTTP_CLIENT, &url, query_params, &db) - .await - .map_err(|e| { - tracing::error!("Failed to get items from hub: {}", e); - ErrorData::internal_error(format!("Failed to get items from hub: {}", e), None) - })?; - - use axum::body::to_bytes; - let body_bytes = to_bytes(response, usize::MAX).await.map_err(|e| { - tracing::error!("Failed to read response body: {}", e); - ErrorData::internal_error(format!("Failed to read response body: {}", e), None) - })?; - let body_str = String::from_utf8(body_bytes.to_vec()).map_err(|e| { - tracing::error!("Failed to decode response body: {}", e); - ErrorData::internal_error(format!("Failed to decode response body: {}", e), None) - })?; - let hub_response: HubResponse = serde_json::from_str(&body_str).map_err(|e| { - tracing::error!("Failed to parse hub response: {}", e); - ErrorData::internal_error(format!("Failed to parse hub response: {}", e), None) - })?; - - Ok(hub_response.asks) -} - -/// Get the schema for a Hub script -pub async fn get_hub_script_schema(path: &str, db: &DB) -> Result, ErrorData> { - let strip_path = StripPath(path.to_string()); - let res = get_full_hub_script_by_path(strip_path, &HTTP_CLIENT, Some(db)) - .await - .map_err(|e| { - tracing::error!("Failed to get hub script: {}", e); - ErrorData::internal_error(format!("Failed to get hub script: {}", e), None) - })?; - match serde_json::from_str::(res.schema.get()) { - Ok(schema) => Ok(Some(schema)), - Err(e) => { - tracing::warn!("Failed to convert schema: {}", e); - Ok(None) - } - } -} diff --git a/backend/windmill-api/src/mcp/utils/mod.rs b/backend/windmill-api/src/mcp/utils/mod.rs deleted file mode 100644 index e4072313a0..0000000000 --- a/backend/windmill-api/src/mcp/utils/mod.rs +++ /dev/null @@ -1,10 +0,0 @@ -//! Utility functions and helpers for MCP server -//! -//! This module contains various utility functions for schema transformation, -//! database operations, data models, and path transformations. - -pub mod models; -pub mod database; -pub mod schema; -pub mod transform; -pub mod scope_matcher; \ No newline at end of file diff --git a/backend/windmill-api/src/mcp/utils/schema.rs b/backend/windmill-api/src/mcp/utils/schema.rs deleted file mode 100644 index b441d94a78..0000000000 --- a/backend/windmill-api/src/mcp/utils/schema.rs +++ /dev/null @@ -1,143 +0,0 @@ -//! Schema transformation utilities for MCP server -//! -//! Contains functions for transforming Windmill schemas into MCP-compatible formats, -//! including resource enrichment and schema conversion utilities. - -use windmill_mcp::server::ErrorData; -use serde_json::Value; -use std::collections::HashMap; -use windmill_common::db::UserDB; -use windmill_common::scripts::Schema; - -use super::database::get_resources; -use super::models::{ResourceInfo, ResourceType, SchemaType}; -use super::transform::apply_key_transformation; -use crate::db::ApiAuthed; - -/// Convert a Windmill Schema to a SchemaType -pub fn convert_schema_to_schema_type(schema: Option) -> SchemaType { - let schema_obj = if let Some(ref s) = schema { - match serde_json::from_str::(s.0.get()) { - Ok(val) => val, - Err(_) => SchemaType::default(), - } - } else { - SchemaType::default() - }; - schema_obj -} - -/// Transform the schema for resources by enriching with resource information -pub async fn transform_schema_for_resources( - schema: &SchemaType, - user_db: &UserDB, - authed: &ApiAuthed, - w_id: &str, - resources_cache: &mut HashMap>, - resources_types: &Vec, -) -> Result { - let mut schema_obj: SchemaType = schema.clone(); - - // replace invalid char in property key with underscore - let replacements: Vec<(String, String, Value)> = schema_obj - .properties - .iter() - .filter_map(|(key, value)| { - if key.chars().any(|c| !c.is_alphanumeric() && c != '_') { - let new_key = apply_key_transformation(key); - Some((key.clone(), new_key, value.clone())) - } else { - None - } - }) - .collect(); - - for (old_key, new_key, value) in replacements { - schema_obj.properties.remove(&old_key); - schema_obj.properties.insert(new_key, value); - } - - for (_key, prop_value) in schema_obj.properties.iter_mut() { - if let Value::Object(prop_map) = prop_value { - // if property is a resource, fetch the resource type infos, and add each available resource to the description - if let Some(format_value) = prop_map.get("format") { - if let Value::String(format_str) = format_value { - if format_str.starts_with("resource-") { - let resource_type_key = - format_str.split("-").last().unwrap_or_default().to_string(); - let resource_type = resources_types - .iter() - .find(|rt| rt.name == resource_type_key); - let resource_type_obj = resource_type.cloned(); - - if !resources_cache.contains_key(&resource_type_key) { - let available_resources = - get_resources(user_db, authed, &w_id, &resource_type_key).await; - - match available_resources { - Ok(cache_data) => { - resources_cache.insert(resource_type_key.clone(), cache_data); - } - Err(e) => { - tracing::error!("Failed to fetch resource cache data: {}", e); - continue; // Skip this property if fetching failed - } - } - } - - if let Some(resource_cache) = resources_cache.get(&resource_type_key) { - let resources_count = resource_cache.len(); - let description = match resource_type_obj { - Some(resource_type_obj) => format!( - "This is a resource named `{}` with the following description: `{}`.\\nThe path of the resource should be used to specify the resource.\\n{}", - resource_type_obj.name, - resource_type_obj.description.as_deref().unwrap_or("No description"), - if resources_count == 0 { - "This resource does not have any available instances, you should create one from your windmill workspace." - } else if resources_count > 1 { - "This resource has multiple available instances, you should precisely select the one you want to use." - } else { - "There is 1 resource available." - } - ), - None => "An object parameter.".to_string() - }; - prop_map - .insert("type".to_string(), Value::String("string".to_string())); - prop_map.insert("description".to_string(), Value::String(description)); - if resources_count > 0 { - let resources_description = resource_cache - .iter() - .map(|resource| { - format!( - "{}: $res:{}", - resource.description.as_deref().unwrap_or("No title"), - resource.path - ) - }) - .collect::>() - .join("\\n"); - - prop_map.insert( - "description".to_string(), - Value::String(format!( - "{}\\nHere are the available resources, in the format title:path. Title can be empty. Path should be used to specify the resource:\\n{}", - prop_map.get("description").unwrap_or(&Value::String("No description".to_string())), - resources_description - )), - ); - } - } - } - } - } - } else { - tracing::warn!( - "Schema property value is not a JSON object: {:?}", - prop_value - ); - } - } - - Ok(schema_obj) -} diff --git a/backend/windmill-mcp/Cargo.toml b/backend/windmill-mcp/Cargo.toml index a3e4bfed5f..afe2547f19 100644 --- a/backend/windmill-mcp/Cargo.toml +++ b/backend/windmill-mcp/Cargo.toml @@ -10,7 +10,7 @@ path = "src/lib.rs" [features] default = [] -server = ["rmcp/transport-streamable-http-server", "rmcp/transport-streamable-http-server-session", "rmcp/transport-worker"] +server = ["rmcp/transport-streamable-http-server", "rmcp/transport-streamable-http-server-session", "rmcp/transport-worker", "dep:sqlx", "dep:async-trait", "dep:http", "dep:tokio-util", "dep:tokio"] auth = ["rmcp/auth", "dep:oauth2"] [dependencies] @@ -22,3 +22,9 @@ serde.workspace = true serde_json.workspace = true tracing.workspace = true rmcp.workspace = true +sqlx = { workspace = true, optional = true } +async-trait = { workspace = true, optional = true } +http = { workspace = true, optional = true } +tokio-util = { workspace = true, features = ["rt"], optional = true } +tokio = { workspace = true, optional = true } +futures.workspace = true diff --git a/backend/windmill-mcp/src/client/mod.rs b/backend/windmill-mcp/src/client/mod.rs new file mode 100644 index 0000000000..2364571486 --- /dev/null +++ b/backend/windmill-mcp/src/client/mod.rs @@ -0,0 +1,209 @@ +//! MCP Client implementation +//! +//! This module provides functionality for connecting to external MCP servers +//! and executing tools on them. + +mod types; + +pub use types::{McpResource, McpToolSource}; + +use anyhow::{Context, Result}; +use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; +use rmcp::{ + model::{ + CallToolRequestParam, ClientCapabilities, ClientInfo, Implementation, + InitializeRequestParam, Tool as McpTool, + }, + service::RunningService, + transport::{ + streamable_http_client::StreamableHttpClientTransportConfig, StreamableHttpClientTransport, + }, + RoleClient, ServiceExt, +}; +use serde_json::{json, Value}; +use std::str::FromStr; +use windmill_common::variables::get_secret_value_as_admin; +use windmill_common::DB; + +/// MCP client for communicating with external MCP servers +pub struct McpClient { + /// The underlying rmcp client + client: RunningService, + /// Cached list of available tools from the server + available_tools: Vec, +} + +impl McpClient { + /// Create a new MCP client from a resource configuration + pub async fn from_resource(resource: McpResource, db: &DB, w_id: &str) -> Result { + // Build custom reqwest client with headers if provided + let mut headers = HeaderMap::new(); + if let Some(token_path) = &resource.token { + if !token_path.trim().is_empty() { + let value = + get_secret_value_as_admin(db, w_id, token_path.trim_start_matches("$var:")) + .await?; + headers.insert( + HeaderName::from_static("authorization"), + HeaderValue::from_str(format!("Bearer {}", value).as_str())?, + ); + } + } + if let Some(resource_headers) = &resource.headers { + for (key, value) in resource_headers { + match (HeaderName::from_str(key), HeaderValue::from_str(value)) { + (Ok(name), Ok(value)) => { + headers.insert(name, value); + } + _ => { + tracing::warn!("Invalid header: {}={}", key, value); + } + } + } + } + + let reqwest_client = reqwest::Client::builder() + .default_headers(headers) + .build() + .context("Failed to build HTTP client")?; + + // Create the HTTP transport with custom client + let config = StreamableHttpClientTransportConfig::with_uri(resource.url.as_str()); + let transport = StreamableHttpClientTransport::with_client(reqwest_client, config); + + // Set up client info + let client_info = ClientInfo { + protocol_version: Default::default(), + capabilities: ClientCapabilities::default(), + client_info: Implementation { + name: "windmill-ai-agent".to_string(), + title: Some("Windmill AI Agent".to_string()), + version: env!("CARGO_PKG_VERSION").to_string(), + website_url: None, + icons: None, + }, + }; + + // Initialize the connection + let client = client_info + .serve(transport) + .await + .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")? + .tools; + + Ok(Self { client, available_tools }) + } + + /// Get the list of available tools from the MCP server + pub fn available_tools(&self) -> &[McpTool] { + &self.available_tools + } + + /// Call a tool on the MCP server, with openai-style arguments + pub async fn call_tool(&self, name: &str, arguments: &str) -> Result { + // Convert OpenAI-style arguments to MCP format + let mcp_args = + Self::openai_args_to_mcp_args(arguments).context("Failed to parse tool arguments")?; + + let result = self + .client + .call_tool(CallToolRequestParam { + name: name.to_string().into(), + arguments: mcp_args, + task: None, + }) + .await + .context(format!("Failed to call MCP tool: {}", name))?; + + // Convert the result to a JSON value + // MCP tools return ToolResult which contains content array + let result_json = + serde_json::to_value(&result).context("Failed to serialize MCP tool result")?; + + Ok(result_json) + } + + /// Close the connection + pub async fn shutdown(self) -> Result<()> { + self.client.cancel().await?; + Ok(()) + } + + /// Fix array schemas to ensure they have the required 'items' property + /// OpenAI requires all array types to have an 'items' field. MCP servers may + /// return schemas without this field, so we add a default. + pub fn fix_array_schemas(schema: &mut Value) { + if let Value::Object(obj) = schema { + // Check if this is an array type + if let Some(type_val) = obj.get("type") { + let is_array = match type_val { + Value::String(s) => s == "array", + Value::Array(arr) => arr.iter().any(|v| v.as_str() == Some("array")), + _ => false, + }; + + // If it's an array and missing 'items', add a default + if is_array && !obj.contains_key("items") { + obj.insert("items".to_string(), json!({})); + } + } + + // Recursively fix nested schemas + if let Some(Value::Object(props)) = obj.get_mut("properties") { + for value in props.values_mut() { + Self::fix_array_schemas(value); + } + } + + // Fix items if present (for nested arrays) + if let Some(items) = obj.get_mut("items") { + Self::fix_array_schemas(items); + } + + // Fix oneOf, anyOf, allOf schemas + for key in &["oneOf", "anyOf", "allOf"] { + if let Some(Value::Array(schemas)) = obj.get_mut(*key) { + for schema in schemas { + Self::fix_array_schemas(schema); + } + } + } + + // Fix additionalProperties if it's a schema + if let Some(additional) = obj.get_mut("additionalProperties") { + if additional.is_object() { + Self::fix_array_schemas(additional); + } + } + } + } + + /// Convert OpenAI-style tool call arguments to MCP format + /// OpenAI sends arguments as a JSON string, MCP expects a Map + fn openai_args_to_mcp_args( + args_str: &str, + ) -> Result>> { + if args_str.trim().is_empty() { + return Ok(None); + } + + let args_value: serde_json::Value = + serde_json::from_str(args_str).context("Failed to parse tool call arguments")?; + + match args_value { + serde_json::Value::Object(map) => Ok(Some(map)), + serde_json::Value::Null => Ok(None), + _ => Ok(Some( + vec![("value".to_string(), args_value)] + .into_iter() + .collect(), + )), + } + } +} diff --git a/backend/windmill-mcp/src/client/types.rs b/backend/windmill-mcp/src/client/types.rs new file mode 100644 index 0000000000..232f2a5a9a --- /dev/null +++ b/backend/windmill-mcp/src/client/types.rs @@ -0,0 +1,32 @@ +//! Client-specific types for MCP +//! +//! Contains configuration and metadata types used for MCP client connections. + +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; + +/// MCP server resource configuration +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct McpResource { + /// Name of the MCP resource (used for prefixing tools) + pub name: String, + /// HTTP URL for the MCP server endpoint + pub url: String, + /// Optional token for authentication + #[serde(skip_serializing_if = "Option::is_none")] + pub token: Option, + /// Optional headers + #[serde(skip_serializing_if = "Option::is_none")] + pub headers: Option>, +} + +/// Metadata for tracking MCP tool sources +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct McpToolSource { + /// Name of the MCP resource this tool comes from + pub name: String, + /// Original tool name in the MCP server + pub tool_name: String, + /// Path of the MCP resource + pub resource_path: String, +} diff --git a/backend/windmill-mcp/src/common/mod.rs b/backend/windmill-mcp/src/common/mod.rs new file mode 100644 index 0000000000..d67c1718cc --- /dev/null +++ b/backend/windmill-mcp/src/common/mod.rs @@ -0,0 +1,16 @@ +//! Common types and utilities for MCP server and client +//! +//! This module contains shared data structures, transformation utilities, +//! and scope parsing functionality used throughout the MCP implementation. + +pub mod schema; +pub mod scope; +pub mod transform; +pub mod types; + +pub use schema::convert_schema_to_schema_type; +pub use scope::{is_resource_allowed, parse_mcp_scopes, McpScopeConfig}; +pub use transform::{ + apply_key_transformation, reverse_transform, reverse_transform_key, transform_path, +}; +pub use types::*; diff --git a/backend/windmill-mcp/src/common/schema.rs b/backend/windmill-mcp/src/common/schema.rs new file mode 100644 index 0000000000..c7cec8849e --- /dev/null +++ b/backend/windmill-mcp/src/common/schema.rs @@ -0,0 +1,41 @@ +//! Schema conversion utilities for MCP server +//! +//! Contains functions for converting Windmill schemas into MCP-compatible formats. + +use std::collections::HashSet; + +use serde_json::Value; + +use super::types::SchemaType; +use windmill_common::scripts::Schema; + +/// Convert a Windmill Schema to a SchemaType +pub fn convert_schema_to_schema_type(schema: Option) -> SchemaType { + let schema_obj = if let Some(ref s) = schema { + match serde_json::from_str::(s.0.get()) { + Ok(val) => val, + Err(_) => SchemaType::default(), + } + } else { + SchemaType::default() + }; + schema_obj +} + +/// Extract resource type keys from a schema +/// +/// Scans the schema properties for fields with format "resource-{type}" +/// and returns a set of all unique resource type names found. +pub fn extract_resource_types_from_schema(schema: &SchemaType) -> HashSet { + let mut resource_types = HashSet::new(); + for (_key, prop_value) in schema.properties.iter() { + if let Value::Object(prop_map) = prop_value { + if let Some(Value::String(format_str)) = prop_map.get("format") { + if let Some(rt) = format_str.strip_prefix("resource-") { + resource_types.insert(rt.to_string()); + } + } + } + } + resource_types +} diff --git a/backend/windmill-api/src/mcp/utils/scope_matcher.rs b/backend/windmill-mcp/src/common/scope.rs similarity index 86% rename from backend/windmill-api/src/mcp/utils/scope_matcher.rs rename to backend/windmill-mcp/src/common/scope.rs index 4b50d747ac..7da4e2cebc 100644 --- a/backend/windmill-api/src/mcp/utils/scope_matcher.rs +++ b/backend/windmill-mcp/src/common/scope.rs @@ -3,8 +3,6 @@ //! Contains utilities for parsing and matching MCP token scopes to determine //! which scripts, flows, and endpoints a token has access to. -use windmill_mcp::server::ErrorData; - /// Configuration for MCP scopes parsed from token scopes #[derive(Debug, Clone, Default)] pub struct McpScopeConfig { @@ -24,8 +22,26 @@ pub struct McpScopeConfig { pub hub_apps: Option, } +impl McpScopeConfig { + /// Check if a resource is allowed based on its type and path + pub fn is_allowed(&self, resource_type: &str, path: &str) -> bool { + if self.all { + return true; + } + + let patterns = match resource_type { + "script" => &self.scripts, + "flow" => &self.flows, + "endpoint" => &self.endpoints, + _ => return false, + }; + + is_resource_allowed(path, patterns) + } +} + /// Parse MCP scopes from token scope strings -pub fn parse_mcp_scopes(scopes: &[String]) -> Result { +pub fn parse_mcp_scopes(scopes: &[String]) -> Result { let mut config = McpScopeConfig::default(); for scope in scopes { @@ -69,19 +85,19 @@ pub fn parse_mcp_scopes(scopes: &[String]) -> Result if let Some(resources) = scope.strip_prefix("mcp:scripts:") { // New granular script scope: mcp:scripts:path1,path2,f/folder/* - config.scripts.extend(parse_resource_list(resources)?); + config.scripts.extend(parse_resource_list(resources)); continue; } if let Some(resources) = scope.strip_prefix("mcp:flows:") { // New granular flow scope: mcp:flows:path1,path2,f/folder/* - config.flows.extend(parse_resource_list(resources)?); + config.flows.extend(parse_resource_list(resources)); continue; } if let Some(resources) = scope.strip_prefix("mcp:endpoints:") { // New granular endpoint scope: mcp:endpoints:name1,name2 - config.endpoints.extend(parse_resource_list(resources)?); + config.endpoints.extend(parse_resource_list(resources)); continue; } @@ -94,16 +110,16 @@ pub fn parse_mcp_scopes(scopes: &[String]) -> Result } /// Parse comma-separated resource list -fn parse_resource_list(resources: &str) -> Result, ErrorData> { +fn parse_resource_list(resources: &str) -> Vec { if resources.is_empty() { - return Ok(vec![]); + return vec![]; } - Ok(resources + resources .split(',') .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()) - .collect()) + .collect() } /// Check if a resource path matches any pattern in the allowed list @@ -226,4 +242,16 @@ mod tests { let empty: Vec = vec![]; assert!(!is_resource_allowed("any/path", &empty)); } + + #[test] + fn test_scope_config_is_allowed() { + let mut config = McpScopeConfig::default(); + config.scripts.push("u/admin/*".to_string()); + config.flows.push("f/automation/*".to_string()); + + assert!(config.is_allowed("script", "u/admin/test")); + assert!(!config.is_allowed("script", "u/other/test")); + assert!(config.is_allowed("flow", "f/automation/test")); + assert!(!config.is_allowed("flow", "f/other/test")); + } } diff --git a/backend/windmill-api/src/mcp/utils/transform.rs b/backend/windmill-mcp/src/common/transform.rs similarity index 78% rename from backend/windmill-api/src/mcp/utils/transform.rs rename to backend/windmill-mcp/src/common/transform.rs index d4ef52fd3f..9a6fa317b1 100644 --- a/backend/windmill-api/src/mcp/utils/transform.rs +++ b/backend/windmill-mcp/src/common/transform.rs @@ -3,9 +3,9 @@ //! Contains functions for transforming paths, keys, and other identifiers //! to make them compatible with MCP tool naming requirements. -use super::models::SchemaType; +use super::types::SchemaType; -// MCP clients do not allow names longer than 60 characters +/// MCP clients do not allow names longer than 60 characters const MAX_PATH_LENGTH: usize = 60; /// Transform the path for workspace scripts/flows @@ -114,3 +114,38 @@ pub fn reverse_transform_key(transformed_key: &str, schema_obj: &Option, @@ -39,16 +42,13 @@ pub struct SchemaType { impl Default for SchemaType { fn default() -> Self { - Self { - r#type: "object".to_string(), - properties: HashMap::new(), - required: vec![], - } + Self { r#type: "object".to_string(), properties: HashMap::new(), required: vec![] } } } /// Script information from database -#[derive(Serialize, FromRow, Debug)] +#[derive(Serialize, Debug)] +#[cfg_attr(feature = "server", derive(FromRow))] pub struct ScriptInfo { pub path: String, pub summary: Option, @@ -57,7 +57,8 @@ pub struct ScriptInfo { } /// Flow information from database -#[derive(Serialize, FromRow, Debug)] +#[derive(Serialize, Debug)] +#[cfg_attr(feature = "server", derive(FromRow))] pub struct FlowInfo { pub path: String, pub summary: Option, @@ -66,7 +67,8 @@ pub struct FlowInfo { } /// Resource information from database -#[derive(Serialize, FromRow, Debug, Clone)] +#[derive(Serialize, Debug, Clone)] +#[cfg_attr(feature = "server", derive(FromRow))] pub struct ResourceInfo { pub path: String, pub description: Option, @@ -74,25 +76,34 @@ pub struct ResourceInfo { } /// Resource type information from database -#[derive(Serialize, FromRow, Debug, Clone)] +#[derive(Serialize, Debug, Clone)] +#[cfg_attr(feature = "server", derive(FromRow))] pub struct ResourceType { pub name: String, pub description: Option, } /// Schema holder for database queries -#[derive(Serialize, FromRow)] +#[derive(Serialize)] +#[cfg_attr(feature = "server", derive(FromRow))] pub struct ItemSchema { pub schema: Option, } /// Trait for objects that can be converted to MCP tools pub trait ToolableItem { + /// Get the path or identifier for this item (transformed for MCP compatibility) fn get_path_or_id(&self) -> String; + /// Get the summary/title of this item fn get_summary(&self) -> &str; + /// Get the description of this item fn get_description(&self) -> &str; + /// Get the JSON schema for this item's parameters fn get_schema(&self) -> SchemaType; + /// Whether this item is from the Hub fn is_hub(&self) -> bool; + /// Get the type of this item ("script" or "flow") fn item_type(&self) -> &'static str; + /// Get the integration type (for hub scripts) fn get_integration_type(&self) -> Option; -} \ No newline at end of file +} diff --git a/backend/windmill-mcp/src/lib.rs b/backend/windmill-mcp/src/lib.rs index bf05dc7c45..9026837493 100644 --- a/backend/windmill-mcp/src/lib.rs +++ b/backend/windmill-mcp/src/lib.rs @@ -1,51 +1,33 @@ -/* - * Author: Ruben Fiszel - * Copyright: Windmill Labs, Inc 2022 - * This file and its contents are licensed under the AGPLv3 License. - * Please see the included NOTICE for copyright information and - * LICENSE-AGPL for a copy of the license. - */ +//! Windmill MCP (Model Context Protocol) implementation +//! +//! This crate provides: +//! - MCP client for connecting to external MCP servers (used by AI agents) +//! - Common types and utilities for MCP implementations +//! - MCP server types (when `server` feature is enabled) +//! - OAuth support (when `auth` feature is enabled) -use anyhow::{Context, Result}; -use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; -use serde_json::{json, Value}; -use std::str::FromStr; -use windmill_common::variables::get_secret_value_as_admin; -use windmill_common::DB; +// Common types and utilities module +pub mod common; + +// Client module +pub mod client; + +// Re-export common types at crate root for convenience +pub use common::{ + convert_schema_to_schema_type, is_resource_allowed, parse_mcp_scopes, transform_path, FlowInfo, + HubResponse, HubScriptInfo, ItemSchema, McpScopeConfig, ResourceInfo, ResourceType, SchemaType, + ScriptInfo, ToolableItem, WorkspaceId, +}; + +// Re-export client types at crate root for backward compatibility +pub use client::{McpClient, McpResource, McpToolSource}; // Re-export rmcp types for client usage pub use rmcp::model::Tool as McpTool; -use rmcp::{ - model::{ - CallToolRequestParam, ClientCapabilities, ClientInfo, Implementation, - InitializeRequestParam, - }, - service::RunningService, - transport::{ - streamable_http_client::StreamableHttpClientTransportConfig, StreamableHttpClientTransport, - }, - RoleClient, ServiceExt, -}; -// Re-export rmcp server types when server feature is enabled +// Server module (when server feature is enabled) #[cfg(feature = "server")] -pub mod server { - //! Re-exports of rmcp server types for MCP server implementations - - pub use rmcp::handler::server::ServerHandler; - pub use rmcp::model::{ - Annotated, CallToolRequestParam, CallToolResult, Content, Implementation, - InitializeRequestParam, InitializeResult, ListPromptsResult, ListResourceTemplatesResult, - ListResourcesResult, ListToolsResult, PaginatedRequestParam, ProtocolVersion, RawContent, - RawTextContent, ServerCapabilities, ServerInfo, Tool, ToolAnnotations, - }; - pub use rmcp::service::{RequestContext, RoleServer}; - pub use rmcp::transport::streamable_http_server::{ - session::local::LocalSessionManager, StreamableHttpService, - }; - pub use rmcp::transport::StreamableHttpServerConfig; - pub use rmcp::ErrorData; -} +pub mod server; // Re-export rmcp auth types when auth feature is enabled #[cfg(feature = "auth")] @@ -60,212 +42,3 @@ pub mod oauth { RedirectUrl, Scope, TokenUrl, }; } - -use std::collections::HashMap; - -use serde::{Deserialize, Serialize}; - -/// MCP server resource configuration -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct McpResource { - /// Name of the MCP resource (used for prefixing tools) - pub name: String, - /// HTTP URL for the MCP server endpoint - pub url: String, - /// Optional token for authentication - #[serde(skip_serializing_if = "Option::is_none")] - pub token: Option, - /// Optional headers - #[serde(skip_serializing_if = "Option::is_none")] - pub headers: Option>, -} - -/// Metadata for tracking MCP tool sources -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct McpToolSource { - /// Name of the MCP resource this tool comes from - pub name: String, - /// Original tool name in the MCP server - pub tool_name: String, - /// Path of the MCP resource - pub resource_path: String, -} - -/// MCP client for communicating with external MCP servers -pub struct McpClient { - /// The underlying rmcp client - client: RunningService, - /// Cached list of available tools from the server - available_tools: Vec, -} - -impl McpClient { - /// Create a new MCP client from a resource configuration - pub async fn from_resource(resource: McpResource, db: &DB, w_id: &str) -> Result { - // Build custom reqwest client with headers if provided - let mut headers = HeaderMap::new(); - if let Some(token_path) = &resource.token { - if !token_path.trim().is_empty() { - let value = - get_secret_value_as_admin(db, w_id, token_path.trim_start_matches("$var:")) - .await?; - headers.insert( - HeaderName::from_static("authorization"), - HeaderValue::from_str(format!("Bearer {}", value).as_str())?, - ); - } - } - if let Some(resource_headers) = &resource.headers { - for (key, value) in resource_headers { - match (HeaderName::from_str(key), HeaderValue::from_str(value)) { - (Ok(name), Ok(value)) => { - headers.insert(name, value); - } - _ => { - tracing::warn!("Invalid header: {}={}", key, value); - } - } - } - } - - let reqwest_client = reqwest::Client::builder() - .default_headers(headers) - .build() - .context("Failed to build HTTP client")?; - - // Create the HTTP transport with custom client - let config = StreamableHttpClientTransportConfig::with_uri(resource.url.as_str()); - let transport = StreamableHttpClientTransport::with_client(reqwest_client, config); - - // Set up client info - let client_info = ClientInfo { - protocol_version: Default::default(), - capabilities: ClientCapabilities::default(), - client_info: Implementation { - name: "windmill-ai-agent".to_string(), - title: Some("Windmill AI Agent".to_string()), - version: env!("CARGO_PKG_VERSION").to_string(), - website_url: None, - icons: None, - }, - }; - - // Initialize the connection - let client = client_info - .serve(transport) - .await - .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")? - .tools; - - Ok(Self { client, available_tools }) - } - - /// Get the list of available tools from the MCP server - pub fn available_tools(&self) -> &[McpTool] { - &self.available_tools - } - - /// Call a tool on the MCP server, with openai-style arguments - pub async fn call_tool(&self, name: &str, arguments: &str) -> Result { - // Convert OpenAI-style arguments to MCP format - let mcp_args = - Self::openai_args_to_mcp_args(arguments).context("Failed to parse tool arguments")?; - - let result = self - .client - .call_tool(CallToolRequestParam { name: name.to_string().into(), arguments: mcp_args, task: None }) - .await - .context(format!("Failed to call MCP tool: {}", name))?; - - // Convert the result to a JSON value - // MCP tools return ToolResult which contains content array - let result_json = - serde_json::to_value(&result).context("Failed to serialize MCP tool result")?; - - Ok(result_json) - } - - /// Close the connection - pub async fn shutdown(self) -> Result<()> { - self.client.cancel().await?; - Ok(()) - } - - /// Fix array schemas to ensure they have the required 'items' property - /// OpenAI requires all array types to have an 'items' field. MCP servers may - /// return schemas without this field, so we add a default. - pub fn fix_array_schemas(schema: &mut Value) { - if let Value::Object(obj) = schema { - // Check if this is an array type - if let Some(type_val) = obj.get("type") { - let is_array = match type_val { - Value::String(s) => s == "array", - Value::Array(arr) => arr.iter().any(|v| v.as_str() == Some("array")), - _ => false, - }; - - // If it's an array and missing 'items', add a default - if is_array && !obj.contains_key("items") { - obj.insert("items".to_string(), json!({})); - } - } - - // Recursively fix nested schemas - if let Some(Value::Object(props)) = obj.get_mut("properties") { - for value in props.values_mut() { - Self::fix_array_schemas(value); - } - } - - // Fix items if present (for nested arrays) - if let Some(items) = obj.get_mut("items") { - Self::fix_array_schemas(items); - } - - // Fix oneOf, anyOf, allOf schemas - for key in &["oneOf", "anyOf", "allOf"] { - if let Some(Value::Array(schemas)) = obj.get_mut(*key) { - for schema in schemas { - Self::fix_array_schemas(schema); - } - } - } - - // Fix additionalProperties if it's a schema - if let Some(additional) = obj.get_mut("additionalProperties") { - if additional.is_object() { - Self::fix_array_schemas(additional); - } - } - } - } - - /// Convert OpenAI-style tool call arguments to MCP format - /// OpenAI sends arguments as a JSON string, MCP expects a Map - fn openai_args_to_mcp_args( - args_str: &str, - ) -> Result>> { - if args_str.trim().is_empty() { - return Ok(None); - } - - let args_value: serde_json::Value = - serde_json::from_str(args_str).context("Failed to parse tool call arguments")?; - - match args_value { - serde_json::Value::Object(map) => Ok(Some(map)), - serde_json::Value::Null => Ok(None), - _ => Ok(Some( - vec![("value".to_string(), args_value)] - .into_iter() - .collect(), - )), - } - } -} diff --git a/backend/windmill-mcp/src/server/backend.rs b/backend/windmill-mcp/src/server/backend.rs new file mode 100644 index 0000000000..fd01b2b26c --- /dev/null +++ b/backend/windmill-mcp/src/server/backend.rs @@ -0,0 +1,159 @@ +//! MCP Backend trait definitions +//! +//! This module defines the traits that must be implemented by the backend +//! (typically windmill-api) to provide the actual functionality for the MCP server. + +use async_trait::async_trait; +use rmcp::ErrorData; +use serde_json::Value; +use std::collections::HashMap; + +use crate::common::types::{ + FlowInfo, HubScriptInfo, ResourceInfo, ResourceType, SchemaType, ScriptInfo, +}; +use crate::server::endpoints::EndpointTool; + +/// Result type for backend operations using rmcp's ErrorData directly +pub type BackendResult = Result; + +/// Authentication context required by the MCP server +pub trait McpAuth: Send + Sync + Clone + 'static { + /// Get the username + fn username(&self) -> &str; + /// Get the email + fn email(&self) -> &str; + /// Check if user is admin + fn is_admin(&self) -> bool; + /// Check if user is operator + fn is_operator(&self) -> bool; + /// Get user's groups + fn groups(&self) -> &[String]; + /// Get user's folders as (name, can_write, is_owner) + fn folders(&self) -> &[(String, bool, bool)]; + /// Get token scopes + fn scopes(&self) -> Option<&[String]>; + + /// Check if the user has an MCP scope + fn has_mcp_scope(&self) -> bool { + self.scopes() + .map(|s| s.iter().any(|scope| scope.starts_with("mcp:"))) + .unwrap_or(false) + } +} + +/// The core backend trait that windmill-api implements +/// +/// This trait abstracts the windmill-api specific operations needed by the MCP server. +/// By implementing this trait, windmill-api can inject its database access, job execution, +/// and other functionality without windmill-mcp needing to depend on windmill-api directly. +#[async_trait] +pub trait McpBackend: Send + Sync + Clone + 'static { + /// The authentication context type + type Auth: McpAuth; + + // ───────────────────────────────────────────────────────────────── + // Listing Operations + // ───────────────────────────────────────────────────────────────── + + /// List scripts, optionally filtered to favorites only + async fn list_scripts( + &self, + auth: &Self::Auth, + workspace_id: &str, + favorites_only: bool, + ) -> BackendResult>; + + /// List flows, optionally filtered to favorites only + async fn list_flows( + &self, + auth: &Self::Auth, + workspace_id: &str, + favorites_only: bool, + ) -> BackendResult>; + + /// List resource types in workspace + async fn list_resource_types( + &self, + auth: &Self::Auth, + workspace_id: &str, + ) -> BackendResult>; + + /// List resources of a specific type + async fn list_resources( + &self, + auth: &Self::Auth, + workspace_id: &str, + resource_type: &str, + ) -> BackendResult>; + + /// List hub scripts, optionally filtered by app integrations + async fn list_hub_scripts(&self, app_filter: Option<&str>) + -> BackendResult>; + + // ───────────────────────────────────────────────────────────────── + // Schema Operations + // ───────────────────────────────────────────────────────────────── + + /// Get schema for a script or flow + async fn get_item_schema( + &self, + auth: &Self::Auth, + workspace_id: &str, + path: &str, + item_type: &str, + ) -> BackendResult>; + + /// Get schema for a hub script + async fn get_hub_script_schema(&self, path: &str) -> BackendResult>; + + // ───────────────────────────────────────────────────────────────── + // Schema Transformation (requires DB access for resources) + // ───────────────────────────────────────────────────────────────── + + /// Transform schema for resources by enriching with available resource information. + /// The resources_cache should be pre-populated with all needed resource types. + fn transform_schema_for_resources( + &self, + schema: &SchemaType, + resources_cache: &HashMap>, + resources_types: &[ResourceType], + ) -> SchemaType; + + // ───────────────────────────────────────────────────────────────── + // Execution Operations + // ───────────────────────────────────────────────────────────────── + + /// Run a script and wait for result + async fn run_script( + &self, + auth: &Self::Auth, + workspace_id: &str, + path: &str, + args: Value, + ) -> BackendResult; + + /// Run a flow and wait for result + async fn run_flow( + &self, + auth: &Self::Auth, + workspace_id: &str, + path: &str, + args: Value, + ) -> BackendResult; + + /// Call an endpoint tool (generated API endpoint) + async fn call_endpoint( + &self, + auth: &Self::Auth, + workspace_id: &str, + endpoint_tool: &EndpointTool, + args: Value, + ) -> BackendResult; + + // ───────────────────────────────────────────────────────────────── + // Endpoint Tools + // ───────────────────────────────────────────────────────────────── + + /// Get all available endpoint tools + fn all_endpoint_tools(&self) -> Vec; +} diff --git a/backend/windmill-mcp/src/server/endpoints.rs b/backend/windmill-mcp/src/server/endpoints.rs new file mode 100644 index 0000000000..ddf531bd1e --- /dev/null +++ b/backend/windmill-mcp/src/server/endpoints.rs @@ -0,0 +1,103 @@ +//! Endpoint tools for MCP server +//! +//! Contains the EndpointTool structure and utilities for converting +//! them to MCP tools. + +use rmcp::model::{Tool, ToolAnnotations}; +use serde::{Deserialize, Serialize}; +use std::borrow::Cow; +use std::sync::Arc; + +/// Represents an auto-generated endpoint tool from OpenAPI specification +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct EndpointTool { + pub name: Cow<'static, str>, + pub description: Cow<'static, str>, + pub instructions: Cow<'static, str>, + pub path: Cow<'static, str>, + pub method: Cow<'static, str>, + pub path_params_schema: Option, + pub query_params_schema: Option, + pub body_schema: Option, +} + +/// Convert a single endpoint tool to MCP tool +pub fn endpoint_tool_to_mcp_tool(tool: &EndpointTool) -> Tool { + let mut combined_properties = serde_json::Map::new(); + let mut combined_required = Vec::new(); + + // Combine all parameter schemas + let schemas = [ + &tool.path_params_schema, + &tool.query_params_schema, + &tool.body_schema, + ]; + + for schema in schemas.iter().filter_map(|s| s.as_ref()) { + merge_schema_into(&mut combined_properties, &mut combined_required, schema); + } + + let combined_schema = serde_json::json!({ + "type": "object", + "properties": combined_properties, + "required": combined_required + }); + + let description = format!("{}. {}", tool.description, tool.instructions); + + // Create annotations based on HTTP method and endpoint characteristics + let annotations = create_endpoint_annotations(tool); + + Tool { + name: tool.name.clone(), + description: Some(description.into()), + input_schema: Arc::new(combined_schema.as_object().unwrap().clone()), + title: Some(tool.name.to_string()), + output_schema: None, + icons: None, + annotations: Some(annotations), + meta: None, + } +} + +/// Create appropriate annotations for endpoint tools based on HTTP method +fn create_endpoint_annotations(tool: &EndpointTool) -> ToolAnnotations { + let method = tool.method.as_ref(); + + // Determine characteristics based on HTTP method + let (read_only, destructive, idempotent, open_world) = match method { + "GET" => (true, false, true, true), // Read-only, safe, idempotent + "POST" => (false, true, false, true), // Can modify, potentially destructive, not idempotent + "PUT" => (false, false, true, true), // Can modify, typically idempotent updates + "DELETE" => (false, true, true, true), // Destructive but idempotent + "PATCH" => (false, false, false, true), // Partial updates, not guaranteed idempotent + _ => (false, true, false, true), // Default: assume can modify and be destructive + }; + + ToolAnnotations { + title: Some(format!("{} {}", method, tool.path)), + read_only_hint: Some(read_only), + destructive_hint: Some(destructive), + idempotent_hint: Some(idempotent), + open_world_hint: Some(open_world), + } +} + +/// Merge schema into combined properties and required fields +fn merge_schema_into( + combined_properties: &mut serde_json::Map, + combined_required: &mut Vec, + schema: &serde_json::Value, +) { + if let Some(props) = schema.get("properties").and_then(|p| p.as_object()) { + for (key, value) in props { + combined_properties.insert(key.clone(), value.clone()); + } + } + + if let Some(required) = schema.get("required").and_then(|r| r.as_array()) { + for req in required.iter().filter_map(|r| r.as_str()) { + combined_required.push(req.to_string()); + } + } +} diff --git a/backend/windmill-mcp/src/server/mod.rs b/backend/windmill-mcp/src/server/mod.rs new file mode 100644 index 0000000000..90e6083435 --- /dev/null +++ b/backend/windmill-mcp/src/server/mod.rs @@ -0,0 +1,32 @@ +//! MCP Server module +//! +//! This module provides the MCP server implementation including: +//! - `McpBackend` trait for backend implementations +//! - `Runner` struct that implements the MCP protocol +//! - Re-exports of rmcp types + +pub mod backend; +pub mod endpoints; +pub mod runner; +pub mod tools; + +// Re-export main types +pub use backend::{BackendResult, McpAuth, McpBackend}; +pub use endpoints::{endpoint_tool_to_mcp_tool, EndpointTool}; +pub use runner::Runner; +pub use tools::create_tool_from_item; + +// Re-export rmcp types for convenience +pub use rmcp::handler::server::ServerHandler; +pub use rmcp::model::{ + Annotated, CallToolRequestParam, CallToolResult, Content, Implementation, + InitializeRequestParam, InitializeResult, ListPromptsResult, ListResourceTemplatesResult, + ListResourcesResult, ListToolsResult, PaginatedRequestParam, ProtocolVersion, RawContent, + RawTextContent, ServerCapabilities, ServerInfo, Tool, ToolAnnotations, +}; +pub use rmcp::service::{RequestContext, RoleServer}; +pub use rmcp::transport::streamable_http_server::{ + session::local::LocalSessionManager, StreamableHttpService, +}; +pub use rmcp::transport::StreamableHttpServerConfig; +pub use rmcp::ErrorData; diff --git a/backend/windmill-mcp/src/server/runner.rs b/backend/windmill-mcp/src/server/runner.rs new file mode 100644 index 0000000000..39fbda5c7a --- /dev/null +++ b/backend/windmill-mcp/src/server/runner.rs @@ -0,0 +1,373 @@ +//! MCP Server Runner implementation +//! +//! Contains the generic Runner that implements the MCP ServerHandler trait +//! and delegates to a McpBackend for actual functionality. + +use crate::common::schema::extract_resource_types_from_schema; +use crate::common::scope::parse_mcp_scopes; +use crate::common::transform::{reverse_transform, reverse_transform_key}; +use crate::common::types::{ResourceInfo, ToolableItem, WorkspaceId}; +use crate::server::backend::{McpAuth, McpBackend}; +use crate::server::endpoints::endpoint_tool_to_mcp_tool; +use crate::server::tools::create_tool_from_item; +use rmcp::handler::server::ServerHandler; +use rmcp::model::{ + CallToolRequestParam, CallToolResult, Content, Implementation, InitializeRequestParam, + InitializeResult, ListPromptsResult, ListResourceTemplatesResult, ListResourcesResult, + ListToolsResult, PaginatedRequestParam, ProtocolVersion, ServerCapabilities, ServerInfo, +}; +use rmcp::service::{RequestContext, RoleServer}; +use rmcp::ErrorData; +use serde_json::Value; +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; + +// Re-export from http crate for extracting request parts +use http::request::Parts as HttpParts; + +/// MCP Server Runner - generic over the backend implementation +/// +/// This struct implements the MCP ServerHandler trait and uses a McpBackend +/// to perform the actual operations (database queries, job execution, etc.) +pub struct Runner { + backend: Arc, +} + +impl Clone for Runner { + fn clone(&self) -> Self { + Self { backend: self.backend.clone() } + } +} + +impl Runner { + /// Create a new Runner with the given backend + pub fn new(backend: B) -> Self { + Self { backend: Arc::new(backend) } + } + + /// Extract authentication and workspace from request context + fn extract_context( + context: &RequestContext, + ) -> Result<(B::Auth, String), ErrorData> { + let http_parts = context.extensions.get::().ok_or_else(|| { + tracing::error!("http::request::Parts not found"); + ErrorData::internal_error("http::request::Parts not found", None) + })?; + + let auth = http_parts.extensions.get::().ok_or_else(|| { + tracing::error!("Auth extension not found"); + ErrorData::internal_error("Auth extension not found", None) + })?; + + let workspace_id = http_parts + .extensions + .get::() + .ok_or_else(|| { + tracing::error!("WorkspaceId not found"); + ErrorData::internal_error("WorkspaceId not found", None) + }) + .map(|w_id| w_id.0.clone())?; + + // Validate MCP scope + if !auth.has_mcp_scope() { + tracing::error!("Unauthorized: missing mcp scope"); + return Err(ErrorData::internal_error( + "Unauthorized: missing mcp scope", + None, + )); + } + + Ok((auth.clone(), workspace_id)) + } +} + +impl ServerHandler for Runner { + fn get_info(&self) -> ServerInfo { + ServerInfo { + protocol_version: ProtocolVersion::default(), + capabilities: ServerCapabilities::builder().enable_tools().build(), + server_info: Implementation::from_build_env(), + instructions: Some( + "This server provides a list of scripts and flows the user can run on Windmill. \ + Each flow and script is a tool callable with their respective arguments." + .to_string(), + ), + } + } + + async fn initialize( + &self, + _request: InitializeRequestParam, + _context: RequestContext, + ) -> Result { + Ok(self.get_info()) + } + + async fn list_tools( + &self, + _request: Option, + context: RequestContext, + ) -> Result { + let (auth, workspace_id) = Self::extract_context(&context)?; + + // Parse MCP scopes to determine what to expose + let scopes = auth.scopes().unwrap_or(&[]); + let scope_config = + parse_mcp_scopes(scopes).map_err(|e| ErrorData::internal_error(e, None))?; + + let favorites_only = scope_config.favorites; + + // Fetch all items concurrently + let (scripts, flows, resource_types, hub_scripts) = tokio::try_join!( + self.backend + .list_scripts(&auth, &workspace_id, favorites_only), + self.backend + .list_flows(&auth, &workspace_id, favorites_only), + self.backend.list_resource_types(&auth, &workspace_id), + async { + if let Some(ref apps) = scope_config.hub_apps { + self.backend.list_hub_scripts(Some(apps)).await + } else { + Ok(vec![]) + } + } + )?; + + // Filter items based on scope + let filtered_scripts: Vec<_> = scripts + .into_iter() + .filter(|s| !scope_config.granular || scope_config.is_allowed("script", &s.path)) + .collect(); + + let filtered_flows: Vec<_> = flows + .into_iter() + .filter(|f| !scope_config.granular || scope_config.is_allowed("flow", &f.path)) + .collect(); + + // Collect all needed resource types from all schemas + let mut needed_resource_types: HashSet = HashSet::new(); + for script in &filtered_scripts { + needed_resource_types.extend(extract_resource_types_from_schema(&script.get_schema())); + } + for flow in &filtered_flows { + needed_resource_types.extend(extract_resource_types_from_schema(&flow.get_schema())); + } + for hub_script in &hub_scripts { + needed_resource_types + .extend(extract_resource_types_from_schema(&hub_script.get_schema())); + } + + // Pre-fetch all resources + let resource_futures: Vec<_> = needed_resource_types + .into_iter() + .map(|rt| { + let backend = self.backend.clone(); + let auth = auth.clone(); + let workspace_id = workspace_id.clone(); + async move { + backend + .list_resources(&auth, &workspace_id, &rt) + .await + .map(|resources| (rt, resources)) + } + }) + .collect(); + + let resource_results = futures::future::try_join_all(resource_futures).await?; + let resources_cache: HashMap> = + resource_results.into_iter().collect(); + + let mut tools = Vec::new(); + + for script in &filtered_scripts { + tools.push(create_tool_from_item( + script, + self.backend.as_ref(), + &resources_cache, + &resource_types, + )); + } + + for flow in &filtered_flows { + tools.push(create_tool_from_item( + flow, + self.backend.as_ref(), + &resources_cache, + &resource_types, + )); + } + + for hub_script in &hub_scripts { + tools.push(create_tool_from_item( + hub_script, + self.backend.as_ref(), + &resources_cache, + &resource_types, + )); + } + + // Add endpoint tools from the generated MCP tools, filtered by scope + let endpoint_tools = self.backend.all_endpoint_tools(); + for endpoint_tool in endpoint_tools { + if scope_config.granular && !scope_config.is_allowed("endpoint", &endpoint_tool.name) { + continue; + } + + tools.push(endpoint_tool_to_mcp_tool(&endpoint_tool)); + } + + Ok(ListToolsResult { tools, next_cursor: None, meta: None }) + } + + async fn call_tool( + &self, + request: CallToolRequestParam, + context: RequestContext, + ) -> Result { + let (auth, workspace_id) = Self::extract_context(&context)?; + + // Parse MCP scopes for authorization + let scopes = auth.scopes().unwrap_or(&[]); + let scope_config = + parse_mcp_scopes(scopes).map_err(|e| ErrorData::internal_error(e, None))?; + + // Handle truncated tool names + if request.name.ends_with("_TRUNC") { + return Ok(CallToolResult::error(vec![rmcp::model::Annotated::new( + rmcp::model::RawContent::Text(rmcp::model::RawTextContent { + text: "Tool path is too long. Consider shortening it to make it compatible with MCP.".to_string(), + meta: None, + }), + None, + )])); + } + + let args = request.arguments.map(Value::Object).unwrap_or(Value::Null); + + // Check if this is an endpoint tool + let endpoint_tools = self.backend.all_endpoint_tools(); + for endpoint_tool in &endpoint_tools { + if endpoint_tool.name.as_ref() == request.name { + // Validate endpoint scope + if scope_config.granular + && !scope_config.is_allowed("endpoint", &endpoint_tool.name) + { + return Err(ErrorData::internal_error( + format!( + "Access denied: endpoint '{}' not in token scope", + endpoint_tool.name + ), + None, + )); + } + + // This is an endpoint tool, call via backend + let result = self + .backend + .call_endpoint(&auth, &workspace_id, endpoint_tool, args) + .await + .map_err(|e| ErrorData::internal_error(e.message, None))?; + + return Ok(CallToolResult::success(vec![Content::text( + serde_json::to_string_pretty(&result).unwrap_or_else(|_| "{}".to_string()), + )])); + } + } + + // Not an endpoint tool - parse as script/flow + let (tool_type, path, is_hub) = reverse_transform(&request.name).map_err(|e| { + ErrorData::internal_error(format!("Failed to parse tool name: {}", e), None) + })?; + + // Validate script/flow scope + if !is_hub && scope_config.granular { + if tool_type == "script" && !scope_config.is_allowed("script", &path) { + return Err(ErrorData::internal_error( + format!("Access denied: script '{}' not in token scope", path), + None, + )); + } else if tool_type == "flow" && !scope_config.is_allowed("flow", &path) { + return Err(ErrorData::internal_error( + format!("Access denied: flow '{}' not in token scope", path), + None, + )); + } + } + + // Get item schema for argument transformation + let item_schema = if is_hub { + self.backend + .get_hub_script_schema(&format!("hub/{}", path)) + .await + .map_err(|e| ErrorData::internal_error(e.message, None))? + } else { + self.backend + .get_item_schema(&auth, &workspace_id, &path, tool_type) + .await + .map_err(|e| ErrorData::internal_error(e.message, None))? + }; + + // Transform arguments back to original key names + let transformed_args = if let Value::Object(map) = args { + let mut args_hash = HashMap::new(); + for (k, v) in map { + let original_key = reverse_transform_key(&k, &item_schema); + args_hash.insert(original_key, v); + } + Value::Object(args_hash.into_iter().collect()) + } else { + args + }; + + let script_or_flow_path = if is_hub { + format!("hub/{}", path) + } else { + path + }; + + // Execute script or flow + let result = if tool_type == "script" { + self.backend + .run_script(&auth, &workspace_id, &script_or_flow_path, transformed_args) + .await + } else { + self.backend + .run_flow(&auth, &workspace_id, &script_or_flow_path, transformed_args) + .await + }; + + match result { + Ok(value) => Ok(CallToolResult::success(vec![Content::text( + serde_json::to_string_pretty(&value).unwrap_or_else(|_| "{}".to_string()), + )])), + Err(e) => Err(ErrorData::internal_error( + format!("Failed to run {}: {}", tool_type, e.message), + None, + )), + } + } + + async fn list_resources( + &self, + _request: Option, + _context: RequestContext, + ) -> Result { + Ok(ListResourcesResult { resources: vec![], next_cursor: None, meta: None }) + } + + async fn list_prompts( + &self, + _request: Option, + _context: RequestContext, + ) -> Result { + Ok(ListPromptsResult::default()) + } + + async fn list_resource_templates( + &self, + _request: Option, + _context: RequestContext, + ) -> Result { + Ok(ListResourceTemplatesResult::default()) + } +} diff --git a/backend/windmill-mcp/src/server/tools.rs b/backend/windmill-mcp/src/server/tools.rs new file mode 100644 index 0000000000..fea875e498 --- /dev/null +++ b/backend/windmill-mcp/src/server/tools.rs @@ -0,0 +1,184 @@ +//! Tool creation utilities for MCP server +//! +//! Contains functionality for converting Windmill items (scripts, flows, hub scripts) +//! into MCP tools. + +use rmcp::model::{Tool, ToolAnnotations}; +use std::borrow::Cow; +use std::collections::HashMap; +use std::sync::Arc; + +use crate::common::schema::convert_schema_to_schema_type; +use crate::common::transform::transform_path; +use crate::common::types::{ + FlowInfo, HubScriptInfo, ResourceInfo, ResourceType, SchemaType, ScriptInfo, ToolableItem, +}; +use crate::server::backend::McpBackend; + +/// Implementation of ToolableItem for ScriptInfo +impl ToolableItem for ScriptInfo { + fn get_path_or_id(&self) -> String { + transform_path(&self.path, "script") + } + + fn get_summary(&self) -> &str { + self.summary.as_deref().unwrap_or("No summary") + } + + fn get_description(&self) -> &str { + self.description.as_deref().unwrap_or("No description") + } + + fn get_schema(&self) -> SchemaType { + convert_schema_to_schema_type(self.schema.clone()) + } + + fn is_hub(&self) -> bool { + false + } + + fn item_type(&self) -> &'static str { + "script" + } + + fn get_integration_type(&self) -> Option { + None + } +} + +/// Implementation of ToolableItem for FlowInfo +impl ToolableItem for FlowInfo { + fn get_path_or_id(&self) -> String { + transform_path(&self.path, "flow") + } + + fn get_summary(&self) -> &str { + self.summary.as_deref().unwrap_or("No summary") + } + + fn get_description(&self) -> &str { + self.description.as_deref().unwrap_or("No description") + } + + fn get_schema(&self) -> SchemaType { + convert_schema_to_schema_type(self.schema.clone()) + } + + fn is_hub(&self) -> bool { + false + } + + fn item_type(&self) -> &'static str { + "flow" + } + + fn get_integration_type(&self) -> Option { + None + } +} + +/// Implementation of ToolableItem for HubScriptInfo +impl ToolableItem for HubScriptInfo { + fn get_path_or_id(&self) -> String { + let id = self.version_id; + let summary = self.summary.as_deref().unwrap_or("No summary"); + format!("hs-{}-{}", id, summary.replace(" ", "_")) + } + + fn get_summary(&self) -> &str { + self.summary.as_deref().unwrap_or("No summary") + } + + fn get_description(&self) -> &str { + self.description.as_deref().unwrap_or("No description") + } + + fn get_schema(&self) -> SchemaType { + match serde_json::from_value::(self.schema.clone().unwrap_or_default()) { + Ok(schema_type) => schema_type, + Err(_) => SchemaType::default(), + } + } + + fn is_hub(&self) -> bool { + true + } + + fn item_type(&self) -> &'static str { + "script" + } + + fn get_integration_type(&self) -> Option { + self.app.clone() + } +} + +/// Create an MCP Tool from a ToolableItem +/// +/// The resources_cache should be pre-populated with all resource types +/// that may be referenced by the item's schema. +pub fn create_tool_from_item( + item: &T, + backend: &B, + resources_cache: &HashMap>, + resources_types: &[ResourceType], +) -> Tool { + let is_hub = item.is_hub(); + let path = item.get_path_or_id(); + let item_type = item.item_type(); + let description = format!( + "This is a {} named `{}` with the following description: `{}`.{}", + item_type, + item.get_summary(), + item.get_description(), + if is_hub { + format!( + " It is a tool used for the following app: {}", + item.get_integration_type() + .unwrap_or("No integration type".to_string()) + ) + } else { + "".to_string() + } + ); + + let schema = item.get_schema(); + let schema_obj = + backend.transform_schema_for_resources(&schema, resources_cache, resources_types); + + let input_schema_map = match serde_json::to_value(schema_obj) { + Ok(serde_json::Value::Object(map)) => map, + Ok(_) => { + tracing::warn!( + "Schema object for tool '{}' did not serialize to a JSON object, using empty schema.", + path + ); + serde_json::Map::new() + } + Err(e) => { + tracing::error!( + "Failed to serialize schema object for tool '{}': {}. Using empty schema.", + path, + e + ); + serde_json::Map::new() + } + }; + + Tool { + name: Cow::Owned(path), + description: Some(Cow::Owned(description)), + input_schema: Arc::new(input_schema_map), + title: Some(item.get_summary().to_string()), + output_schema: None, + icons: None, + annotations: Some(ToolAnnotations { + title: Some(item.get_summary().to_string()), + read_only_hint: Some(false), // Can modify environment + destructive_hint: Some(true), // Can potentially be destructive + idempotent_hint: Some(false), // Are not guaranteed to be idempotent + open_world_hint: Some(true), // Can interact with external services + }), + meta: None, + } +}