Initial commit: Custom Proxmox MCP with SSE wrapper

server.py

@@ -0,0 +1,104 @@
+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_USER:
+    try:
+        if PROXMOX_TOKEN_ID and PROXMOX_PASSWORD:
+            # Token-based authentication
+            # PROXMOX_USER should be like "user@pam" or "user@pve"
+            # PROXMOX_TOKEN_ID is the token name (e.g., "mcp-token")
+            # PROXMOX_PASSWORD is the token secret value
+            proxmox = ProxmoxAPI(
+                PROXMOX_URL,
+                user=PROXMOX_USER,
+                token_name=PROXMOX_TOKEN_ID,
+                token_value=PROXMOX_PASSWORD,
+                verify_ssl=PROXMOX_VERIFY_SSL
+            )
+            logger.info(f"Proxmox API client configured with token auth for {PROXMOX_USER}")
+        elif PROXMOX_PASSWORD:
+            # Password-based authentication
+            proxmox = ProxmoxAPI(
+                PROXMOX_URL,
+                user=PROXMOX_USER,
+                password=PROXMOX_PASSWORD,
+                verify_ssl=PROXMOX_VERIFY_SSL
+            )
+            logger.info(f"Proxmox API client configured with password auth for {PROXMOX_USER}")
+        else:
+            logger.warning("PROXMOX_PASSWORD (or token secret) not set. Tools may fail.")
+    except Exception as e:
+        logger.error(f"Failed to configure Proxmox API: {e}")
+else:
+    logger.warning("PROXMOX_URL or PROXMOX_USER not set. Tools may fail.")
+
+# --- FastMCP Server ---
+# Note: Host header validation is handled by reverse proxy in production
+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__":
+    import uvicorn
+    # Use the exposed sse_app from FastMCP and run it with uvicorn
+    uvicorn.run(mcp.sse_app, host="0.0.0.0", port=8000)