Workflows MCP Server

Instructions

List all available workflow scripts.

Returns:
    dict: List of workflows with their metadata

Implementation Reference

• src/server.py:186-227 (handler)

  Primary handler for the 'list_workflows' MCP tool, decorated with @mcp.tool() for FastMCP registration and execution.

  @mcp.tool()
  def list_workflows() -> dict:
      """
      List all available workflow scripts.
      
      Returns:
          dict: List of workflows with their metadata
      """
      try:
          workflows = []
          workflows_path = Path(WORKFLOWS_DIR)
          
          for file in workflows_path.glob("*.py"):
              # Read the file to extract metadata
              content = file.read_text()
              
              # Extract description from docstring
              description = ""
              lines = content.split("\n")
              for line in lines:
                  if line.startswith("Description:"):
                      description = line.replace("Description:", "").strip()
                      break
              
              workflows.append({
                  "name": file.stem,
                  "path": str(file),
                  "description": description,
                  "modified": datetime.fromtimestamp(file.stat().st_mtime).isoformat()
              })
          
          return {
              "status": "success",
              "count": len(workflows),
              "workflows": workflows
          }
      except Exception as e:
          return {
              "status": "error",
              "message": f"Failed to list workflows: {str(e)}"
          }

• src/mcp_http_server.py:127-134 (schema)

  Input schema definition for the 'list_workflows' tool in the MCP_TOOLS configuration list.

      "name": "list_workflows",
      "description": "List all available workflow scripts with their metadata.",
      "inputSchema": {
          "type": "object",
          "properties": {}
      }
  },
  {

• src/mcp_http_server.py:266-289 (handler)

  Handler function for 'list_workflows' tool in the MCP HTTP server implementation.

  def tool_list_workflows(arguments: dict) -> dict:
      """List all workflows."""
      try:
          workflows = []
          workflows_path = Path(WORKFLOWS_DIR)
          
          for file in workflows_path.glob("*.py"):
              content = file.read_text()
              description = ""
              for line in content.split("\n"):
                  if line.startswith("Description:"):
                      description = line.replace("Description:", "").strip()
                      break
              
              workflows.append({
                  "name": file.stem,
                  "description": description,
                  "modified": datetime.fromtimestamp(file.stat().st_mtime).isoformat()
              })
          
          return {"workflows": workflows, "count": len(workflows)}
      except Exception as e:
          return {"error": str(e)}

• src/mcp_http_server.py:378-385 (registration)

  Registration of the 'list_workflows' handler in the TOOL_HANDLERS dictionary for the 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/http_server.py:175-199 (handler)

  HTTP endpoint handler for listing workflows in the simple HTTP server.

  async def list_workflows_handler():
      """List all available workflow scripts."""
      try:
          workflows = []
          workflows_path = Path(WORKFLOWS_DIR)
          
          for file in workflows_path.glob("*.py"):
              content = file.read_text()
              description = ""
              for line in content.split("\n"):
                  if line.startswith("Description:"):
                      description = line.replace("Description:", "").strip()
                      break
              
              workflows.append({
                  "name": file.stem,
                  "path": str(file),
                  "description": description,
                  "modified": datetime.fromtimestamp(file.stat().st_mtime).isoformat()
              })
          
          return {"status": "success", "count": len(workflows), "workflows": workflows}
      except Exception as e:
          return {"status": "error", "message": str(e)}