Initial commit: Custom Proxmox MCP with SSE wrapper

server.py

@@ -0,0 +1,85 @@
+import os
+import logging
+from mcp.server.fastmcp import FastMCP
+from proxmoxer import ProxmoxAPI
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+# Logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# --- Proxmox API Configuration ---
+PROXMOX_URL = os.getenv("PROXMOX_URL")
+PROXMOX_USER = os.getenv("PROXMOX_USER")
+PROXMOX_PASSWORD = os.getenv("PROXMOX_PASSWORD")
+PROXMOX_TOKEN_ID = os.getenv("PROXMOX_TOKEN_ID")
+PROXMOX_VERIFY_SSL = os.getenv("PROXMOX_VERIFY_SSL", "false").lower() == "true"
+
+# --- Initialize Proxmox Client ---
+proxmox = None
+if PROXMOX_URL and (PROXMOX_PASSWORD and PROXMOX_TOKEN_ID):
+    try:
+        proxmox = ProxmoxAPI(PROXMOX_URL, user=f"{PROXMOX_USER}@{PROXMOX_TOKEN_ID}", token_name=PROXMOX_TOKEN_ID, token_value=PROXMOX_PASSWORD, verify_ssl=PROXMOX_VERIFY_SSL)
+        logger.info("Proxmox API client configured.")
+    except Exception as e:
+        logger.error(f"Failed to configure Proxmox API: {e}")
+elif PROXMOX_URL and PROXMOX_USER and PROXMOX_PASSWORD:
+     try:
+        proxmox = ProxmoxAPI(PROXMOX_URL, user=PROXMOX_USER, password=PROXMOX_PASSWORD, verify_ssl=PROXMOX_VERIFY_SSL)
+        logger.info("Proxmox API client configured.")
+     except Exception as e:
+        logger.error(f"Failed to configure Proxmox API: {e}")
+else:
+    logger.warning("Proxmox API credentials not fully set. Tools may fail.")
+
+# --- FastMCP Server ---
+mcp = FastMCP("Proxmox MCP")
+
+@mcp.tool()
+def list_nodes() -> dict:
+    """Lists all Proxmox nodes."""
+    if not proxmox: return {"error": "Proxmox API not configured"}
+    try:
+        nodes = proxmox.nodes.get()
+        return {"nodes": nodes}
+    except Exception as e:
+        return {"error": str(e)}
+
+@mcp.tool()
+def get_cluster_resources() -> dict:
+    """Gets a summary of all cluster resources."""
+    if not proxmox: return {"error": "Proxmox API not configured"}
+    try:
+        resources = proxmox.cluster.resources.get()
+        return {"resources": resources}
+    except Exception as e:
+        return {"error": str(e)}
+
+@mcp.tool()
+def proxmox_api_call(path: str, method: str = "GET", data: dict = {}, node: str = None, vmid: int = None, lxcid: int = None) -> dict:
+    """Executes a raw Proxmox API call."""
+    if not proxmox: return {"error": "Proxmox API not configured"}
+    
+    try:
+        # Build path
+        api_path = proxmox
+        for segment in path.strip('/').split('/'):
+            if segment == "nodes" and node: api_path = api_path.nodes(node)
+            elif segment == "qemu" and vmid: api_path = api_path.qemu(vmid)
+            elif segment == "lxc" and lxcid: api_path = api_path.lxc(lxcid)
+            elif segment: api_path = api_path(segment)
+        
+        # Execute
+        method_func = getattr(api_path, method.lower())
+        if method.upper() in ["POST", "PUT"]:
+            return {"result": method_func(**data)}
+        else:
+            return {"result": method_func()}
+    except Exception as e:
+        return {"error": str(e)}
+
+if __name__ == "__main__":
+    mcp.run(transport="sse")