starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-23 08:00:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: move mcp logic to windmill-mcp (#7584)

Browse Source 
* draft

* clean up mcp logic

* cleaning

* cleaning

* better code

* error logging

* cleaning
This commit is contained in:
centdix
2026-01-16 21:55:14 +01:00
committed by GitHub
parent dcee9fe7b1
commit 437bad4cb7
30 changed files with 2161 additions and 1719 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+6
View File
@@ -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",
]
backend/generate_mcp_endpoints_tools/generate_mcp_tools.py
+5 -23
View File
@@ -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<serde_json::Value>,
pub query_params_schema: Option<serde_json::Value>,
pub body_schema: Option<serde_json::Value>,
}
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<EndpointTool> {
pub fn all_tools() -> Vec<EndpointTool> {{
vec![]
}
}}
"""
tool_definitions = []
@@ -386,9 +370,7 @@ pub fn all_tools() -> Vec<EndpointTool> {
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<EndpointTool> {{
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():
backend/windmill-api/src/lib.rs
+3 -2
View File
@@ -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),
backend/windmill-api/src/mcp/tools/auto_generated_endpoints.rs → backend/windmill-api/src/mcp/auto_generated_endpoints.rs
+1 -15
View File
@@ -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<serde_json::Value>,
pub query_params_schema: Option<serde_json::Value>,
pub body_schema: Option<serde_json::Value>,
}
use windmill_mcp::server::EndpointTool;
pub fn all_tools() -> Vec<EndpointTool> {
vec![
backend/windmill-api/src/mcp/core.rs
+448
View File
@@ -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<Vec<ScriptInfo>> {
let scope_type = if favorites_only { "favorites" } else { "all" };
get_items::<ScriptInfo>(&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<Vec<FlowInfo>> {
let scope_type = if favorites_only { "favorites" } else { "all" };
get_items::<FlowInfo>(&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<Vec<ResourceType>> {
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<Vec<ResourceInfo>> {
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<Vec<HubScriptInfo>> {
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<Option<SchemaType>> {
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::<SchemaType>(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<Option<SchemaType>> {
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::<SchemaType>(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<String, Vec<ResourceInfo>>,
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::<Vec<String>>()
.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<Value> {
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<Value> {
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<Value> {
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<EndpointTool> {
all_tools()
}
}
/// Extract workspace ID from path and store it in request extensions
pub async fn extract_and_store_workspace_id(
Path(params): Path<String>,
mut request: Request<axum::body::Body>,
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<Vec<EndpointTool>> {
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))
}
backend/windmill-api/src/mcp/mod.rs
+5 -5
View File
@@ -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};
backend/windmill-api/src/mcp/server.rs
-561
View File
@@ -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<T: ToolableItem>(
item: &T,
user_db: &UserDB,
authed: &ApiAuthed,
workspace_id: &str,
resources_cache: &mut HashMap<String, Vec<ResourceInfo>>,
resources_types: &Vec<ResourceType>,
) -> Result<Tool, ErrorData> {
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<RoleServer>,
) -> Result<CallToolResult, ErrorData> {
let http_parts = context
.extensions
.get::<axum::http::request::Parts>()
.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::<ApiAuthed>().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::<DB>().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::<UserDB>().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::<WorkspaceId>()
.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::<SchemaType>(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<PaginatedRequestParam>,
mut _context: RequestContext<RoleServer>,
) -> Result<ListToolsResult, ErrorData> {
let http_parts = _context
.extensions
.get::<axum::http::request::Parts>()
.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::<ApiAuthed>().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::<DB>().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::<UserDB>().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::<WorkspaceId>()
.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::<ScriptInfo>(user_db, authed, &workspace_id, scope_type, "script");
let flows_fn = get_items::<FlowInfo>(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<String, Vec<ResourceInfo>> = HashMap::new();
let mut tools: Vec<Tool> = 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<RoleServer>,
) -> Result<InitializeResult, ErrorData> {
Ok(self.get_info())
}
async fn list_resources(
&self,
_request: Option<PaginatedRequestParam>,
_context: RequestContext<RoleServer>,
) -> Result<ListResourcesResult, ErrorData> {
Ok(ListResourcesResult { resources: vec![], next_cursor: None, meta: None })
}
async fn list_prompts(
&self,
_request: Option<PaginatedRequestParam>,
_context: RequestContext<RoleServer>,
) -> Result<ListPromptsResult, ErrorData> {
Ok(ListPromptsResult::default())
}
async fn list_resource_templates(
&self,
_request: Option<PaginatedRequestParam>,
_context: RequestContext<RoleServer>,
) -> Result<ListResourceTemplatesResult, ErrorData> {
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<String>,
mut request: Request<axum::body::Body>,
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<Vec<EndpointTool>> {
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))
}
backend/windmill-api/src/mcp/tools/endpoint_tools.rs
-307
View File
@@ -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<EndpointTool> {
all_tools()
}
/// Convert endpoint tools to MCP tools
pub fn endpoint_tools_to_mcp_tools(endpoint_tools: Vec<EndpointTool>) -> Vec<Tool> {
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<String, serde_json::Value>,
combined_required: &mut Vec<String>,
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<serde_json::Value, ErrorData> {
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<String, serde_json::Value>,
path_schema: &Option<serde_json::Value>,
) -> Result<String, ErrorData> {
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<String, serde_json::Value>,
query_schema: &Option<serde_json::Value>,
) -> 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<String> = 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<String, serde_json::Value>,
body_schema: &Option<serde_json::Value>,
) -> Option<serde_json::Value> {
if method == "GET" {
return None;
}
let schema = body_schema.as_ref()?;
let props = schema.get("properties")?.as_object()?;
let body_map: serde_json::Map<String, serde_json::Value> = 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<serde_json::Value>,
) -> Result<reqwest::Response, ErrorData> {
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))
}
backend/windmill-api/src/mcp/tools/flow_tools.rs
-40
View File
@@ -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<String> {
None
}
}
backend/windmill-api/src/mcp/tools/hub_tools.rs
-43
View File
@@ -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::<SchemaType>(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<String> {
self.app.clone()
}
}
backend/windmill-api/src/mcp/tools/mod.rs
-10
View File
@@ -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;
backend/windmill-api/src/mcp/tools/script_tools.rs
-40
View File
@@ -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<String> {
None
}
}
backend/windmill-api/src/mcp/utils.rs
+414
View File
@@ -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<Option<Schema>, 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<Vec<ResourceType>, 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<Vec<ResourceInfo>, 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<T: for<'a> sqlx::FromRow<'a, sqlx::postgres::PgRow> + Send + Unpin>(
user_db: &UserDB,
authed: &ApiAuthed,
workspace_id: &str,
scope_type: &str,
item_type: &str,
) -> Result<Vec<T>, 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<Vec<HubScriptInfo>, 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<Option<Schema>, 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::<Schema>(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<String, Value>,
path_schema: &Option<Value>,
) -> BackendResult<String> {
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<String, Value>,
query_schema: &Option<Value>,
) -> 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<String> = 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<String, Value>,
body_schema: &Option<Value>,
) -> Option<Value> {
if method == "GET" {
return None;
}
let schema = body_schema.as_ref()?;
let props = schema.get("properties")?.as_object()?;
let body_map: serde_json::Map<String, Value> = 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<Value>,
) -> BackendResult<reqwest::Response> {
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<Body>) -> BackendResult<Value> {
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)))
}
backend/windmill-api/src/mcp/utils/database.rs
-243
View File
@@ -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<Option<Schema>, 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<Vec<ResourceType>, 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<Vec<ResourceInfo>, 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<T: for<'a> sqlx::FromRow<'a, sqlx::postgres::PgRow> + Send + Unpin>(
user_db: &UserDB,
authed: &ApiAuthed,
workspace_id: &str,
scope_type: &str,
item_type: &str,
) -> Result<Vec<T>, 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<Vec<HubScriptInfo>, 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<Option<Schema>, 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::<Schema>(res.schema.get()) {
Ok(schema) => Ok(Some(schema)),
Err(e) => {
tracing::warn!("Failed to convert schema: {}", e);
Ok(None)
}
}
}
backend/windmill-api/src/mcp/utils/mod.rs
-10
View File
@@ -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;
backend/windmill-api/src/mcp/utils/schema.rs
-143
View File
@@ -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<Schema>) -> SchemaType {
let schema_obj = if let Some(ref s) = schema {
match serde_json::from_str::<SchemaType>(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<String, Vec<ResourceInfo>>,
resources_types: &Vec<ResourceType>,
) -> Result<SchemaType, ErrorData> {
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::<Vec<String>>()
.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)
}
backend/windmill-mcp/Cargo.toml
+7 -1
View File
@@ -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
backend/windmill-mcp/src/client/mod.rs
+209
View File
@@ -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<RoleClient, InitializeRequestParam>,
/// Cached list of available tools from the server
available_tools: Vec<McpTool>,
}
impl McpClient {
/// Create a new MCP client from a resource configuration
pub async fn from_resource(resource: McpResource, db: &DB, w_id: &str) -> Result<Self> {
// 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<serde_json::Value> {
// 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<Option<serde_json::Map<String, serde_json::Value>>> {
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(),
)),
}
}
}
backend/windmill-mcp/src/client/types.rs
+32
View File
@@ -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<String>,
/// Optional headers
#[serde(skip_serializing_if = "Option::is_none")]
pub headers: Option<HashMap<String, String>>,
}
/// 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,
}
backend/windmill-mcp/src/common/mod.rs
+16
View File
@@ -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::*;
backend/windmill-mcp/src/common/schema.rs
+41
View File
@@ -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<Schema>) -> SchemaType {
let schema_obj = if let Some(ref s) = schema {
match serde_json::from_str::<SchemaType>(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<String> {
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
}
backend/windmill-api/src/mcp/utils/scope_matcher.rs → backend/windmill-mcp/src/common/scope.rs
+38 -10
View File
@@ -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<String>,
}
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<McpScopeConfig, ErrorData> {
pub fn parse_mcp_scopes(scopes: &[String]) -> Result<McpScopeConfig, String> {
let mut config = McpScopeConfig::default();
for scope in scopes {
@@ -69,19 +85,19 @@ pub fn parse_mcp_scopes(scopes: &[String]) -> Result<McpScopeConfig, ErrorData>
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<McpScopeConfig, ErrorData>
}
/// Parse comma-separated resource list
fn parse_resource_list(resources: &str) -> Result<Vec<String>, ErrorData> {
fn parse_resource_list(resources: &str) -> Vec<String> {
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<String> = 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"));
}
}
backend/windmill-api/src/mcp/utils/transform.rs → backend/windmill-mcp/src/common/transform.rs
+37 -2
View File
@@ -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<SchemaTy
transformed_key.to_string()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_transform_path() {
assert_eq!(
transform_path("u/admin/script", "script"),
"s-u_admin_script"
);
assert_eq!(transform_path("f/folder/flow", "flow"), "f-f_folder_flow");
assert_eq!(transform_path("my_script", "script"), "s-my__script");
}
#[test]
fn test_reverse_transform() {
let (type_str, path, is_hub) = reverse_transform("s-u_admin_script").unwrap();
assert_eq!(type_str, "script");
assert_eq!(path, "u/admin/script");
assert!(!is_hub);
let (type_str, path, is_hub) = reverse_transform("f-f_folder_flow").unwrap();
assert_eq!(type_str, "flow");
assert_eq!(path, "f/folder/flow");
assert!(!is_hub);
}
#[test]
fn test_apply_key_transformation() {
assert_eq!(apply_key_transformation("my key"), "my_key");
assert_eq!(apply_key_transformation("key!@#"), "key");
assert_eq!(apply_key_transformation("key_123"), "key_123");
}
}
backend/windmill-api/src/mcp/utils/models.rs → backend/windmill-mcp/src/common/types.rs
+24 -13
View File
@@ -5,10 +5,12 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sqlx::FromRow;
use std::collections::HashMap;
use windmill_common::scripts::Schema;
#[cfg(feature = "server")]
use sqlx::FromRow;
/// Workspace ID wrapper for Axum extensions
#[derive(Clone, Debug)]
pub struct WorkspaceId(pub String);
@@ -30,7 +32,8 @@ pub struct HubScriptInfo {
}
/// Schema type structure for JSON schemas
#[derive(Serialize, FromRow, Deserialize, Debug, Clone)]
#[derive(Serialize, Deserialize, Debug, Clone)]
#[cfg_attr(feature = "server", derive(FromRow))]
pub struct SchemaType {
pub r#type: String,
pub properties: HashMap<String, Value>,
@@ -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<String>,
@@ -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<String>,
@@ -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<String>,
@@ -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<String>,
}
/// Schema holder for database queries
#[derive(Serialize, FromRow)]
#[derive(Serialize)]
#[cfg_attr(feature = "server", derive(FromRow))]
pub struct ItemSchema {
pub schema: Option<Schema>,
}
/// 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<String>;
}
}
backend/windmill-mcp/src/lib.rs
+24 -251
View File
@@ -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<String>,
/// Optional headers
#[serde(skip_serializing_if = "Option::is_none")]
pub headers: Option<HashMap<String, String>>,
}
/// 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<RoleClient, InitializeRequestParam>,
/// Cached list of available tools from the server
available_tools: Vec<McpTool>,
}
impl McpClient {
/// Create a new MCP client from a resource configuration
pub async fn from_resource(resource: McpResource, db: &DB, w_id: &str) -> Result<Self> {
// 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<serde_json::Value> {
// 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<Option<serde_json::Map<String, serde_json::Value>>> {
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(),
)),
}
}
}
backend/windmill-mcp/src/server/backend.rs
+159
View File
@@ -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<T> = Result<T, ErrorData>;
/// 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<Vec<ScriptInfo>>;
/// List flows, optionally filtered to favorites only
async fn list_flows(
&self,
auth: &Self::Auth,
workspace_id: &str,
favorites_only: bool,
) -> BackendResult<Vec<FlowInfo>>;
/// List resource types in workspace
async fn list_resource_types(
&self,
auth: &Self::Auth,
workspace_id: &str,
) -> BackendResult<Vec<ResourceType>>;
/// List resources of a specific type
async fn list_resources(
&self,
auth: &Self::Auth,
workspace_id: &str,
resource_type: &str,
) -> BackendResult<Vec<ResourceInfo>>;
/// List hub scripts, optionally filtered by app integrations
async fn list_hub_scripts(&self, app_filter: Option<&str>)
-> BackendResult<Vec<HubScriptInfo>>;
// ─────────────────────────────────────────────────────────────────
// 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<Option<SchemaType>>;
/// Get schema for a hub script
async fn get_hub_script_schema(&self, path: &str) -> BackendResult<Option<SchemaType>>;
// ─────────────────────────────────────────────────────────────────
// 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<String, Vec<ResourceInfo>>,
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<Value>;
/// Run a flow and wait for result
async fn run_flow(
&self,
auth: &Self::Auth,
workspace_id: &str,
path: &str,
args: Value,
) -> BackendResult<Value>;
/// Call an endpoint tool (generated API endpoint)
async fn call_endpoint(
&self,
auth: &Self::Auth,
workspace_id: &str,
endpoint_tool: &EndpointTool,
args: Value,
) -> BackendResult<Value>;
// ─────────────────────────────────────────────────────────────────
// Endpoint Tools
// ─────────────────────────────────────────────────────────────────
/// Get all available endpoint tools
fn all_endpoint_tools(&self) -> Vec<EndpointTool>;
}
backend/windmill-mcp/src/server/endpoints.rs
+103
View File
@@ -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<serde_json::Value>,
pub query_params_schema: Option<serde_json::Value>,
pub body_schema: Option<serde_json::Value>,
}
/// 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<String, serde_json::Value>,
combined_required: &mut Vec<String>,
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());
}
}
}
backend/windmill-mcp/src/server/mod.rs
+32
View File
@@ -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;
backend/windmill-mcp/src/server/runner.rs
+373
View File
@@ -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<B: McpBackend> {
backend: Arc<B>,
}
impl<B: McpBackend> Clone for Runner<B> {
fn clone(&self) -> Self {
Self { backend: self.backend.clone() }
}
}
impl<B: McpBackend> Runner<B> {
/// 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<RoleServer>,
) -> Result<(B::Auth, String), ErrorData> {
let http_parts = context.extensions.get::<HttpParts>().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::<B::Auth>().ok_or_else(|| {
tracing::error!("Auth extension not found");
ErrorData::internal_error("Auth extension not found", None)
})?;
let workspace_id = http_parts
.extensions
.get::<WorkspaceId>()
.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<B: McpBackend> ServerHandler for Runner<B> {
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<RoleServer>,
) -> Result<InitializeResult, ErrorData> {
Ok(self.get_info())
}
async fn list_tools(
&self,
_request: Option<PaginatedRequestParam>,
context: RequestContext<RoleServer>,
) -> Result<ListToolsResult, ErrorData> {
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<String> = 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<String, Vec<ResourceInfo>> =
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<RoleServer>,
) -> Result<CallToolResult, ErrorData> {
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<PaginatedRequestParam>,
_context: RequestContext<RoleServer>,
) -> Result<ListResourcesResult, ErrorData> {
Ok(ListResourcesResult { resources: vec![], next_cursor: None, meta: None })
}
async fn list_prompts(
&self,
_request: Option<PaginatedRequestParam>,
_context: RequestContext<RoleServer>,
) -> Result<ListPromptsResult, ErrorData> {
Ok(ListPromptsResult::default())
}
async fn list_resource_templates(
&self,
_request: Option<PaginatedRequestParam>,
_context: RequestContext<RoleServer>,
) -> Result<ListResourceTemplatesResult, ErrorData> {
Ok(ListResourceTemplatesResult::default())
}
}
backend/windmill-mcp/src/server/tools.rs
+184
View File
@@ -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<String> {
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<String> {
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::<SchemaType>(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<String> {
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<T: ToolableItem, B: McpBackend>(
item: &T,
backend: &B,
resources_cache: &HashMap<String, Vec<ResourceInfo>>,
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,
}
}