Initial proxmox-mcp: MCP server for vessel troubleshoot stack.

Part of GeneseasX golden-vessel tooling (see homelab-vault Projects/vessel-troubleshoot-stack).

.env.example

@@ -0,0 +1,9 @@
+# Proxmox VE API (token from Datacenter → Permissions → API Tokens)
+PROXMOX_URL=https://10.20.30.254:8006
+PROXMOX_TOKEN_ID=root@pam!cursor
+PROXMOX_TOKEN_SECRET=
+PROXMOX_VERIFY_SSL=false
+PROXMOX_TIMEOUT=60
+
+PROXMOX_ALLOW_WRITES=false
+PROXMOX_TARGETS_PATH=~/.local/share/proxmox-mcp/targets.json

.gitignore

@@ -0,0 +1,7 @@
+__pycache__/
+*.py[cod]
+*.egg-info/
+.env
+.venv/
+dist/
+build/

README.md

@@ -0,0 +1,57 @@
+# proxmox-mcp
+
+MCP server for **Proxmox VE 9** — nodes, VMs, LXCs, status, exec. Sibling of `pfsense-mcp`.
+
+## Quick start
+
+```bash
+cd ~/Projects/proxmox-mcp
+python3 -m venv .venv && .venv/bin/pip install -e .
+```
+
+### Cursor (`~/.cursor/mcp.json`)
+
+```json
+"proxmox": {
+  "command": "/Users/nearxos/Projects/proxmox-mcp/.venv/bin/proxmox-mcp",
+  "env": {
+    "PROXMOX_VERIFY_SSL": "false",
+    "PROXMOX_ALLOW_WRITES": "false",
+    "PROXMOX_TARGETS_PATH": "/Users/nearxos/.local/share/proxmox-mcp/targets.json"
+  }
+}
+```
+
+### Connect
+
+```text
+proxmox_connect(
+  url="https://10.20.30.254:8006",
+  token_id="root@pam!cursor",
+  token_secret="...",
+  name="golden-proxmox"
+)
+proxmox_test_connection()
+proxmox_list_qemu(node="pve")
+```
+
+## Tools
+
+| Tool | Description |
+|------|-------------|
+| `proxmox_connect` / `proxmox_use_target` | Multi-target registry (persisted) |
+| `proxmox_test_connection` | Version + nodes |
+| `proxmox_list_nodes` | Cluster nodes |
+| `proxmox_list_qemu` / `proxmox_list_lxc` | Guest inventory |
+| `proxmox_get_qemu_status` / `proxmox_get_lxc_status` | Live status |
+| `proxmox_get_qemu_config` | VM config |
+| `proxmox_lxc_exec` / `proxmox_qemu_agent_exec` | Guest exec (writes) |
+| `proxmox_get_task_log` | Async task output |
+
+## Golden target
+
+Saved as `golden-proxmox` in `~/.local/share/proxmox-mcp/targets.json` (add API token locally).
+
+## License
+
+MIT

pyproject.toml

@@ -0,0 +1,24 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "proxmox-mcp"
+version = "1.0.0"
+description = "Model Context Protocol server for Proxmox VE 9 management"
+readme = "README.md"
+requires-python = ">=3.10"
+license = "MIT"
+keywords = ["mcp", "proxmox", "pve", "hermes"]
+dependencies = [
+    "mcp>=1.0.0",
+    "httpx>=0.27.0",
+    "pydantic>=2.0.0",
+    "pydantic-settings>=2.0.0",
+]
+
+[project.scripts]
+proxmox-mcp = "proxmox_mcp.server:main"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/proxmox_mcp"]

src/proxmox_mcp/__init__.py

@@ -0,0 +1 @@
+__version__ = "1.0.0"

src/proxmox_mcp/__main__.py

@@ -0,0 +1,4 @@
+from proxmox_mcp.server import main
+
+if __name__ == "__main__":
+    main()

