This is page 2 of 2. Use http://codebase.md/rekklesna/proxmoxmcp-plus?lines=true&page={x} to view the full context.
# Directory Structure
```
├── .gitignore
├── docker-compose.yml
├── Dockerfile
├── LICENSE
├── proxmox-config
│ └── config.example.json
├── pyproject.toml
├── README.md
├── requirements-dev.in
├── requirements.in
├── setup.py
├── src
│ └── proxmox_mcp
│ ├── __init__.py
│ ├── config
│ │ ├── __init__.py
│ │ ├── loader.py
│ │ └── models.py
│ ├── core
│ │ ├── __init__.py
│ │ ├── logging.py
│ │ └── proxmox.py
│ ├── formatting
│ │ ├── __init__.py
│ │ ├── colors.py
│ │ ├── components.py
│ │ ├── formatters.py
│ │ ├── templates.py
│ │ └── theme.py
│ ├── server.py
│ ├── tools
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── cluster.py
│ │ ├── console
│ │ │ ├── __init__.py
│ │ │ └── manager.py
│ │ ├── containers.py
│ │ ├── definitions.py
│ │ ├── node.py
│ │ ├── storage.py
│ │ └── vm.py
│ └── utils
│ ├── __init__.py
│ ├── auth.py
│ └── logging.py
├── start_openapi.sh
├── start_server.sh
├── test_scripts
│ ├── README.md
│ ├── test_common.py
│ ├── test_create_vm.py
│ ├── test_openapi.py
│ ├── test_vm_power.py
│ └── test_vm_start.py
└── tests
├── __init__.py
├── test_server.py
└── test_vm_console.py
```
# Files
--------------------------------------------------------------------------------
/src/proxmox_mcp/tools/vm.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | VM-related tools for Proxmox MCP.
3 |
4 | This module provides tools for managing and interacting with Proxmox VMs:
5 | - Listing all VMs across the cluster with their status
6 | - Retrieving detailed VM information including:
7 | * Resource allocation (CPU, memory)
8 | * Runtime status
9 | * Node placement
10 | - Executing commands within VMs via QEMU guest agent
11 | - Handling VM console operations
12 | - VM power management (start, stop, shutdown, reset)
13 | - VM creation with customizable specifications
14 |
15 | The tools implement fallback mechanisms for scenarios where
16 | detailed VM information might be temporarily unavailable.
17 | """
18 | from typing import List, Optional
19 | from mcp.types import TextContent as Content
20 | from .base import ProxmoxTool
21 | from .definitions import GET_VMS_DESC, EXECUTE_VM_COMMAND_DESC
22 | from .console.manager import VMConsoleManager
23 |
24 | class VMTools(ProxmoxTool):
25 | """Tools for managing Proxmox VMs.
26 |
27 | Provides functionality for:
28 | - Retrieving cluster-wide VM information
29 | - Getting detailed VM status and configuration
30 | - Executing commands within VMs
31 | - Managing VM console operations
32 | - VM power management (start, stop, shutdown, reset)
33 | - VM creation with customizable specifications
34 |
35 | Implements fallback mechanisms for scenarios where detailed
36 | VM information might be temporarily unavailable. Integrates
37 | with QEMU guest agent for VM command execution.
38 | """
39 |
40 | def __init__(self, proxmox_api):
41 | """Initialize VM tools.
42 |
43 | Args:
44 | proxmox_api: Initialized ProxmoxAPI instance
45 | """
46 | super().__init__(proxmox_api)
47 | self.console_manager = VMConsoleManager(proxmox_api)
48 |
49 | def get_vms(self) -> List[Content]:
50 | """List all virtual machines across the cluster with detailed status.
51 |
52 | Retrieves comprehensive information for each VM including:
53 | - Basic identification (ID, name)
54 | - Runtime status (running, stopped)
55 | - Resource allocation and usage:
56 | * CPU cores
57 | * Memory allocation and usage
58 | - Node placement
59 |
60 | Implements a fallback mechanism that returns basic information
61 | if detailed configuration retrieval fails for any VM.
62 |
63 | Returns:
64 | List of Content objects containing formatted VM information:
65 | {
66 | "vmid": "100",
67 | "name": "vm-name",
68 | "status": "running/stopped",
69 | "node": "node-name",
70 | "cpus": core_count,
71 | "memory": {
72 | "used": bytes,
73 | "total": bytes
74 | }
75 | }
76 |
77 | Raises:
78 | RuntimeError: If the cluster-wide VM query fails
79 | """
80 | try:
81 | result = []
82 | for node in self.proxmox.nodes.get():
83 | node_name = node["node"]
84 | vms = self.proxmox.nodes(node_name).qemu.get()
85 | for vm in vms:
86 | vmid = vm["vmid"]
87 | # Get VM config for CPU cores
88 | try:
89 | config = self.proxmox.nodes(node_name).qemu(vmid).config.get()
90 | result.append({
91 | "vmid": vmid,
92 | "name": vm["name"],
93 | "status": vm["status"],
94 | "node": node_name,
95 | "cpus": config.get("cores", "N/A"),
96 | "memory": {
97 | "used": vm.get("mem", 0),
98 | "total": vm.get("maxmem", 0)
99 | }
100 | })
101 | except Exception:
102 | # Fallback if can't get config
103 | result.append({
104 | "vmid": vmid,
105 | "name": vm["name"],
106 | "status": vm["status"],
107 | "node": node_name,
108 | "cpus": "N/A",
109 | "memory": {
110 | "used": vm.get("mem", 0),
111 | "total": vm.get("maxmem", 0)
112 | }
113 | })
114 | return self._format_response(result, "vms")
115 | except Exception as e:
116 | self._handle_error("get VMs", e)
117 |
118 | def create_vm(self, node: str, vmid: str, name: str, cpus: int, memory: int,
119 | disk_size: int, storage: Optional[str] = None, ostype: Optional[str] = None) -> List[Content]:
120 | """Create a new virtual machine with specified configuration.
121 |
122 | Args:
123 | node: Host node name (e.g., 'pve')
124 | vmid: New VM ID number (e.g., '200')
125 | name: VM name (e.g., 'my-new-vm')
126 | cpus: Number of CPU cores (e.g., 1, 2, 4)
127 | memory: Memory size in MB (e.g., 2048 for 2GB)
128 | disk_size: Disk size in GB (e.g., 10, 20, 50)
129 | storage: Storage name (e.g., 'local-lvm', 'vm-storage'). If None, will auto-detect
130 | ostype: OS type (e.g., 'l26' for Linux, 'win10' for Windows). Default: 'l26'
131 |
132 | Returns:
133 | List of Content objects containing creation result
134 |
135 | Raises:
136 | ValueError: If VM ID already exists or invalid parameters
137 | RuntimeError: If VM creation fails
138 | """
139 | try:
140 | # Check if VM ID already exists
141 | try:
142 | existing_vm = self.proxmox.nodes(node).qemu(vmid).config.get()
143 | raise ValueError(f"VM {vmid} already exists on node {node}")
144 | except Exception as e:
145 | if "does not exist" not in str(e).lower():
146 | raise e
147 |
148 | # Get storage information
149 | storage_list = self.proxmox.nodes(node).storage.get()
150 | storage_info = {}
151 | for s in storage_list:
152 | storage_info[s["storage"]] = s
153 |
154 | # Auto-detect storage if not specified
155 | if storage is None:
156 | # Prefer local-lvm for VM images first
157 | for s in storage_list:
158 | if s["storage"] == "local-lvm" and "images" in s.get("content", ""):
159 | storage = s["storage"]
160 | break
161 | if storage is None:
162 | # Then try vm-storage
163 | for s in storage_list:
164 | if s["storage"] == "vm-storage" and "images" in s.get("content", ""):
165 | storage = s["storage"]
166 | break
167 | if storage is None:
168 | # Fallback to any storage that supports images
169 | for s in storage_list:
170 | if "images" in s.get("content", ""):
171 | storage = s["storage"]
172 | break
173 | if storage is None:
174 | raise ValueError("No suitable storage found for VM images")
175 |
176 | # Validate storage exists and supports images
177 | if storage not in storage_info:
178 | raise ValueError(f"Storage '{storage}' not found on node {node}")
179 |
180 | if "images" not in storage_info[storage].get("content", ""):
181 | raise ValueError(f"Storage '{storage}' does not support VM images")
182 |
183 | # Determine appropriate disk format based on storage type
184 | storage_type = storage_info[storage]["type"]
185 |
186 | if storage_type in ["lvm", "lvmthin"]:
187 | # LVM storages use raw format and no cloudinit
188 | disk_format = "raw"
189 | vm_config_storage = {
190 | "scsi0": f"{storage}:{disk_size},format={disk_format}",
191 | }
192 | elif storage_type in ["dir", "nfs", "cifs"]:
193 | # File-based storages can use qcow2
194 | disk_format = "qcow2"
195 | vm_config_storage = {
196 | "scsi0": f"{storage}:{disk_size},format={disk_format}",
197 | "ide2": f"{storage}:cloudinit",
198 | }
199 | else:
200 | # Default to raw for unknown storage types
201 | disk_format = "raw"
202 | vm_config_storage = {
203 | "scsi0": f"{storage}:{disk_size},format={disk_format}",
204 | }
205 |
206 | # Set default OS type
207 | if ostype is None:
208 | ostype = "l26" # Linux 2.6+ kernel
209 |
210 | # Prepare VM configuration
211 | vm_config = {
212 | "vmid": vmid,
213 | "name": name,
214 | "cores": cpus,
215 | "memory": memory,
216 | "ostype": ostype,
217 | "scsihw": "virtio-scsi-pci",
218 | "boot": "order=scsi0",
219 | "agent": "1", # Enable QEMU guest agent
220 | "vga": "std",
221 | "net0": "virtio,bridge=vmbr0",
222 | }
223 |
224 | # Add storage configuration
225 | vm_config.update(vm_config_storage)
226 |
227 | # Create the VM
228 | task_result = self.proxmox.nodes(node).qemu.create(**vm_config)
229 |
230 | cloudinit_note = ""
231 | if storage_type in ["lvm", "lvmthin"]:
232 | cloudinit_note = "\n ⚠️ Note: LVM storage doesn't support cloud-init image"
233 |
234 | result_text = f"""🎉 VM {vmid} created successfully!
235 |
236 | 📋 VM Configuration:
237 | • Name: {name}
238 | • Node: {node}
239 | • VM ID: {vmid}
240 | • CPU Cores: {cpus}
241 | • Memory: {memory} MB ({memory/1024:.1f} GB)
242 | • Disk: {disk_size} GB ({storage}, {disk_format} format)
243 | • Storage Type: {storage_type}
244 | • OS Type: {ostype}
245 | • Network: virtio (bridge=vmbr0)
246 | • QEMU Agent: Enabled{cloudinit_note}
247 |
248 | 🔧 Task ID: {task_result}
249 |
250 | 💡 Next steps:
251 | 1. Upload an ISO to install the operating system
252 | 2. Start the VM using start_vm tool
253 | 3. Access the console to complete OS installation"""
254 |
255 | return [Content(type="text", text=result_text)]
256 |
257 | except ValueError as e:
258 | raise e
259 | except Exception as e:
260 | self._handle_error(f"create VM {vmid}", e)
261 |
262 | def start_vm(self, node: str, vmid: str) -> List[Content]:
263 | """Start a virtual machine.
264 |
265 | Args:
266 | node: Host node name (e.g., 'pve1', 'proxmox-node2')
267 | vmid: VM ID number (e.g., '100', '101')
268 |
269 | Returns:
270 | List of Content objects containing operation result
271 |
272 | Raises:
273 | ValueError: If VM is not found
274 | RuntimeError: If start operation fails
275 | """
276 | try:
277 | # Check if VM exists and get current status
278 | vm_status = self.proxmox.nodes(node).qemu(vmid).status.current.get()
279 | current_status = vm_status.get("status")
280 |
281 | if current_status == "running":
282 | result_text = f"🟢 VM {vmid} is already running"
283 | else:
284 | # Start the VM
285 | task_result = self.proxmox.nodes(node).qemu(vmid).status.start.post()
286 | result_text = f"🚀 VM {vmid} start initiated successfully\nTask ID: {task_result}"
287 |
288 | return [Content(type="text", text=result_text)]
289 |
290 | except Exception as e:
291 | if "does not exist" in str(e).lower() or "not found" in str(e).lower():
292 | raise ValueError(f"VM {vmid} not found on node {node}")
293 | self._handle_error(f"start VM {vmid}", e)
294 |
295 | def stop_vm(self, node: str, vmid: str) -> List[Content]:
296 | """Stop a virtual machine (force stop).
297 |
298 | Args:
299 | node: Host node name (e.g., 'pve1', 'proxmox-node2')
300 | vmid: VM ID number (e.g., '100', '101')
301 |
302 | Returns:
303 | List of Content objects containing operation result
304 |
305 | Raises:
306 | ValueError: If VM is not found
307 | RuntimeError: If stop operation fails
308 | """
309 | try:
310 | # Check if VM exists and get current status
311 | vm_status = self.proxmox.nodes(node).qemu(vmid).status.current.get()
312 | current_status = vm_status.get("status")
313 |
314 | if current_status == "stopped":
315 | result_text = f"🔴 VM {vmid} is already stopped"
316 | else:
317 | # Stop the VM
318 | task_result = self.proxmox.nodes(node).qemu(vmid).status.stop.post()
319 | result_text = f"🛑 VM {vmid} stop initiated successfully\nTask ID: {task_result}"
320 |
321 | return [Content(type="text", text=result_text)]
322 |
323 | except Exception as e:
324 | if "does not exist" in str(e).lower() or "not found" in str(e).lower():
325 | raise ValueError(f"VM {vmid} not found on node {node}")
326 | self._handle_error(f"stop VM {vmid}", e)
327 |
328 | def shutdown_vm(self, node: str, vmid: str) -> List[Content]:
329 | """Shutdown a virtual machine gracefully.
330 |
331 | Args:
332 | node: Host node name (e.g., 'pve1', 'proxmox-node2')
333 | vmid: VM ID number (e.g., '100', '101')
334 |
335 | Returns:
336 | List of Content objects containing operation result
337 |
338 | Raises:
339 | ValueError: If VM is not found
340 | RuntimeError: If shutdown operation fails
341 | """
342 | try:
343 | # Check if VM exists and get current status
344 | vm_status = self.proxmox.nodes(node).qemu(vmid).status.current.get()
345 | current_status = vm_status.get("status")
346 |
347 | if current_status == "stopped":
348 | result_text = f"🔴 VM {vmid} is already stopped"
349 | else:
350 | # Shutdown the VM gracefully
351 | task_result = self.proxmox.nodes(node).qemu(vmid).status.shutdown.post()
352 | result_text = f"💤 VM {vmid} graceful shutdown initiated\nTask ID: {task_result}"
353 |
354 | return [Content(type="text", text=result_text)]
355 |
356 | except Exception as e:
357 | if "does not exist" in str(e).lower() or "not found" in str(e).lower():
358 | raise ValueError(f"VM {vmid} not found on node {node}")
359 | self._handle_error(f"shutdown VM {vmid}", e)
360 |
361 | def reset_vm(self, node: str, vmid: str) -> List[Content]:
362 | """Reset (restart) a virtual machine.
363 |
364 | Args:
365 | node: Host node name (e.g., 'pve1', 'proxmox-node2')
366 | vmid: VM ID number (e.g., '100', '101')
367 |
368 | Returns:
369 | List of Content objects containing operation result
370 |
371 | Raises:
372 | ValueError: If VM is not found
373 | RuntimeError: If reset operation fails
374 | """
375 | try:
376 | # Check if VM exists and get current status
377 | vm_status = self.proxmox.nodes(node).qemu(vmid).status.current.get()
378 | current_status = vm_status.get("status")
379 |
380 | if current_status == "stopped":
381 | result_text = f"⚠️ Cannot reset VM {vmid}: VM is currently stopped\nUse start_vm to start it first"
382 | else:
383 | # Reset the VM
384 | task_result = self.proxmox.nodes(node).qemu(vmid).status.reset.post()
385 | result_text = f"🔄 VM {vmid} reset initiated successfully\nTask ID: {task_result}"
386 |
387 | return [Content(type="text", text=result_text)]
388 |
389 | except Exception as e:
390 | if "does not exist" in str(e).lower() or "not found" in str(e).lower():
391 | raise ValueError(f"VM {vmid} not found on node {node}")
392 | self._handle_error(f"reset VM {vmid}", e)
393 |
394 | async def execute_command(self, node: str, vmid: str, command: str) -> List[Content]:
395 | """Execute a command in a VM via QEMU guest agent.
396 |
397 | Uses the QEMU guest agent to execute commands within a running VM.
398 | Requires:
399 | - VM must be running
400 | - QEMU guest agent must be installed and running in the VM
401 | - Command execution permissions must be enabled
402 |
403 | Args:
404 | node: Host node name (e.g., 'pve1', 'proxmox-node2')
405 | vmid: VM ID number (e.g., '100', '101')
406 | command: Shell command to run (e.g., 'uname -a', 'systemctl status nginx')
407 |
408 | Returns:
409 | List of Content objects containing formatted command output:
410 | {
411 | "success": true/false,
412 | "output": "command output",
413 | "error": "error message if any"
414 | }
415 |
416 | Raises:
417 | ValueError: If VM is not found, not running, or guest agent is not available
418 | RuntimeError: If command execution fails due to permissions or other issues
419 | """
420 | try:
421 | result = await self.console_manager.execute_command(node, vmid, command)
422 | # Use the command output formatter from ProxmoxFormatters
423 | from ..formatting import ProxmoxFormatters
424 | formatted = ProxmoxFormatters.format_command_output(
425 | success=result["success"],
426 | command=command,
427 | output=result["output"],
428 | error=result.get("error")
429 | )
430 | return [Content(type="text", text=formatted)]
431 | except Exception as e:
432 | self._handle_error(f"execute command on VM {vmid}", e)
433 |
434 | def delete_vm(self, node: str, vmid: str, force: bool = False) -> List[Content]:
435 | """Delete/remove a virtual machine completely.
436 |
437 | This will permanently delete the VM and all its associated data including:
438 | - VM configuration
439 | - Virtual disks
440 | - Snapshots
441 |
442 | WARNING: This operation cannot be undone!
443 |
444 | Args:
445 | node: Host node name (e.g., 'pve1', 'proxmox-node2')
446 | vmid: VM ID number (e.g., '100', '101')
447 | force: Force deletion even if VM is running (will stop first)
448 |
449 | Returns:
450 | List of Content objects containing deletion result
451 |
452 | Raises:
453 | ValueError: If VM is not found or is running and force=False
454 | RuntimeError: If deletion fails
455 | """
456 | try:
457 | # Check if VM exists and get current status
458 | try:
459 | vm_status = self.proxmox.nodes(node).qemu(vmid).status.current.get()
460 | current_status = vm_status.get("status")
461 | vm_name = vm_status.get("name", f"VM-{vmid}")
462 | except Exception as e:
463 | if "does not exist" in str(e).lower() or "not found" in str(e).lower():
464 | raise ValueError(f"VM {vmid} not found on node {node}")
465 | raise e
466 |
467 | # Check if VM is running
468 | if current_status == "running":
469 | if not force:
470 | raise ValueError(f"VM {vmid} ({vm_name}) is currently running. "
471 | f"Please stop it first or use force=True to stop and delete.")
472 | else:
473 | # Force stop the VM first
474 | self.proxmox.nodes(node).qemu(vmid).status.stop.post()
475 | result_text = f"🛑 Stopping VM {vmid} ({vm_name}) before deletion...\n"
476 | else:
477 | result_text = f"🗑️ Deleting VM {vmid} ({vm_name})...\n"
478 |
479 | # Delete the VM
480 | task_result = self.proxmox.nodes(node).qemu(vmid).delete()
481 |
482 | result_text += f"""🗑️ VM {vmid} ({vm_name}) deletion initiated successfully!
483 |
484 | ⚠️ WARNING: This operation will permanently remove:
485 | • VM configuration
486 | • All virtual disks
487 | • All snapshots
488 | • Cannot be undone!
489 |
490 | 🔧 Task ID: {task_result}
491 |
492 | ✅ VM {vmid} ({vm_name}) is being deleted from node {node}"""
493 |
494 | return [Content(type="text", text=result_text)]
495 |
496 | except ValueError as e:
497 | raise e
498 | except Exception as e:
499 | self._handle_error(f"delete VM {vmid}", e)
500 |
```
--------------------------------------------------------------------------------
/src/proxmox_mcp/tools/containers.py:
--------------------------------------------------------------------------------
```python
1 | from typing import List, Dict, Optional, Tuple, Any, Union
2 | import json
3 | from mcp.types import TextContent as Content
4 | from .base import ProxmoxTool
5 |
6 |
7 | def _b2h(n: Union[int, float, str]) -> str:
8 | """bytes -> human (binary units)."""
9 | try:
10 | n = float(n)
11 | except Exception:
12 | return "0.00 B"
13 | units = ("B", "KiB", "MiB", "GiB", "TiB", "PiB")
14 | i = 0
15 | while n >= 1024.0 and i < len(units) - 1:
16 | n /= 1024.0
17 | i += 1
18 | return f"{n:.2f} {units[i]}"
19 |
20 | # The rest of the helpers were preserved from your original file; no changes needed
21 |
22 |
23 | def _get(d: Any, key: str, default: Any = None) -> Any:
24 | """dict.get with None guard."""
25 | if isinstance(d, dict):
26 | return d.get(key, default)
27 | return default
28 |
29 |
30 | def _as_dict(maybe: Any) -> Dict:
31 | """Return dict; unwrap {'data': dict}; else {}."""
32 | if isinstance(maybe, dict):
33 | data = maybe.get("data")
34 | if isinstance(data, dict):
35 | return data
36 | return maybe
37 | return {}
38 |
39 |
40 | def _as_list(maybe: Any) -> List:
41 | """Return list; unwrap {'data': list}; else []."""
42 | if isinstance(maybe, list):
43 | return maybe
44 | if isinstance(maybe, dict):
45 | data = maybe.get("data")
46 | if isinstance(data, list):
47 | return data
48 | return []
49 |
50 |
51 | class ContainerTools(ProxmoxTool):
52 | """
53 | LXC container tools for Proxmox MCP.
54 |
55 | - Lists containers cluster-wide (or by node)
56 | - Live stats via /status/current
57 | - Limit fallback via /config (memory MiB, cores/cpulimit)
58 | - RRD fallback when live returns zeros
59 | - Pretty output rendered here; JSON path is raw & sanitized
60 | """
61 |
62 | # ---------- error / output ----------
63 | def _json_fmt(self, data: Any) -> List[Content]:
64 | """Return raw JSON string (never touch project formatters)."""
65 | return [Content(type="text", text=json.dumps(data, indent=2, sort_keys=True))]
66 |
67 | def _err(self, action: str, e: Exception) -> List[Content]:
68 | if hasattr(self, "handle_error"):
69 | return self.handle_error(e, action) # type: ignore[attr-defined]
70 | if hasattr(self, "_handle_error"):
71 | return self._handle_error(action, e) # type: ignore[attr-defined]
72 | return [Content(type="text", text=json.dumps({"error": str(e), "action": action}))]
73 |
74 | # ---------- helpers ----------
75 | def _list_ct_pairs(self, node: Optional[str]) -> List[Tuple[str, Dict]]:
76 | """Yield (node_name, ct_dict). Coerce odd shapes into dicts with vmid."""
77 | out: List[Tuple[str, Dict]] = []
78 | if node:
79 | raw = self.proxmox.nodes(node).lxc.get()
80 | for it in _as_list(raw):
81 | if isinstance(it, dict):
82 | out.append((node, it))
83 | else:
84 | try:
85 | vmid = int(it)
86 | out.append((node, {"vmid": vmid}))
87 | except Exception:
88 | continue
89 | else:
90 | nodes = _as_list(self.proxmox.nodes.get())
91 | for n in nodes:
92 | nname = _get(n, "node")
93 | if not nname:
94 | continue
95 | raw = self.proxmox.nodes(nname).lxc.get()
96 | for it in _as_list(raw):
97 | if isinstance(it, dict):
98 | out.append((nname, it))
99 | else:
100 | try:
101 | vmid = int(it)
102 | out.append((nname, {"vmid": vmid}))
103 | except Exception:
104 | continue
105 | return out
106 |
107 | def _rrd_last(self, node: str, vmid: int) -> Tuple[Optional[float], Optional[int], Optional[int]]:
108 | """Return (cpu_pct, mem_bytes, maxmem_bytes) from the most recent RRD sample."""
109 | try:
110 | rrd = _as_list(self.proxmox.nodes(node).lxc(vmid).rrddata.get(timeframe="hour", ds="cpu,mem,maxmem"))
111 | if not rrd or not isinstance(rrd[-1], dict):
112 | return None, None, None
113 | last = rrd[-1]
114 | # Proxmox RRD cpu is fraction already (0..1). Convert to percent.
115 | cpu_pct = float(_get(last, "cpu", 0.0) or 0.0) * 100.0
116 | mem_bytes = int(_get(last, "mem", 0) or 0)
117 | maxmem_bytes = int(_get(last, "maxmem", 0) or 0)
118 | return cpu_pct, mem_bytes, maxmem_bytes
119 | except Exception:
120 | return None, None, None
121 |
122 | def _status_and_config(self, node: str, vmid: int) -> Tuple[Dict, Dict]:
123 | """Return (status_current_dict, config_dict)."""
124 | raw_status: Dict = {}
125 | raw_config: Dict = {}
126 | try:
127 | raw_status = _as_dict(self.proxmox.nodes(node).lxc(vmid).status.current.get())
128 | except Exception:
129 | raw_status = {}
130 | try:
131 | raw_config = _as_dict(self.proxmox.nodes(node).lxc(vmid).config.get())
132 | except Exception:
133 | raw_config = {}
134 | return raw_status, raw_config
135 |
136 | def _render_pretty(self, rows: List[Dict]) -> List[Content]:
137 | lines: List[str] = ["📦 Containers", ""]
138 | for r in rows:
139 | name = r.get("name") or f"ct-{r.get('vmid')}"
140 | vmid = r.get("vmid")
141 | status = (r.get("status") or "").upper()
142 | node = r.get("node") or "?"
143 | cores = r.get("cores")
144 | cpu_pct = r.get("cpu_pct", 0.0)
145 | mem_bytes = int(r.get("mem_bytes") or 0)
146 | maxmem_bytes = int(r.get("maxmem_bytes") or 0)
147 | mem_pct = r.get("mem_pct")
148 | unlimited = bool(r.get("unlimited_memory", False))
149 |
150 | lines.append(f"📦 {name} (ID: {vmid})")
151 | lines.append(f" • Status: {status}")
152 | lines.append(f" • Node: {node}")
153 | lines.append(f" • CPU: {cpu_pct:.1f}%")
154 | lines.append(f" • CPU Cores: {cores if cores is not None else 'N/A'}")
155 |
156 | if unlimited:
157 | lines.append(f" • Memory: {_b2h(mem_bytes)} (unlimited)")
158 | else:
159 | if maxmem_bytes > 0:
160 | pct_str = f" ({mem_pct:.1f}%)" if isinstance(mem_pct, (int, float)) else ""
161 | lines.append(f" • Memory: {_b2h(mem_bytes)} / {_b2h(maxmem_bytes)}{pct_str}")
162 | else:
163 | lines.append(f" • Memory: {_b2h(mem_bytes)} / 0.00 B")
164 | lines.append("")
165 | return [Content(type="text", text="\n".join(lines).rstrip())]
166 |
167 | # ---------- tool ----------
168 | def get_containers(
169 | self,
170 | node: Optional[str] = None,
171 | include_stats: bool = True,
172 | include_raw: bool = False,
173 | format_style: str = "pretty",
174 | ) -> List[Content]:
175 | """
176 | List containers cluster-wide or by node.
177 |
178 | - `include_stats=True` fetches live CPU/mem from /status/current
179 | - RRD fallback is used if live returns zeros
180 | - `format_style='json'` returns raw JSON list (sanitized)
181 | - `format_style='pretty'` renders a human-friendly table
182 | """
183 | try:
184 | pairs = self._list_ct_pairs(node)
185 | rows: List[Dict] = []
186 |
187 | for nname, ct in pairs:
188 | vmid_val = _get(ct, "vmid")
189 | vmid_int: Optional[int] = None
190 | try:
191 | if vmid_val is not None:
192 | vmid_int = int(vmid_val)
193 | except Exception:
194 | vmid_int = None
195 |
196 | rec: Dict = {
197 | "vmid": str(vmid_val) if vmid_val is not None else None,
198 | "name": _get(ct, "name") or _get(ct, "hostname") or (f"ct-{vmid_val}" if vmid_val is not None else "ct-?"),
199 | "node": nname,
200 | "status": _get(ct, "status"),
201 | }
202 |
203 | if include_stats and vmid_int is not None:
204 | raw_status, raw_config = self._status_and_config(nname, vmid_int)
205 |
206 | cpu_frac = float(_get(raw_status, "cpu", 0.0) or 0.0)
207 | cpu_pct = round(cpu_frac * 100.0, 2)
208 | mem_bytes = int(_get(raw_status, "mem", 0) or 0)
209 | maxmem_bytes = int(_get(raw_status, "maxmem", 0) or 0)
210 |
211 | memory_mib = 0
212 | cores: Optional[Union[int, float]] = None
213 | unlimited_memory = False
214 |
215 | try:
216 | cfg_mem = _get(raw_config, "memory")
217 | if cfg_mem is None:
218 | cfg_mem = _get(raw_config, "ram")
219 | if cfg_mem is None:
220 | cfg_mem = _get(raw_config, "maxmem")
221 | if cfg_mem is None:
222 | cfg_mem = _get(raw_config, "memoryMiB")
223 | if cfg_mem is not None:
224 | try:
225 | memory_mib = int(cfg_mem)
226 | except Exception:
227 | memory_mib = 0
228 | else:
229 | memory_mib = 0
230 |
231 | unlimited_memory = bool(_get(raw_config, "swap", 0) == 0 and memory_mib == 0)
232 |
233 | cfg_cores = _get(raw_config, "cores")
234 | cfg_cpulimit = _get(raw_config, "cpulimit")
235 | if cfg_cores is not None:
236 | cores = int(cfg_cores)
237 | elif cfg_cpulimit is not None and float(cfg_cpulimit) > 0:
238 | cores = float(cfg_cpulimit)
239 | except Exception:
240 | cores = None
241 |
242 | # --- NEW: fallbacks for stopped / missing maxmem ---
243 | status_str = str(_get(raw_status, "status") or _get(ct, "status") or "").lower()
244 |
245 | if status_str == "stopped":
246 | try:
247 | mem_bytes = 0
248 | except Exception:
249 | mem_bytes = 0
250 |
251 | if (not maxmem_bytes or int(maxmem_bytes) == 0) and memory_mib and int(memory_mib) > 0:
252 | try:
253 | maxmem_bytes = int(memory_mib) * 1024 * 1024
254 | except Exception:
255 | maxmem_bytes = 0
256 |
257 | # RRD fallback if zeros
258 | if (mem_bytes == 0) or (maxmem_bytes == 0) or (cpu_pct == 0.0):
259 | rrd_cpu, rrd_mem, rrd_maxmem = self._rrd_last(nname, vmid_int)
260 | if cpu_pct == 0.0 and rrd_cpu is not None:
261 | cpu_pct = rrd_cpu
262 | if mem_bytes == 0 and rrd_mem is not None:
263 | mem_bytes = rrd_mem
264 | if maxmem_bytes == 0 and rrd_maxmem:
265 | maxmem_bytes = rrd_maxmem
266 | if memory_mib == 0:
267 | try:
268 | memory_mib = int(round(maxmem_bytes / (1024 * 1024)))
269 | except Exception:
270 | memory_mib = 0
271 |
272 | rec.update({
273 | "cores": cores,
274 | "memory": memory_mib,
275 | "cpu_pct": cpu_pct,
276 | "mem_bytes": mem_bytes,
277 | "maxmem_bytes": maxmem_bytes,
278 | "mem_pct": (
279 | round((mem_bytes / maxmem_bytes * 100.0), 2)
280 | if (maxmem_bytes and maxmem_bytes > 0)
281 | else None
282 | ),
283 | "unlimited_memory": unlimited_memory,
284 | })
285 |
286 | # For PRETTY only: allow raw blobs to be attached if requested.
287 | if include_raw and format_style != "json":
288 | rec["raw_status"] = raw_status
289 | rec["raw_config"] = raw_config
290 |
291 | rows.append(rec)
292 |
293 | if format_style == "json":
294 | # JSON path must be immune to any formatter assumptions; no raw payloads.
295 | return self._json_fmt(rows)
296 | return self._render_pretty(rows)
297 |
298 | except Exception as e:
299 | return self._err("Failed to list containers", e)
300 |
301 | # ---------- target resolution for control ops ----------
302 | def _resolve_targets(self, selector: str) -> List[Tuple[str, int, str]]:
303 | """
304 | Turn a selector string into a list of (node, vmid, label).
305 | Supports:
306 | - '123' (vmid across cluster)
307 | - 'pve1:123' (node:vmid)
308 | - 'pve1/name' (node/name)
309 | - 'name' (by name/hostname across the cluster)
310 | - comma-separated list of any of the above
311 | """
312 | if not selector:
313 | return []
314 | tokens = [t.strip() for t in selector.split(",") if t.strip()]
315 | inventory: List[Tuple[str, Dict[str, Any]]] = self._list_ct_pairs(node=None)
316 |
317 | resolved: List[Tuple[str, int, str]] = []
318 | for tok in tokens:
319 | if ":" in tok and "/" not in tok:
320 | node, vmid_s = tok.split(":", 1)
321 | try:
322 | vmid = int(vmid_s)
323 | except Exception:
324 | continue
325 | for n, ct in inventory:
326 | if n == node and int(_get(ct, "vmid", -1)) == vmid:
327 | label = _get(ct, "name") or _get(ct, "hostname") or f"ct-{vmid}"
328 | resolved.append((node, vmid, label))
329 | break
330 | continue
331 |
332 | if "/" in tok and ":" not in tok:
333 | node, name = tok.split("/", 1)
334 | name = name.strip()
335 | for n, ct in inventory:
336 | if n == node and (_get(ct, "name") == name or _get(ct, "hostname") == name):
337 | vmid = int(_get(ct, "vmid", -1))
338 | if vmid >= 0:
339 | resolved.append((node, vmid, name))
340 | continue
341 |
342 | if tok.isdigit():
343 | vmid = int(tok)
344 | for n, ct in inventory:
345 | if int(_get(ct, "vmid", -1)) == vmid:
346 | label = _get(ct, "name") or _get(ct, "hostname") or f"ct-{vmid}"
347 | resolved.append((n, vmid, label))
348 | continue
349 |
350 | name = tok
351 | for n, ct in inventory:
352 | if _get(ct, "name") == name or _get(ct, "hostname") == name:
353 | vmid = int(_get(ct, "vmid", -1))
354 | if vmid >= 0:
355 | resolved.append((n, vmid, name))
356 |
357 | uniq = {}
358 | for n, v, lbl in resolved:
359 | uniq[(n, v)] = lbl
360 | return [(n, v, uniq[(n, v)]) for (n, v) in uniq.keys()]
361 |
362 | def _render_action_result(self, title: str, results: List[Dict[str, Any]]) -> List[Content]:
363 | """Pretty-print an action result; JSON stays raw."""
364 | lines = [f"📦 {title}", ""]
365 | for r in results:
366 | status = "✅ OK" if r.get("ok") else "❌ FAIL"
367 | node = r.get("node")
368 | vmid = r.get("vmid")
369 | name = r.get("name") or f"ct-{vmid}"
370 | msg = r.get("message") or r.get("error") or ""
371 | lines.append(f"{status} {name} (ID: {vmid}, node: {node}) {('- ' + str(msg)) if msg else ''}")
372 | return [Content(type="text", text="\n".join(lines).rstrip())]
373 |
374 | # ---------- container control tools ----------
375 | def start_container(self, selector: str, format_style: str = "pretty") -> List[Content]:
376 | """
377 | Start LXC containers matching `selector`.
378 | selector examples: '123', 'pve1:123', 'pve1/name', 'name', 'pve1:101,pve2/web'
379 | """
380 | try:
381 | targets = self._resolve_targets(selector)
382 | if not targets:
383 | return self._err("No containers matched the selector", ValueError(selector))
384 |
385 | results: List[Dict[str, Any]] = []
386 | for node, vmid, label in targets:
387 | try:
388 | resp = self.proxmox.nodes(node).lxc(vmid).status.start.post()
389 | results.append({"ok": True, "node": node, "vmid": vmid, "name": label, "message": resp})
390 | except Exception as e:
391 | results.append({"ok": False, "node": node, "vmid": vmid, "name": label, "error": str(e)})
392 |
393 | if format_style == "json":
394 | return self._json_fmt(results)
395 | return self._render_action_result("Start Containers", results)
396 |
397 | except Exception as e:
398 | return self._err("Failed to start container(s)", e)
399 |
400 | def stop_container(self, selector: str, graceful: bool = True, timeout_seconds: int = 10,
401 | format_style: str = "pretty") -> List[Content]:
402 | """
403 | Stop LXC containers.
404 | graceful=True → POST .../status/shutdown (graceful stop)
405 | graceful=False → POST .../status/stop (force stop)
406 | """
407 | try:
408 | targets = self._resolve_targets(selector)
409 | if not targets:
410 | return self._err("No containers matched the selector", ValueError(selector))
411 |
412 | results: List[Dict[str, Any]] = []
413 | for node, vmid, label in targets:
414 | try:
415 | if graceful:
416 | resp = self.proxmox.nodes(node).lxc(vmid).status.shutdown.post(timeout=timeout_seconds)
417 | else:
418 | resp = self.proxmox.nodes(node).lxc(vmid).status.stop.post()
419 | results.append({"ok": True, "node": node, "vmid": vmid, "name": label, "message": resp})
420 | except Exception as e:
421 | results.append({"ok": False, "node": node, "vmid": vmid, "name": label, "error": str(e)})
422 |
423 | if format_style == "json":
424 | return self._json_fmt(results)
425 | return self._render_action_result("Stop Containers", results)
426 |
427 | except Exception as e:
428 | return self._err("Failed to stop container(s)", e)
429 |
430 | def restart_container(self, selector: str, timeout_seconds: int = 10,
431 | format_style: str = "pretty") -> List[Content]:
432 | """
433 | Restart LXC containers via POST .../status/reboot.
434 | """
435 | try:
436 | targets = self._resolve_targets(selector)
437 | if not targets:
438 | return self._err("No containers matched the selector", ValueError(selector))
439 |
440 | results: List[Dict[str, Any]] = []
441 | for node, vmid, label in targets:
442 | try:
443 | resp = self.proxmox.nodes(node).lxc(vmid).status.reboot.post()
444 | results.append({"ok": True, "node": node, "vmid": vmid, "name": label, "message": resp})
445 | except Exception as e:
446 | results.append({"ok": False, "node": node, "vmid": vmid, "name": label, "error": str(e)})
447 |
448 | if format_style == "json":
449 | return self._json_fmt(results)
450 | return self._render_action_result("Restart Containers", results)
451 |
452 | except Exception as e:
453 | return self._err("Failed to restart container(s)", e)
454 |
455 | def update_container_resources(
456 | self,
457 | selector: str,
458 | cores: Optional[int] = None,
459 | memory: Optional[int] = None,
460 | swap: Optional[int] = None,
461 | disk_gb: Optional[int] = None,
462 | disk: str = "rootfs",
463 | format_style: str = "pretty",
464 | ) -> List[Content]:
465 | """Update container CPU/memory/swap limits and/or extend disk size.
466 |
467 | Parameters:
468 | selector: Container selector (same grammar as start_container)
469 | cores: New CPU core count
470 | memory: New memory limit in MiB
471 | swap: New swap limit in MiB
472 | disk_gb: Additional disk size to add in GiB
473 | disk: Disk identifier to resize (default 'rootfs')
474 | format_style: Output format ('pretty' or 'json')
475 | """
476 |
477 | try:
478 | targets = self._resolve_targets(selector)
479 | if not targets:
480 | return self._err("No containers matched the selector", ValueError(selector))
481 |
482 | results: List[Dict[str, Any]] = []
483 | for node, vmid, label in targets:
484 | rec: Dict[str, Any] = {"ok": True, "node": node, "vmid": vmid, "name": label}
485 | changes: List[str] = []
486 |
487 | try:
488 | update_params: Dict[str, Any] = {}
489 | if cores is not None:
490 | update_params["cores"] = cores
491 | changes.append(f"cores={cores}")
492 | if memory is not None:
493 | update_params["memory"] = memory
494 | changes.append(f"memory={memory}MiB")
495 | if swap is not None:
496 | update_params["swap"] = swap
497 | changes.append(f"swap={swap}MiB")
498 |
499 | if update_params:
500 | self.proxmox.nodes(node).lxc(vmid).config.put(**update_params)
501 |
502 | if disk_gb is not None:
503 | size_str = f"+{disk_gb}G"
504 | # Use PUT for disk resize - some Proxmox versions reject POST
505 | self.proxmox.nodes(node).lxc(vmid).resize.put(disk=disk, size=size_str)
506 | changes.append(f"{disk}+={disk_gb}G")
507 |
508 | rec["message"] = ", ".join(changes) if changes else "no changes"
509 | except Exception as e:
510 | rec["ok"] = False
511 | rec["error"] = str(e)
512 |
513 | results.append(rec)
514 |
515 | if format_style == "json":
516 | return self._json_fmt(results)
517 | return self._render_action_result("Update Container Resources", results)
518 |
519 | except Exception as e:
520 | return self._err("Failed to update container(s)", e)
521 |
```