Proxmox MCP Tool w Pythonie.

Proxmox VE to popularny hypervisor open-source, który świetnie nadaje się do zarządzania maszynami wirtualnymi w środowisku domowym lub firmowym. W tym artykule pokażę, jak zbudować serwer MCP, który pozwoli Claude’owi (lub innemu asystentowi AI) zarządzać Proxmoxem przez rozmowę — listować VM-ki, sprawdzać konfigurację klastra, a nawet uruchamiać i zatrzymywać maszyny.

Jeśli nie wiesz jeszcze czym jest MCP, zacznij od mojego wcześniejszego artykułu: Czym jest MCP (Model Context Protocol)?

---

Po co MCP, skoro można napisać skrypt?

To dobre pytanie. Jeśli masz powtarzalną czynność — np. codzienny raport z backupów wysyłany mailem — klasyczny skrypt w Pythonie uruchamiany przez cron będzie lepszym wyborem. Skrypt jest przewidywalny, testowalny i nie wymaga modelu AI.

MCP ma sens tam, gdzie czynność jest jednorazowa lub ad-hoc. Zamiast pisać skrypt, uruchamiać go, przetwarzać output i zastanawiać się co znaczą poszczególne wartości — po prostu pytasz Clauda. Kilka przykładów:

• „Które VM-ki nie miały backupu od ponad tygodnia?” — Claude odpytuje Proxmox, porównuje timestampy i zwraca gotową listę.
• „Zrób mi raport zużycia zasobów na wszystkich nodach w formacie tabeli” — dane z API trafiają prosto do modelu, który formatuje je tak jak chcesz.
• „Sprawdź czy VM 105 ma poprawnie ustawiony cloud-init przed klonowaniem” — jednorazowa weryfikacja bez pisania czegokolwiek.

Drugi argument to integracja z innymi narzędziami. Claude z podłączonym MCP do Proxmoxa może w tej samej rozmowie sięgnąć do innych serwerów MCP — np. do Jiry żeby założyć ticket, do Slacka żeby wysłać powiadomienie, albo do serwera plików żeby zapisać raport. Taka integracja skryptami wymagałaby dodatkowego kleju, tu dzieje się naturalnie w jednym prompcie.

---

Czego użyjemy

• Python 3.10+
• fastmcp — framework do budowania serwerów MCP w Pythonie
• proxmoxer — klient REST API dla Proxmox VE
• Proxmox VE 7.x lub 8.x z dostępem do API

Instalacja zależności:

pip install fastmcp proxmoxer

---

Krok 1: Tworzenie użytkownika i tokenu API w Proxmox

Zamiast używać hasła roota, Proxmox pozwala tworzyć dedykowanych użytkowników i tokeny API z ograniczonymi uprawnieniami. To bezpieczniejsze podejście — jeśli token wycieknie, wystarczy go unieważnić.

Pełna dokumentacja: Proxmox User Management (oficjalna wiki)

Przez interfejs webowy