src/proxmox_mcp/client.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+import httpx
+
+from proxmox_mcp.config import settings
+from proxmox_mcp.guards import GuardError
+
+
+class ProxmoxAPIError(Exception):
+    def __init__(self, message: str, status_code: int | None = None, body: Any = None):
+        super().__init__(message)
+        self.status_code = status_code
+        self.body = body
+
+
+class ProxmoxClient:
+    """Proxmox VE API client (/api2/json/...)."""
+
+    def __init__(
+        self,
+        url: str | None = None,
+        token_id: str | None = None,
+        token_secret: str | None = None,
+        verify_ssl: bool | None = None,
+        timeout: float | None = None,
+    ):
+        self.base_url = (url or settings.url).rstrip("/")
+        self.token_id = token_id if token_id is not None else settings.token_id
+        self.token_secret = token_secret if token_secret is not None else settings.token_secret
+        self.verify_ssl = verify_ssl if verify_ssl is not None else settings.verify_ssl
+        self.timeout = timeout if timeout is not None else settings.timeout
+
+    def _require_credentials(self) -> None:
+        if not self.token_id or not self.token_secret:
+            raise GuardError(
+                "No API token configured. Call proxmox_connect(url, token_id, token_secret) "
+                "or set PROXMOX_TOKEN_ID and PROXMOX_TOKEN_SECRET."
+            )
+
+    def _headers(self) -> dict[str, str]:
+        return {"Authorization": f"PVEAPIToken={self.token_id}={self.token_secret}"}
+
+    def _request(self, method: str, endpoint: str, *, data: dict[str, Any] | None = None) -> Any:
+        self._require_credentials()
+        path = endpoint.lstrip("/")
+        url = f"{self.base_url}/api2/json/{path}"
+        with httpx.Client(verify=self.verify_ssl, timeout=self.timeout) as client:
+            response = client.request(method, url, headers=self._headers(), data=data)
+        if response.status_code >= 400:
+            try:
+                body = response.json()
+            except json.JSONDecodeError:
+                body = response.text
+            raise ProxmoxAPIError(
+                f"Proxmox API error {response.status_code} for {method} {url}",
+                status_code=response.status_code,
+                body=body,
+            )
+        if not response.content:
+            return {}
+        payload = response.json()
+        return payload.get("data", payload)
+
+    def get(self, endpoint: str) -> Any:
+        return self._request("GET", endpoint)
+
+    def post(self, endpoint: str, data: dict[str, Any] | None = None) -> Any:
+        return self._request("POST", endpoint, data=data or {})
+
+    def delete(self, endpoint: str) -> Any:
+        return self._request("DELETE", endpoint)
+
+
+def get_client(
+    target: str = "",
+    url: str = "",
+    token_id: str = "",
+    token_secret: str = "",
+    verify_ssl: bool | None = None,
+) -> ProxmoxClient:
+    from proxmox_mcp.targets import resolve_client
+
+    return resolve_client(
+        target=target,
+        url=url,
+        token_id=token_id,
+        token_secret=token_secret,
+        verify_ssl=verify_ssl,
+    )

src/proxmox_mcp/config.py

@@ -0,0 +1,16 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="PROXMOX_", extra="ignore")
+
+    url: str = "https://127.0.0.1:8006"
+    token_id: str = ""
+    token_secret: str = ""
+    verify_ssl: bool = False
+    timeout: float = 60.0
+    allow_writes: bool = False
+    targets_path: str = "~/.local/share/proxmox-mcp/targets.json"
+
+
+settings = Settings()

src/proxmox_mcp/guards.py

@@ -0,0 +1,9 @@
+class GuardError(Exception):
+    """Safety or configuration error."""
+
+
+def require_writes() -> None:
+    from proxmox_mcp.config import settings
+
+    if not settings.allow_writes:
+        raise GuardError("Write operations disabled. Set PROXMOX_ALLOW_WRITES=true to enable.")

