Workflows MCP Server

Instructions

Implementation Reference

• src/server.py:264-330 (handler)

  Primary MCP tool handler for update_workflow, decorated with @mcp.tool(), implements the core logic for updating workflow scripts by regenerating the file content.
  @mcp.tool() def update_workflow(name: str, description: str = None, code: str = None) -> dict: """ Update an existing workflow script. Args: name: The name of the workflow to update description: New description (optional, keeps existing if not provided) code: New Python code (optional, keeps existing if not provided) Returns: dict: Status of the operation """ try: workflow_path = get_workflow_path(name) if not workflow_path.exists(): return { "status": "error", "message": f"Workflow '{name}' not found. Use create_workflow to create it." } # Read existing content to preserve metadata if needed existing_content = workflow_path.read_text() # Extract existing description if not provided if description is None: for line in existing_content.split("\n"): if line.startswith("Description:"): description = line.replace("Description:", "").strip() break description = description or "No description" # If no new code provided, extract existing code if code is None: # Find the code after the imports lines = existing_content.split("\n") code_start = 0 for i, line in enumerate(lines): if line.startswith("def run(") or line.startswith("async def run("): code_start = i break # Extract from run function to before if __name__ code_lines = [] for i in range(code_start, len(lines)): if lines[i].startswith('if __name__'): break code_lines.append(lines[i]) code = "\n".join(code_lines) # Generate updated 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}' updated successfully", "path": str(workflow_path) } except Exception as e: return { "status": "error", "message": f"Failed to update workflow: {str(e)}" }
• src/mcp_http_server.py:148-169 (schema)

  Input schema definition for the update_workflow tool in the MCP HTTP server.
  { "name": "update_workflow", "description": "Update an existing workflow script's description or code.", "inputSchema": { "type": "object", "properties": { "name": { "type": "string", "description": "The name of the workflow to update" }, "description": { "type": "string", "description": "New description (optional)" }, "code": { "type": "string", "description": "New Python code (optional)" } }, "required": ["name"] } },
• src/mcp_http_server.py:378-385 (registration)

  Registration of tool handlers including update_workflow in the TOOL_HANDLERS dictionary for MCP HTTP server.
  TOOL_HANDLERS = { "create_workflow": tool_create_workflow, "execute_workflow": tool_execute_workflow, "list_workflows": tool_list_workflows, "read_workflow": tool_read_workflow, "update_workflow": tool_update_workflow, "delete_workflow": tool_delete_workflow, }
• src/mcp_http_server.py:311-356 (handler)

  Handler function for update_workflow in the MCP HTTP server implementation.
  def tool_update_workflow(arguments: dict) -> dict: """Update an existing workflow.""" try: name = arguments.get("name") description = arguments.get("description") code = arguments.get("code") if not name: return {"error": "Workflow name is required"} workflow_path = get_workflow_path(name) if not workflow_path.exists(): return {"error": f"Workflow '{name}' not found"} existing_content = workflow_path.read_text() if description is None: for line in existing_content.split("\n"): if line.startswith("Description:"): description = line.replace("Description:", "").strip() break description = description or "No description" if code is None: lines = existing_content.split("\n") code_start = 0 for i, line in enumerate(lines): if line.startswith("def run("): code_start = i break code_lines = [] for i in range(code_start, len(lines)): if lines[i].startswith('if __name__'): break code_lines.append(lines[i]) code = "\n".join(code_lines) script_content = generate_workflow_script(name, description, code) workflow_path.write_text(script_content) return {"status": "success", "message": f"Workflow '{name}' updated successfully"} except Exception as e: return {"error": str(e)}
• src/http_server.py:318-367 (handler)

  REST HTTP handler for update_workflow tool call.
  async def update_workflow_handler(body: dict): """Handle update_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 not workflow_path.exists(): return {"status": "error", "message": f"Workflow '{name}' not found"} existing_content = workflow_path.read_text() if description is None: for line in existing_content.split("\n"): if line.startswith("Description:"): description = line.replace("Description:", "").strip() break description = description or "No description" if code is None: lines = existing_content.split("\n") code_start = 0 for i, line in enumerate(lines): if line.startswith("def run("): code_start = i break code_lines = [] for i in range(code_start, len(lines)): if lines[i].startswith('if __name__'): break code_lines.append(lines[i]) code = "\n".join(code_lines) script_content = generate_workflow_script(name, description, code) workflow_path.write_text(script_content) return { "status": "success", "message": f"Workflow '{name}' updated successfully", "path": str(workflow_path) } except Exception as e: return {"status": "error", "message": str(e)}