1. Zaloguj się do Proxmox Web UI (np. https://192.168.1.10:8006)
2. Przejdź do Datacenter → Permissions → Users
3. Kliknij Add i utwórz użytkownika, np. mcp@pve (realm: pve)
4. Przejdź do Datacenter → Permissions → API Tokens
5. Kliknij Add, wybierz użytkownika mcp@pve, nadaj nazwę tokenu np. mcp-token
6. Odznacz „Privilege Separation” jeśli chcesz, żeby token dziedziczył uprawnienia usera, lub zostaw zaznaczone dla dodatkowej izolacji
7. Skopiuj i zapisz wyświetlony secret tokenu — zobaczysz go tylko raz

Nadanie uprawnień

Przejdź do Datacenter → Permissions, kliknij Add → User Permission i przypisz rolę:

• PVEAuditor — wystarczy do odczytu (list_vms, cluster_info, itd.)
• PVEVMAdmin — potrzebne do operacji zapisu (start/stop/clone/delete)
• Path: / — dla całego klastra

Możesz też zrobić to przez CLI na nodzie Proxmox:

# Utwórz usera
pveum user add mcp@pve --comment "MCP API user"

# Utwórz token
pveum user token add mcp@pve mcp-token --privsep 0

# Nadaj uprawnienia (rola PVEAuditor na całym klastrze)
pveum aclmod / -user mcp@pve -role PVEAuditor

---

Krok 2: Budowa serwera MCP — kod z wyjaśnieniem

Inicjalizacja i połączenie z Proxmox

import os
from typing import Any, Dict, List, Tuple, Optional

from fastmcp import FastMCP
from proxmoxer import ProxmoxAPI

# Flaga globalna: czy zezwalamy na operacje zapisu (stop/start/delete/clone)?
ALLOW_WRITE = os.getenv("ALLOW_WRITE", "false").lower() == "true"

# Tworzymy instancję serwera MCP — nazwa widoczna w kliencie (np. Claude Desktop)
mcp = FastMCP("Proxmox MCP Server")


def ensure_write_allowed(confirm: str, expected: str):
    """Sprawdza czy operacje zapisu są włączone i czy potwierdzenie jest prawidłowe."""
    if not ALLOW_WRITE:
        raise RuntimeError("Write operations are disabled (ALLOW_WRITE=false)")
    if confirm != expected:
        raise ValueError(f"Invalid confirmation. Expected: {expected}")


def env_bool(name: str, default: bool = False) -> bool:
    """Pomocnik do odczytu zmiennych bool z env."""
    v = os.getenv(name)
    if v is None:
        return default
    return v.strip().lower() in ("1", "true", "yes", "y", "on")


def split_token_id(token_id: str) -> Tuple[str, str]:
    """Rozdziela token_id w formacie 'user@realm!tokenname' na dwie części."""
    user, token_name = token_id.split("!", 1)
    return user, token_name


def proxmox() -> ProxmoxAPI:
    """Tworzy połączenie z Proxmox API przy użyciu tokenu z env variables."""
    host = os.environ["PVE_HOST"]
    token_id = os.environ["PVE_TOKEN_ID"]
    token_secret = os.environ["PVE_TOKEN_SECRET"]

    user, token_name = split_token_id(token_id)

    return ProxmoxAPI(
        host,
        user=user,
        token_name=token_name,
        token_value=token_secret,          # secret zawsze z env, nie hardkodowany!
        verify_ssl=env_bool("PVE_VERIFY_SSL", default=True),
    )

Kilka kluczowych decyzji w tym fragmencie:

• Połączenie przy każdym wywołaniu — proxmox() to fabryka, nie singleton. ProxmoxAPI jest bezstanowy (REST), więc nie ma sensu trzymać połączenia.
• Token zamiast hasła — proxmoxer wspiera token API bezpośrednio, co jest bezpieczniejsze niż przekazywanie hasła.
• ALLOW_WRITE jako zabezpieczenie — domyślnie serwer jest read-only. Żeby AI mogła cokolwiek zmodyfikować, trzeba świadomie ustawić tę flagę.

---

Narzędzia odczytu (read-only tools)

@mcp.tool()
def cluster_info() -> Dict[str, Any]:
    """Podstawowe info o klastrze: wersja PVE, status quorum, lista nodów."""
    p = proxmox()
    return {
        "version": p.version.get(),           # GET /version
        "cluster_status": p.cluster.status.get(),  # GET /cluster/status
    }


@mcp.tool()
def vm_inventory() -> List[Dict[str, Any]]:
    """
    Lista wszystkich VM (QEMU) w klastrze.
    Zwraca: vmid, name, node, status, cpu, mem, uptime, tags.
    Używa /cluster/resources — jednego wywołania API dla całego klastra.
    """
    p = proxmox()
    return [r for r in p.cluster.resources.get() if r.get("type") == "qemu"]


@mcp.tool()
def vm_config(vmid: int) -> Dict[str, Any]:
    """
    Szczegółowa konfiguracja konkretnej VM: dyski, sieć, BIOS, cloud-init, tagi.
    Przydatne do audytu.
    """
    p = proxmox()
    # Najpierw szukamy na którym nodzie jest dana VM
    node = None
    for r in p.cluster.resources.get():
        if r.get("type") == "qemu" and r.get("vmid") == vmid:
            node = r.get("node")
            break
    if not node:
        raise ValueError(f"VMID {vmid} not found")

    config = p.nodes(node).qemu(vmid).config.get()  # GET /nodes/{node}/qemu/{vmid}/config
    return {"vmid": vmid, "node": node, "config": config}


@mcp.tool()
def node_inventory() -> List[Dict[str, Any]]:
    """Status wszystkich nodów: kernel, uptime, load, CPU, RAM."""
    p = proxmox()
    out = []
    for n in p.nodes.get():
        node = n["node"]
        status = p.nodes(node).status.get()
        out.append({"node": node, "summary": n, "status": status})
    return out


@mcp.tool()
def storage_inventory() -> Dict[str, Any]:
    """
    Inventory storage: definicje z /storage + status per node (free/used/active).
    """
    p = proxmox()
    storages = p.storage.get()
    per_node = []
    for n in p.nodes.get():
        node = n["node"]
        per_node.append({"node": node, "storage": p.nodes(node).storage.get()})
    return {"storages": storages, "per_node": per_node}


@mcp.tool()
def network_inventory() -> List[Dict[str, Any]]:
    """Sieci na nodach: bridges, bonds, VLANy. Przydatne do audytu segmentacji."""
    p = proxmox()
    out = []
    for n in p.nodes.get():
        node = n["node"]
        out.append({"node": node, "network": p.nodes(node).network.get()})
    return out


@mcp.tool()
def user_permissions() -> Dict[str, Any]:
    """Użytkownicy, ACL i tokeny API — do audytu uprawnień."""
    p = proxmox()
    tokens = None
    try:
        tokens = p.access.tokens.get()  # może być niedostępne dla roli Auditor
    except Exception:
        pass
    return {
        "users": p.access.users.get(),
        "acl": p.access.acl.get(),
        "tokens": tokens,
    }

Dekorator @mcp.tool() to magia FastMCP — na podstawie sygnatury funkcji i docstringa automatycznie generuje schemat JSON, który klient MCP (np. Claude) używa do zrozumienia co dane narzędzie robi i jakie przyjmuje parametry.

---

Narzędzia zapisu (write tools) z potwierdzeniem

Operacje destruktywne (stop, start, delete, clone) wymagają dwóch warunków: flagi ALLOW_WRITE=true oraz przesłania specjalnego stringa potwierdzającego. Dzięki temu AI nie może przypadkowo skasować VM-ki.

@mcp.tool()
def stop_vm(vmid: int, confirm: str) -> str:
    """
    Zatrzymuje VM.
    Aby wykonać, ustaw ALLOW_WRITE=true i przekaż confirm='STOP {vmid}'.
    """
    ensure_write_allowed(confirm, f"STOP {vmid}")
    p = proxmox()
    for r in p.cluster.resources.get():
        if r.get("type") == "qemu" and r.get("vmid") == vmid:
            p.nodes(r["node"]).qemu(vmid).status.stop.post()
            return f"VM {vmid} stopped on node {r['node']}"
    raise ValueError(f"VMID {vmid} not found")


@mcp.tool()
def start_vm(vmid: int, confirm: str) -> str:
    """
    Uruchamia VM.
    Wymaga confirm='START {vmid}'.
    """
    ensure_write_allowed(confirm, f"START {vmid}")
    p = proxmox()
    for r in p.cluster.resources.get():
        if r.get("type") == "qemu" and r.get("vmid") == vmid:
            p.nodes(r["node"]).qemu(vmid).status.start.post()
            return f"VM {vmid} start issued on node {r['node']}"
    raise ValueError(f"VMID {vmid} not found")


@mcp.tool()
def delete_vm(vmid: int, confirm: str) -> str:
    """
    Kasuje VM. Operacja nieodwracalna!
    Wymaga confirm='DELETE {vmid}'.
    """
    ensure_write_allowed(confirm, f"DELETE {vmid}")
    p = proxmox()
    for r in p.cluster.resources.get():
        if r.get("type") == "qemu" and r.get("vmid") == vmid:
            p.nodes(r["node"]).qemu(vmid).delete()
            return f"VM {vmid} deleted from node {r['node']}"
    raise ValueError(f"VMID {vmid} not found")


@mcp.tool()
def clone_vm(vmid: int, newid: int, name: str, full: bool, storage: str, confirm: str) -> str:
    """
    Klonuje VM lub szablon.
    full=True oznacza pełny klon (niezależny), False = linked clone.
    Wymaga confirm='CLONE {vmid} {newid}'.
    """
    ensure_write_allowed(confirm, f"CLONE {vmid} {newid}")
    p = proxmox()
    for r in p.cluster.resources.get():
        if r.get("type") == "qemu" and r.get("vmid") == vmid:
            p.nodes(r["node"]).qemu(vmid).clone.post(
                newid=newid,
                name=name,
                full=1 if full else 0,
                storage=storage,
            )
            return f"Clone {vmid} -> {newid} ({name}) started on node {r['node']}"
    raise ValueError(f"VMID {vmid} not found")


@mcp.tool()
def set_vm_cloudinit(vmid: int, ciuser: str, cipassword: str,
                     ipconfig0: str, nameserver: str, confirm: str) -> str:
    """
    Konfiguruje cloud-init na VM.
    ipconfig0 format: 'ip=10.10.10.5/24,gw=10.10.10.1'
    nameserver format: '9.9.9.9 8.8.8.8'
    Wymaga confirm='CLOUDINIT {vmid}'.
    """
    ensure_write_allowed(confirm, f"CLOUDINIT {vmid}")
    p = proxmox()
    for r in p.cluster.resources.get():
        if r.get("type") == "qemu" and r.get("vmid") == vmid:
            p.nodes(r["node"]).qemu(vmid).config.post(
                ciuser=ciuser,
                cipassword=cipassword,
                ipconfig0=ipconfig0,
                nameserver=nameserver,
            )
            return f"Cloud-Init set on VM {vmid}"
    raise ValueError(f"VMID {vmid} not found")

---

Uruchomienie serwera

if __name__ == "__main__":
    # FastMCP używa stdio jako domyślnego transportu —
    # klient (np. Claude Desktop) komunikuje się z serwerem przez stdin/stdout
    mcp.run(show_banner=False, log_level="ERROR")

---

Krok 3: Podłączenie do Claude Desktop

Dodaj wpis do pliku konfiguracyjnego Claude Desktop (~/Library/Application Support/Claude/claude_desktop_config.json na macOS):

{
  "mcpServers": {
    "proxmox": {
      "command": "python",
      "args": ["/ścieżka/do/mcp_server_proxmox.py"],
      "env": {
        "PVE_HOST": "192.168.1.10",
        "PVE_TOKEN_ID": "mcp@pve!mcp-token",
        "PVE_TOKEN_SECRET": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
        "PVE_VERIFY_SSL": "false",
        "ALLOW_WRITE": "false"
      }
    }
  }
}

Po restarcie Claude Desktop zobaczysz nowe narzędzia dostępne w oknie czatu. Możesz zapytać Clauda np. „Które VM-ki są aktualnie uruchomione?” albo „Pokaż mi konfigurację sieci na nodach”.

---

Podsumowanie: dostępne narzędzia MCP

Narzędzie	Typ	Opis
cluster_info	odczyt	Wersja PVE, status klastra i quorum
vm_inventory	odczyt	Lista wszystkich VM z całego klastra
vm_config	odczyt	Szczegółowa konfiguracja konkretnej VM
node_inventory	odczyt	Status wszystkich nodów (CPU, RAM, uptime)
storage_inventory	odczyt	Definicje i status storage
network_inventory	odczyt	Sieci, bridges, VLANy na nodach
user_permissions	odczyt	Użytkownicy, ACL, tokeny API
backup_inventory	odczyt	Lista backupów z /cluster/backup
stop_vm	zapis*	Zatrzymuje VM (wymaga potwierdzenia)
start_vm	zapis*	Uruchamia VM (wymaga potwierdzenia)
delete_vm	zapis*	Kasuje VM — nieodwracalne! (wymaga potwierdzenia)
clone_vm	zapis*	Klonuje VM lub szablon (wymaga potwierdzenia)
set_vm_cloudinit	zapis*	Ustawia cloud-init na VM (wymaga potwierdzenia)

* Narzędzia zapisu wymagają ALLOW_WRITE=true oraz przesłania stringa potwierdzającego (np. "STOP 100").

---

Uwagi bezpieczeństwa

• Nigdy nie hardkoduj tokenu w kodzie. Zawsze używaj zmiennych środowiskowych (PVE_TOKEN_SECRET).
• Zasada minimalnych uprawnień — jeśli używasz serwera tylko do odczytu, przypisz roli PVEAuditor. Nie dawaj Administrator.
• ALLOW_WRITE=false domyślnie — włącz zapis tylko gdy naprawdę potrzebujesz i na czas sesji.
• Mechanizm potwierdzenia chroni przed przypadkowym wywołaniem destruktywnej akcji przez AI — model musi świadomie wygenerować odpowiedni string (np. "DELETE 105").