src/proxmox_mcp/server.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import json
+
+from mcp.server.fastmcp import FastMCP
+
+from proxmox_mcp import __version__
+from proxmox_mcp.config import settings
+from proxmox_mcp.tools import register_all
+
+mcp = FastMCP(
+    "proxmox",
+    instructions=(
+        "Proxmox VE 9 MCP. Saved targets load from PROXMOX_TARGETS_PATH. "
+        "Call proxmox_connect(url, token_id, token_secret, name=...) or proxmox_use_target. "
+        "Exec tools require PROXMOX_ALLOW_WRITES=true."
+    ),
+)
+
+
+@mcp.tool()
+def proxmox_get_config() -> str:
+    from proxmox_mcp.targets import registry
+
+    return json.dumps(
+        {
+            "allow_writes": settings.allow_writes,
+            "active_target": registry.active().name if registry.active() else None,
+            "targets": registry.list(),
+            "version": __version__,
+        },
+        indent=2,
+    )
+
+
+register_all(mcp)
+
+
+def main() -> None:
+    mcp.run()
+
+
+if __name__ == "__main__":
+    main()

src/proxmox_mcp/targets.py

@@ -0,0 +1,205 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+
+from proxmox_mcp.client import ProxmoxClient
+from proxmox_mcp.config import settings
+from proxmox_mcp.guards import GuardError
+
+
+@dataclass
+class TargetConfig:
+    name: str
+    url: str
+    token_id: str = ""
+    token_secret: str = ""
+    verify_ssl: bool = False
+    description: str = ""
+    registered_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
+
+    def normalized_url(self) -> str:
+        url = self.url.strip().rstrip("/")
+        if not url.startswith(("http://", "https://")):
+            url = f"https://{url}"
+        return url
+
+    def has_credentials(self) -> bool:
+        return bool(self.token_id and self.token_secret)
+
+    def public_view(self) -> dict[str, Any]:
+        return {
+            "name": self.name,
+            "url": self.normalized_url(),
+            "verify_ssl": self.verify_ssl,
+            "description": self.description,
+            "token_configured": bool(self.token_id and self.token_secret),
+            "registered_at": self.registered_at,
+        }
+
+
+class TargetRegistry:
+    def __init__(self) -> None:
+        self._targets: dict[str, TargetConfig] = {}
+        self._active: str | None = None
+        self._loading: bool = False
+
+    @staticmethod
+    def _normalize_name(name: str) -> str:
+        name = re.sub(r"[^a-z0-9_-]+", "-", name.strip().lower())
+        name = re.sub(r"-+", "-", name).strip("-")
+        if not name:
+            raise GuardError("Target name must contain at least one alphanumeric character.")
+        return name
+
+    def register(
+        self,
+        *,
+        name: str = "",
+        url: str,
+        token_id: str = "",
+        token_secret: str = "",
+        verify_ssl: bool = False,
+        description: str = "",
+        registered_at: str = "",
+        set_active: bool = True,
+    ) -> TargetConfig:
+        if not url.strip():
+            raise GuardError("url is required.")
+        host = url.strip().rstrip("/").split("//")[-1].split(":")[0]
+        target_name = self._normalize_name(name) if name.strip() else self._normalize_name(host or "proxmox")
+        cfg_kwargs: dict[str, Any] = {
+            "name": target_name,
+            "url": url,
+            "token_id": token_id,
+            "token_secret": token_secret,
+            "verify_ssl": verify_ssl,
+            "description": description,
+        }
+        if registered_at.strip():
+            cfg_kwargs["registered_at"] = registered_at.strip()
+        cfg = TargetConfig(**cfg_kwargs)
+        if not cfg.has_credentials():
+            raise GuardError("token_id and token_secret are required.")
+        self._targets[target_name] = cfg
+        if set_active:
+            self._active = target_name
+        self._persist()
+        return cfg
+
+    def unregister(self, name: str) -> None:
+        key = self._normalize_name(name)
+        if key not in self._targets:
+            raise GuardError(f"Unknown target '{name}'.")
+        del self._targets[key]
+        if self._active == key:
+            self._active = next(iter(self._targets), None)
+        self._persist()
+
+    def set_active(self, name: str) -> TargetConfig:
+        key = self._normalize_name(name)
+        if key not in self._targets:
+            raise GuardError(f"Unknown target '{name}'.")
+        self._active = key
+        self._persist()
+        return self._targets[key]
+
+    def get(self, name: str) -> TargetConfig:
+        key = self._normalize_name(name)
+        if key not in self._targets:
+            raise GuardError(f"Unknown target '{name}'.")
+        return self._targets[key]
+
+    def active(self) -> TargetConfig | None:
+        if self._active and self._active in self._targets:
+            return self._targets[self._active]
+        return None
+
+    def list(self) -> list[dict[str, Any]]:
+        active = self._active
+        rows = []
+        for cfg in self._targets.values():
+            row = cfg.public_view()
+            row["active"] = cfg.name == active
+            rows.append(row)
+        return sorted(rows, key=lambda r: r["name"])
+
+    def resolve(
+        self,
+        *,
+        target: str = "",
+        url: str = "",
+        token_id: str = "",
+        token_secret: str = "",
+        verify_ssl: bool | None = None,
+    ) -> TargetConfig:
+        if url.strip() or token_id.strip() or token_secret.strip():
+            if not url.strip():
+                raise GuardError("url is required when passing inline credentials.")
+            cfg = TargetConfig(
+                name="__inline__",
+                url=url,
+                token_id=token_id,
+                token_secret=token_secret,
+                verify_ssl=settings.verify_ssl if verify_ssl is None else verify_ssl,
+            )
+            if not cfg.has_credentials():
+                raise GuardError("Inline connection requires token_id and token_secret.")
+            return cfg
+        if target.strip():
+            return self.get(target)
+        active = self.active()
+        if active:
+            return active
+        if settings.token_id and settings.token_secret and settings.url:
+            return TargetConfig(
+                name="__env__",
+                url=settings.url,
+                token_id=settings.token_id,
+                token_secret=settings.token_secret,
+                verify_ssl=settings.verify_ssl,
+            )
+        raise GuardError("No Proxmox target connected. Call proxmox_connect first.")
+
+    def _persist(self) -> None:
+        if self._loading:
+            return
+        from proxmox_mcp.targets_store import save_registry
+
+        save_registry(self)
+
+
+registry = TargetRegistry()
+
+
+def _load_persisted_targets() -> None:
+    from proxmox_mcp.targets_store import load_registry
+
+    load_registry(registry)
+
+
+_load_persisted_targets()
+
+
+def resolve_client(
+    target: str = "",
+    url: str = "",
+    token_id: str = "",
+    token_secret: str = "",
+    verify_ssl: bool | None = None,
+) -> ProxmoxClient:
+    cfg = registry.resolve(
+        target=target,
+        url=url,
+        token_id=token_id,
+        token_secret=token_secret,
+        verify_ssl=verify_ssl,
+    )
+    return ProxmoxClient(
+        url=cfg.normalized_url(),
+        token_id=cfg.token_id,
+        token_secret=cfg.token_secret,
+        verify_ssl=cfg.verify_ssl,
+    )

