commit dab31bf8d3debdf6a241ad4a0d0af388bb634375 Author: nearxos Date: Fri Jun 12 22:54:08 2026 +0300 Initial proxmox-mcp: MCP server for vessel troubleshoot stack. Part of GeneseasX golden-vessel tooling (see homelab-vault Projects/vessel-troubleshoot-stack). diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..06d763a --- /dev/null +++ b/.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 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c068ad8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +.env +.venv/ +dist/ +build/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..01f6585 --- /dev/null +++ b/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 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..94434eb --- /dev/null +++ b/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"] diff --git a/src/proxmox_mcp/__init__.py b/src/proxmox_mcp/__init__.py new file mode 100644 index 0000000..5becc17 --- /dev/null +++ b/src/proxmox_mcp/__init__.py @@ -0,0 +1 @@ +__version__ = "1.0.0" diff --git a/src/proxmox_mcp/__main__.py b/src/proxmox_mcp/__main__.py new file mode 100644 index 0000000..c311350 --- /dev/null +++ b/src/proxmox_mcp/__main__.py @@ -0,0 +1,4 @@ +from proxmox_mcp.server import main + +if __name__ == "__main__": + main() diff --git a/src/proxmox_mcp/client.py b/src/proxmox_mcp/client.py new file mode 100644 index 0000000..c668853 --- /dev/null +++ b/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, + ) diff --git a/src/proxmox_mcp/config.py b/src/proxmox_mcp/config.py new file mode 100644 index 0000000..ac82d99 --- /dev/null +++ b/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() diff --git a/src/proxmox_mcp/guards.py b/src/proxmox_mcp/guards.py new file mode 100644 index 0000000..2e19e00 --- /dev/null +++ b/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.") diff --git a/src/proxmox_mcp/server.py b/src/proxmox_mcp/server.py new file mode 100644 index 0000000..0477033 --- /dev/null +++ b/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() diff --git a/src/proxmox_mcp/targets.py b/src/proxmox_mcp/targets.py new file mode 100644 index 0000000..0f13991 --- /dev/null +++ b/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, + ) diff --git a/src/proxmox_mcp/targets_store.py b/src/proxmox_mcp/targets_store.py new file mode 100644 index 0000000..e55618b --- /dev/null +++ b/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 diff --git a/src/proxmox_mcp/tools/__init__.py b/src/proxmox_mcp/tools/__init__.py new file mode 100644 index 0000000..14628e1 --- /dev/null +++ b/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) diff --git a/src/proxmox_mcp/tools/guests.py b/src/proxmox_mcp/tools/guests.py new file mode 100644 index 0000000..120655f --- /dev/null +++ b/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) diff --git a/src/proxmox_mcp/tools/nodes.py b/src/proxmox_mcp/tools/nodes.py new file mode 100644 index 0000000..50aacd1 --- /dev/null +++ b/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) diff --git a/src/proxmox_mcp/tools/targets.py b/src/proxmox_mcp/tools/targets.py new file mode 100644 index 0000000..007658f --- /dev/null +++ b/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) diff --git a/src/proxmox_mcp/tools/tasks.py b/src/proxmox_mcp/tools/tasks.py new file mode 100644 index 0000000..c767544 --- /dev/null +++ b/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, + )