Workflows MCP Server

Instructions

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": "..."}

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)}