mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-20 16:02:19 +00:00
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
This commit is contained in:
centdix
GitHub
+2
-1
@@ -7,4 +7,5 @@ heaptrack*
|
||||
index/
|
||||
windmill-api/openapi-*.*
|
||||
.duckdb/*
|
||||
*ee.rs
|
||||
*ee.rs
|
||||
generate_mcp_endpoints_tools/venv
|
||||
@@ -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
|
||||
```
|
||||
@@ -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<serde_json::Value>,
|
||||
pub query_params_schema: Option<serde_json::Value>,
|
||||
pub body_schema: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
pub fn all_tools() -> Vec<EndpointTool> {
|
||||
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<serde_json::Value>,
|
||||
pub query_params_schema: Option<serde_json::Value>,
|
||||
pub body_schema: Option<serde_json::Value>,
|
||||
}}
|
||||
|
||||
pub fn all_tools() -> Vec<EndpointTool> {{
|
||||
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()
|
||||
@@ -0,0 +1 @@
|
||||
PyYAML>=6.0
|
||||
@@ -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/<path>, flow/<path>)
|
||||
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/<path>, flow/<path>)
|
||||
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:
|
||||
|
||||
@@ -854,6 +854,22 @@ impl From<ApiAuthed> for Authed {
|
||||
}
|
||||
}
|
||||
|
||||
impl From<Authed> 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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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<SchemaType>,
|
||||
) -> 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::<serde_json::Value>(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 })
|
||||
}
|
||||
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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<EndpointTool>) -> Vec<Tool> {
|
||||
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<String, serde_json::Value>,
|
||||
combined_required: &mut Vec<String>,
|
||||
schema: &serde_json::Value,
|
||||
) {
|
||||
if let Some(props) = schema.get("properties").and_then(|p| p.as_object()) {
|
||||
for (key, value) in props {
|
||||
combined_properties.insert(key.clone(), value.clone());
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(required) = schema.get("required").and_then(|r| r.as_array()) {
|
||||
for req in required.iter().filter_map(|r| r.as_str()) {
|
||||
combined_required.push(req.to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn call_endpoint_tool(
|
||||
tool: &EndpointTool,
|
||||
args: serde_json::Value,
|
||||
workspace_id: &str,
|
||||
api_authed: &ApiAuthed,
|
||||
) -> Result<serde_json::Value, Error> {
|
||||
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<String, serde_json::Value>,
|
||||
path_schema: &Option<serde_json::Value>,
|
||||
) -> Result<String, Error> {
|
||||
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<String, serde_json::Value>,
|
||||
query_schema: &Option<serde_json::Value>,
|
||||
) -> String {
|
||||
let Some(schema) = query_schema else { return String::new() };
|
||||
let Some(props) = schema.get("properties").and_then(|p| p.as_object()) else { return String::new() };
|
||||
|
||||
let query_params: Vec<String> = props
|
||||
.keys()
|
||||
.filter_map(|param_name| {
|
||||
args_map.get(param_name)
|
||||
.filter(|v| !v.is_null())
|
||||
.map(|value| {
|
||||
let value_str = value.to_string();
|
||||
let str_val = value_str.trim_matches('"');
|
||||
format!("{}={}",
|
||||
urlencoding::encode(param_name),
|
||||
urlencoding::encode(str_val)
|
||||
)
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
|
||||
if query_params.is_empty() {
|
||||
String::new()
|
||||
} else {
|
||||
format!("?{}", query_params.join("&"))
|
||||
}
|
||||
}
|
||||
|
||||
fn build_request_body(
|
||||
method: &http::Method,
|
||||
args_map: &serde_json::Map<String, serde_json::Value>,
|
||||
body_schema: &Option<serde_json::Value>,
|
||||
) -> Option<serde_json::Value> {
|
||||
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<String, serde_json::Value> = props
|
||||
.keys()
|
||||
.filter_map(|param_name| {
|
||||
args_map.get(param_name)
|
||||
.map(|value| (param_name.clone(), value.clone()))
|
||||
})
|
||||
.collect();
|
||||
|
||||
if body_map.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(serde_json::Value::Object(body_map))
|
||||
}
|
||||
}
|
||||
|
||||
async fn create_http_request(
|
||||
method: &http::Method,
|
||||
url: &str,
|
||||
workspace_id: &str,
|
||||
api_authed: &ApiAuthed,
|
||||
body_json: Option<serde_json::Value>,
|
||||
) -> Result<reqwest::Response, Error> {
|
||||
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)
|
||||
})
|
||||
}
|
||||
@@ -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<Uuid>,
|
||||
label: Option<String>,
|
||||
audit_span: Option<String>,
|
||||
scopes: Option<Vec<String>>,
|
||||
) -> crate::error::Result<String> {
|
||||
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))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user