From e2ef3317efaddaddbe436ec99733ecf7135cb575 Mon Sep 17 00:00:00 2001 From: centdix <40307056+centdix@users.noreply.github.com> Date: Thu, 7 Aug 2025 17:55:35 +0200 Subject: [PATCH] feat(mcp): add api endpoints as tools (#6329) * working list tools * working call tool * add schema * implement calling the endpoint * use openapi instead * correctly implement call_tool * provide workspace from context * cleaning * add more endpoints * remove resolved hack * add missing properties description * add list scripts and flows * add instructions * remove bacon.toml * cleaning * remove bacon.toml * nit * nit * cleaning * fix openapi file * nit * better error handling --- backend/.gitignore | 3 +- .../generate_mcp_endpoints_tools/README.md | 35 + .../generate_mcp_tools.py | 307 +++++ .../requirements.txt | 1 + backend/windmill-api/openapi.yaml | 104 +- backend/windmill-api/src/db.rs | 16 + backend/windmill-api/src/lib.rs | 4 + backend/windmill-api/src/mcp.rs | 121 +- backend/windmill-api/src/mcp_tools.rs | 1161 +++++++++++++++++ backend/windmill-api/src/mcp_utils.rs | 241 ++++ backend/windmill-common/src/auth.rs | 42 +- 11 files changed, 1939 insertions(+), 96 deletions(-) create mode 100644 backend/generate_mcp_endpoints_tools/README.md create mode 100644 backend/generate_mcp_endpoints_tools/generate_mcp_tools.py create mode 100644 backend/generate_mcp_endpoints_tools/requirements.txt create mode 100644 backend/windmill-api/src/mcp_tools.rs create mode 100644 backend/windmill-api/src/mcp_utils.rs diff --git a/backend/.gitignore b/backend/.gitignore index 2a3262acac..264bef360e 100644 --- a/backend/.gitignore +++ b/backend/.gitignore @@ -7,4 +7,5 @@ heaptrack* index/ windmill-api/openapi-*.* .duckdb/* -*ee.rs \ No newline at end of file +*ee.rs +generate_mcp_endpoints_tools/venv \ No newline at end of file diff --git a/backend/generate_mcp_endpoints_tools/README.md b/backend/generate_mcp_endpoints_tools/README.md new file mode 100644 index 0000000000..b19245fcdd --- /dev/null +++ b/backend/generate_mcp_endpoints_tools/README.md @@ -0,0 +1,35 @@ +## MCP Tools Generator + +The `generate_mcp_tools.py` script parses the OpenAPI specification and generates Rust code for MCP (Model Context Protocol) tools. + +### Setup + +```bash +cd backend/generate_mcp_endpoints_tools +pip install -r requirements.txt +``` + +### Usage + +```bash +python3 generate_mcp_tools.py +``` + +The script will: +1. Parse `backend/windmill-api/openapi.yaml` +2. Find all endpoints marked with `x-mcp-tool: true` +3. Generate `backend/windmill-api/src/mcp_tools.rs` with a const array of tools + +### Adding MCP Tools + +To mark an endpoint as an MCP tool, add `x-mcp-tool: true` to the operation in the OpenAPI spec. You can also add `x-mcp-instructions` to complete the description of the tool with instructions on how to correctly use the endpoint: + +```yaml +/w/{workspace}/scripts/list: + get: + x-mcp-tool: true + x-mcp-instructions: you should call that with this or that arg + summary: list scripts in workspace + operationId: listScripts + # ... rest of endpoint definition +``` \ No newline at end of file diff --git a/backend/generate_mcp_endpoints_tools/generate_mcp_tools.py b/backend/generate_mcp_endpoints_tools/generate_mcp_tools.py new file mode 100644 index 0000000000..a20ef45f06 --- /dev/null +++ b/backend/generate_mcp_endpoints_tools/generate_mcp_tools.py @@ -0,0 +1,307 @@ +#!/usr/bin/env python3 +""" +Script to parse the OpenAPI YAML file and generate Rust code with MCP tools. +Searches for endpoints tagged with 'x-mcp-tool: true' and creates a const array. +""" + +import json +import sys +from pathlib import Path +from typing import Dict, List, Any, Optional + +def load_openapi_spec(file_path: str) -> Dict[str, Any]: + """Load and parse the OpenAPI YAML specification.""" + try: + import yaml + with open(file_path, 'r', encoding='utf-8') as f: + return yaml.safe_load(f) + except ImportError: + print("PyYAML not found. Please install it with: pip install PyYAML", file=sys.stderr) + sys.exit(1) + except Exception as e: + print(f"Error loading OpenAPI spec: {e}", file=sys.stderr) + sys.exit(1) + +def extract_separate_schemas(parameters: List[Dict[str, Any]], request_body: Optional[Dict[str, Any]], spec: Dict[str, Any]) -> tuple: + """Extract separate schemas for path parameters, query parameters, and request body.""" + path_params_schema = { + "type": "object", + "properties": {}, + "required": [] + } + + query_params_schema = { + "type": "object", + "properties": {}, + "required": [] + } + + body_schema = None + + # Process parameters + for param in parameters: + # Resolve $ref if present + if '$ref' in param: + param = resolve_schema_refs(param, spec) + + param_name = param.get('name', '') + param_schema = param.get('schema', {'type': 'string'}) + param_required = param.get('required', False) + param_description = param.get('description', '') + param_in = param.get('in', 'query') + + # Resolve any refs in the parameter schema + param_schema = resolve_schema_refs(param_schema, spec) + + # Add description if available + if param_description: + param_schema = dict(param_schema) + param_schema['description'] = param_description + + # Route to appropriate schema based on parameter location + if param_in == 'path': + # Skip 'workspace' path parameter as it's automatically provided by the MCP context + if param_name != 'workspace': + path_params_schema['properties'][param_name] = param_schema + if param_required: + path_params_schema['required'].append(param_name) + elif param_in == 'query': + query_params_schema['properties'][param_name] = param_schema + if param_required: + query_params_schema['required'].append(param_name) + + # Process request body if present + if request_body: + body_schema = extract_request_body_schema(request_body, spec) + + # Return None for empty schemas + path_params_schema = path_params_schema if path_params_schema['properties'] else None + query_params_schema = query_params_schema if query_params_schema['properties'] else None + + return (path_params_schema, query_params_schema, body_schema) + +def resolve_ref(ref_path: str, spec: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """Resolve a $ref path to the actual schema definition.""" + if not ref_path.startswith('#/'): + return None + + # Remove the '#/' prefix and split by '/' + path_parts = ref_path[2:].split('/') + + # Navigate through the spec following the path + current = spec + for part in path_parts: + if isinstance(current, dict) and part in current: + current = current[part] + else: + return None + + return current if isinstance(current, dict) else None + +def resolve_schema_refs(schema: Dict[str, Any], spec: Dict[str, Any]) -> Dict[str, Any]: + """Recursively resolve all $ref references in a schema.""" + if not isinstance(schema, dict): + return schema + + # If this is a $ref, resolve it + if '$ref' in schema: + ref_path = schema['$ref'] + resolved = resolve_ref(ref_path, spec) + if resolved: + # Recursively resolve any refs in the resolved schema + return resolve_schema_refs(resolved, spec) + else: + print(f"Warning: Could not resolve $ref: {ref_path}") + return schema + + # Recursively process all values in the schema + resolved_schema = {} + for key, value in schema.items(): + if isinstance(value, dict): + resolved_schema[key] = resolve_schema_refs(value, spec) + elif isinstance(value, list): + resolved_schema[key] = [ + resolve_schema_refs(item, spec) if isinstance(item, dict) else item + for item in value + ] + else: + resolved_schema[key] = value + + return resolved_schema + +def extract_request_body_schema(request_body: Dict[str, Any], spec: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """Extract request body schema from OpenAPI requestBody definition and resolve refs.""" + if not request_body: + return None + + content = request_body.get('content', {}) + json_content = content.get('application/json', {}) + schema = json_content.get('schema', {}) + + if schema: + # Resolve any $ref references in the schema + return resolve_schema_refs(schema, spec) + + return None + +def http_method_to_rust(method: str) -> str: + """Convert HTTP method string to Rust http::Method enum.""" + method_map = { + 'get': 'http::Method::GET', + 'post': 'http::Method::POST', + 'put': 'http::Method::PUT', + 'delete': 'http::Method::DELETE', + 'patch': 'http::Method::PATCH', + 'head': 'http::Method::HEAD', + 'options': 'http::Method::OPTIONS' + } + return method_map.get(method.lower(), f'http::Method::{method.upper()}') + +def schema_to_rust_value(schema: Optional[Dict[str, Any]]) -> str: + """Convert a schema dict to a Rust serde_json::json! expression.""" + if schema is None: + return "None" + return f"Some(serde_json::json!({json.dumps(schema, indent=8)}))" + +def find_mcp_tools(spec: Dict[str, Any]) -> List[Dict[str, Any]]: + """Find all endpoints marked with x-mcp-tool: true.""" + tools = [] + paths = spec.get('paths', {}) + + for path, path_item in paths.items(): + for method, operation in path_item.items(): + if isinstance(operation, dict) and operation.get('x-mcp-tool') is True: + # Extract tool information + tool = { + 'name': operation.get('operationId', f"{method}_{path.replace('/', '_').replace('{', '').replace('}', '')}"), + 'description': operation.get('summary', operation.get('description', f'{method.upper()} {path}')), + 'instructions': operation.get('x-mcp-instructions', ''), + 'path': path, + 'method': method.upper(), + 'parameters': operation.get('parameters', []), + 'requestBody': operation.get('requestBody'), + } + tools.append(tool) + + return tools + +def generate_rust_code(tools: List[Dict[str, Any]], spec: Dict[str, Any]) -> str: + """Generate the complete Rust code with MCP tools.""" + if not tools: + return """// No MCP tools found in the OpenAPI specification + +use std::borrow::Cow; + +#[derive(Debug, Clone)] +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: http::Method, + pub path_params_schema: Option, + pub query_params_schema: Option, + pub body_schema: Option, +} + +pub fn all_tools() -> Vec { + vec![] +} +""" + + tool_definitions = [] + + for tool in tools: + tool_name = tool['name'] + description = tool['description'] + instructions = tool['instructions'] + path = tool['path'] + method = http_method_to_rust(tool['method']) + + # Generate separate schemas + path_params_schema, query_params_schema, body_schema = extract_separate_schemas( + tool['parameters'], tool['requestBody'], spec + ) + + path_params_rust = schema_to_rust_value(path_params_schema) + query_params_rust = schema_to_rust_value(query_params_schema) + body_schema_rust = schema_to_rust_value(body_schema) + + # Generate tool definition + tool_def = f""" EndpointTool {{ + name: Cow::Borrowed("{tool_name}"), + description: Cow::Borrowed("{description}"), + instructions: Cow::Borrowed("{instructions}"), + path: Cow::Borrowed("{path}"), + method: {method}, + path_params_schema: {path_params_rust}, + query_params_schema: {query_params_rust}, + body_schema: {body_schema_rust}, + }}""" + tool_definitions.append(tool_def) + + # Combine everything + tool_definitions_str = ",\n".join(tool_definitions) + + rust_code = f"""// Auto-generated MCP tools from OpenAPI specification +// This file is generated by generate_mcp_tools.py - DO NOT EDIT MANUALLY + +use std::borrow::Cow; + +#[derive(Debug, Clone)] +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: http::Method, + pub path_params_schema: Option, + pub query_params_schema: Option, + pub body_schema: Option, +}} + +pub fn all_tools() -> Vec {{ + vec![ +{tool_definitions_str} + ] +}} +""" + + return rust_code + +def main(): + """Main function to parse OpenAPI and generate Rust code.""" + script_dir = Path(__file__).parent + backend_dir = script_dir.parent + openapi_file = backend_dir / "windmill-api" / "openapi.yaml" + output_file = backend_dir / "windmill-api" / "src" / "mcp_tools.rs" + + if not openapi_file.exists(): + print(f"OpenAPI file not found: {openapi_file}", file=sys.stderr) + sys.exit(1) + + print(f"Loading OpenAPI specification from: {openapi_file}") + spec = load_openapi_spec(str(openapi_file)) + + print("Searching for endpoints with x-mcp-tool: true...") + tools = find_mcp_tools(spec) + + if tools: + print(f"Found {len(tools)} MCP tool(s):") + for tool in tools: + print(f" - {tool['name']}: {tool['method']} {tool['path']}") + else: + print("No MCP tools found (no endpoints with x-mcp-tool: true)") + + print(f"Generating Rust code...") + rust_code = generate_rust_code(tools, spec) + + print(f"Writing to: {output_file}") + output_file.parent.mkdir(parents=True, exist_ok=True) + with open(output_file, 'w', encoding='utf-8') as f: + f.write(rust_code) + + print("Done!") + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/backend/generate_mcp_endpoints_tools/requirements.txt b/backend/generate_mcp_endpoints_tools/requirements.txt new file mode 100644 index 0000000000..043876c0b8 --- /dev/null +++ b/backend/generate_mcp_endpoints_tools/requirements.txt @@ -0,0 +1 @@ +PyYAML>=6.0 \ No newline at end of file diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 831af7111a..f29088f0b9 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -3068,11 +3068,13 @@ paths: post: summary: create variable operationId: createVariable + x-mcp-tool: true tags: - variable parameters: - $ref: "#/components/parameters/WorkspaceId" - name: already_encrypted + description: whether the variable is already encrypted (default false) in: query schema: type: boolean @@ -3118,6 +3120,7 @@ paths: delete: summary: delete variable operationId: deleteVariable + x-mcp-tool: true tags: - variable parameters: @@ -3135,12 +3138,14 @@ paths: post: summary: update variable operationId: updateVariable + x-mcp-tool: true tags: - variable parameters: - $ref: "#/components/parameters/WorkspaceId" - $ref: "#/components/parameters/Path" - name: already_encrypted + description: whether the variable is already encrypted (default false) in: query schema: type: boolean @@ -3163,6 +3168,7 @@ paths: get: summary: get variable operationId: getVariable + x-mcp-tool: true tags: - variable parameters: @@ -3227,11 +3233,13 @@ paths: get: summary: list variables operationId: listVariable + x-mcp-tool: true tags: - variable parameters: - $ref: "#/components/parameters/WorkspaceId" - name: path_start + description: filter variables by path prefix in: query schema: type: string @@ -3832,11 +3840,13 @@ paths: post: summary: create resource operationId: createResource + x-mcp-tool: true tags: - resource parameters: - $ref: "#/components/parameters/WorkspaceId" - name: update_if_exists + description: update the resource if it already exists (default false) in: query schema: type: boolean @@ -3859,6 +3869,7 @@ paths: delete: summary: delete resource operationId: deleteResource + x-mcp-tool: true tags: - resource parameters: @@ -3876,6 +3887,7 @@ paths: post: summary: update resource operationId: updateResource + x-mcp-tool: true tags: - resource parameters: @@ -3926,6 +3938,7 @@ paths: get: summary: get resource operationId: getResource + x-mcp-tool: true tags: - resource parameters: @@ -3998,6 +4011,7 @@ paths: get: summary: list resources operationId: listResource + x-mcp-tool: true tags: - resource parameters: @@ -4015,6 +4029,7 @@ paths: schema: type: string - name: path_start + description: filter resources by path prefix in: query schema: type: string @@ -4650,6 +4665,7 @@ paths: get: summary: list all scripts operationId: listScripts + x-mcp-tool: true tags: - script parameters: @@ -5025,6 +5041,7 @@ paths: get: summary: get script by path operationId: getScriptByPath + x-mcp-tool: true tags: - script parameters: @@ -5615,6 +5632,7 @@ paths: get: summary: list all flows operationId: listFlows + x-mcp-tool: true tags: - flow parameters: @@ -5796,6 +5814,7 @@ paths: get: summary: get flow by path operationId: getFlowByPath + x-mcp-tool: true tags: - flow parameters: @@ -7246,6 +7265,7 @@ paths: get: summary: list all queued jobs operationId: listQueue + x-mcp-tool: true tags: - job parameters: @@ -7585,6 +7605,7 @@ paths: get: summary: list all jobs operationId: listJobs + x-mcp-tool: true tags: - job parameters: @@ -8545,6 +8566,11 @@ paths: post: summary: create schedule operationId: createSchedule + x-mcp-tool: true + x-mcp-instructions: | + Creates a new schedule. + The schedule should include seconds. + You should get the schema of the script or flow before creating the schedule to correctly specify the arguments needed. tags: - schedule parameters: @@ -8568,6 +8594,11 @@ paths: post: summary: update schedule operationId: updateSchedule + x-mcp-tool: true + x-mcp-instructions: | + Updates a schedule. + The schedule should include seconds. + You should get the schema of the script or flow before updating the schedule to correctly specify the arguments needed. tags: - schedule parameters: @@ -8622,6 +8653,7 @@ paths: delete: summary: delete schedule operationId: deleteSchedule + x-mcp-tool: true tags: - schedule parameters: @@ -8639,6 +8671,7 @@ paths: get: summary: get schedule operationId: getSchedule + x-mcp-tool: true tags: - schedule parameters: @@ -8673,6 +8706,7 @@ paths: get: summary: list schedules operationId: listSchedules + x-mcp-tool: true tags: - schedule parameters: @@ -8686,10 +8720,12 @@ paths: schema: type: string - name: is_flow + description: filter schedules by whether they target a flow in: query schema: type: boolean - name: path_start + description: filter schedules by path prefix in: query schema: type: string @@ -11294,6 +11330,7 @@ paths: get: summary: list workers operationId: listWorkers + x-mcp-tool: true tags: - worker parameters: @@ -13375,6 +13412,7 @@ components: name: publication in: path required: true + description: The name of the publication schema: type: string VersionId: @@ -14125,6 +14163,7 @@ components: ScriptArgs: type: object + description: The arguments to pass to the script or flow additionalProperties: {} Input: @@ -14627,18 +14666,25 @@ components: properties: path: type: string + description: The path to the variable value: type: string + description: The value of the variable is_secret: type: boolean + description: Whether the variable is a secret description: type: string + description: The description of the variable account: type: integer + description: The account identifier is_oauth: type: boolean + description: Whether the variable is an OAuth variable expires_at: type: string + description: The expiration date of the variable format: date-time required: - path @@ -14651,12 +14697,16 @@ components: properties: path: type: string + description: The path to the variable value: type: string + description: The new value of the variable is_secret: type: boolean + description: Whether the variable is a secret description: type: string + description: The new description of the variable AuditLog: type: object @@ -14989,11 +15039,14 @@ components: properties: path: type: string + description: The path to the resource value: {} description: type: string + description: The description of the resource resource_type: type: string + description: The resource_type associated with the resource required: - path - value @@ -15004,9 +15057,14 @@ components: properties: path: type: string + description: The path to the resource description: type: string + description: The new description of the resource value: {} + resource_type: + type: string + description: The new resource_type to be associated with the resource Resource: type: object @@ -15215,54 +15273,78 @@ components: properties: path: type: string + description: The path where the schedule will be created schedule: type: string + description: The cron schedule to trigger the script or flow. Should include seconds. timezone: type: string + description: The timezone to use for the cron schedule script_path: type: string + description: The path to the script or flow to trigger is_flow: type: boolean + description: Whether the schedule is for a flow args: $ref: "#/components/schemas/ScriptArgs" + description: The arguments to pass to the script or flow enabled: type: boolean + description: Whether the schedule is enabled on_failure: # a reference to a script path, flow path, or webhook (script/, flow/) type: string + description: The path to the script or flow to trigger on failure on_failure_times: type: number + description: The number of times to retry on failure on_failure_exact: type: boolean + description: Whether the schedule should only run on the exact time on_failure_extra_args: $ref: "#/components/schemas/ScriptArgs" + description: The arguments to pass to the script or flow on failure on_recovery: type: string + description: The path to the script or flow to trigger on recovery on_recovery_times: type: number + description: The number of times to retry on recovery on_recovery_extra_args: $ref: "#/components/schemas/ScriptArgs" + description: The arguments to pass to the script or flow on recovery on_success: type: string + description: The path to the script or flow to trigger on success on_success_extra_args: $ref: "#/components/schemas/ScriptArgs" + description: The arguments to pass to the script or flow on success ws_error_handler_muted: type: boolean + description: Whether the WebSocket error handler is muted retry: $ref: "../../openflow.openapi.yaml#/components/schemas/Retry" + description: The retry configuration for the schedule no_flow_overlap: type: boolean + description: Whether the schedule should not run if a flow is already running summary: type: string + description: The summary of the schedule description: type: string + description: The description of the schedule tag: type: string + description: The tag of the schedule paused_until: type: string + description: The date and time the schedule will be paused until format: date-time cron_version: type: string + description: The version of the cron schedule to use (last is v2) required: - path - schedule @@ -15276,51 +15358,69 @@ components: properties: schedule: type: string + description: The cron schedule to trigger the script or flow. Should include seconds. timezone: type: string + description: The timezone to use for the cron schedule args: $ref: "#/components/schemas/ScriptArgs" + description: The arguments to pass to the script or flow on_failure: # a reference to a script path, flow path, or webhook (script/, flow/) type: string + description: The path to the script or flow to trigger on failure on_failure_times: type: number + description: The number of times to retry on failure on_failure_exact: type: boolean + description: Whether the schedule should only run on the exact time on_failure_extra_args: $ref: "#/components/schemas/ScriptArgs" + description: The arguments to pass to the script or flow on failure on_recovery: type: string + description: The path to the script or flow to trigger on recovery on_recovery_times: type: number + description: The number of times to retry on recovery on_recovery_extra_args: $ref: "#/components/schemas/ScriptArgs" + description: The arguments to pass to the script or flow on recovery on_success: type: string + description: The path to the script or flow to trigger on success on_success_extra_args: $ref: "#/components/schemas/ScriptArgs" + description: The arguments to pass to the script or flow on success ws_error_handler_muted: type: boolean + description: Whether the WebSocket error handler is muted retry: $ref: "../../openflow.openapi.yaml#/components/schemas/Retry" + description: The retry configuration for the schedule no_flow_overlap: type: boolean + description: Whether the schedule should not run if a flow is already running summary: type: string + description: The summary of the schedule description: type: string + description: The description of the schedule tag: type: string + description: The tag of the schedule paused_until: type: string + description: The date and time the schedule will be paused until format: date-time cron_version: type: string + description: The version of the cron schedule to use (last is v2) required: - schedule - timezone - - script_path - - is_flow - args TriggerExtraProperty: diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index bf04b515a5..7fa7c61386 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -854,6 +854,22 @@ impl From for Authed { } } +impl From for ApiAuthed { + fn from(value: Authed) -> Self { + Self { + email: value.email, + username: value.username, + is_admin: value.is_admin, + is_operator: value.is_operator, + groups: value.groups, + folders: value.folders, + scopes: value.scopes, + username_override: None, // Authed doesn't have this field, so default to None + token_prefix: value.token_prefix, + } + } +} + impl From<&ApiAuthed> for AuditAuthor { fn from(value: &ApiAuthed) -> Self { Self { diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 018a95d469..fe9b7af3f7 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -21,6 +21,8 @@ use crate::smtp_server_oss::SmtpServer; #[cfg(feature = "mcp")] use crate::mcp::{extract_and_store_workspace_id, setup_mcp_server, shutdown_mcp_server}; #[cfg(feature = "mcp")] +mod mcp_utils; +#[cfg(feature = "mcp")] use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; use crate::tracing_init::MyOnFailure; @@ -201,6 +203,8 @@ mod workspaces_oss; #[cfg(feature = "mcp")] mod mcp; +#[cfg(feature = "mcp")] +mod mcp_tools; pub const DEFAULT_BODY_LIMIT: usize = 2097152 * 100; // 200MB diff --git a/backend/windmill-api/src/mcp.rs b/backend/windmill-api/src/mcp.rs index ca345ba934..eba9907bb4 100644 --- a/backend/windmill-api/src/mcp.rs +++ b/backend/windmill-api/src/mcp.rs @@ -2,7 +2,7 @@ use std::borrow::Cow; use std::collections::HashMap; use std::sync::Arc; -use axum::body::to_bytes; +use axum::body::{to_bytes}; use axum::Router; use axum::{extract::Path, http::Request, middleware::Next, response::Response}; use rmcp::{ @@ -32,6 +32,10 @@ use rmcp::transport::streamable_http_server::{ }; use windmill_common::utils::{query_elems_from_hub, StripPath}; +use crate::mcp_tools::all_tools; +use crate::mcp_utils::{endpoint_tools_to_mcp_tools, call_endpoint_tool}; + + /// Transforms the path for workspace scripts/flows. /// /// This function takes a path and a type string. @@ -480,54 +484,6 @@ impl Runner { Ok(hub_response.asks) } - /// Transforms a value if it's an object. - /// - /// This function takes a key and a value, and a schema object. - /// If the value is a string that starts with "$res:", it returns the value as is. - /// Otherwise, it checks if the key is defined in the schema and if it's an object type. - /// If it is, it transforms the value to a string. This is because some clients do not support object types. - /// # Parameters - /// - `key`: The key of the value to transform. - /// - `value`: The value to transform. - /// - `schema_obj`: The schema object. - /// - /// # Returns - /// - `Value`: The transformed value. - fn transform_value_if_object( - key: &str, - value: &Value, - schema_obj: &Option, - ) -> Value { - if value.is_string() && value.as_str().unwrap().starts_with("$res:") { - return value.clone(); - } - - let schema_obj = match schema_obj { - Some(s) => s, - None => return value.clone(), - }; - - // Check if property is defined in schema and is an object type - let is_obj_type = match schema_obj.properties.get(key) { - Some(property) => { - let prop_type = property.get("type").and_then(|t| t.as_str()); - prop_type == Some("object") - } - None => false, - }; - - // If it's an object type and we received a string, try to parse it - if is_obj_type && value.is_string() { - if let Some(str_val) = value.as_str() { - if let Ok(obj_val) = serde_json::from_str::(str_val) { - return obj_val; - } - } - } - - value.clone() - } - /// Reverses the transformation of a key. /// /// This function takes a transformed key and a schema object. @@ -626,17 +582,6 @@ impl Runner { for (_key, prop_value) in schema_obj.properties.iter_mut() { if let serde_json::Value::Object(prop_map) = prop_value { - // transform object properties to string because some client does not support object, might change in the future - if let Some(type_value) = prop_map.get("type") { - if let serde_json::Value::String(type_str) = type_value { - if type_str == "object" { - prop_map.insert( - "type".to_string(), - serde_json::Value::String("string".to_string()), - ); - } - } - } // 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 serde_json::Value::String(format_str) = format_value { @@ -646,10 +591,7 @@ impl Runner { let resource_type = resources_types .iter() .find(|rt| rt.name == resource_type_key); - let resource_type_obj = resource_type.cloned().unwrap_or_else(|| { - tracing::info!("Resource type not found: {}", resource_type_key); - ResourceType { name: resource_type_key.clone(), description: None } - }); + let resource_type_obj = resource_type.cloned(); if !resources_cache.contains_key(&resource_type_key) { let available_resources = Runner::inner_get_resources( @@ -677,18 +619,21 @@ impl Runner { if let Some(resource_cache) = resources_cache.get(&resource_type_key) { let resources_count = resource_cache.len(); - let description = 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." - } - ); + 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(), serde_json::Value::String("string".to_string()), @@ -842,6 +787,7 @@ impl Runner { } } + impl ServerHandler for Runner { /// Handles the `CallTool` request from the MCP client. /// @@ -907,6 +853,19 @@ impl ServerHandler for Runner { }) .map(|w_id| w_id.0.clone())?; + // Check if this is a generated endpoint tool + let endpoint_tools = all_tools(); + 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?; + 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) = Runner::reverse_transform(&request.name).unwrap_or_default(); @@ -933,10 +892,7 @@ impl ServerHandler for Runner { for (k, v) in map { // need to transform back the key without invalid characters to the original key let original_key = Runner::reverse_transform_key(&k, &schema_obj); - - // object properties are transformed to string because some client does not support object, might change in the future - let transformed_v = Runner::transform_value_if_object(&k, &v, &schema_obj); - args_hash.insert(original_key, to_raw_value(&transformed_v)); + args_hash.insert(original_key, to_raw_value(&v)); } windmill_queue::PushArgsOwned { extra: None, args: args_hash } } else { @@ -1132,6 +1088,11 @@ impl ServerHandler for Runner { ); } + // Add endpoint tools from the generated MCP tools + let endpoint_tools = all_tools(); + let mcp_tools_converted = endpoint_tools_to_mcp_tools(endpoint_tools); + tools.extend(mcp_tools_converted); + Ok(ListToolsResult { tools, next_cursor: None }) } diff --git a/backend/windmill-api/src/mcp_tools.rs b/backend/windmill-api/src/mcp_tools.rs new file mode 100644 index 0000000000..b106a97df9 --- /dev/null +++ b/backend/windmill-api/src/mcp_tools.rs @@ -0,0 +1,1161 @@ +// Auto-generated MCP tools from OpenAPI specification +// This file is generated by generate_mcp_tools.py - DO NOT EDIT MANUALLY + +use std::borrow::Cow; + +#[derive(Debug, Clone)] +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: http::Method, + pub path_params_schema: Option, + pub query_params_schema: Option, + pub body_schema: Option, +} + +pub fn all_tools() -> Vec { + vec![ + EndpointTool { + name: Cow::Borrowed("createVariable"), + description: Cow::Borrowed("create variable"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/variables/create"), + method: http::Method::POST, + path_params_schema: None, + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "already_encrypted": { + "type": "boolean", + "description": "whether the variable is already encrypted (default false)" + } + }, + "required": [] +})), + body_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "The path to the variable" + }, + "value": { + "type": "string", + "description": "The value of the variable" + }, + "is_secret": { + "type": "boolean", + "description": "Whether the variable is a secret" + }, + "description": { + "type": "string", + "description": "The description of the variable" + }, + "account": { + "type": "integer", + "description": "The account identifier" + }, + "is_oauth": { + "type": "boolean", + "description": "Whether the variable is an OAuth variable" + }, + "expires_at": { + "type": "string", + "description": "The expiration date of the variable", + "format": "date-time" + } + }, + "required": [ + "path", + "value", + "is_secret", + "description" + ] +})), + }, + EndpointTool { + name: Cow::Borrowed("deleteVariable"), + description: Cow::Borrowed("delete variable"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/variables/delete/{path}"), + method: http::Method::DELETE, + path_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string" + } + }, + "required": [ + "path" + ] +})), + query_params_schema: None, + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("updateVariable"), + description: Cow::Borrowed("update variable"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/variables/update/{path}"), + method: http::Method::POST, + path_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string" + } + }, + "required": [ + "path" + ] +})), + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "already_encrypted": { + "type": "boolean", + "description": "whether the variable is already encrypted (default false)" + } + }, + "required": [] +})), + body_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "The path to the variable" + }, + "value": { + "type": "string", + "description": "The new value of the variable" + }, + "is_secret": { + "type": "boolean", + "description": "Whether the variable is a secret" + }, + "description": { + "type": "string", + "description": "The new description of the variable" + } + } +})), + }, + EndpointTool { + name: Cow::Borrowed("getVariable"), + description: Cow::Borrowed("get variable"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/variables/get/{path}"), + method: http::Method::GET, + path_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string" + } + }, + "required": [ + "path" + ] +})), + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "decrypt_secret": { + "type": "boolean", + "description": "ask to decrypt secret if this variable is secret\n(if not secret no effect, default: true)\n" + }, + "include_encrypted": { + "type": "boolean", + "description": "ask to include the encrypted value if secret and decrypt secret is not true (default: false)\n" + } + }, + "required": [] +})), + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("listVariable"), + description: Cow::Borrowed("list variables"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/variables/list"), + method: http::Method::GET, + path_params_schema: None, + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path_start": { + "type": "string", + "description": "filter variables by path prefix" + }, + "page": { + "type": "integer", + "description": "which page to return (start at 1, default 1)" + }, + "per_page": { + "type": "integer", + "description": "number of items to return for a given page (default 30, max 100)" + } + }, + "required": [] +})), + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("createResource"), + description: Cow::Borrowed("create resource"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/resources/create"), + method: http::Method::POST, + path_params_schema: None, + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "update_if_exists": { + "type": "boolean", + "description": "update the resource if it already exists (default false)" + } + }, + "required": [] +})), + body_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "The path to the resource" + }, + "value": {}, + "description": { + "type": "string", + "description": "The description of the resource" + }, + "resource_type": { + "type": "string", + "description": "The resource_type associated with the resource" + } + }, + "required": [ + "path", + "value", + "resource_type" + ] +})), + }, + EndpointTool { + name: Cow::Borrowed("deleteResource"), + description: Cow::Borrowed("delete resource"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/resources/delete/{path}"), + method: http::Method::DELETE, + path_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string" + } + }, + "required": [ + "path" + ] +})), + query_params_schema: None, + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("updateResource"), + description: Cow::Borrowed("update resource"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/resources/update/{path}"), + method: http::Method::POST, + path_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string" + } + }, + "required": [ + "path" + ] +})), + query_params_schema: None, + body_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "The path to the resource" + }, + "description": { + "type": "string", + "description": "The new description of the resource" + }, + "value": {}, + "resource_type": { + "type": "string", + "description": "The new resource_type to be associated with the resource" + } + } +})), + }, + EndpointTool { + name: Cow::Borrowed("getResource"), + description: Cow::Borrowed("get resource"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/resources/get/{path}"), + method: http::Method::GET, + path_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string" + } + }, + "required": [ + "path" + ] +})), + query_params_schema: None, + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("listResource"), + description: Cow::Borrowed("list resources"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/resources/list"), + method: http::Method::GET, + path_params_schema: None, + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "page": { + "type": "integer", + "description": "which page to return (start at 1, default 1)" + }, + "per_page": { + "type": "integer", + "description": "number of items to return for a given page (default 30, max 100)" + }, + "resource_type": { + "type": "string", + "description": "resource_types to list from, separated by ','," + }, + "resource_type_exclude": { + "type": "string", + "description": "resource_types to not list from, separated by ','," + }, + "path_start": { + "type": "string", + "description": "filter resources by path prefix" + } + }, + "required": [] +})), + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("listScripts"), + description: Cow::Borrowed("list all scripts"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/scripts/list"), + method: http::Method::GET, + path_params_schema: None, + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "page": { + "type": "integer", + "description": "which page to return (start at 1, default 1)" + }, + "per_page": { + "type": "integer", + "description": "number of items to return for a given page (default 30, max 100)" + }, + "order_desc": { + "type": "boolean", + "description": "order by desc order (default true)" + }, + "created_by": { + "type": "string", + "description": "mask to filter exact matching user creator" + }, + "path_start": { + "type": "string", + "description": "mask to filter matching starting path" + }, + "path_exact": { + "type": "string", + "description": "mask to filter exact matching path" + }, + "first_parent_hash": { + "type": "string", + "description": "mask to filter scripts whom first direct parent has exact hash" + }, + "last_parent_hash": { + "type": "string", + "description": "mask to filter scripts whom last parent in the chain has exact hash.\nBeware that each script stores only a limited number of parents. Hence\nthe last parent hash for a script is not necessarily its top-most parent.\nTo find the top-most parent you will have to jump from last to last hash\n until finding the parent\n" + }, + "parent_hash": { + "type": "string", + "description": "is the hash present in the array of stored parent hashes for this script.\nThe same warning applies than for last_parent_hash. A script only store a\nlimited number of direct parent\n" + }, + "show_archived": { + "type": "boolean", + "description": "(default false)\nshow only the archived files.\nwhen multiple archived hash share the same path, only the ones with the latest create_at\nare\ned.\n" + }, + "include_without_main": { + "type": "boolean", + "description": "(default false)\ninclude scripts without an exported main function\n" + }, + "include_draft_only": { + "type": "boolean", + "description": "(default false)\ninclude scripts that have no deployed version\n" + }, + "is_template": { + "type": "boolean", + "description": "(default regardless)\nif true show only the templates\nif false show only the non templates\nif not defined, show all regardless of if the script is a template\n" + }, + "kinds": { + "type": "string", + "description": "(default regardless)\nscript kinds to filter, split by comma\n" + }, + "starred_only": { + "type": "boolean", + "description": "(default false)\nshow only the starred items\n" + }, + "with_deployment_msg": { + "type": "boolean", + "description": "(default false)\ninclude deployment message\n" + }, + "languages": { + "type": "string", + "description": "Filter to only include scripts written in the given languages.\nAccepts multiple values as a comma-separated list.\n" + } + }, + "required": [] +})), + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("getScriptByPath"), + description: Cow::Borrowed("get script by path"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/scripts/get/p/{path}"), + method: http::Method::GET, + path_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string" + } + }, + "required": [ + "path" + ] +})), + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "with_starred_info": { + "type": "boolean" + } + }, + "required": [] +})), + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("listFlows"), + description: Cow::Borrowed("list all flows"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/flows/list"), + method: http::Method::GET, + path_params_schema: None, + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "page": { + "type": "integer", + "description": "which page to return (start at 1, default 1)" + }, + "per_page": { + "type": "integer", + "description": "number of items to return for a given page (default 30, max 100)" + }, + "order_desc": { + "type": "boolean", + "description": "order by desc order (default true)" + }, + "created_by": { + "type": "string", + "description": "mask to filter exact matching user creator" + }, + "path_start": { + "type": "string", + "description": "mask to filter matching starting path" + }, + "path_exact": { + "type": "string", + "description": "mask to filter exact matching path" + }, + "show_archived": { + "type": "boolean", + "description": "(default false)\nshow only the archived files.\nwhen multiple archived hash share the same path, only the ones with the latest create_at\nare displayed.\n" + }, + "starred_only": { + "type": "boolean", + "description": "(default false)\nshow only the starred items\n" + }, + "include_draft_only": { + "type": "boolean", + "description": "(default false)\ninclude items that have no deployed version\n" + }, + "with_deployment_msg": { + "type": "boolean", + "description": "(default false)\ninclude deployment message\n" + } + }, + "required": [] +})), + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("getFlowByPath"), + description: Cow::Borrowed("get flow by path"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/flows/get/{path}"), + method: http::Method::GET, + path_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string" + } + }, + "required": [ + "path" + ] +})), + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "with_starred_info": { + "type": "boolean" + } + }, + "required": [] +})), + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("listQueue"), + description: Cow::Borrowed("list all queued jobs"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/jobs/queue/list"), + method: http::Method::GET, + path_params_schema: None, + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "order_desc": { + "type": "boolean", + "description": "order by desc order (default true)" + }, + "created_by": { + "type": "string", + "description": "mask to filter exact matching user creator" + }, + "parent_job": { + "type": "string", + "format": "uuid", + "description": "The parent job that is at the origin and responsible for the execution of this script if any" + }, + "worker": { + "type": "string", + "description": "worker this job was ran on" + }, + "script_path_exact": { + "type": "string", + "description": "mask to filter exact matching path" + }, + "script_path_start": { + "type": "string", + "description": "mask to filter matching starting path" + }, + "schedule_path": { + "type": "string", + "description": "mask to filter by schedule path" + }, + "script_hash": { + "type": "string", + "description": "mask to filter exact matching path" + }, + "started_before": { + "type": "string", + "format": "date-time", + "description": "filter on started before (inclusive) timestamp" + }, + "started_after": { + "type": "string", + "format": "date-time", + "description": "filter on started after (exclusive) timestamp" + }, + "success": { + "type": "boolean", + "description": "filter on successful jobs" + }, + "scheduled_for_before_now": { + "type": "boolean", + "description": "filter on jobs scheduled_for before now (hence waitinf for a worker)" + }, + "job_kinds": { + "type": "string", + "description": "filter on job kind (values 'preview', 'script', 'dependencies', 'flow') separated by," + }, + "suspended": { + "type": "boolean", + "description": "filter on suspended jobs" + }, + "running": { + "type": "boolean", + "description": "filter on running jobs" + }, + "args": { + "type": "string", + "description": "filter on jobs containing those args as a json subset (@> in postgres)" + }, + "result": { + "type": "string", + "description": "filter on jobs containing those result as a json subset (@> in postgres)" + }, + "allow_wildcards": { + "type": "boolean", + "description": "allow wildcards (*) in the filter of label, tag, worker" + }, + "tag": { + "type": "string", + "description": "filter on jobs with a given tag/worker group" + }, + "page": { + "type": "integer", + "description": "which page to return (start at 1, default 1)" + }, + "per_page": { + "type": "integer", + "description": "number of items to return for a given page (default 30, max 100)" + }, + "all_workspaces": { + "type": "boolean", + "description": "get jobs from all workspaces (only valid if request come from the `admins` workspace)" + }, + "is_not_schedule": { + "type": "boolean", + "description": "is not a scheduled job" + } + }, + "required": [] +})), + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("listJobs"), + description: Cow::Borrowed("list all jobs"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/jobs/list"), + method: http::Method::GET, + path_params_schema: None, + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "created_by": { + "type": "string", + "description": "mask to filter exact matching user creator" + }, + "label": { + "type": "string", + "description": "mask to filter exact matching job's label (job labels are completed jobs with as a result an object containing a string in the array at key 'wm_labels')" + }, + "worker": { + "type": "string", + "description": "worker this job was ran on" + }, + "parent_job": { + "type": "string", + "format": "uuid", + "description": "The parent job that is at the origin and responsible for the execution of this script if any" + }, + "script_path_exact": { + "type": "string", + "description": "mask to filter exact matching path" + }, + "script_path_start": { + "type": "string", + "description": "mask to filter matching starting path" + }, + "schedule_path": { + "type": "string", + "description": "mask to filter by schedule path" + }, + "script_hash": { + "type": "string", + "description": "mask to filter exact matching path" + }, + "started_before": { + "type": "string", + "format": "date-time", + "description": "filter on started before (inclusive) timestamp" + }, + "started_after": { + "type": "string", + "format": "date-time", + "description": "filter on started after (exclusive) timestamp" + }, + "created_before": { + "type": "string", + "format": "date-time", + "description": "filter on created before (inclusive) timestamp" + }, + "created_after": { + "type": "string", + "format": "date-time", + "description": "filter on created after (exclusive) timestamp" + }, + "created_or_started_before": { + "type": "string", + "format": "date-time", + "description": "filter on created_at for non non started job and started_at otherwise before (inclusive) timestamp" + }, + "running": { + "type": "boolean", + "description": "filter on running jobs" + }, + "scheduled_for_before_now": { + "type": "boolean", + "description": "filter on jobs scheduled_for before now (hence waitinf for a worker)" + }, + "created_or_started_after": { + "type": "string", + "format": "date-time", + "description": "filter on created_at for non non started job and started_at otherwise after (exclusive) timestamp" + }, + "created_or_started_after_completed_jobs": { + "type": "string", + "format": "date-time", + "description": "filter on created_at for non non started job and started_at otherwise after (exclusive) timestamp but only for the completed jobs" + }, + "job_kinds": { + "type": "string", + "description": "filter on job kind (values 'preview', 'script', 'dependencies', 'flow') separated by," + }, + "suspended": { + "type": "boolean", + "description": "filter on suspended jobs" + }, + "args": { + "type": "string", + "description": "filter on jobs containing those args as a json subset (@> in postgres)" + }, + "tag": { + "type": "string", + "description": "filter on jobs with a given tag/worker group" + }, + "result": { + "type": "string", + "description": "filter on jobs containing those result as a json subset (@> in postgres)" + }, + "allow_wildcards": { + "type": "boolean", + "description": "allow wildcards (*) in the filter of label, tag, worker" + }, + "page": { + "type": "integer", + "description": "which page to return (start at 1, default 1)" + }, + "per_page": { + "type": "integer", + "description": "number of items to return for a given page (default 30, max 100)" + }, + "is_skipped": { + "type": "boolean", + "description": "is the job skipped" + }, + "is_flow_step": { + "type": "boolean", + "description": "is the job a flow step" + }, + "has_null_parent": { + "type": "boolean", + "description": "has null parent" + }, + "success": { + "type": "boolean", + "description": "filter on successful jobs" + }, + "all_workspaces": { + "type": "boolean", + "description": "get jobs from all workspaces (only valid if request come from the `admins` workspace)" + }, + "is_not_schedule": { + "type": "boolean", + "description": "is not a scheduled job" + } + }, + "required": [] +})), + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("createSchedule"), + description: Cow::Borrowed("create schedule"), + instructions: Cow::Borrowed("Creates a new schedule. +The schedule should include seconds. +You should get the schema of the script or flow before creating the schedule to correctly specify the arguments needed. +"), + path: Cow::Borrowed("/w/{workspace}/schedules/create"), + method: http::Method::POST, + path_params_schema: None, + query_params_schema: None, + body_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "The path where the schedule will be created" + }, + "schedule": { + "type": "string", + "description": "The cron schedule to trigger the script or flow. Should include seconds." + }, + "timezone": { + "type": "string", + "description": "The timezone to use for the cron schedule" + }, + "script_path": { + "type": "string", + "description": "The path to the script or flow to trigger" + }, + "is_flow": { + "type": "boolean", + "description": "Whether the schedule is for a flow" + }, + "args": { + "type": "object", + "description": "The arguments to pass to the script or flow", + "additionalProperties": {} + }, + "enabled": { + "type": "boolean", + "description": "Whether the schedule is enabled" + }, + "on_failure": { + "type": "string", + "description": "The path to the script or flow to trigger on failure" + }, + "on_failure_times": { + "type": "number", + "description": "The number of times to retry on failure" + }, + "on_failure_exact": { + "type": "boolean", + "description": "Whether the schedule should only run on the exact time" + }, + "on_failure_extra_args": { + "type": "object", + "description": "The arguments to pass to the script or flow", + "additionalProperties": {} + }, + "on_recovery": { + "type": "string", + "description": "The path to the script or flow to trigger on recovery" + }, + "on_recovery_times": { + "type": "number", + "description": "The number of times to retry on recovery" + }, + "on_recovery_extra_args": { + "type": "object", + "description": "The arguments to pass to the script or flow", + "additionalProperties": {} + }, + "on_success": { + "type": "string", + "description": "The path to the script or flow to trigger on success" + }, + "on_success_extra_args": { + "type": "object", + "description": "The arguments to pass to the script or flow", + "additionalProperties": {} + }, + "ws_error_handler_muted": { + "type": "boolean", + "description": "Whether the WebSocket error handler is muted" + }, + "retry": { + "$ref": "../../openflow.openapi.yaml#/components/schemas/Retry", + "description": "The retry configuration for the schedule" + }, + "no_flow_overlap": { + "type": "boolean", + "description": "Whether the schedule should not run if a flow is already running" + }, + "summary": { + "type": "string", + "description": "The summary of the schedule" + }, + "description": { + "type": "string", + "description": "The description of the schedule" + }, + "tag": { + "type": "string", + "description": "The tag of the schedule" + }, + "paused_until": { + "type": "string", + "description": "The date and time the schedule will be paused until", + "format": "date-time" + }, + "cron_version": { + "type": "string", + "description": "The version of the cron schedule to use (last is v2)" + } + }, + "required": [ + "path", + "schedule", + "timezone", + "script_path", + "is_flow", + "args" + ] +})), + }, + EndpointTool { + name: Cow::Borrowed("updateSchedule"), + description: Cow::Borrowed("update schedule"), + instructions: Cow::Borrowed("Updates a schedule. +The schedule should include seconds. +You should get the schema of the script or flow before updating the schedule to correctly specify the arguments needed. +"), + path: Cow::Borrowed("/w/{workspace}/schedules/update/{path}"), + method: http::Method::POST, + path_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string" + } + }, + "required": [ + "path" + ] +})), + query_params_schema: None, + body_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "schedule": { + "type": "string", + "description": "The cron schedule to trigger the script or flow. Should include seconds." + }, + "timezone": { + "type": "string", + "description": "The timezone to use for the cron schedule" + }, + "args": { + "type": "object", + "description": "The arguments to pass to the script or flow", + "additionalProperties": {} + }, + "on_failure": { + "type": "string", + "description": "The path to the script or flow to trigger on failure" + }, + "on_failure_times": { + "type": "number", + "description": "The number of times to retry on failure" + }, + "on_failure_exact": { + "type": "boolean", + "description": "Whether the schedule should only run on the exact time" + }, + "on_failure_extra_args": { + "type": "object", + "description": "The arguments to pass to the script or flow", + "additionalProperties": {} + }, + "on_recovery": { + "type": "string", + "description": "The path to the script or flow to trigger on recovery" + }, + "on_recovery_times": { + "type": "number", + "description": "The number of times to retry on recovery" + }, + "on_recovery_extra_args": { + "type": "object", + "description": "The arguments to pass to the script or flow", + "additionalProperties": {} + }, + "on_success": { + "type": "string", + "description": "The path to the script or flow to trigger on success" + }, + "on_success_extra_args": { + "type": "object", + "description": "The arguments to pass to the script or flow", + "additionalProperties": {} + }, + "ws_error_handler_muted": { + "type": "boolean", + "description": "Whether the WebSocket error handler is muted" + }, + "retry": { + "$ref": "../../openflow.openapi.yaml#/components/schemas/Retry", + "description": "The retry configuration for the schedule" + }, + "no_flow_overlap": { + "type": "boolean", + "description": "Whether the schedule should not run if a flow is already running" + }, + "summary": { + "type": "string", + "description": "The summary of the schedule" + }, + "description": { + "type": "string", + "description": "The description of the schedule" + }, + "tag": { + "type": "string", + "description": "The tag of the schedule" + }, + "paused_until": { + "type": "string", + "description": "The date and time the schedule will be paused until", + "format": "date-time" + }, + "cron_version": { + "type": "string", + "description": "The version of the cron schedule to use (last is v2)" + } + }, + "required": [ + "schedule", + "timezone", + "args" + ] +})), + }, + EndpointTool { + name: Cow::Borrowed("deleteSchedule"), + description: Cow::Borrowed("delete schedule"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/schedules/delete/{path}"), + method: http::Method::DELETE, + path_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string" + } + }, + "required": [ + "path" + ] +})), + query_params_schema: None, + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("getSchedule"), + description: Cow::Borrowed("get schedule"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/schedules/get/{path}"), + method: http::Method::GET, + path_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "path": { + "type": "string" + } + }, + "required": [ + "path" + ] +})), + query_params_schema: None, + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("listSchedules"), + description: Cow::Borrowed("list schedules"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/w/{workspace}/schedules/list"), + method: http::Method::GET, + path_params_schema: None, + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "page": { + "type": "integer", + "description": "which page to return (start at 1, default 1)" + }, + "per_page": { + "type": "integer", + "description": "number of items to return for a given page (default 30, max 100)" + }, + "args": { + "type": "string", + "description": "filter on jobs containing those args as a json subset (@> in postgres)" + }, + "path": { + "type": "string", + "description": "filter by path" + }, + "is_flow": { + "type": "boolean", + "description": "filter schedules by whether they target a flow" + }, + "path_start": { + "type": "string", + "description": "filter schedules by path prefix" + } + }, + "required": [] +})), + body_schema: None, + }, + EndpointTool { + name: Cow::Borrowed("listWorkers"), + description: Cow::Borrowed("list workers"), + instructions: Cow::Borrowed(""), + path: Cow::Borrowed("/workers/list"), + method: http::Method::GET, + path_params_schema: None, + query_params_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "page": { + "type": "integer", + "description": "which page to return (start at 1, default 1)" + }, + "per_page": { + "type": "integer", + "description": "number of items to return for a given page (default 30, max 100)" + }, + "ping_since": { + "type": "integer", + "description": "number of seconds the worker must have had a last ping more recent of (default to 300)" + } + }, + "required": [] +})), + body_schema: None, + } + ] +} diff --git a/backend/windmill-api/src/mcp_utils.rs b/backend/windmill-api/src/mcp_utils.rs new file mode 100644 index 0000000000..1bad72ad79 --- /dev/null +++ b/backend/windmill-api/src/mcp_utils.rs @@ -0,0 +1,241 @@ +use rmcp::{model::Tool, Error}; +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; +use crate::mcp_tools::EndpointTool; + +pub fn endpoint_tools_to_mcp_tools(endpoint_tools: Vec) -> Vec { + endpoint_tools.into_iter().map(|tool| endpoint_tool_to_mcp_tool(&tool)).collect() +} + +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); + + Tool { + name: tool.name.clone(), + description: Some(description.into()), + input_schema: Arc::new(combined_schema.as_object().unwrap().clone()), + annotations: Some(rmcp::model::ToolAnnotations { + title: Some(format!("{} {}", + match tool.method { + http::Method::GET => "GET", + http::Method::POST => "POST", + http::Method::PUT => "PUT", + http::Method::DELETE => "DELETE", + http::Method::PATCH => "PATCH", + _ => "UNKNOWN" + }, + tool.path + )), + read_only_hint: None, + destructive_hint: None, + idempotent_hint: None, + open_world_hint: None, + }), + } +} + +fn merge_schema_into( + combined_properties: &mut serde_json::Map, + combined_required: &mut Vec, + schema: &serde_json::Value, +) { + if let Some(props) = schema.get("properties").and_then(|p| p.as_object()) { + for (key, value) in props { + combined_properties.insert(key.clone(), value.clone()); + } + } + + if let Some(required) = schema.get("required").and_then(|r| r.as_array()) { + for req in required.iter().filter_map(|r| r.as_str()) { + combined_required.push(req.to_string()); + } + } +} + +pub async fn call_endpoint_tool( + tool: &EndpointTool, + args: serde_json::Value, + workspace_id: &str, + api_authed: &ApiAuthed, +) -> Result { + let args_map = match &args { + serde_json::Value::Object(map) => map, + _ => return Err(Error::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_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 status = response.status(); + let response_text = response.text().await.map_err(|e| { + Error::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(Error::internal_error( + format!("HTTP {} {}: {}", status.as_u16(), status.canonical_reason().unwrap_or(""), response_text), + None + )) + } +} + +fn substitute_path_params( + path: &str, + workspace_id: &str, + args_map: &serde_json::Map, + path_schema: &Option, +) -> Result { + let mut path_template = path.replace("{workspace}", workspace_id); + + if let Some(schema) = path_schema { + if let Some(props) = schema.get("properties").and_then(|p| p.as_object()) { + for (param_name, _) in props { + let placeholder = format!("{{{}}}", param_name); + match args_map.get(param_name) { + Some(param_value) => { + if let Some(str_val) = param_value.as_str() { + path_template = path_template.replace(&placeholder, str_val); + } + }, + None => { + tracing::warn!("Missing required path parameter: {}", param_name); + return Err(Error::invalid_params( + format!("Missing required path parameter: {}", param_name), + None + )); + } + } + } + } + } + + Ok(path_template) +} + +fn build_query_string( + args_map: &serde_json::Map, + query_schema: &Option, +) -> String { + let Some(schema) = query_schema else { return String::new() }; + let Some(props) = schema.get("properties").and_then(|p| p.as_object()) else { return String::new() }; + + let query_params: Vec = props + .keys() + .filter_map(|param_name| { + args_map.get(param_name) + .filter(|v| !v.is_null()) + .map(|value| { + let value_str = value.to_string(); + let str_val = value_str.trim_matches('"'); + format!("{}={}", + urlencoding::encode(param_name), + urlencoding::encode(str_val) + ) + }) + }) + .collect(); + + if query_params.is_empty() { + String::new() + } else { + format!("?{}", query_params.join("&")) + } +} + +fn build_request_body( + method: &http::Method, + args_map: &serde_json::Map, + body_schema: &Option, +) -> Option { + if method == &http::Method::GET { + return None; + } + + let schema = body_schema.as_ref()?; + let props = schema.get("properties")?.as_object()?; + + let body_map: serde_json::Map = props + .keys() + .filter_map(|param_name| { + args_map.get(param_name) + .map(|value| (param_name.clone(), value.clone())) + }) + .collect(); + + if body_map.is_empty() { + None + } else { + Some(serde_json::Value::Object(body_map)) + } +} + +async fn create_http_request( + method: &http::Method, + url: &str, + workspace_id: &str, + api_authed: &ApiAuthed, + body_json: Option, +) -> Result { + let client = &crate::HTTP_CLIENT; + let mut request_builder = match method { + &http::Method::GET => client.get(url), + &http::Method::POST => client.post(url), + &http::Method::PUT => client.put(url), + &http::Method::DELETE => client.delete(url), + &http::Method::PATCH => client.patch(url), + _ => return Err(Error::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))?; + 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| { + Error::internal_error(format!("Failed to execute request: {}", e), None) + }) +} \ No newline at end of file diff --git a/backend/windmill-common/src/auth.rs b/backend/windmill-common/src/auth.rs index a2de086e22..543d6eb462 100644 --- a/backend/windmill-common/src/auth.rs +++ b/backend/windmill-common/src/auth.rs @@ -297,25 +297,41 @@ pub async fn create_token_for_owner( } }; + create_jwt_token(job_authed, w_id, expires_in, Some(*job_id), Some(label.to_string()), audit_span, None) + .await +} + +pub async fn create_jwt_token( + authed: Authed, + workspace_id: &str, + expires_in_seconds: u64, + job_id: Option, + label: Option, + audit_span: Option, + scopes: Option>, +) -> crate::error::Result { let payload = JWTAuthClaims { - email: job_authed.email, - username: job_authed.username, - is_admin: job_authed.is_admin, - is_operator: job_authed.is_operator, - groups: job_authed.groups, - folders: job_authed.folders, - label: Some(label.to_string()), - workspace_id: w_id.to_string(), - exp: (chrono::Utc::now() + chrono::Duration::seconds(expires_in as i64)).timestamp() - as usize, - job_id: Some(job_id.to_string()), - scopes: None, + email: authed.email.clone(), + username: authed.username.clone(), + is_admin: authed.is_admin, + is_operator: authed.is_operator, + groups: authed.groups.clone(), + folders: authed.folders.clone(), + label, + workspace_id: workspace_id.to_string(), + exp: (chrono::Utc::now() + chrono::Duration::seconds(expires_in_seconds as i64)) + .timestamp() as usize, + job_id: job_id.map(|id| id.to_string()), + scopes, audit_span, }; let token = jwt::encode_with_internal_secret(&payload) .await - .with_context(|| format!("Could not encode JWT token for job {job_id}"))?; + .with_context(|| match job_id { + Some(job_id) => format!("Could not encode JWT token for job {job_id}"), + None => "Could not encode JWT token".to_string(), + })?; Ok(format!("jwt_{}", token)) }