Initial commit: Custom Proxmox MCP with SSE wrapper

.gitea/workflows/build.yaml

@@ -0,0 +1,30 @@
+name: Build and Push Proxmox MCP Docker Image
+
+on:
+  push:
+    branches:
+      - main
+      - master
+
+jobs:
+  build:
+    runs-on: docker
+    container:
+      image: catthehacker/ubuntu:act-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+
+      - name: Login to Gitea Container Registry
+        uses: docker/login-action@v2
+        with:
+          registry: gitea.ext.ben.io
+          username: ${{ gitea.actor }}
+          password: ${{ secrets.CR_PAT }} # Using CR_PAT secret
+
+      - name: Build and Push
+        uses: docker/build-push-action@v4
+        with:
+          context: .
+          push: true
+          tags: gitea.ext.ben.io/${{ gitea.repository }}:latest

Dockerfile

@@ -0,0 +1,37 @@
+# Stage 1: Builder for dependencies
+FROM python:3.11-slim-bullseye AS builder
+
+WORKDIR /app
+
+# Install uv
+RUN pip install uv
+
+# Copy only dependency files first to leverage Docker cache
+COPY pyproject.toml ./
+
+# Install dependencies with uv
+# Use 'uv pip install --system' to install directly into the system python environment
+# This avoids creating a virtualenv inside the Docker build stage
+RUN uv pip install --system .
+
+# Stage 2: Final image
+FROM python:3.11-slim-bullseye
+
+WORKDIR /app
+
+# Copy installed dependencies from builder
+# We copy the entire site-packages directory
+COPY --from=builder /usr/local/lib/python3.11/site-packages/ /usr/local/lib/python3.11/site-packages/
+# Also copy uv itself (optional, but good for debugging)
+COPY --from=builder /usr/local/bin/uv /usr/local/bin/uv
+# And the bin directory for executables like 'uvicorn'
+COPY --from=builder /usr/local/bin/ /usr/local/bin/
+
+# Copy application code
+COPY server.py ./
+
+# Expose the port Uvicorn will listen on
+EXPOSE 8000
+
+# Run the Uvicorn server
+CMD ["python", "server.py"]

README.md

@@ -0,0 +1,46 @@
+# Custom Proxmox MCP Server
+
+A robust, maintenance-free Model Context Protocol (MCP) server for Proxmox VE, built with Python and `proxmoxer`.
+
+## Philosophy: The Hybrid Approach
+
+Most MCP servers suffer from "feature rot" where the author implements 10 tools (`start_vm`, `stop_vm`) but misses 500 others. This project takes a hybrid approach:
+
+1.  **Core Tools:** A small set of high-value tools for discovery and context (e.g., `list_nodes`, `get_cluster_resources`).
+2.  **Raw API Access:** A single powerful tool `proxmox_api` that allows the LLM to call *any* Proxmox API endpoint dynamically. This ensures 100% API coverage without writing wrappers for every function.
+
+## Tech Stack
+
+*   **Language:** Python 3.11+
+*   **MCP SDK:** `mcp` (Official Python SDK) with `FastMCP` (if available) or standard server implementation.
+*   **Proxmox Client:** `proxmoxer` (Community standard, stable).
+*   **Transport:** SSE (Server-Sent Events) for Docker/Remote compatibility.
+*   **Deployment:** Docker (built via Gitea Actions).
+
+## Tools
+
+### `list_nodes`
+Returns a list of all nodes in the cluster with their status.
+
+### `get_resources`
+Returns a summary of all resources (VMs, LXC containers, storage) across the cluster.
+
+### `proxmox_api`
+Executes a raw API call to Proxmox.
+*   `service`: Service path (e.g., `nodes/pve1/qemu`).
+*   `method`: HTTP Method (GET, POST, PUT, DELETE).
+*   `data`: Optional JSON payload.
+
+## Configuration
+
+Required Environment Variables:
+*   `PROXMOX_URL`: Base URL (e.g., `https://192.168.1.1:8006`).
+*   `PROXMOX_USER`: User (e.g., `root@pam`).
+*   `PROXMOX_PASSWORD`: Password or Token Secret.
+*   `PROXMOX_TOKEN_ID`: (Optional) Token ID if using tokens (e.g., `mcp-token`).
+*   `PROXMOX_VERIFY_SSL`: `true` or `false` (Default: `false` for homelabs).
+
+## Build & Deploy
+
+This project is built using Gitea Actions and pushed to the internal registry.
+Deploy using Portainer.

docker-compose.yml

@@ -0,0 +1,22 @@
+services:
+  proxmox-mcp:
+    image: gitea.ext.ben.io/b3nw/proxmox-mcp-custom:latest
+    container_name: proxmox-mcp
+    environment:
+      # --- Proxmox API Credentials ---
+      # Base URL of your Proxmox VE instance (e.g., pve.local.ben.io:8006)
+      - PROXMOX_URL=your_proxmox_host:8006
+      # Proxmox API User (e.g., root@pam or user@pve)
+      - PROXMOX_USER=your_proxmox_user
+      # Proxmox API Token Secret (usually starts with "pve_mcp!")
+      - PROXMOX_PASSWORD=your_proxmox_token_secret
+      # Proxmox API Token ID (e.g., "mcp-token")
+      - PROXMOX_TOKEN_ID=your_proxmox_token_id
+      # Set to 'true' to skip SSL verification (e.g., for self-signed certs in homelab)
+      - PROXMOX_VERIFY_SSL=false
+      
+      # --- MCP Server Settings ---
+      - PORT=8000 # Internal container port
+    ports:
+      - "8000:8000" # Host Port : Container Port
+    restart: unless-stopped

pyproject.toml

@@ -0,0 +1,20 @@
+[project]
+name = "proxmox-mcp-custom"
+version = "0.1.0"
+description = "Custom Proxmox MCP Server"
+dependencies = [
+    "mcp",
+    "proxmoxer",
+    "requests", # Required for proxmoxer https backend
+    "uvicorn",
+    "fastapi",
+    "python-dotenv", # For local development
+]
+requires-python = ">=3.11"
+
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[tool.uv]
+python-version = "3.11" # Specify desired Python version for uv

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