src/proxmox_mcp/targets_store.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from proxmox_mcp.targets import TargetRegistry
+
+
+def targets_path() -> Path:
+    from proxmox_mcp.config import settings
+
+    return Path(os.path.expanduser(settings.targets_path))
+
+
+def load_registry(registry: TargetRegistry) -> None:
+    path = targets_path()
+    if not path.exists():
+        return
+    try:
+        data = json.loads(path.read_text())
+    except (OSError, json.JSONDecodeError):
+        return
+
+    registry._loading = True
+    loaded_active: str | None = None
+    for entry in data.get("targets", []):
+        if not isinstance(entry, dict):
+            continue
+        name = str(entry.get("name", "")).strip()
+        url = str(entry.get("url", "")).strip()
+        if not name or not url:
+            continue
+        try:
+            registry.register(
+                name=name,
+                url=url,
+                token_id=str(entry.get("token_id", "")),
+                token_secret=str(entry.get("token_secret", "")),
+                verify_ssl=bool(entry.get("verify_ssl", False)),
+                description=str(entry.get("description", "")),
+                registered_at=str(entry.get("registered_at", "")),
+                set_active=False,
+            )
+        except Exception:
+            continue
+
+    active = str(data.get("active", "")).strip()
+    if active:
+        try:
+            registry.set_active(active)
+            loaded_active = active
+        except Exception:
+            pass
+    if not loaded_active and registry.list():
+        registry.set_active(registry.list()[0]["name"])
+    registry._loading = False
+
+
+def save_registry(registry: TargetRegistry) -> None:
+    path = targets_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    targets: list[dict[str, Any]] = []
+    for cfg in registry._targets.values():
+        targets.append(
+            {
+                "name": cfg.name,
+                "url": cfg.normalized_url(),
+                "token_id": cfg.token_id,
+                "token_secret": cfg.token_secret,
+                "verify_ssl": cfg.verify_ssl,
+                "description": cfg.description,
+                "registered_at": cfg.registered_at,
+            }
+        )
+    payload = {"active": registry._active, "targets": sorted(targets, key=lambda r: r["name"])}
+    path.write_text(json.dumps(payload, indent=2) + "\n")
+    try:
+        path.chmod(0o600)
+    except OSError:
+        pass

