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

fix(mcp): use stateless mode for openai sdk compatibility (#6656)

Browse Source 
* update crate

* use non stateful mode

* fix

* fix
This commit is contained in:
centdix
2025-09-23 10:16:26 +00:00
committed by GitHub
parent 3f66314419
commit 389b692523
6 changed files with 255 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+21 -8
View File
@@ -10856,9 +10856,9 @@ dependencies = [
[[package]]
name = "rmcp"
version = "0.2.1"
version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "37f2048a81a7ff7e8ef6bc5abced70c3d9114c8f03d85d7aaaafd9fd04f12e9e"
checksum = "41ab0892f4938752b34ae47cb53910b1b0921e55e77ddb6e44df666cab17939f"
dependencies = [
"base64 0.22.1",
"bytes",
@@ -10871,7 +10871,7 @@ dependencies = [
"pin-project-lite",
"rand 0.9.0",
"rmcp-macros",
"schemars 0.8.22",
"schemars 1.0.4",
"serde",
"serde_json",
"sse-stream",
@@ -10886,11 +10886,11 @@ dependencies = [
[[package]]
name = "rmcp-macros"
version = "0.2.1"
version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72398e694b9f6dbb5de960cf158c8699e6a1854cb5bbaac7de0646b2005763c4"
checksum = "1827cd98dab34cade0513243c6fe0351f0f0b2c9d6825460bcf45b42804bdda0"
dependencies = [
"darling 0.20.11",
"darling 0.21.3",
"proc-macro2",
"quote",
"serde_json",
@@ -11431,9 +11431,8 @@ version = "0.8.22"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3fbf2ae1b8bc8e02df939598064d22402220cd5bbcca1c76f7d6a310974d5615"
dependencies = [
"chrono",
"dyn-clone",
"schemars_derive",
"schemars_derive 0.8.22",
"serde",
"serde_json",
]
@@ -11456,8 +11455,10 @@ version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "82d20c4491bc164fa2f6c5d44565947a52ad80b9505d8e36f8d54c27c739fcd0"
dependencies = [
"chrono",
"dyn-clone",
"ref-cast",
"schemars_derive 1.0.4",
"serde",
"serde_json",
]
@@ -11474,6 +11475,18 @@ dependencies = [
"syn 2.0.106",
]
[[package]]
name = "schemars_derive"
version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "33d020396d1d138dc19f1165df7545479dcd58d93810dc5d646a16e55abefa80"
dependencies = [
"proc-macro2",
"quote",
"serde_derive_internals",
"syn 2.0.106",
]
[[package]]
name = "scoped-tls"
version = "1.0.1"
backend/windmill-api/Cargo.toml
+1 -1
View File
@@ -40,7 +40,7 @@ mcp = ["dep:rmcp"]
python = []
[dependencies]
rmcp = { version = "0.2.1", features=["transport-streamable-http-server", "transport-streamable-http-server-session", "transport-worker"], optional = true }
rmcp = { version = "0.6.4", features=["transport-streamable-http-server", "transport-streamable-http-server-session", "transport-worker"], optional = true }
windmill-queue.workspace = true
windmill-common = { workspace = true, default-features = false }
windmill-audit.workspace = true
backend/windmill-api/src/mcp/server.rs
+80 -56
View File
@@ -4,16 +4,17 @@
//! specification. This is a thin orchestration layer that delegates to the appropriate
//! modules for tool management, database operations, and schema transformation.
use std::borrow::Cow;
use std::collections::HashMap;
use std::sync::Arc;
use std::{borrow::Cow, time::Duration};
use axum::body::to_bytes;
use rmcp::{
handler::server::ServerHandler,
model::*,
service::{RequestContext, RoleServer},
Error,
transport::StreamableHttpServerConfig,
ErrorData,
};
use serde_json::Value;
use tokio::try_join;
@@ -26,35 +27,29 @@ 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_items, get_resources_types, get_scripts_from_hub, get_item_schema, get_hub_script_schema
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,
},
models::{ScriptInfo, FlowInfo, ResourceInfo, ResourceType, SchemaType, ToolableItem, WorkspaceId},
schema::transform_schema_for_resources,
transform::{reverse_transform, reverse_transform_key},
};
use super::tools::{
endpoint_tools::{all_endpoint_tools, endpoint_tools_to_mcp_tools, call_endpoint_tool, EndpointTool},
};
use axum::{
extract::Path,
http::Request,
middleware::Next,
response::Response,
routing::get,
Json,
Router,
extract::Path, http::Request, middleware::Next, response::Response, routing::get, Json, Router,
};
use rmcp::transport::streamable_http_server::{
session::local::LocalSessionManager,
SessionManager,
StreamableHttpService,
session::local::LocalSessionManager, SessionManager, StreamableHttpService,
};
use windmill_common::error::JsonResult;
/// MCP Server Runner - implements the core MCP protocol handlers
#[derive(Clone)]
pub struct Runner {}
@@ -72,7 +67,7 @@ impl Runner {
workspace_id: &str,
resources_cache: &mut HashMap<String, Vec<ResourceInfo>>,
resources_types: &Vec<ResourceType>,
) -> Result<Tool, Error> {
) -> Result<Tool, ErrorData> {
let is_hub = item.is_hub();
let path = item.get_path_or_id();
let item_type = item.item_type();
@@ -120,52 +115,57 @@ impl Runner {
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
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
open_world_hint: Some(true), // Can interact with external services
}),
})
}
}
impl ServerHandler for Runner {
/// Handles the `CallTool` request from the MCP client
async fn call_tool(
&self,
request: CallToolRequestParam,
context: RequestContext<RoleServer>,
) -> Result<CallToolResult, Error> {
) -> Result<CallToolResult, ErrorData> {
let http_parts = context
.extensions
.get::<axum::http::request::Parts>()
.ok_or_else(|| {
tracing::error!("http::request::Parts not found");
Error::internal_error("http::request::Parts not found", None)
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");
Error::internal_error("ApiAuthed Axum extension not found", None)
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");
Error::internal_error("DB Axum extension not found", None)
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");
Error::internal_error("UserDB Axum extension not found", None)
ErrorData::internal_error("UserDB Axum extension not found", None)
})?;
let args = request.arguments.map(Value::Object).ok_or_else(|| {
Error::invalid_params("Missing arguments for tool", Some(request.name.clone().into()))
ErrorData::invalid_params(
"Missing arguments for tool",
Some(request.name.clone().into()),
)
})?;
let workspace_id = http_parts
@@ -173,7 +173,7 @@ impl ServerHandler for Runner {
.get::<WorkspaceId>()
.ok_or_else(|| {
tracing::error!("WorkspaceId not found");
Error::internal_error("WorkspaceId not found", None)
ErrorData::internal_error("WorkspaceId not found", None)
})
.map(|w_id| w_id.0.clone())?;
@@ -182,18 +182,20 @@ impl ServerHandler for Runner {
for endpoint_tool in endpoint_tools {
if endpoint_tool.name.as_ref() == request.name {
// This is an endpoint tool, forward to the actual HTTP endpoint
let result = call_endpoint_tool(&endpoint_tool, args.clone(), &workspace_id, &authed).await?;
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())
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| {
Error::internal_error(format!("Failed to reverse transform path: {}", e), None)
ErrorData::internal_error(format!("Failed to reverse transform path: {}", e), None)
})?;
let item_schema = if is_hub {
get_hub_script_schema(&format!("hub/{}", path), db).await?
} else {
@@ -259,14 +261,20 @@ impl ServerHandler for Runner {
let body_bytes = to_bytes(response.into_body(), usize::MAX)
.await
.map_err(|e| {
Error::internal_error(format!("Failed to read response body: {}", e), None)
ErrorData::internal_error(
format!("Failed to read response body: {}", e),
None,
)
})?;
let body_str = String::from_utf8(body_bytes.to_vec()).map_err(|e| {
Error::internal_error(format!("Failed to decode response body: {}", e), None)
ErrorData::internal_error(
format!("Failed to decode response body: {}", e),
None,
)
})?;
Ok(CallToolResult::success(vec![Content::text(body_str)]))
}
Err(e) => Err(Error::internal_error(
Err(e) => Err(ErrorData::internal_error(
format!("Failed to run script: {}", e),
None,
)),
@@ -278,30 +286,30 @@ impl ServerHandler for Runner {
&self,
_request: Option<PaginatedRequestParam>,
mut _context: RequestContext<RoleServer>,
) -> Result<ListToolsResult, Error> {
) -> Result<ListToolsResult, ErrorData> {
let http_parts = _context
.extensions
.get::<axum::http::request::Parts>()
.ok_or_else(|| {
tracing::error!("http::request::Parts not found");
Error::internal_error("http::request::Parts not found", None)
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");
Error::internal_error("ApiAuthed Axum extension not found", None)
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");
Error::internal_error("DB Axum extension not found", None)
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");
Error::internal_error("UserDB Axum extension not found", None)
ErrorData::internal_error("UserDB Axum extension not found", None)
})?;
let workspace_id = http_parts
@@ -309,7 +317,7 @@ impl ServerHandler for Runner {
.get::<WorkspaceId>()
.ok_or_else(|| {
tracing::error!("WorkspaceId not found");
Error::internal_error("WorkspaceId not found", None)
ErrorData::internal_error("WorkspaceId not found", None)
})
.map(|w_id| w_id.0.clone())?;
@@ -319,10 +327,18 @@ impl ServerHandler for Runner {
.iter()
.find(|scope| scope.starts_with("mcp:") && !scope.contains("hub"))
});
let hub_scope = scopes.and_then(|scopes| scopes.iter().find(|scope| scope.starts_with("mcp:hub")));
let hub_scope =
scopes.and_then(|scopes| scopes.iter().find(|scope| scope.starts_with("mcp:hub")));
let (scope_type, scope_path) = owned_scope.map_or(("all", None), |scope| {
let parts = scope.split(":").collect::<Vec<&str>>();
(parts[1], if parts.len() == 3 { Some(parts[2]) } else { None })
(
parts[1],
if parts.len() == 3 {
Some(parts[2])
} else {
None
},
)
});
let scope_integrations = hub_scope.and_then(|scope| {
let parts = scope.split(":").collect::<Vec<&str>>();
@@ -341,8 +357,14 @@ impl ServerHandler for Runner {
"script",
scope_path.as_deref(),
);
let flows_fn =
get_items::<FlowInfo>(user_db, authed, &workspace_id, scope_type, "flow", scope_path.as_deref());
let flows_fn = get_items::<FlowInfo>(
user_db,
authed,
&workspace_id,
scope_type,
"flow",
scope_path.as_deref(),
);
let resources_types_fn = get_resources_types(user_db, authed, &workspace_id);
let hub_scripts_fn = get_scripts_from_hub(db, scope_integrations.as_deref());
let (scripts, flows, resources_types, hub_scripts) = if scope_integrations.is_some() {
@@ -410,7 +432,7 @@ impl ServerHandler for Runner {
fn get_info(&self) -> ServerInfo {
ServerInfo {
protocol_version: Default::default(),
protocol_version: ProtocolVersion::default(),
capabilities: ServerCapabilities::builder()
.enable_tools()
.enable_tool_list_changed()
@@ -424,7 +446,7 @@ impl ServerHandler for Runner {
&self,
_request: InitializeRequestParam,
_context: RequestContext<RoleServer>,
) -> Result<InitializeResult, Error> {
) -> Result<InitializeResult, ErrorData> {
Ok(self.get_info())
}
@@ -432,7 +454,7 @@ impl ServerHandler for Runner {
&self,
_request: Option<PaginatedRequestParam>,
_context: RequestContext<RoleServer>,
) -> Result<ListResourcesResult, Error> {
) -> Result<ListResourcesResult, ErrorData> {
Ok(ListResourcesResult { resources: vec![], next_cursor: None })
}
@@ -440,7 +462,7 @@ impl ServerHandler for Runner {
&self,
_request: Option<PaginatedRequestParam>,
_context: RequestContext<RoleServer>,
) -> Result<ListPromptsResult, Error> {
) -> Result<ListPromptsResult, ErrorData> {
Ok(ListPromptsResult::default())
}
@@ -448,7 +470,7 @@ impl ServerHandler for Runner {
&self,
_request: Option<PaginatedRequestParam>,
_context: RequestContext<RoleServer>,
) -> Result<ListResourceTemplatesResult, Error> {
) -> Result<ListResourceTemplatesResult, ErrorData> {
Ok(ListResourceTemplatesResult::default())
}
}
@@ -467,7 +489,10 @@ pub async fn extract_and_store_workspace_id(
/// Setup the MCP server with HTTP transport
pub async fn setup_mcp_server() -> anyhow::Result<(Router, Arc<LocalSessionManager>)> {
let session_manager = Arc::new(LocalSessionManager::default());
let service_config = Default::default();
let service_config = StreamableHttpServerConfig {
sse_keep_alive: Some(Duration::from_secs(15)),
stateful_mode: false,
};
let service = StreamableHttpService::new(
|| Ok(Runner::new()),
session_manager.clone(),
@@ -513,6 +538,5 @@ async fn list_mcp_tools_handler() -> JsonResult<Vec<EndpointTool>> {
/// Creates a router service for listing MCP tools
pub fn list_tools_service() -> Router {
Router::new()
.route("/", get(list_mcp_tools_handler))
}
Router::new().route("/", get(list_mcp_tools_handler))
}
backend/windmill-api/src/mcp/tools/endpoint_tools.rs
+90 -55
View File
@@ -3,16 +3,16 @@
//! Contains the auto-generated endpoint tools and utilities for converting
//! them to MCP tools and handling HTTP calls to Windmill API endpoints.
use rmcp::{model::Tool, Error};
use crate::db::ApiAuthed;
use rmcp::{model::Tool, ErrorData};
use std::sync::Arc;
use windmill_common::auth::create_jwt_token;
use windmill_common::db::Authed;
use windmill_common::BASE_URL;
use crate::db::ApiAuthed;
// Import the auto-generated tools
use super::auto_generated_endpoints;
pub use auto_generated_endpoints::{EndpointTool, all_tools};
pub use auto_generated_endpoints::{all_tools, EndpointTool};
/// Get all available endpoint tools
pub fn all_endpoint_tools() -> Vec<EndpointTool> {
@@ -21,25 +21,28 @@ pub fn all_endpoint_tools() -> Vec<EndpointTool> {
/// 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()
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.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,
@@ -47,7 +50,7 @@ pub fn endpoint_tool_to_mcp_tool(tool: &EndpointTool) -> Tool {
});
let description = format!("{}. {}", tool.description, tool.instructions);
// Create annotations based on HTTP method and endpoint characteristics
let annotations = create_endpoint_annotations(tool);
@@ -55,6 +58,9 @@ pub fn endpoint_tool_to_mcp_tool(tool: &EndpointTool) -> 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),
}
}
@@ -62,15 +68,15 @@ pub fn endpoint_tool_to_mcp_tool(tool: &EndpointTool) -> Tool {
/// Create appropriate annotations for endpoint tools based on HTTP method
fn create_endpoint_annotations(tool: &EndpointTool) -> rmcp::model::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
"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
_ => (false, true, false, true), // Default: assume can modify and be destructive
};
rmcp::model::ToolAnnotations {
@@ -93,7 +99,7 @@ fn merge_schema_into(
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());
@@ -107,34 +113,52 @@ pub async fn call_endpoint_tool(
args: serde_json::Value,
workspace_id: &str,
api_authed: &ApiAuthed,
) -> Result<serde_json::Value, Error> {
) -> Result<serde_json::Value, ErrorData> {
let args_map = match &args {
serde_json::Value::Object(map) => map,
_ => return Err(Error::invalid_params("Arguments must be an object", Some(tool.name.clone().into()))),
_ => {
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 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_URL.read().await, path_template, query_string);
let full_url = format!(
"{}/api{}{}",
BASE_URL.read().await,
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 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| {
Error::internal_error(format!("Failed to read response text: {}", e), None)
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)))
Ok(serde_json::from_str(&response_text)
.unwrap_or_else(|_| serde_json::Value::String(response_text)))
} else {
Err(Error::internal_error(
format!("HTTP {} {}: {}", status.as_u16(), status.canonical_reason().unwrap_or(""), response_text),
None
Err(ErrorData::internal_error(
format!(
"HTTP {} {}: {}",
status.as_u16(),
status.canonical_reason().unwrap_or(""),
response_text
),
None,
))
}
}
@@ -145,9 +169,9 @@ fn substitute_path_params(
workspace_id: &str,
args_map: &serde_json::Map<String, serde_json::Value>,
path_schema: &Option<serde_json::Value>,
) -> Result<String, Error> {
) -> 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 {
@@ -157,19 +181,19 @@ fn substitute_path_params(
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(Error::invalid_params(
return Err(ErrorData::invalid_params(
format!("Missing required path parameter: {}", param_name),
None
None,
));
}
}
}
}
}
Ok(path_template)
}
@@ -178,25 +202,31 @@ 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 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)
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),
format!(
"{}={}",
urlencoding::encode(param_name),
urlencoding::encode(str_val)
)
})
})
.collect();
if query_params.is_empty() {
String::new()
} else {
@@ -213,18 +243,19 @@ fn build_request_body(
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)
args_map
.get(param_name)
.map(|value| (param_name.clone(), value.clone()))
})
.collect();
if body_map.is_empty() {
None
} else {
@@ -239,7 +270,7 @@ async fn create_http_request(
workspace_id: &str,
api_authed: &ApiAuthed,
body_json: Option<serde_json::Value>,
) -> Result<reqwest::Response, Error> {
) -> Result<reqwest::Response, ErrorData> {
let client = &crate::HTTP_CLIENT;
let mut request_builder = match method {
"GET" => client.get(url),
@@ -247,16 +278,19 @@ async fn create_http_request(
"PUT" => client.put(url),
"DELETE" => client.delete(url),
"PATCH" => client.patch(url),
_ => return Err(Error::invalid_params(
format!("Unsupported HTTP method: {}", method),
None
)),
_ => {
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| Error::internal_error(e.to_string(), None))?;
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
@@ -266,7 +300,8 @@ async fn create_http_request(
.json(&body);
}
request_builder.send().await.map_err(|e| {
Error::internal_error(format!("Failed to execute request: {}", e), None)
})
}
request_builder
.send()
.await
.map_err(|e| ErrorData::internal_error(format!("Failed to execute request: {}", e), None))
}
backend/windmill-api/src/mcp/utils/database.rs
+50 -40
View File
@@ -3,28 +3,32 @@
//! Contains all database query functions and database-related utilities
//! used by the MCP server implementation.
use rmcp::Error;
use rmcp::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;
use super::models::*;
/// Check if the user has proper MCP scopes
pub fn check_scopes(authed: &ApiAuthed) -> Result<(), Error> {
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:all") && !scope.starts_with("mcp:favorites") && !scope.starts_with("mcp:hub:"))
|| scopes.unwrap().iter().all(|scope| {
!scope.starts_with("mcp:all")
&& !scope.starts_with("mcp:favorites")
&& !scope.starts_with("mcp:hub:")
})
{
tracing::error!("Unauthorized: missing mcp scope");
return Err(Error::internal_error("Unauthorized: missing mcp scope".to_string(), None));
return Err(ErrorData::internal_error(
"Unauthorized: missing mcp scope".to_string(),
None,
));
}
Ok(())
}
@@ -36,7 +40,7 @@ pub async fn get_item_schema(
authed: &ApiAuthed,
workspace_id: &str,
item_type: &str,
) -> Result<Option<Schema>, Error> {
) -> 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));
@@ -45,23 +49,23 @@ pub async fn get_item_schema(
sqlb.and_where("o.draft_only IS NOT TRUE");
let sql = sqlb.sql().map_err(|_e| {
tracing::error!("failed to build sql: {}", _e);
Error::internal_error("failed to build sql", None)
ErrorData::internal_error("failed to build sql", None)
})?;
let mut tx = user_db
.clone()
.begin(authed)
.await
.map_err(|_e| Error::internal_error("failed to begin transaction", None))?;
.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);
Error::internal_error("failed to fetch item schema", None)
ErrorData::internal_error("failed to fetch item schema", None)
})?;
tx.commit()
.await
.map_err(|_e| Error::internal_error("failed to commit transaction", None))?;
.map_err(|_e| ErrorData::internal_error("failed to commit transaction", None))?;
Ok(item.schema)
}
@@ -70,29 +74,29 @@ pub async fn get_resources_types(
user_db: &UserDB,
authed: &ApiAuthed,
workspace_id: &str,
) -> Result<Vec<ResourceType>, Error> {
) -> 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);
Error::internal_error("failed to build sql", None)
ErrorData::internal_error("failed to build sql", None)
})?;
let mut tx = user_db
.clone()
.begin(authed)
.await
.map_err(|_e| Error::internal_error("failed to begin transaction", None))?;
.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);
Error::internal_error("failed to fetch resource types", None)
ErrorData::internal_error("failed to fetch resource types", None)
})?;
tx.commit()
.await
.map_err(|_e| Error::internal_error("failed to commit transaction", None))?;
.map_err(|_e| ErrorData::internal_error("failed to commit transaction", None))?;
Ok(rows)
}
@@ -102,30 +106,30 @@ pub async fn get_resources(
authed: &ApiAuthed,
workspace_id: &str,
resource_type: &str,
) -> Result<Vec<ResourceInfo>, Error> {
) -> 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);
Error::internal_error("failed to build sql", None)
ErrorData::internal_error("failed to build sql", None)
})?;
let mut tx = user_db
.clone()
.begin(authed)
.await
.map_err(|_e| Error::internal_error("failed to begin transaction", None))?;
.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);
Error::internal_error("failed to fetch resources", None)
ErrorData::internal_error("failed to fetch resources", None)
})?;
tx.commit()
.await
.map_err(|_e| Error::internal_error("failed to commit transaction", None))?;
.map_err(|_e| ErrorData::internal_error("failed to commit transaction", None))?;
Ok(rows)
}
@@ -138,7 +142,7 @@ pub async fn get_items<T: for<'a> sqlx::FromRow<'a, sqlx::postgres::PgRow> + Sen
scope_type: &str,
item_type: &str,
scope_path: Option<&str>,
) -> Result<Vec<T>, Error> {
) -> 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);
@@ -157,9 +161,15 @@ pub async fn get_items<T: for<'a> sqlx::FromRow<'a, sqlx::postgres::PgRow> + Sen
// scope path is always a folder path, format is f/my_folder/*
if let Some(scope_path) = scope_path {
if scope_path.split("/").count() != 3 || !scope_path.starts_with("f/") || !scope_path.ends_with("/*") {
return Err(Error::internal_error(
format!("Invalid folder format: {}, expected format is f/my_folder/*", scope_path),
if scope_path.split("/").count() != 3
|| !scope_path.starts_with("f/")
|| !scope_path.ends_with("/*")
{
return Err(ErrorData::internal_error(
format!(
"Invalid folder format: {}, expected format is f/my_folder/*",
scope_path
),
None,
));
}
@@ -177,23 +187,23 @@ pub async fn get_items<T: for<'a> sqlx::FromRow<'a, sqlx::postgres::PgRow> + Sen
.limit(100);
let sql = sqlb.sql().map_err(|_e| {
tracing::error!("failed to build sql: {}", _e);
Error::internal_error("failed to build sql", None)
ErrorData::internal_error("failed to build sql", None)
})?;
let mut tx = user_db
.clone()
.begin(authed)
.await
.map_err(|_e| Error::internal_error("failed to begin transaction", None))?;
.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);
Error::internal_error(format!("failed to fetch {}", item_type), None)
ErrorData::internal_error(format!("failed to fetch {}", item_type), None)
})?;
tx.commit()
.await
.map_err(|_e| Error::internal_error("failed to commit transaction", None))?;
.map_err(|_e| ErrorData::internal_error("failed to commit transaction", None))?;
Ok(rows)
}
@@ -201,7 +211,7 @@ pub async fn get_items<T: for<'a> sqlx::FromRow<'a, sqlx::postgres::PgRow> + Sen
pub async fn get_scripts_from_hub(
db: &DB,
scope_integrations: Option<&str>,
) -> Result<Vec<HubScriptInfo>, Error> {
) -> Result<Vec<HubScriptInfo>, ErrorData> {
let query_params = Some(vec![
("limit", "100".to_string()),
("with_schema", "true".to_string()),
@@ -213,34 +223,34 @@ pub async fn get_scripts_from_hub(
.await
.map_err(|e| {
tracing::error!("Failed to get items from hub: {}", e);
Error::internal_error(format!("Failed to get items from hub: {}", e), None)
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);
Error::internal_error(format!("Failed to read response body: {}", e), None)
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);
Error::internal_error(format!("Failed to decode response body: {}", e), None)
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);
Error::internal_error(format!("Failed to parse hub response: {}", e), None)
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>, Error> {
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);
Error::internal_error(format!("Failed to get hub script: {}", e), None)
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)),
@@ -249,4 +259,4 @@ pub async fn get_hub_script_schema(path: &str, db: &DB) -> Result<Option<Schema>
Ok(None)
}
}
}
}
backend/windmill-api/src/mcp/utils/schema.rs
+13 -30
View File
@@ -3,16 +3,16 @@
//! Contains functions for transforming Windmill schemas into MCP-compatible formats,
//! including resource enrichment and schema conversion utilities.
use rmcp::Error;
use rmcp::ErrorData;
use serde_json::Value;
use std::collections::HashMap;
use windmill_common::db::UserDB;
use windmill_common::scripts::Schema;
use crate::db::ApiAuthed;
use super::models::{SchemaType, ResourceInfo, ResourceType};
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 {
@@ -35,7 +35,7 @@ pub async fn transform_schema_for_resources(
w_id: &str,
resources_cache: &mut HashMap<String, Vec<ResourceInfo>>,
resources_types: &Vec<ResourceType>,
) -> Result<SchemaType, Error> {
) -> Result<SchemaType, ErrorData> {
let mut schema_obj: SchemaType = schema.clone();
// replace invalid char in property key with underscore
@@ -71,24 +71,15 @@ pub async fn transform_schema_for_resources(
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;
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);
resources_cache.insert(resource_type_key.clone(), cache_data);
}
Err(e) => {
tracing::error!(
"Failed to fetch resource cache data: {}",
e
);
tracing::error!("Failed to fetch resource cache data: {}", e);
continue; // Skip this property if fetching failed
}
}
@@ -111,24 +102,16 @@ pub async fn transform_schema_for_resources(
),
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),
);
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.description.as_deref().unwrap_or("No title"),
resource.path
)
})
@@ -157,4 +140,4 @@ pub async fn transform_schema_for_resources(
}
Ok(schema_obj)
}
}