execute_workflow
Run a named workflow script with optional parameters to automate tasks and processes programmatically.
Instructions
Execute a workflow script by name.
Args:
name: The name of the workflow to execute
params: Optional dictionary of parameters to pass to the workflow's run() function
Returns:
dict: The result of the workflow execution
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| name | Yes | ||
| params | No |
Implementation Reference
- src/server.py:116-184 (handler)Core MCP tool handler for 'execute_workflow'. Registered via @mcp.tool(). Executes the named workflow script in the workflows directory using subprocess, passing params as JSON argument, captures output, parses as JSON, and returns structured result or error.@mcp.tool() def execute_workflow(name: str, params: dict = None) -> dict: """ Execute a workflow script by name. Args: name: The name of the workflow to execute params: Optional dictionary of parameters to pass to the workflow's run() function Returns: dict: The result of the workflow execution """ try: workflow_path = get_workflow_path(name) if not workflow_path.exists(): return { "status": "error", "message": f"Workflow '{name}' not found" } # Prepare the command params_json = json.dumps(params or {}) # Execute the workflow result = subprocess.run( [sys.executable, str(workflow_path), params_json], capture_output=True, text=True, timeout=300, # 5 minute timeout cwd=WORKFLOWS_DIR ) # Parse the output output = result.stdout.strip() error = result.stderr.strip() if result.returncode != 0: return { "status": "error", "message": f"Workflow execution failed", "error": error, "output": output, "return_code": result.returncode } # Try to parse output as JSON try: output_data = json.loads(output) if output else {} except json.JSONDecodeError: output_data = {"raw_output": output} return { "status": "success", "message": f"Workflow '{name}' executed successfully", "result": output_data, "stderr": error if error else None } except subprocess.TimeoutExpired: return { "status": "error", "message": f"Workflow '{name}' timed out after 5 minutes" } except Exception as e: return { "status": "error", "message": f"Failed to execute workflow: {str(e)}" }
- src/server.py:118-127 (schema)Input schema and description defined in the tool's docstring, specifying 'name' (str, required) and 'params' (dict, optional).""" Execute a workflow script by name. Args: name: The name of the workflow to execute params: Optional dictionary of parameters to pass to the workflow's run() function Returns: dict: The result of the workflow execution """
- src/server.py:29-34 (helper)Helper function to resolve sanitized workflow filename to Path in WORKFLOWS_DIR, used by execute_workflow.def get_workflow_path(name: str) -> Path: """Get the full path for a workflow file.""" # Sanitize the name to prevent directory traversal safe_name = "".join(c for c in name if c.isalnum() or c in "_-").lower() return Path(WORKFLOWS_DIR) / f"{safe_name}.py"
- src/mcp_http_server.py:109-124 (schema)Explicit JSON schema for 'execute_workflow' tool in HTTP MCP server."name": "execute_workflow", "description": "Execute a workflow script by name with optional parameters.", "inputSchema": { "type": "object", "properties": { "name": { "type": "string", "description": "The name of the workflow to execute" }, "params": { "type": "object", "description": "Optional dictionary of parameters to pass to the workflow's run() function" } }, "required": ["name"] }
- src/mcp_http_server.py:378-380 (registration)Registration of 'execute_workflow' handler (tool_execute_workflow) in the HTTP MCP server's TOOL_HANDLERS dict.TOOL_HANDLERS = { "create_workflow": tool_create_workflow, "execute_workflow": tool_execute_workflow,