diff --git a/proxmox_tools.py b/proxmox_tools.py index d32357b..62ab6fa 100644 --- a/proxmox_tools.py +++ b/proxmox_tools.py @@ -1,166 +1,194 @@ -#!/usr/bin/env python3 """ -proxmox_tools.py - Proxmox API Tools -Description: Full toolset for Proxmox VE management: - - LXC (create, start, stop, delete) - - VM (clone from template, create from ISO, manage) - - Monitoring (nodes, storage, resources, tasks) -Usage: - from proxmox_tools import Tools - tools = Tools() - print(tools.connection()) - print(tools.vm_create(vmid=200, name="ubuntu", ciuser="ubuntu", cipassword="1234567")) +title: Proxmox Tools (PVE) +author: (your name) +description: Инструменты для управления Proxmox VE: LXC (создание, старт, стоп, удаление), VM (клонирование из шаблона, создание из ISO, управление), мониторинг (ноды, хранилище, ресурсы, задачи). +version: 1.0.0 """ -import json -import re import time import requests - -# ============================================================ -# CONFIG - change as needed -# ============================================================ -PROXMOX_HOST = "https://192.168.31.2:8006" -PROXMOX_USER = "root@pam" -PROXMOX_TOKEN_NAME = "openwebui-token" -PROXMOX_TOKEN_VALUE = "7aba7743-0133-494b-9dd7-9715f55e5dca" -VERIFY_SSL = False -DEFAULT_NODE = "pve" -DEFAULT_STORAGE = "local" -DEFAULT_BRIDGE = "vmbr0" - -# ============================================================ -# INTERNAL HELPERS -# ============================================================ -_base_url = PROXMOX_HOST.rstrip("/") + "/api2/json" -_auth_header = { - "Authorization": f"PVEAPIToken={PROXMOX_USER}!{PROXMOX_TOKEN_NAME}={PROXMOX_TOKEN_VALUE}" -} +import urllib3 +from pydantic import BaseModel, Field +from typing import Optional -def _api(method, path, params=None, data=None): - """Execute a Proxmox API request.""" - url = _base_url + path - resp = requests.request( - method=method, - url=url, - headers=_auth_header, - params=params, - json=data, - verify=VERIFY_SSL, - timeout=30, - ) - resp.raise_for_status() - return resp.json().get("data", {}) - - -def _fmt(b): - """Format bytes to human-readable.""" - if b is None: - return "N/A" - b = float(b) - for u in ["B", "KB", "MB", "GB", "TB"]: - if b < 1024: - return f"{b:.1f} {u}" - b /= 1024 - return f"{b:.1f} PB" - - -def _pct(u, t): - """Format used/total as '1.2 GB / 4.0 GB (30.0%)'.""" - if u is None or t is None or t == 0: - return "N/A" - return f"{_fmt(u)} / {_fmt(t)} ({u/t*100:.1f}%)" - - -def _uptime(s): - """Format uptime seconds to '1d 2h 3m 4s'.""" - if s is None: - return "N/A" - d, r = divmod(int(s), 86400) - h, r = divmod(r, 3600) - m, s2 = divmod(r, 60) - parts = [] - if d: - parts.append(f"{d}d") - if h: - parts.append(f"{h}h") - if m: - parts.append(f"{m}m") - parts.append(f"{s2}s") - return " ".join(parts) - - -def _wait_for_task(upid, timeout=60): - """ - Wait for a Proxmox task to complete by UPID. - Returns (ok: bool, message: str). - """ - if not upid or not isinstance(upid, str) or not upid.startswith("UPID:"): - return True, "no UPID to wait" - node = DEFAULT_NODE - start = time.time() - while time.time() - start < timeout: - try: - tasks = _api("GET", f"/nodes/{node}/tasks") - for t in tasks: - if t.get("upid") == upid: - status = t.get("status", "") - if status == "stopped": - if t.get("exitstatus") == "OK": - return True, "OK" - else: - return False, str(t) - break - except Exception: - pass - time.sleep(2) - return False, "timeout waiting for task" - - -# ============================================================ -# MAIN TOOLS CLASS -# ============================================================ class Tools: - """Proxmox VE management toolset.""" + class Valves(BaseModel): + proxmox_host: str = Field( + default="https://192.168.31.2:8006", + description="Base URL of the Proxmox API (e.g. https://192.168.31.2:8006)", + ) + proxmox_user: str = Field( + default="root@pam", description="Proxmox API user (e.g. root@pam)" + ) + proxmox_token_name: str = Field( + default="openwebui-token", description="Proxmox API token name/id" + ) + proxmox_token_value: str = Field( + default="", + description="Proxmox API token secret (set this in the Valves UI, never commit it to code)", + ) + verify_ssl: bool = Field( + default=False, description="Verify TLS certificate of the Proxmox host" + ) + default_node: str = Field( + default="pve", description="Default Proxmox node name" + ) + default_storage: str = Field( + default="local", description="Default storage for LXC rootfs / templates" + ) + default_bridge: str = Field( + default="vmbr0", description="Default network bridge" + ) + default_template_vmid: int = Field( + default=1000, + description="Default VMID of the cloud-init template used for VM cloning", + ) + request_timeout: int = Field( + default=30, description="HTTP request timeout in seconds" + ) - # ---- CONNECTION & NODES -------------------------------------------- + def __init__(self): + self.valves = self.Valves() + # Silence "InsecureRequestWarning" spam when verify_ssl is False + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + # ============================================================ + # INTERNAL HELPERS (not exposed to the LLM) + # ============================================================ + + def _base_url(self) -> str: + return self.valves.proxmox_host.rstrip("/") + "/api2/json" + + def _auth_header(self) -> dict: + return { + "Authorization": ( + f"PVEAPIToken={self.valves.proxmox_user}!" + f"{self.valves.proxmox_token_name}={self.valves.proxmox_token_value}" + ) + } + + def _api( + self, + method: str, + path: str, + params: Optional[dict] = None, + data: Optional[dict] = None, + ): + """Execute a Proxmox API request.""" + url = self._base_url() + path + resp = requests.request( + method=method, + url=url, + headers=self._auth_header(), + params=params, + json=data, + verify=self.valves.verify_ssl, + timeout=self.valves.request_timeout, + ) + resp.raise_for_status() + return resp.json().get("data", {}) @staticmethod - def connection(): - """Check Proxmox connection.""" + def _fmt(b) -> str: + """Format bytes to human-readable.""" + if b is None: + return "N/A" + b = float(b) + for u in ["B", "KB", "MB", "GB", "TB"]: + if b < 1024: + return f"{b:.1f} {u}" + b /= 1024 + return f"{b:.1f} PB" + + @classmethod + def _pct(cls, u, t) -> str: + """Format used/total as '1.2 GB / 4.0 GB (30.0%)'.""" + if u is None or t is None or t == 0: + return "N/A" + return f"{cls._fmt(u)} / {cls._fmt(t)} ({u/t*100:.1f}%)" + + @staticmethod + def _uptime(s) -> str: + """Format uptime seconds to '1d 2h 3m 4s'.""" + if s is None: + return "N/A" + d, r = divmod(int(s), 86400) + h, r = divmod(r, 3600) + m, s2 = divmod(r, 60) + parts = [] + if d: + parts.append(f"{d}d") + if h: + parts.append(f"{h}h") + if m: + parts.append(f"{m}m") + parts.append(f"{s2}s") + return " ".join(parts) + + def _wait_for_task(self, upid: str, node: str, timeout: int = 60): + """Wait for a Proxmox task to complete by UPID, on the given node.""" + if not upid or not isinstance(upid, str) or not upid.startswith("UPID:"): + return True, "no UPID to wait" + start = time.time() + while time.time() - start < timeout: + try: + tasks = self._api("GET", f"/nodes/{node}/tasks") + for t in tasks: + if t.get("upid") == upid: + status = t.get("status", "") + if status == "stopped": + return ( + (True, "OK") + if t.get("exitstatus") == "OK" + else (False, str(t)) + ) + break + except Exception: + pass + time.sleep(2) + return False, "timeout waiting for task" + + # ============================================================ + # TOOLS (exposed to the LLM) + # ============================================================ + + def connection(self) -> str: + """Check the connection to the Proxmox VE API and report its version.""" try: - v = _api("GET", "/version") + v = self._api("GET", "/version") return ( "OK Proxmox VE " + v.get("version", "?") - + " | node: pve | IP: 192.168.31.2" + + f" | node: {self.valves.default_node} | host: {self.valves.proxmox_host}" ) except Exception as e: return "ERR " + str(e) - @staticmethod - def nodes(): - """Show all cluster nodes.""" + def nodes(self) -> str: + """List all cluster nodes with their CPU/RAM usage and uptime.""" try: - items = _api("GET", "/nodes") + items = self._api("GET", "/nodes") out = ["Nodes:"] for n in items: out.append(f"- {n.get('node', '?')} ({n.get('status', '?')})") out.append( - f" CPU: {n.get('cpu', 0)*100:.1f}% | " - f"RAM: {_pct(n.get('mem'), n.get('maxmem'))}" + f" CPU: {n.get('cpu', 0)*100:.1f}% | RAM: {self._pct(n.get('mem'), n.get('maxmem'))}" ) - out.append(f" Uptime: {_uptime(n.get('uptime'))}") + out.append(f" Uptime: {self._uptime(n.get('uptime'))}") return "\n".join(out) except Exception as e: return "ERR " + str(e) - @staticmethod - def node_status(node=DEFAULT_NODE): - """Show detailed node status.""" + def node_status(self, node: Optional[str] = None) -> str: + """Show detailed status of a Proxmox node: CPU model, load, RAM, swap, disk, uptime, kernel and PVE version. + + Args: + node: Node name. Defaults to the configured default node if omitted. + """ + node = node or self.valves.default_node try: - s = _api("GET", f"/nodes/{node}/status") + s = self._api("GET", f"/nodes/{node}/status") out = [f"Node {node}:"] ci = s.get("cpuinfo", {}) out.append(f"CPU: {ci.get('model', '?')} ({s.get('cpus', 0)} cores)") @@ -169,14 +197,14 @@ class Tools: if la: out.append(f"Loadavg: {' / '.join(f'{float(v):.2f}' for v in la)}") m = s.get("memory", {}) - out.append(f"RAM: {_pct(m.get('used'), m.get('total'))}") + out.append(f"RAM: {self._pct(m.get('used'), m.get('total'))}") sw = s.get("swap", {}) if sw.get("total", 0): - out.append(f"SWAP: {_pct(sw.get('used'), sw.get('total'))}") + out.append(f"SWAP: {self._pct(sw.get('used'), sw.get('total'))}") r = s.get("rootfs", {}) if r: - out.append(f"Disk: {_pct(r.get('used'), r.get('total'))}") - out.append(f"Uptime: {_uptime(s.get('uptime'))}") + out.append(f"Disk: {self._pct(r.get('used'), r.get('total'))}") + out.append(f"Uptime: {self._uptime(s.get('uptime'))}") out.append(f"Kernel: {s.get('kversion', '?')}") if s.get("pveversion"): out.append(f"PVE: {s['pveversion']}") @@ -184,11 +212,10 @@ class Tools: except Exception as e: return "ERR " + str(e) - @staticmethod - def cluster(): - """Show cluster status (quorum, members).""" + def cluster(self) -> str: + """Show cluster status: quorum state and cluster members.""" try: - items = _api("GET", "/cluster/status") + items = self._api("GET", "/cluster/status") out = ["Cluster:"] has_cluster_info = False for item in items: @@ -196,13 +223,11 @@ class Tools: if t == "cluster": has_cluster_info = True out.append( - f"Cluster: {item.get('name', '?')} | " - f"Quorum: {'OK' if item.get('quorate') else 'FAIL'}" + f"Cluster: {item.get('name', '?')} | Quorum: {'OK' if item.get('quorate') else 'FAIL'}" ) elif t == "node": out.append( - f" {item.get('name', '?')} - {item.get('status', '?')} " - f"({item.get('ip', '-')})" + f" {item.get('name', '?')} - {item.get('status', '?')} ({item.get('ip', '-')})" ) if not has_cluster_info: node_count = sum(1 for i in items if i.get("type") == "node") @@ -212,16 +237,15 @@ class Tools: except Exception as e: return "ERR " + str(e) - # ---- RESOURCES & STORAGE ------------------------------------------ + def resources(self, ftype: str = "") -> str: + """List all cluster resources (nodes, QEMU VMs, LXC containers, storage), optionally filtered by type. - @staticmethod - def resources(ftype=""): - """Show all cluster resources (VMs, LXC, nodes, storage).""" + Args: + ftype: Optional filter: "node", "qemu", "lxc", or "storage". Empty string returns all types. + """ try: - p = {} - if ftype: - p["type"] = ftype - items = _api("GET", "/cluster/resources", params=p) + p = {"type": ftype} if ftype else {} + items = self._api("GET", "/cluster/resources", params=p) out = ["Resources:"] g = {} for r in items: @@ -241,7 +265,7 @@ class Tools: if t in ("qemu", "lxc"): line += f" (ID:{r.get('vmid', '?')})" if r.get("maxmem"): - line += f" RAM:{_pct(r.get('mem'), r.get('maxmem'))}" + line += f" RAM:{self._pct(r.get('mem'), r.get('maxmem'))}" if r.get("cpu"): line += f" CPU:{r.get('cpu')*100:.1f}%" out.append(line) @@ -250,11 +274,10 @@ class Tools: except Exception as e: return "ERR " + str(e) - @staticmethod - def storage(): - """Show all storages.""" + def storage(self) -> str: + """List all configured storages with their content types and usage.""" try: - items = _api("GET", "/storage") + items = self._api("GET", "/storage") if not items: return "No storage" out = ["Storage:"] @@ -263,70 +286,79 @@ class Tools: out.append(f"- {name} ({s.get('type', '?')})") out.append(f" Content: {s.get('content', '?')}") if s.get("total"): - out.append(f" Used: {_pct(s.get('used'), s.get('total'))}") - out.append(f" Free: {_fmt(s.get('avail'))}") + out.append(f" Used: {self._pct(s.get('used'), s.get('total'))}") + out.append(f" Free: {self._fmt(s.get('avail'))}") out.append("") return "\n".join(out) except Exception as e: return "ERR " + str(e) - @staticmethod - def tasks(limit=10, node=DEFAULT_NODE): - """Show recent tasks on node.""" + def tasks(self, limit: int = 10, node: Optional[str] = None) -> str: + """Show recent tasks on a node. + + Args: + limit: Maximum number of tasks to return. + node: Node name. Defaults to the configured default node if omitted. + """ + node = node or self.valves.default_node try: - items = _api("GET", f"/nodes/{node}/tasks", params={"limit": limit}) + items = self._api("GET", f"/nodes/{node}/tasks", params={"limit": limit}) if not items: return f"No tasks on {node}" out = [f"Tasks on {node}:"] for t in items: start = time.ctime(t.get("starttime", 0)) if t.get("starttime") else "?" out.append( - f"- {t.get('type', '?')} | {t.get('status', '?')} | " - f"{t.get('user', '?')} | {start}" + f"- {t.get('type', '?')} | {t.get('status', '?')} | {t.get('user', '?')} | {start}" ) return "\n".join(out) except Exception as e: return "ERR " + str(e) - # ---- LXC CONTAINERS ----------------------------------------------- + def lxc_list(self, node: Optional[str] = None) -> str: + """List LXC containers on a node with status, CPU, RAM, disk and uptime. - @staticmethod - def lxc_list(node=DEFAULT_NODE): - """Show LXC container list.""" + Args: + node: Node name. Defaults to the configured default node if omitted. + """ + node = node or self.valves.default_node try: - items = _api("GET", f"/nodes/{node}/lxc") + items = self._api("GET", f"/nodes/{node}/lxc") if not items: return f"No LXC on {node}" out = [f"LXC on {node}:"] for c in items: out.append( - f"- {c.get('name', '?')} (ID:{c.get('vmid', '?')}) - " - f"{c.get('status', '?')}" + f"- {c.get('name', '?')} (ID:{c.get('vmid', '?')}) - {c.get('status', '?')}" ) if c.get("cpu"): out.append(f" CPU:{c['cpu']*100:.1f}%") if c.get("maxmem"): - out.append(f" RAM:{_pct(c.get('mem'), c.get('maxmem'))}") + out.append(f" RAM:{self._pct(c.get('mem'), c.get('maxmem'))}") if c.get("maxdisk"): - out.append(f" Disk:{_pct(c.get('disk'), c.get('maxdisk'))}") + out.append(f" Disk:{self._pct(c.get('disk'), c.get('maxdisk'))}") if c.get("uptime"): - out.append(f" Uptime:{_uptime(c['uptime'])}") + out.append(f" Uptime:{self._uptime(c['uptime'])}") out.append("") return "\n".join(out) except Exception as e: return "ERR " + str(e) - @staticmethod - def lxc_detail(vmid, node=DEFAULT_NODE): - """Show detailed LXC info.""" + def lxc_detail(self, vmid: int, node: Optional[str] = None) -> str: + """Show detailed info about a specific LXC container: OS, CPU, RAM, disk, network, usage and uptime. + + Args: + vmid: The numeric ID of the LXC container. + node: Node name. Defaults to the configured default node if omitted. + """ + node = node or self.valves.default_node try: - s = _api("GET", f"/nodes/{node}/lxc/{vmid}/status/current") - c = _api("GET", f"/nodes/{node}/lxc/{vmid}/config") + s = self._api("GET", f"/nodes/{node}/lxc/{vmid}/status/current") + c = self._api("GET", f"/nodes/{node}/lxc/{vmid}/config") name = s.get("name", c.get("hostname", "?")) out = [f"LXC #{vmid}: {name} - {s.get('status', '?')}"] out.append( - f"OS: {c.get('ostype', '?')} | CPU: {c.get('cores', '?')} | " - f"RAM: {c.get('memory', '?')}MB" + f"OS: {c.get('ostype', '?')} | CPU: {c.get('cores', '?')} | RAM: {c.get('memory', '?')}MB" ) out.append(f"Disk: {c.get('rootfs', '?')} | SWAP: {c.get('swap', '?')}MB") out.append(f"Hostname: {c.get('hostname', '?')}") @@ -334,20 +366,24 @@ class Tools: if nets: out.append("Network:\n" + "\n".join(nets)) if s.get("maxmem"): - out.append(f"RAM use: {_pct(s.get('mem'), s.get('maxmem'))}") + out.append(f"RAM use: {self._pct(s.get('mem'), s.get('maxmem'))}") if s.get("cpu"): out.append(f"CPU: {s['cpu']*100:.1f}%") if s.get("uptime"): - out.append(f"Uptime: {_uptime(s['uptime'])}") + out.append(f"Uptime: {self._uptime(s['uptime'])}") return "\n".join(out) except Exception as e: return "ERR " + str(e) - @staticmethod - def lxc_action(vmid, action, node=DEFAULT_NODE): - """ - Manage LXC: start, stop, shutdown, restart, suspend, resume, delete. + def lxc_action(self, vmid: int, action: str, node: Optional[str] = None) -> str: + """Perform a lifecycle action on an LXC container. + + Args: + vmid: The numeric ID of the LXC container. + action: One of "start", "stop", "shutdown", "restart", "suspend", "resume", "delete". + node: Node name. Defaults to the configured default node if omitted. """ + node = node or self.valves.default_node try: acts = { "start": "start", @@ -361,51 +397,88 @@ class Tools: if action not in acts: return f"Unknown action: {action}. Allowed: {', '.join(acts.keys())}" if action == "delete": - result = _api("DELETE", f"/nodes/{node}/lxc/{vmid}") + result = self._api("DELETE", f"/nodes/{node}/lxc/{vmid}") else: - result = _api("POST", f"/nodes/{node}/lxc/{vmid}/status/{acts[action]}") + result = self._api( + "POST", f"/nodes/{node}/lxc/{vmid}/status/{acts[action]}" + ) return f"OK LXC #{vmid} {action}\n{result}" except Exception as e: return "ERR " + str(e) - @staticmethod - def templates(storage="local", node=DEFAULT_NODE): - """Show available LXC templates in storage.""" + def templates( + self, storage: Optional[str] = None, node: Optional[str] = None + ) -> str: + """List available LXC templates in a storage. + + Args: + storage: Storage name. Defaults to the configured default storage if omitted. + node: Node name. Defaults to the configured default node if omitted. + """ + node = node or self.valves.default_node + storage = storage or self.valves.default_storage try: - content = _api("GET", f"/nodes/{node}/storage/{storage}/content") + content = self._api("GET", f"/nodes/{node}/storage/{storage}/content") tmpl = [x for x in content if x.get("content") == "vztmpl"] if not tmpl: return f"No templates in {storage}" out = [f"Templates in {storage}:"] for t in tmpl: - out.append(f" {t.get('volid', '?')} - {_fmt(t.get('size'))}") + out.append(f" {t.get('volid', '?')} - {self._fmt(t.get('size'))}") return "\n".join(out) except Exception as e: return "ERR " + str(e) - @staticmethod def lxc_create( - vmid, - hostname, - ostemplate, - password="", - sshkey="", - storage=DEFAULT_STORAGE, - cores=1, - memory=512, - swap=0, - disk="8G", - bridge=DEFAULT_BRIDGE, - ip="dhcp", - netmask=24, - gw="", - dns="", - domain="", - node=DEFAULT_NODE, - unpriv=True, - start_now=True, - ): - """Create a new LXC container.""" + self, + vmid: int, + hostname: str, + ostemplate: str, + password: str = "", + sshkey: str = "", + storage: Optional[str] = None, + cores: int = 1, + memory: int = 512, + swap: int = 0, + disk: str = "8G", + bridge: Optional[str] = None, + ip: str = "dhcp", + netmask: int = 24, + gw: str = "", + dns: str = "", + domain: str = "", + node: Optional[str] = None, + unpriv: bool = True, + start_now: bool = True, + wait: bool = True, + ) -> str: + """Create a new LXC container from a template. + + Args: + vmid: New numeric ID for the container (must not already exist). + hostname: Hostname for the container. + ostemplate: Template volume id, e.g. "local:vztmpl/debian-12-standard_12.7-1_amd64.tar.zst". + password: Root password (required if sshkey is not provided). + sshkey: SSH public key (required if password is not provided). + storage: Storage for the rootfs. Defaults to the configured default storage if omitted. + cores: Number of CPU cores. + memory: RAM in MB. + swap: Swap in MB. + disk: Root disk size, e.g. "8G". + bridge: Network bridge. Defaults to the configured default bridge if omitted. + ip: IP address, or "dhcp". + netmask: Netmask prefix length, used only if ip is not "dhcp". + gw: Gateway IP, used only if ip is not "dhcp". + dns: Nameserver IP. + domain: Search domain. + node: Node name. Defaults to the configured default node if omitted. + unpriv: Whether to create an unprivileged container. + start_now: Whether to start the container right after creation. + wait: Whether to wait for the creation task to finish before returning. + """ + node = node or self.valves.default_node + storage = storage or self.valves.default_storage + bridge = bridge or self.valves.default_bridge try: if not hostname: return "Error: hostname required" @@ -413,8 +486,10 @@ class Tools: return "Error: ostemplate required" if not password and not sshkey: return "Error: password or sshkey required" - exist = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] - exist += [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] + exist = [str(x.get("vmid")) for x in self._api("GET", f"/nodes/{node}/lxc")] + exist += [ + str(x.get("vmid")) for x in self._api("GET", f"/nodes/{node}/qemu") + ] if str(vmid) in exist: return f"Error: VMID {vmid} already exists" params = { @@ -446,71 +521,93 @@ class Tools: params["searchdomain"] = domain if start_now: params["start"] = 1 - result = _api("POST", f"/nodes/{node}/lxc", data=params) - out = [f"OK LXC #{vmid} creating!"] - out.append(f" Name: {hostname}") - out.append(f" Template: {ostemplate}") - out.append(f" CPU: {cores} cores, RAM: {memory}MB, Disk: {disk}") - if ip and ip.lower() != "dhcp": - out.append(f" Net: {bridge}, IP: {ip}/{netmask}") - else: - out.append(f" Net: {bridge}, IP: DHCP") - if isinstance(result, str) and result: - out.append(f"UPID: {result}") + result = self._api("POST", f"/nodes/{node}/lxc", data=params) + out = [ + f"OK LXC #{vmid} creating!", + f" Name: {hostname}", + f" Template: {ostemplate}", + f" CPU: {cores} cores, RAM: {memory}MB, Disk: {disk}", + ] + out.append( + f" Net: {bridge}, IP: {'DHCP' if ip == 'dhcp' else f'{ip}/{netmask}'}" + ) + upid = result if isinstance(result, str) else "" + if upid: + out.append(f"UPID: {upid}") + if wait and upid: + out.append("Waiting for creation to finish...") + ok, msg = self._wait_for_task(upid, node=node, timeout=120) + out.append( + " Creation done" + if ok + else f" WARN Creation may not have finished: {msg}" + ) return "\n".join(out) except Exception as e: return "ERR " + str(e) - # ---- VIRTUAL MACHINES (QEMU) ------------------------------------- + def vm_list(self, node: Optional[str] = None) -> str: + """List QEMU VMs on a node with status, CPU, RAM and uptime. - @staticmethod - def vm_list(node=DEFAULT_NODE): - """Show all VM (QEMU) list.""" + Args: + node: Node name. Defaults to the configured default node if omitted. + """ + node = node or self.valves.default_node try: - items = _api("GET", f"/nodes/{node}/qemu") + items = self._api("GET", f"/nodes/{node}/qemu") if not items: return f"No VMs on {node}" out = [f"VMs on {node}:"] for v in items: out.append( - f"- {v.get('name', '?')} (ID:{v.get('vmid', '?')}) - " - f"{v.get('status', '?')}" + f"- {v.get('name', '?')} (ID:{v.get('vmid', '?')}) - {v.get('status', '?')}" ) if v.get("cpu"): out.append(f" CPU:{v['cpu']*100:.1f}%") if v.get("maxmem"): - out.append(f" RAM:{_pct(v.get('mem'), v.get('maxmem'))}") + out.append(f" RAM:{self._pct(v.get('mem'), v.get('maxmem'))}") if v.get("uptime"): - out.append(f" Uptime:{_uptime(v['uptime'])}") + out.append(f" Uptime:{self._uptime(v['uptime'])}") out.append("") return "\n".join(out) except Exception as e: return "ERR " + str(e) - @staticmethod - def vm_detail(vmid, node=DEFAULT_NODE): - """Show detailed VM info.""" + def vm_detail(self, vmid: int, node: Optional[str] = None) -> str: + """Show detailed status of a specific QEMU VM: RAM, CPU, disk and uptime. + + Args: + vmid: The numeric ID of the VM. + node: Node name. Defaults to the configured default node if omitted. + """ + node = node or self.valves.default_node try: - s = _api("GET", f"/nodes/{node}/qemu/{vmid}/status/current") - out = [f"VM #{vmid}: {s.get('name', '?')}"] - out.append(f"Status: {s.get('status', '?')}") + s = self._api("GET", f"/nodes/{node}/qemu/{vmid}/status/current") + out = [ + f"VM #{vmid}: {s.get('name', '?')}", + f"Status: {s.get('status', '?')}", + ] if s.get("maxmem"): - out.append(f"RAM: {_pct(s.get('mem'), s.get('maxmem'))}") + out.append(f"RAM: {self._pct(s.get('mem'), s.get('maxmem'))}") if s.get("cpu"): out.append(f"CPU: {s['cpu']*100:.1f}%") if s.get("maxdisk") and s.get("maxdisk", 0) > 0: - out.append(f"Disk: {_pct(s.get('disk'), s.get('maxdisk'))}") + out.append(f"Disk: {self._pct(s.get('disk'), s.get('maxdisk'))}") if s.get("uptime"): - out.append(f"Uptime: {_uptime(s['uptime'])}") + out.append(f"Uptime: {self._uptime(s['uptime'])}") return "\n".join(out) except Exception as e: return "ERR " + str(e) - @staticmethod - def vm_action(vmid, action, node=DEFAULT_NODE): - """ - Manage VM: start, stop, shutdown, restart, suspend, resume, delete. + def vm_action(self, vmid: int, action: str, node: Optional[str] = None) -> str: + """Perform a lifecycle action on a QEMU VM. + + Args: + vmid: The numeric ID of the VM. + action: One of "start", "stop", "shutdown", "restart", "suspend", "resume", "delete". + node: Node name. Defaults to the configured default node if omitted. """ + node = node or self.valves.default_node try: acts = { "start": "start", @@ -524,91 +621,96 @@ class Tools: if action not in acts: return f"Unknown action: {action}. Allowed: {', '.join(acts.keys())}" if action == "delete": - result = _api("DELETE", f"/nodes/{node}/qemu/{vmid}") + result = self._api("DELETE", f"/nodes/{node}/qemu/{vmid}") else: - result = _api( + result = self._api( "POST", f"/nodes/{node}/qemu/{vmid}/status/{acts[action]}" ) return f"OK VM #{vmid} {action}\n{result}" except Exception as e: return "ERR " + str(e) - @staticmethod def vm_create_from_template( - vmid, - name, - template_vmid=1000, - full=True, - memory=None, - cores=None, - storage=None, - target_node=None, - pool=None, - start_now=True, - ciuser="", - cipassword="", - sshkeys="", - node=DEFAULT_NODE, - ): - """ - Create a new VM by cloning from a template. - Default template: VM #1000 (cloud-init). + self, + vmid: int, + name: str, + template_vmid: Optional[int] = None, + full: bool = True, + memory: Optional[int] = None, + cores: Optional[int] = None, + storage: Optional[str] = None, + target_node: Optional[str] = None, + pool: Optional[str] = None, + start_now: bool = True, + ciuser: str = "", + cipassword: str = "", + sshkeys: str = "", + node: Optional[str] = None, + ) -> str: + """Create a new VM by cloning it from an existing template VM (cloud-init). - Cloud-init params (for guest OS user setup): - ciuser: username in the guest (e.g. "ubuntu") - cipassword: password for the user - sshkeys: SSH public keys (newline-separated string) + The clone step never sends a "start" flag to the API — the VM is started + via a separate call afterwards, once cloning and cloud-init settings are applied. - FIXED: "start" parameter is NOT sent to /clone - - startup is done via separate API call after clone completes. + Args: + vmid: New numeric ID for the cloned VM (must not already exist). + name: Name for the new VM. + template_vmid: VMID of the source template to clone from. Defaults to the configured default template if omitted. + full: True for a full clone, False for a linked clone. + memory: RAM in MB to set after cloning. Leave unset to keep the template's value. + cores: CPU cores to set after cloning. Leave unset to keep the template's value. + storage: Target storage for the clone. Leave unset to use the template's storage. + target_node: Target node if cloning to a different node than the source. + pool: Resource pool to assign the clone to. + start_now: Whether to start the VM right after cloning. + ciuser: Cloud-init user to set (e.g. "ubuntu"). + cipassword: Cloud-init password to set. + sshkeys: Cloud-init SSH public keys to set (newline-separated). + node: Node name where the template lives. Defaults to the configured default node if omitted. """ + node = node or self.valves.default_node + template_vmid = template_vmid or self.valves.default_template_vmid try: if not vmid: return "Error: vmid required" if not name: return "Error: name required" - - # Check VMID availability - exist_vm = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] - exist_lxc = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] + exist_vm = [ + str(x.get("vmid")) for x in self._api("GET", f"/nodes/{node}/qemu") + ] + exist_lxc = [ + str(x.get("vmid")) for x in self._api("GET", f"/nodes/{node}/lxc") + ] if str(vmid) in exist_vm + exist_lxc: return f"Error: VMID {vmid} already exists" - - # Step 1: Clone (WITHOUT "start" parameter!) - params = { - "newid": vmid, - "name": name, - "full": 1 if full else 0, - } + # Step 1: clone (without "start" - startup is a separate call below) + params = {"newid": vmid, "name": name, "full": 1 if full else 0} if storage: params["storage"] = storage if target_node: params["target"] = target_node if pool: params["pool"] = pool - - result = _api( + result = self._api( "POST", f"/nodes/{node}/qemu/{template_vmid}/clone", data=params ) - - out = [f"OK VM #{vmid}: {name} cloning from template #{template_vmid}!"] - out.append(f" Type: {'full' if full else 'linked'} clone") + out = [ + f"OK VM #{vmid}: {name} cloning from template #{template_vmid}!", + f" Type: {'full' if full else 'linked'} clone", + ] if storage: out.append(f" Storage: {storage}") - upid = result if isinstance(result, str) else "" if upid: out.append(f"UPID: {upid}") - - # Step 2: Wait for clone to finish + # Step 2: wait for clone to finish out.append("Waiting for clone to finish...") - ok, msg = _wait_for_task(upid, timeout=120) - if not ok: - out.append(f" WARN Clone may not have finished: {msg}") - else: - out.append(" Clone done") - - # Step 3: Apply settings (resources + cloud-init) + clone_node = target_node or node + ok, msg = self._wait_for_task(upid, node=clone_node, timeout=120) + out.append( + " Clone done" if ok else f" WARN Clone may not have finished: {msg}" + ) + # Step 3: apply settings (resources + cloud-init) config_params = {} if memory is not None: config_params["memory"] = memory @@ -620,13 +722,12 @@ class Tools: config_params["cipassword"] = cipassword if sshkeys: config_params["sshkeys"] = sshkeys - if config_params: out.append("Applying settings...") try: - _api( + self._api( "PUT", - f"/nodes/{node}/qemu/{vmid}/config", + f"/nodes/{clone_node}/qemu/{vmid}/config", data=config_params, ) if memory is not None: @@ -642,83 +743,77 @@ class Tools: out.append(" Settings applied") except Exception as e2: out.append(f" WARN Could not apply settings: {e2}") - - # Step 4: Start VM (separate call!) + # Step 4: start VM (separate call) if start_now: out.append("Starting VM...") try: - _api("POST", f"/nodes/{node}/qemu/{vmid}/status/start") + self._api("POST", f"/nodes/{clone_node}/qemu/{vmid}/status/start") out.append(f" VM #{vmid} started!") except Exception as e3: out.append(f" WARN Could not start: {e3}") else: out.append(f" VM #{vmid} created (not started)") - return "\n".join(out) - except Exception as e: return "ERR " + str(e) - @staticmethod def vm_create( - vmid, - name, - template_vmid=1000, - memory=4096, - cores=2, - disk_size="32G", - disk_storage="local-lvm", - iso="", - bridge=DEFAULT_BRIDGE, - ip="dhcp", - netmask=24, - gw="", - ostype="l26", - agent=1, - ciuser="", - cipassword="", - sshkeys="", - node=DEFAULT_NODE, - start_now=True, - ): - """ - Universal VM creation. + self, + vmid: int, + name: str, + template_vmid: Optional[int] = None, + memory: int = 4096, + cores: int = 2, + disk_size: str = "32G", + disk_storage: str = "local-lvm", + iso: str = "", + bridge: Optional[str] = None, + ip: str = "dhcp", + netmask: int = 24, + gw: str = "", + ostype: str = "l26", + agent: int = 1, + ciuser: str = "", + cipassword: str = "", + sshkeys: str = "", + node: Optional[str] = None, + start_now: bool = True, + wait: bool = True, + ) -> str: + """Create a new VM. Two modes: clone from a template (default), or create fresh from an ISO. - Modes: - 1. From template (default): clones VM #1000 (cloud-init) - 2. From ISO: template_vmid=None, iso="local:iso/image.iso" + Mode 1 (default): clones the configured cloud-init template. Use ciuser/cipassword/sshkeys + to set up the guest user via cloud-init. + Mode 2 (ISO): set template_vmid=0 and provide iso, ip, gw as needed. - Cloud-init params (for mode 1): - ciuser: guest OS username - cipassword: guest OS password - sshkeys: SSH public keys - - Examples: - # Clone from template with cloud-init user: - tools.vm_create( - vmid=200, - name="web-server", - memory=8192, - cores=4, - ciuser="ubuntu", - cipassword="mypassword", - start_now=True, - ) - - # Create from ISO: - tools.vm_create( - vmid=202, - name="debian-server", - template_vmid=None, - iso="local:iso/debian-12.iso", - ip="192.168.31.60", - gw="192.168.31.1", - ) + Args: + vmid: New numeric ID for the VM (must not already exist). + name: Name for the new VM. + template_vmid: VMID of a template to clone from. Set to 0 to instead create from an ISO. Defaults to the configured default template if omitted. + memory: RAM in MB. + cores: CPU cores. + disk_size: Disk size, e.g. "32G". Only used in ISO mode. + disk_storage: Storage for the disk. + iso: ISO volume id to boot from, only used in ISO mode, e.g. "local:iso/debian-12.iso". + bridge: Network bridge. Defaults to the configured default bridge if omitted. + ip: IP address, or "dhcp". Only used in ISO mode. + netmask: Netmask prefix length. Only used in ISO mode. + gw: Gateway IP. Only used in ISO mode. + ostype: Guest OS type, e.g. "l26" for Linux 2.6+/3.x/4.x/5.x/6.x. + agent: Whether to enable the QEMU guest agent (1 or 0). Only used in ISO mode. + ciuser: Cloud-init user (clone mode only). + cipassword: Cloud-init password (clone mode only). + sshkeys: Cloud-init SSH public keys (clone mode only). + node: Node name. Defaults to the configured default node if omitted. + start_now: Whether to start the VM right after creation. + wait: Whether to wait for the creation task to finish before returning (ISO mode only; clone mode always waits). """ + node = node or self.valves.default_node + bridge = bridge or self.valves.default_bridge try: # Mode 1: clone from template - if template_vmid: - return Tools.vm_create_from_template( + if template_vmid is None or template_vmid: + return self.vm_create_from_template( vmid=vmid, name=name, template_vmid=template_vmid, @@ -732,18 +827,17 @@ class Tools: sshkeys=sshkeys, node=node, ) - # Mode 2: create from scratch (ISO) - if not name: - return "Error: name required" - if not vmid: - return "Error: vmid required" - - exist_vm = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] - exist_lxc = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] + if not name or not vmid: + return "Error: name and vmid required" + exist_vm = [ + str(x.get("vmid")) for x in self._api("GET", f"/nodes/{node}/qemu") + ] + exist_lxc = [ + str(x.get("vmid")) for x in self._api("GET", f"/nodes/{node}/lxc") + ] if str(vmid) in exist_vm + exist_lxc: return f"Error: VMID {vmid} already exists" - params = { "vmid": vmid, "name": name, @@ -753,19 +847,12 @@ class Tools: "ostype": ostype, "agent": str(agent), } - - # Disk - if ":" in disk_size: - params["virtio0"] = disk_size - else: - params["virtio0"] = f"{disk_storage}:{disk_size}" - - # ISO + params["virtio0"] = ( + disk_size if ":" in disk_size else f"{disk_storage}:{disk_size}" + ) if iso: params["ide2"] = f"{iso},media=cdrom" params["boot"] = "order=ide2;virtio0" - - # Network net = f"name=eth0,bridge={bridge}" if ip and ip.lower() != "dhcp": net += f",ip={ip}/{netmask}" @@ -774,37 +861,31 @@ class Tools: else: net += ",ip=dhcp" params["net0"] = net - if start_now: params["start"] = 1 - - result = _api("POST", f"/nodes/{node}/qemu", data=params) - - out = [f"OK VM #{vmid}: {name} creating!"] - out.append(f" CPU: {cores} cores, RAM: {memory} MB") - out.append(f" Disk: {params['virtio0']}") + result = self._api("POST", f"/nodes/{node}/qemu", data=params) + out = [ + f"OK VM #{vmid}: {name} creating!", + f" CPU: {cores} cores, RAM: {memory} MB", + f" Disk: {params['virtio0']}", + ] if iso: out.append(f" ISO: {iso}") out.append( - f" Net: {bridge}, " - f"IP: {'DHCP' if ip == 'dhcp' else f'{ip}/{netmask}'}" + f" Net: {bridge}, IP: {'DHCP' if ip == 'dhcp' else f'{ip}/{netmask}'}" ) out.append(f" QEMU Agent: {'on' if agent else 'off'}") - if isinstance(result, str) and result: - out.append(f"UPID: {result}") + upid = result if isinstance(result, str) else "" + if upid: + out.append(f"UPID: {upid}") + if wait and upid: + out.append("Waiting for creation to finish...") + ok, msg = self._wait_for_task(upid, node=node, timeout=120) + out.append( + " Creation done" + if ok + else f" WARN Creation may not have finished: {msg}" + ) return "\n".join(out) - except Exception as e: return "ERR " + str(e) - - -# ============================================================ -# TEST -# ============================================================ -if __name__ == "__main__": - tools = Tools() - print(tools.connection()) - print() - print(tools.vm_list()) - print() - print(tools.storage())