diff --git a/src/proxmox_mcp/tools/guests.py b/src/proxmox_mcp/tools/guests.py index 44880c8..a4f8556 100644 --- a/src/proxmox_mcp/tools/guests.py +++ b/src/proxmox_mcp/tools/guests.py @@ -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) diff --git a/src/proxmox_mcp/tools/nodes.py b/src/proxmox_mcp/tools/nodes.py index 66b9a0a..3517669 100644 --- a/src/proxmox_mcp/tools/nodes.py +++ b/src/proxmox_mcp/tools/nodes.py @@ -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)