src/proxmox_mcp/tools/__init__.py

@@ -0,0 +1,10 @@
+from mcp.server.fastmcp import FastMCP
+
+from proxmox_mcp.tools import guests, nodes, targets, tasks
+
+
+def register_all(mcp: FastMCP) -> None:
+    targets.register(mcp)
+    nodes.register(mcp)
+    guests.register(mcp)
+    tasks.register(mcp)

src/proxmox_mcp/tools/guests.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import json
+
+from mcp.server.fastmcp import FastMCP
+
+from proxmox_mcp.client import get_client
+from proxmox_mcp.guards import GuardError, require_writes
+
+
+def register(mcp: FastMCP) -> None:
+    @mcp.tool()
+    def proxmox_list_qemu(node: str, target: str = "") -> str:
+        """List QEMU VMs on a node."""
+        client = get_client(target=target)
+        return json.dumps(client.get(f"nodes/{node}/qemu"), indent=2, default=str)
+
+    @mcp.tool()
+    def proxmox_list_lxc(node: str, target: str = "") -> str:
+        """List LXC containers on a node."""
+        client = get_client(target=target)
+        return json.dumps(client.get(f"nodes/{node}/lxc"), indent=2, default=str)
+
+    @mcp.tool()
+    def proxmox_get_qemu_status(node: str, vmid: int, target: str = "") -> str:
+        client = get_client(target=target)
+        return json.dumps(client.get(f"nodes/{node}/qemu/{vmid}/status/current"), indent=2, default=str)
+
+    @mcp.tool()
+    def proxmox_get_lxc_status(node: str, vmid: int, target: str = "") -> str:
+        client = get_client(target=target)
+        return json.dumps(client.get(f"nodes/{node}/lxc/{vmid}/status/current"), indent=2, default=str)
+
+    @mcp.tool()
+    def proxmox_get_qemu_config(node: str, vmid: int, target: str = "") -> str:
+        client = get_client(target=target)
+        return json.dumps(client.get(f"nodes/{node}/qemu/{vmid}/config"), indent=2, default=str)
+
+    @mcp.tool()
+    def proxmox_lxc_exec(
+        node: str,
+        vmid: int,
+        command: str,
+        args: str = "",
+        target: str = "",
+    ) -> str:
+        """Execute a command inside an LXC (requires root@pam permissions)."""
+        require_writes()
+        client = get_client(target=target)
+        data: dict[str, str] = {"command": command}
+        if args.strip():
+            data["extra-args"] = args
+        result = client.post(f"nodes/{node}/lxc/{vmid}/exec", data=data)
+        return json.dumps(result, indent=2, default=str)
+
+    @mcp.tool()
+    def proxmox_qemu_agent_exec(
+        node: str,
+        vmid: int,
+        command: str,
+        args: list[str] | None = None,
+        target: str = "",
+    ) -> str:
+        """Execute via QEMU guest agent (VM must have agent enabled)."""
+        require_writes()
+        client = get_client(target=target)
+        payload = {"command": [command] + (args or [])}
+        result = client.post(f"nodes/{node}/qemu/{vmid}/agent/exec", data={"command": json.dumps(payload["command"])})
+        return json.dumps(result, indent=2, default=str)

