nearxos/proxmox-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Proxmox MCP fixes: P1 snapshot tools, P4 error wrapping

Browse Source
This commit is contained in:
root
2026-07-04 20:15:58 +00:00
parent 2e371b2709
commit 4c62d830a7
2 changed files with 248 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

355
src/proxmox_mcp/tools/guests.py
View File

@@ -12,81 +12,111 @@ 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)
try:
client = get_client(target=target)
return json.dumps(client.get(f"nodes/{node}/qemu"), indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@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)
try:
client = get_client(target=target)
return json.dumps(client.get(f"nodes/{node}/lxc"), indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@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)
try:
client = get_client(target=target)
return json.dumps(client.get(f"nodes/{node}/qemu/{vmid}/status/current"), indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@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)
try:
client = get_client(target=target)
return json.dumps(client.get(f"nodes/{node}/lxc/{vmid}/status/current"), indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@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)
try:
client = get_client(target=target)
return json.dumps(client.get(f"nodes/{node}/qemu/{vmid}/config"), indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@mcp.tool()
def proxmox_get_lxc_config(node: str, vmid: int, target: str = "") -> str:
"""Get LXC container configuration."""
client = get_client(target=target)
return json.dumps(client.get(f"nodes/{node}/lxc/{vmid}/config"), indent=2, default=str)
try:
client = get_client(target=target)
return json.dumps(client.get(f"nodes/{node}/lxc/{vmid}/config"), indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
# ── Lifecycle tools ──────────────────────────────────────────
@mcp.tool()
def proxmox_start_vm(node: str, vmid: int, target: str = "") -> str:
"""Start a stopped VM or LXC container."""
require_writes()
client = get_client(target=target)
try:
result = client.post(f"nodes/{node}/qemu/{vmid}/status/start")
except ProxmoxAPIError:
result = client.post(f"nodes/{node}/lxc/{vmid}/status/start")
return json.dumps(result, indent=2, default=str)
require_writes()
client = get_client(target=target)
try:
result = client.post(f"nodes/{node}/qemu/{vmid}/status/start")
except ProxmoxAPIError:
result = client.post(f"nodes/{node}/lxc/{vmid}/status/start")
return json.dumps(result, indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@mcp.tool()
def proxmox_stop_vm(node: str, vmid: int, target: str = "") -> str:
"""Stop a VM or LXC container (immediate stop)."""
require_writes()
client = get_client(target=target)
try:
result = client.post(f"nodes/{node}/qemu/{vmid}/status/stop")
except ProxmoxAPIError:
result = client.post(f"nodes/{node}/lxc/{vmid}/status/stop")
return json.dumps(result, indent=2, default=str)
require_writes()
client = get_client(target=target)
try:
result = client.post(f"nodes/{node}/qemu/{vmid}/status/stop")
except ProxmoxAPIError:
result = client.post(f"nodes/{node}/lxc/{vmid}/status/stop")
return json.dumps(result, indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@mcp.tool()
def proxmox_shutdown_vm(node: str, vmid: int, timeout: int = 60, target: str = "") -> str:
"""Gracefully shutdown a VM or LXC (ACPI shutdown). Timeout in seconds."""
require_writes()
client = get_client(target=target)
try:
result = client.post(f"nodes/{node}/qemu/{vmid}/status/shutdown", data={"timeout": timeout})
except ProxmoxAPIError:
result = client.post(f"nodes/{node}/lxc/{vmid}/status/shutdown", data={"timeout": timeout})
return json.dumps(result, indent=2, default=str)
require_writes()
client = get_client(target=target)
try:
result = client.post(f"nodes/{node}/qemu/{vmid}/status/shutdown", data={"timeout": timeout})
except ProxmoxAPIError:
result = client.post(f"nodes/{node}/lxc/{vmid}/status/shutdown", data={"timeout": timeout})
return json.dumps(result, indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@mcp.tool()
def proxmox_reboot_vm(node: str, vmid: int, target: str = "") -> str:
"""Reboot a VM or LXC container."""
require_writes()
client = get_client(target=target)
def proxmox_restart_vm(node: str, vmid: int, target: str = "") -> str:
"""Restart a VM or LXC container."""
try:
result = client.post(f"nodes/{node}/qemu/{vmid}/status/reboot")
except ProxmoxAPIError:
result = client.post(f"nodes/{node}/lxc/{vmid}/status/reboot")
return json.dumps(result, indent=2, default=str)
require_writes()
client = get_client(target=target)
try:
result = client.post(f"nodes/{node}/qemu/{vmid}/status/reboot")
except ProxmoxAPIError:
result = client.post(f"nodes/{node}/lxc/{vmid}/status/reboot")
return json.dumps(result, indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
# ── Guest exec tools ─────────────────────────────────────────
@@ -99,13 +129,16 @@ def register(mcp: FastMCP) -> None:
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)
try:
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)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@mcp.tool()
def proxmox_qemu_agent_exec(
@@ -116,11 +149,14 @@ def register(mcp: FastMCP) -> 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)
try:
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)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
# ── Create tools ─────────────────────────────────────────────
@@ -148,28 +184,31 @@ def register(mcp: FastMCP) -> None:
net0: e.g. 'name=eth0,bridge=vmbr0,ip=10.77.30.100/24,gw=10.77.30.1'
features: e.g. 'nesting=1'
"""
require_writes()
client = get_client(target=target)
data: dict = {
"vmid": vmid,
"ostemplate": ostemplate,
"storage": storage,
"hostname": hostname,
"cores": cores,
"memory": memory,
"swap": swap,
"rootfs": f"{storage}:{disk}",
"unprivileged": int(unprivileged),
"start": int(start),
}
if password:
data["password"] = password
if net0:
data["net0"] = net0
if features:
data["features"] = features
result = client.post(f"nodes/{node}/lxc", data=data)
return json.dumps(result, indent=2, default=str)
try:
require_writes()
client = get_client(target=target)
data: dict = {
"vmid": vmid,
"ostemplate": ostemplate,
"storage": storage,
"hostname": hostname,
"cores": cores,
"memory": memory,
"swap": swap,
"rootfs": f"{storage}:{disk}",
"unprivileged": int(unprivileged),
"start": int(start),
}
if password:
data["password"] = password
if net0:
data["net0"] = net0
if features:
data["features"] = features
result = client.post(f"nodes/{node}/lxc", data=data)
return json.dumps(result, indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@mcp.tool()
def proxmox_create_qemu(
@@ -192,25 +231,28 @@ def register(mcp: FastMCP) -> None:
ide2: e.g. 'local:iso/debian-12-netinst.iso,media=cdrom'
net0: e.g. 'virtio,bridge=vmbr0'
"""
require_writes()
client = get_client(target=target)
data: dict = {
"vmid": vmid,
"name": name,
"ostype": ostype,
"cores": cores,
"memory": memory,
"scsihw": scsihw,
"start": int(start),
}
if storage and disk:
data["scsi0"] = f"{storage}:{disk}"
if net0:
data["net0"] = net0
if ide2:
data["ide2"] = ide2
result = client.post(f"nodes/{node}/qemu", data=data)
return json.dumps(result, indent=2, default=str)
try:
require_writes()
client = get_client(target=target)
data: dict = {
"vmid": vmid,
"name": name,
"ostype": ostype,
"cores": cores,
"memory": memory,
"scsihw": scsihw,
"start": int(start),
}
if storage and disk:
data["scsi0"] = f"{storage}:{disk}"
if net0:
data["net0"] = net0
if ide2:
data["ide2"] = ide2
result = client.post(f"nodes/{node}/qemu", data=data)
return json.dumps(result, indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
# ── Delete / clone tools ─────────────────────────────────────
@@ -227,26 +269,29 @@ def register(mcp: FastMCP) -> None:
Set purge=true to remove from disk (destroys all data).
Set destroy_unreferenced=true to also remove unreferenced disks.
"""
require_writes()
client = get_client(target=target)
# Determine type — try QEMU first
try:
client.get(f"nodes/{node}/qemu/{vmid}/status/current")
endpoint = f"nodes/{node}/qemu/{vmid}"
except ProxmoxAPIError:
endpoint = f"nodes/{node}/lxc/{vmid}"
require_writes()
client = get_client(target=target)
# Determine type — try QEMU first
try:
client.get(f"nodes/{node}/qemu/{vmid}/status/current")
endpoint = f"nodes/{node}/qemu/{vmid}"
except ProxmoxAPIError:
endpoint = f"nodes/{node}/lxc/{vmid}"
params = {}
if purge:
params["purge"] = 1
if destroy_unreferenced:
params["destroy-unreferenced-disks"] = 1
# Append query params to endpoint
if params:
qs = "&".join(f"{k}={v}" for k, v in params.items())
endpoint = f"{endpoint}?{qs}"
result = client.delete(endpoint)
return json.dumps(result, indent=2, default=str)
params = {}
if purge:
params["purge"] = 1
if destroy_unreferenced:
params["destroy-unreferenced-disks"] = 1
# Append query params to endpoint
if params:
qs = "&".join(f"{k}={v}" for k, v in params.items())
endpoint = f"{endpoint}?{qs}"
result = client.delete(endpoint)
return json.dumps(result, indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@mcp.tool()
def proxmox_clone_vm(
@@ -258,18 +303,84 @@ def register(mcp: FastMCP) -> None:
target: str = "",
) -> str:
"""Clone a VM or LXC to a new VMID."""
require_writes()
client = get_client(target=target)
try:
client.get(f"nodes/{node}/qemu/{vmid}/status/current")
endpoint = f"nodes/{node}/qemu/{vmid}/clone"
except ProxmoxAPIError:
endpoint = f"nodes/{node}/lxc/{vmid}/clone"
require_writes()
client = get_client(target=target)
try:
client.get(f"nodes/{node}/qemu/{vmid}/status/current")
endpoint = f"nodes/{node}/qemu/{vmid}/clone"
except ProxmoxAPIError:
endpoint = f"nodes/{node}/lxc/{vmid}/clone"
data: dict = {"newid": newid}
if name:
data["name"] = name
if storage:
data["storage"] = storage
result = client.post(endpoint, data=data)
return json.dumps(result, indent=2, default=str)
data: dict = {"newid": newid}
if name:
data["name"] = name
if storage:
data["storage"] = storage
result = client.post(endpoint, data=data)
return json.dumps(result, indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
# ── Snapshot tools ───────────────────────────────────────────
@mcp.tool()
def proxmox_create_snapshot(
node: str,
vmid: int,
snapshot_name: str,
description: str = "",
target: str = "",
) -> str:
"""Create a VM snapshot. VM must be running or stopped."""
try:
require_writes()
client = get_client(target=target)
data: dict = {"snapname": snapshot_name}
if description:
data["description"] = description
result = client.post(f"nodes/{node}/qemu/{vmid}/snapshot", data=data)
return json.dumps(result, indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@mcp.tool()
def proxmox_list_snapshots(node: str, vmid: int, target: str = "") -> str:
"""List all snapshots for a VM."""
try:
client = get_client(target=target)
return json.dumps(client.get(f"nodes/{node}/qemu/{vmid}/snapshot"), indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@mcp.tool()
def proxmox_rollback_vm(
node: str,
vmid: int,
snapshot_name: str,
target: str = "",
) -> str:
"""Rollback a VM to a snapshot. VM must be stopped."""
try:
require_writes()
client = get_client(target=target)
result = client.post(f"nodes/{node}/qemu/{vmid}/snapshot/{snapshot_name}/rollback")
return json.dumps(result, indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@mcp.tool()
def proxmox_delete_snapshot(
node: str,
vmid: int,
snapshot_name: str,
target: str = "",
) -> str:
"""Delete a VM snapshot."""
try:
require_writes()
client = get_client(target=target)
result = client.delete(f"nodes/{node}/qemu/{vmid}/snapshot/{snapshot_name}")
return json.dumps(result, indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)

21
src/proxmox_mcp/tools/nodes.py
View File

@@ -26,16 +26,25 @@ def register(mcp: FastMCP) -> None:
@mcp.tool()
def proxmox_list_nodes(target: str = "") -> str:
client = get_client(target=target)
return json.dumps(client.get("nodes"), indent=2, default=str)
try:
client = get_client(target=target)
return json.dumps(client.get("nodes"), indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@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)
try:
client = get_client(target=target)
return json.dumps(client.get(f"nodes/{node}/status"), indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)
@mcp.tool()
def proxmox_get_next_vmid(target: str = "") -> str:
"""Get the next available VMID from the cluster."""
client = get_client(target=target)
return json.dumps(client.get("cluster/nextid"), indent=2, default=str)
try:
client = get_client(target=target)
return json.dumps(client.get("cluster/nextid"), indent=2, default=str)
except (GuardError, ProxmoxAPIError) as exc:
return json.dumps({"status": "error", "message": str(exc)}, indent=2)