Workflows MCP Server

Instructions

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

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