src/proxmox_mcp/tools/nodes.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import json
+
+from mcp.server.fastmcp import FastMCP
+
+from proxmox_mcp.client import ProxmoxAPIError, get_client
+from proxmox_mcp.guards import GuardError
+
+
+def register(mcp: FastMCP) -> None:
+    @mcp.tool()
+    def proxmox_test_connection(
+        target: str = "",
+        url: str = "",
+        token_id: str = "",
+        token_secret: str = "",
+    ) -> str:
+        try:
+            client = get_client(target=target, url=url, token_id=token_id, token_secret=token_secret)
+            version = client.get("version")
+            nodes = client.get("nodes")
+            return json.dumps({"status": "ok", "version": version, "nodes": nodes}, indent=2, default=str)
+        except (GuardError, ProxmoxAPIError) as exc:
+            return json.dumps({"status": "error", "message": str(exc)}, indent=2)
+
+    @mcp.tool()
+    def proxmox_list_nodes(target: str = "") -> str:
+        client = get_client(target=target)
+        return json.dumps(client.get("nodes"), indent=2, default=str)
+
+    @mcp.tool()
+    def proxmox_get_node_status(node: str, target: str = "") -> str:
+        client = get_client(target=target)
+        return json.dumps(client.get(f"nodes/{node}/status"), indent=2, default=str)

src/proxmox_mcp/tools/targets.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import json
+
+from mcp.server.fastmcp import FastMCP
+
+from proxmox_mcp.targets import registry
+
+
+def register(mcp: FastMCP) -> None:
+    @mcp.tool()
+    def proxmox_connect(
+        url: str,
+        token_id: str,
+        token_secret: str,
+        name: str = "",
+        verify_ssl: bool = False,
+        description: str = "",
+        set_active: bool = True,
+    ) -> str:
+        """Register a Proxmox VE target. token_id format: user@realm!tokenname."""
+        cfg = registry.register(
+            name=name,
+            url=url,
+            token_id=token_id,
+            token_secret=token_secret,
+            verify_ssl=verify_ssl,
+            description=description,
+            set_active=set_active,
+        )
+        return json.dumps(
+            {"status": "connected", "target": cfg.public_view(), "active": registry.active().name},
+            indent=2,
+        )
+
+    @mcp.tool()
+    def proxmox_use_target(name: str) -> str:
+        cfg = registry.set_active(name)
+        return json.dumps({"status": "ok", "active": cfg.public_view()}, indent=2)
+
+    @mcp.tool()
+    def proxmox_list_targets() -> str:
+        return json.dumps(
+            {"active": registry.active().name if registry.active() else None, "targets": registry.list()},
+            indent=2,
+        )
+
+    @mcp.tool()
+    def proxmox_disconnect_target(name: str) -> str:
+        registry.unregister(name)
+        return json.dumps({"status": "removed", "name": name}, indent=2)

src/proxmox_mcp/tools/tasks.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+import json
+
+from mcp.server.fastmcp import FastMCP
+
+from proxmox_mcp.client import get_client
+
+
+def register(mcp: FastMCP) -> None:
+    @mcp.tool()
+    def proxmox_get_task_status(node: str, upid: str, target: str = "") -> str:
+        client = get_client(target=target)
+        return json.dumps(client.get(f"nodes/{node}/tasks/{upid}/status"), indent=2, default=str)
+
+    @mcp.tool()
+    def proxmox_get_task_log(node: str, upid: str, limit: int = 50, target: str = "") -> str:
+        client = get_client(target=target)
+        return json.dumps(
+            client.get(f"nodes/{node}/tasks/{upid}/log?limit={limit}"),
+            indent=2,
+            default=str,
+        )