Workflows MCP Server

Instructions

Implementation Reference

• src/server.py:68-114 (handler)

  Primary MCP tool handler for create_workflow. Creates a new workflow script file using provided name, description, and code. Uses helpers get_workflow_path and generate_workflow_script.
  @mcp.tool() def create_workflow(name: str, description: str, code: str) -> dict: """ Create a new Python workflow script. Args: name: The name of the workflow (will be used as filename, e.g., "meeting_review_to_slack") description: A description of what the workflow does code: The Python code for the workflow. Must include a `run(params: dict = None) -> dict` function. Returns: dict: Status of the operation with the file path Example code structure: def run(params: dict = None) -> dict: params = params or {} # Your workflow logic here return {"status": "success", "result": "..."} """ try: workflow_path = get_workflow_path(name) # Check if workflow already exists if workflow_path.exists(): return { "status": "error", "message": f"Workflow '{name}' already exists. Use update_workflow to modify it." } # Generate the full script script_content = generate_workflow_script(name, description, code) # Write the file workflow_path.write_text(script_content) return { "status": "success", "message": f"Workflow '{name}' created successfully", "path": str(workflow_path), "name": name } except Exception as e: return { "status": "error", "message": f"Failed to create workflow: {str(e)}" }
• src/server.py:36-66 (helper)

  Helper function to generate the full Python script content for a workflow, including metadata docstring, standard imports, and boilerplate main block.
  def generate_workflow_script(name: str, description: str, code: str) -> str: """Generate a complete workflow script with metadata.""" template = f'''""" Workflow: {name} Description: {description} Created: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} """ import os import sys import json import requests from datetime import datetime from typing import Any, Dict, Optional {code} if __name__ == "__main__": # Allow passing params as JSON via command line params = {{}} if len(sys.argv) > 1: try: params = json.loads(sys.argv[1]) except json.JSONDecodeError: print("Warning: Could not parse params as JSON") result = run(params) print(json.dumps(result, indent=2, default=str)) ''' return template
• src/server.py:29-34 (helper)

  Helper to compute the sanitized file path for a workflow in the workflows directory.
  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:86-107 (schema)

  Explicit JSON schema definition for the create_workflow tool input parameters in the MCP HTTP server.
  { "name": "create_workflow", "description": "Create a new Python workflow script. The code parameter must include a `run(params: dict = None) -> dict` function that serves as the entry point.", "inputSchema": { "type": "object", "properties": { "name": { "type": "string", "description": "The name of the workflow (will be used as filename, e.g., 'meeting_review_to_slack')" }, "description": { "type": "string", "description": "A description of what the workflow does" }, "code": { "type": "string", "description": "The Python code for the workflow. Must include a `run(params: dict = None) -> dict` function." } }, "required": ["name", "description", "code"] } },
• src/http_server.py:212-238 (handler)

  HTTP handler for create_workflow tool call in the FastAPI HTTP server.
  async def create_workflow_handler(body: dict): """Handle create_workflow tool call.""" try: name = body.get("name") description = body.get("description", "") code = body.get("code", "") if not name: return {"status": "error", "message": "Workflow name is required"} workflow_path = get_workflow_path(name) if workflow_path.exists(): return {"status": "error", "message": f"Workflow '{name}' already exists"} script_content = generate_workflow_script(name, description, code) workflow_path.write_text(script_content) return { "status": "success", "message": f"Workflow '{name}' created successfully", "path": str(workflow_path), "name": name } except Exception as e: return {"status": "error", "message": str(e)}