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,33 @@
+# 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
+RUN uv sync --system # --system to install into the system site-packages
+
+# Stage 2: Final image
+FROM python:3.11-slim-bullseye
+
+WORKDIR /app
+
+# Copy installed dependencies from builder
+COPY --from=builder /usr/local/lib/python3.11/site-packages/ /usr/local/lib/python3.11/site-packages/
+# Also copy uv itself
+COPY --from=builder /usr/local/bin/uv /usr/local/bin/uv
+
+# Copy application code
+COPY server.py ./
+
+# Expose the port Uvicorn will listen on
+EXPOSE 8000
+
+# Run the Uvicorn server
+# The server.py script runs uvicorn with host 0.0.0.0 and port 8000
+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,19 @@
+[project]
+name = "proxmox-mcp-custom"
+version = "0.1.0"
+description = "Custom Proxmox MCP Server"
+dependencies = [
+    "mcp",
+    "proxmoxer",
+    "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,159 @@
+from fastapi import FastAPI, Request
+from fastapi.responses import StreamingResponse
+from uvicorn import Config, Server
+from mcp.server.fastapi import FastAPIAppBuilder
+from mcp.server.models import CallToolRequest, CallToolResponse, Tool
+from mcp.server.transports.sse import SSEServerTransport
+from mcp.server.server import MCPServer
+from mcp.exceptions import MCPException
+import os
+import json
+import logging
+from proxmoxer import ProxmoxAPI, ProxmoxAPIException
+from dotenv import load_dotenv
+
+# Load environment variables for local development
+load_dotenv()
+
+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") # Or token secret
+PROXMOX_TOKEN_ID = os.getenv("PROXMOX_TOKEN_ID")
+PROXMOX_VERIFY_SSL = os.getenv("PROXMOX_VERIFY_SSL", "false").lower() == "true"
+
+if not PROXMOX_URL or not (PROXMOX_PASSWORD and PROXMOX_TOKEN_ID):
+    logger.fatal("Proxmox API credentials (URL, User, Password/Token) not fully set.")
+    exit(1)
+
+# Initialize ProxmoxAPI client
+try:
+    if PROXMOX_TOKEN_ID and PROXMOX_PASSWORD: # Using API Token
+        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)
+    else: # Using Username/Password
+        proxmox = ProxmoxAPI(PROXMOX_URL, user=PROXMOX_USER, password=PROXMOX_PASSWORD, verify_ssl=PROXMOX_VERIFY_SSL)
+    # Test connection
+    proxmox.version.get()
+    logger.info("Successfully connected to Proxmox API.")
+except ProxmoxAPIException as e:
+    logger.fatal(f"Failed to connect to Proxmox API: {e}")
+    exit(1)
+except Exception as e:
+    logger.fatal(f"An unexpected error occurred during Proxmox API connection: {e}")
+    exit(1)
+
+# --- MCP Server Setup ---
+mcp_server = MCPServer(
+    name="Proxmox MCP Server",
+    version="1.0",
+    description="MCP Server for Proxmox VE via proxmoxer",
+    capabilities={"tools": True}
+)
+
+# --- Core Tools ---
+@mcp_server.tool(
+    name="list_nodes",
+    description="Returns a list of all Proxmox nodes in the cluster with their status.",
+    parameters={} # No parameters needed
+)
+def list_nodes():
+    """Lists all Proxmox nodes."""
+    try:
+        nodes = proxmox.nodes.get()
+        return {"nodes": nodes}
+    except ProxmoxAPIException as e:
+        raise MCPException(f"Proxmox API Error: {e}")
+    except Exception as e:
+        raise MCPException(f"Error listing nodes: {e}")
+
+@mcp_server.tool(
+    name="get_cluster_resources",
+    description="Returns a summary of all resources (VMs, LXC containers, storage, etc.) across the cluster.",
+    parameters={} # No parameters needed
+)
+def get_cluster_resources():
+    """Gets a summary of all cluster resources."""
+    try:
+        resources = proxmox.cluster.resources.get()
+        return {"resources": resources}
+    except ProxmoxAPIException as e:
+        raise MCPException(f"Proxmox API Error: {e}")
+    except Exception as e:
+        raise MCPException(f"Error getting cluster resources: {e}")
+
+# --- Raw API Access Tool ---
+@mcp_server.tool(
+    name="proxmox_api_call",
+    description="Executes a raw API call to Proxmox VE. Use carefully!",
+    parameters={
+        "path": {"type": "string", "description": "The API path (e.g., 'nodes/pve1/qemu/100/status/start')."},
+        "method": {"type": "string", "enum": ["GET", "POST", "PUT", "DELETE"], "description": "HTTP Method."},
+        "data": {"type": "object", "description": "Optional JSON payload for POST/PUT methods.", "default": {}},
+        "node": {"type": "string", "description": "Optional node name if path is relative to a node.", "default": None},
+        "vmid": {"type": "integer", "description": "Optional VM ID if path is relative to a VM.", "default": None},
+        "lxcid": {"type": "integer", "description": "Optional LXC ID if path is relative to an LXC container.", "default": None},
+    },
+)
+def proxmox_api_call(path: str, method: str, data: dict = {}, node: str = None, vmid: int = None, lxcid: int = None):
+    """Executes a raw Proxmox API call."""
+    try:
+        # Dynamically build the ProxmoxAPI path
+        api_path = proxmox
+        path_segments = path.strip('/').split('/')
+        
+        for segment in path_segments:
+            if segment: # Skip empty segments
+                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)
+                else:
+                    api_path = api_path(segment)
+        
+        method_func = getattr(api_path, method.lower(), None)
+        if not method_func:
+            raise MCPException(f"Unsupported method '{method}' for path segment '{path}'")
+
+        if method.upper() in ["POST", "PUT"]:
+            result = method_func(**data)
+        else:
+            result = method_func()
+        
+        return {"result": result}
+    except ProxmoxAPIException as e:
+        raise MCPException(f"Proxmox API Error: {e}")
+    except Exception as e:
+        raise MCPException(f"Error executing raw API call: {e}")
+
+
+# --- FastAPI Integration ---
+app_builder = FastAPIAppBuilder(mcp_server)
+app = app_builder.build_app()
+
+@app.get("/")
+async def root():
+    return {"message": "Proxmox MCP Server is running. Access /sse for events."}
+
+@app.get("/sse")
+async def sse(request: Request):
+    """Server-Sent Events endpoint for MCP communication."""
+    logger.info("New SSE connection from client.")
+    return StreamingResponse(
+        SSEServerTransport(mcp_server).get_response_generator(request),
+        media_type="text/event-stream"
+    )
+
+@app.post("/messages")
+async def messages(request: Request):
+    """Endpoint for MCP client messages."""
+    logger.info("New POST /messages from client.")
+    return await SSEServerTransport(mcp_server).handle_request(request)
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)