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,
```

Workflows MCP Server