d-bis/mcp-proxmox
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor

Browse Source
This commit is contained in:
canvrno
2025-02-18 23:06:04 -07:00
parent 088588b3b3
commit 101fe332de
15 changed files with 548 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
src/proxmox_mcp/config/__init__.py Normal file
View File

33
src/proxmox_mcp/config/loader.py Normal file
View File

@@ -0,0 +1,33 @@
"""
Configuration loading utilities for the Proxmox MCP server.
"""
import json
import os
from typing import Optional
from .models import Config
def load_config(config_path: Optional[str] = None) -> Config:
"""Load configuration from file.
Args:
config_path: Path to the configuration file
Returns:
Config object containing the loaded configuration
Raises:
ValueError: If config path is not provided or config is invalid
"""
if not config_path:
raise ValueError("PROXMOX_MCP_CONFIG environment variable must be set")
try:
with open(config_path) as f:
config_data = json.load(f)
if not config_data.get('proxmox', {}).get('host'):
raise ValueError("Proxmox host cannot be empty")
return Config(**config_data)
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON in config file: {e}")
except Exception as e:
raise ValueError(f"Failed to load config: {e}")

34
src/proxmox_mcp/config/models.py Normal file
View File

@@ -0,0 +1,34 @@
"""
Configuration models for the Proxmox MCP server.
"""
from typing import Optional, Annotated
from pydantic import BaseModel, Field
class NodeStatus(BaseModel):
node: Annotated[str, Field(description="Name/ID of node to query (e.g. 'pve1', 'proxmox-node2')")]
class VMCommand(BaseModel):
node: Annotated[str, Field(description="Host node name (e.g. 'pve1', 'proxmox-node2')")]
vmid: Annotated[str, Field(description="VM ID number (e.g. '100', '101')")]
command: Annotated[str, Field(description="Shell command to run (e.g. 'uname -a', 'systemctl status nginx')")]
class ProxmoxConfig(BaseModel):
host: str
port: int = 8006
verify_ssl: bool = True
service: str = "PVE"
class AuthConfig(BaseModel):
user: str
token_name: str
token_value: str
class LoggingConfig(BaseModel):
level: str = "INFO"
format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
file: Optional[str] = None
class Config(BaseModel):
proxmox: ProxmoxConfig
auth: AuthConfig
logging: LoggingConfig

0
src/proxmox_mcp/core/__init__.py Normal file
View File

55
src/proxmox_mcp/core/logging.py Normal file
View File

@@ -0,0 +1,55 @@
"""
Logging configuration for the Proxmox MCP server.
"""
import logging
import os
from typing import Optional
from ..config.models import LoggingConfig
def setup_logging(config: LoggingConfig) -> logging.Logger:
"""Configure logging based on settings.
Args:
config: Logging configuration
Returns:
Configured logger instance
"""
# Convert relative path to absolute
log_file = config.file
if log_file and not os.path.isabs(log_file):
log_file = os.path.join(os.getcwd(), log_file)
# Create handlers
handlers = []
if log_file:
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(getattr(logging, config.level.upper()))
handlers.append(file_handler)
# Console handler for errors only
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.ERROR)
handlers.append(console_handler)
# Configure formatters
formatter = logging.Formatter(config.format)
for handler in handlers:
handler.setFormatter(formatter)
# Configure root logger
root_logger = logging.getLogger()
root_logger.setLevel(getattr(logging, config.level.upper()))
# Remove any existing handlers
for handler in root_logger.handlers[:]:
root_logger.removeHandler(handler)
# Add new handlers
for handler in handlers:
root_logger.addHandler(handler)
# Create and return server logger
logger = logging.getLogger("proxmox-mcp")
return logger

71
src/proxmox_mcp/core/proxmox.py Normal file
View File

@@ -0,0 +1,71 @@
"""
Proxmox API setup and management.
"""
import logging
from typing import Dict, Any
from proxmoxer import ProxmoxAPI
from ..config.models import ProxmoxConfig, AuthConfig
class ProxmoxManager:
"""Manager class for Proxmox API operations."""
def __init__(self, proxmox_config: ProxmoxConfig, auth_config: AuthConfig):
"""Initialize the Proxmox API manager.
Args:
proxmox_config: Proxmox connection configuration
auth_config: Authentication configuration
"""
self.logger = logging.getLogger("proxmox-mcp.proxmox")
self.config = self._create_config(proxmox_config, auth_config)
self.api = self._setup_api()
def _create_config(self, proxmox_config: ProxmoxConfig, auth_config: AuthConfig) -> Dict[str, Any]:
"""Create a configuration dictionary for ProxmoxAPI.
Args:
proxmox_config: Proxmox connection configuration
auth_config: Authentication configuration
Returns:
Dictionary containing merged configuration
"""
return {
'host': proxmox_config.host,
'port': proxmox_config.port,
'user': auth_config.user,
'token_name': auth_config.token_name,
'token_value': auth_config.token_value,
'verify_ssl': proxmox_config.verify_ssl,
'service': proxmox_config.service
}
def _setup_api(self) -> ProxmoxAPI:
"""Initialize and test Proxmox API connection.
Returns:
Initialized ProxmoxAPI instance
Raises:
RuntimeError: If connection fails
"""
try:
self.logger.info(f"Connecting to Proxmox host: {self.config['host']}")
api = ProxmoxAPI(**self.config)
# Test connection
api.version.get()
self.logger.info("Successfully connected to Proxmox API")
return api
except Exception as e:
self.logger.error(f"Failed to connect to Proxmox: {e}")
raise RuntimeError(f"Failed to connect to Proxmox: {e}")
def get_api(self) -> ProxmoxAPI:
"""Get the initialized Proxmox API instance.
Returns:
ProxmoxAPI instance
"""
return self.api

293
src/proxmox_mcp/server.py
View File

@@ -1,243 +1,96 @@
#!/usr/bin/env python3
import json
"""
Main server implementation for Proxmox MCP.
"""
import logging
import os
import sys
import signal
from pathlib import Path
from typing import Dict, Any, Optional, List, Annotated
from pydantic import BaseModel, Field
from urllib.parse import urljoin
from typing import Optional, List, Annotated
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.tools import Tool
from mcp.server.fastmcp.tools.base import Tool as BaseTool
from mcp.types import CallToolResult as Response, TextContent as Content
from proxmoxer import ProxmoxAPI
from mcp.types import TextContent as Content
from pydantic import Field
# Import from the same directory
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from tools.vm_console import VMConsoleManager
class NodeStatus(BaseModel):
node: Annotated[str, Field(description="Name/ID of node to query (e.g. 'pve1', 'proxmox-node2')")]
class VMCommand(BaseModel):
node: Annotated[str, Field(description="Host node name (e.g. 'pve1', 'proxmox-node2')")]
vmid: Annotated[str, Field(description="VM ID number (e.g. '100', '101')")]
command: Annotated[str, Field(description="Shell command to run (e.g. 'uname -a', 'systemctl status nginx')")]
class ProxmoxConfig(BaseModel):
host: str
port: int = 8006
verify_ssl: bool = True
service: str = "PVE"
class AuthConfig(BaseModel):
user: str
token_name: str
token_value: str
class LoggingConfig(BaseModel):
level: str = "INFO"
format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
file: Optional[str] = None
class Config(BaseModel):
proxmox: ProxmoxConfig
auth: AuthConfig
logging: LoggingConfig
from .config.loader import load_config
from .core.logging import setup_logging
from .core.proxmox import ProxmoxManager
from .tools.node import NodeTools
from .tools.vm import VMTools
from .tools.storage import StorageTools
from .tools.cluster import ClusterTools
from .tools.definitions import (
GET_NODES_DESC,
GET_NODE_STATUS_DESC,
GET_VMS_DESC,
EXECUTE_VM_COMMAND_DESC,
GET_CONTAINERS_DESC,
GET_STORAGE_DESC,
GET_CLUSTER_STATUS_DESC
)
class ProxmoxMCPServer:
"""Main server class for Proxmox MCP."""
def __init__(self, config_path: Optional[str] = None):
self.config = self._load_config(config_path)
self._setup_logging()
self.proxmox = self._setup_proxmox()
self.vm_console = VMConsoleManager(self.proxmox)
"""Initialize the server.
Args:
config_path: Path to configuration file
"""
self.config = load_config(config_path)
self.logger = setup_logging(self.config.logging)
# Initialize core components
self.proxmox_manager = ProxmoxManager(self.config.proxmox, self.config.auth)
self.proxmox = self.proxmox_manager.get_api()
# Initialize tools
self.node_tools = NodeTools(self.proxmox)
self.vm_tools = VMTools(self.proxmox)
self.storage_tools = StorageTools(self.proxmox)
self.cluster_tools = ClusterTools(self.proxmox)
# Initialize MCP server
self.mcp = FastMCP("ProxmoxMCP")
self._setup_tools()
def _load_config(self, config_path: Optional[str]) -> Config:
"""Load configuration from file or environment variables."""
if not config_path:
raise ValueError("PROXMOX_MCP_CONFIG environment variable must be set")
try:
with open(config_path) as f:
config_data = json.load(f)
if not config_data.get('proxmox', {}).get('host'):
raise ValueError("Proxmox host cannot be empty")
return Config(**config_data)
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON in config file: {e}")
except Exception as e:
raise ValueError(f"Failed to load config: {e}")
def _setup_logging(self) -> None:
"""Configure logging based on settings."""
# Convert relative path to absolute
log_file = self.config.logging.file
if log_file and not os.path.isabs(log_file):
log_file = os.path.join(os.getcwd(), log_file)
# Create handlers
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(getattr(logging, self.config.logging.level.upper()))
# Console handler for errors only
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.ERROR)
# Configure formatters
formatter = logging.Formatter(self.config.logging.format)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
# Configure root logger
root_logger = logging.getLogger()
root_logger.setLevel(getattr(logging, self.config.logging.level.upper()))
root_logger.addHandler(file_handler)
root_logger.addHandler(console_handler)
self.logger = logging.getLogger("proxmox-mcp")
def _setup_proxmox(self) -> ProxmoxAPI:
"""Initialize Proxmox API connection."""
try:
self.logger.info(f"Connecting to Proxmox with config: {self.config.proxmox}")
# Store the working configuration for reuse
self.proxmox_config = {
'host': self.config.proxmox.host,
'user': self.config.auth.user,
'token_name': self.config.auth.token_name,
'token_value': self.config.auth.token_value,
'verify_ssl': self.config.proxmox.verify_ssl,
'service': 'PVE'
}
# Create and test CustomProxmoxAPI instance
api = ProxmoxAPI(**self.proxmox_config)
api.version.get() # Test connection
return api
except Exception as e:
self.logger.error(f"Failed to connect to Proxmox: {e}")
raise
def _setup_tools(self) -> None:
"""Register MCP tools."""
# Node tools
@self.mcp.tool(description=GET_NODES_DESC)
def get_nodes():
return self.node_tools.get_nodes()
@self.mcp.tool(
description="List all nodes in the Proxmox cluster with their status, CPU, memory, and role information.\n\n"
"Example:\n"
'{"node": "pve1", "status": "online", "cpu_usage": 0.15, "memory": {"used": "8GB", "total": "32GB"}}')
def get_nodes() -> List[Content]:
try:
result = self.proxmox.nodes.get()
nodes = [{"node": node["node"], "status": node["status"]} for node in result]
return [Content(type="text", text=json.dumps(nodes))]
except Exception as e:
self.logger.error(f"Failed to get nodes: {e}")
raise
@self.mcp.tool(description=GET_NODE_STATUS_DESC)
def get_node_status(
node: Annotated[str, Field(description="Name/ID of node to query (e.g. 'pve1', 'proxmox-node2')")]
):
return self.node_tools.get_node_status(node)
@self.mcp.tool(
description="Get detailed status information for a specific Proxmox node.\n\n"
"Parameters:\n"
"node* - Name/ID of node to query (e.g. 'pve1')\n\n"
"Example:\n"
'{"cpu": {"usage": 0.15}, "memory": {"used": "8GB", "total": "32GB"}}')
def get_node_status(node: Annotated[str, Field(description="Name/ID of node to query (e.g. 'pve1')")]) -> List[Content]:
try:
result = self.proxmox.nodes(node).status.get()
return [Content(type="text", text=json.dumps(result))]
except Exception as e:
self.logger.error(f"Failed to get node status: {e}")
raise
# VM tools
@self.mcp.tool(description=GET_VMS_DESC)
def get_vms():
return self.vm_tools.get_vms()
@self.mcp.tool(
description="List all virtual machines across the cluster with their status and resource usage.\n\n"
"Example:\n"
'{"vmid": "100", "name": "ubuntu", "status": "running", "cpu": 2, "memory": 4096}')
def get_vms() -> List[Content]:
try:
result = []
for node in self.proxmox.nodes.get():
vms = self.proxmox.nodes(node["node"]).qemu.get()
result.extend([{
"vmid": vm["vmid"],
"name": vm["name"],
"status": vm["status"],
"node": node["node"]
} for vm in vms])
return [Content(type="text", text=json.dumps(result))]
except Exception as e:
self.logger.error(f"Failed to get VMs: {e}")
raise
@self.mcp.tool(
description="List all LXC containers across the cluster with their status and configuration.\n\n"
"Example:\n"
'{"vmid": "200", "name": "nginx", "status": "running", "template": "ubuntu-20.04"}')
def get_containers() -> List[Content]:
try:
result = []
for node in self.proxmox.nodes.get():
containers = self.proxmox.nodes(node["node"]).lxc.get()
result.extend([{
"vmid": container["vmid"],
"name": container["name"],
"status": container["status"],
"node": node["node"]
} for container in containers])
return [Content(type="text", text=json.dumps(result))]
except Exception as e:
self.logger.error(f"Failed to get containers: {e}")
raise
@self.mcp.tool(
description="List storage pools across the cluster with their usage and configuration.\n\n"
"Example:\n"
'{"storage": "local-lvm", "type": "lvm", "used": "500GB", "total": "1TB"}')
def get_storage() -> List[Content]:
try:
result = self.proxmox.storage.get()
storage = [{"storage": storage["storage"], "type": storage["type"]} for storage in result]
return [Content(type="text", text=json.dumps(storage))]
except Exception as e:
self.logger.error(f"Failed to get storage: {e}")
raise
@self.mcp.tool(
description="Get overall Proxmox cluster health and configuration status.\n\n"
"Example:\n"
'{"name": "proxmox", "quorum": "ok", "nodes": 3, "ha_status": "active"}')
def get_cluster_status() -> List[Content]:
try:
result = self.proxmox.cluster.status.get()
return [Content(type="text", text=json.dumps(result))]
except Exception as e:
self.logger.error(f"Failed to get cluster status: {e}")
raise
@self.mcp.tool(
description="Execute commands in a VM via QEMU guest agent.\n\n"
"Parameters:\n"
"node* - Host node name (e.g. 'pve1')\n"
"vmid* - VM ID number (e.g. '100')\n"
"command* - Shell command to run (e.g. 'uname -a')\n\n"
"Example:\n"
'{"success": true, "output": "Linux vm1 5.4.0", "exit_code": 0}')
@self.mcp.tool(description=EXECUTE_VM_COMMAND_DESC)
async def execute_vm_command(
node: Annotated[str, Field(description="Host node name (e.g. 'pve1')")],
vmid: Annotated[str, Field(description="VM ID number (e.g. '100')")],
command: Annotated[str, Field(description="Shell command to run (e.g. 'uname -a')")]
) -> List[Content]:
try:
result = await self.vm_console.execute_command(node, vmid, command)
return [Content(type="text", text=json.dumps(result))]
except Exception as e:
self.logger.error(f"Failed to execute VM command: {e}")
raise
node: Annotated[str, Field(description="Host node name (e.g. 'pve1', 'proxmox-node2')")],
vmid: Annotated[str, Field(description="VM ID number (e.g. '100', '101')")],
command: Annotated[str, Field(description="Shell command to run (e.g. 'uname -a', 'systemctl status nginx')")]
):
return await self.vm_tools.execute_command(node, vmid, command)
# Storage tools
@self.mcp.tool(description=GET_STORAGE_DESC)
def get_storage():
return self.storage_tools.get_storage()
# Cluster tools
@self.mcp.tool(description=GET_CLUSTER_STATUS_DESC)
def get_cluster_status():
return self.cluster_tools.get_cluster_status()
def start(self) -> None:
"""Start the MCP server."""

54
src/proxmox_mcp/tools/base.py Normal file
View File

@@ -0,0 +1,54 @@
"""
Base classes and utilities for Proxmox MCP tools.
"""
import logging
from typing import Any, Dict, List, Optional
from mcp.types import TextContent as Content
from proxmoxer import ProxmoxAPI
class ProxmoxTool:
"""Base class for Proxmox MCP tools."""
def __init__(self, proxmox_api: ProxmoxAPI):
"""Initialize the tool.
Args:
proxmox_api: Initialized ProxmoxAPI instance
"""
self.proxmox = proxmox_api
self.logger = logging.getLogger(f"proxmox-mcp.{self.__class__.__name__.lower()}")
def _format_response(self, data: Any) -> List[Content]:
"""Format response data into MCP content.
Args:
data: Data to format
Returns:
List of Content objects
"""
import json
return [Content(type="text", text=json.dumps(data, indent=2))]
def _handle_error(self, operation: str, error: Exception) -> None:
"""Handle and log errors.
Args:
operation: Description of the operation that failed
error: The exception that occurred
Raises:
ValueError: For invalid input or state
RuntimeError: For other errors
"""
error_msg = str(error)
self.logger.error(f"Failed to {operation}: {error_msg}")
if "not found" in error_msg.lower():
raise ValueError(f"Resource not found: {error_msg}")
if "permission denied" in error_msg.lower():
raise ValueError(f"Permission denied: {error_msg}")
if "invalid" in error_msg.lower():
raise ValueError(f"Invalid input: {error_msg}")
raise RuntimeError(f"Failed to {operation}: {error_msg}")

31
src/proxmox_mcp/tools/cluster.py Normal file
View File

@@ -0,0 +1,31 @@
"""
Cluster-related tools for Proxmox MCP.
"""
from typing import List
from mcp.types import TextContent as Content
from .base import ProxmoxTool
from .definitions import GET_CLUSTER_STATUS_DESC
class ClusterTools(ProxmoxTool):
"""Tools for managing Proxmox cluster."""
def get_cluster_status(self) -> List[Content]:
"""Get overall Proxmox cluster health and configuration status.
Returns:
List of Content objects containing cluster status
Raises:
RuntimeError: If the operation fails
"""
try:
result = self.proxmox.cluster.status.get()
status = {
"name": result[0].get("name") if result else None,
"quorum": result[0].get("quorate"),
"nodes": len([node for node in result if node.get("type") == "node"]),
"resources": [res for res in result if res.get("type") == "resource"]
}
return self._format_response(status)
except Exception as e:
self._handle_error("get cluster status", e)

6
src/proxmox_mcp/tools/console/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
"""
Console management package for Proxmox MCP.
"""
from .manager import VMConsoleManager
__all__ = ['VMConsoleManager']

0
src/proxmox_mcp/tools/vm_console.py → src/proxmox_mcp/tools/console/manager.py
View File

51
src/proxmox_mcp/tools/definitions.py Normal file
View File

@@ -0,0 +1,51 @@
"""
Tool descriptions for Proxmox MCP tools.
"""
# Node tool descriptions
GET_NODES_DESC = """List all nodes in the Proxmox cluster with their status, CPU, memory, and role information.
Example:
{"node": "pve1", "status": "online", "cpu_usage": 0.15, "memory": {"used": "8GB", "total": "32GB"}}"""
GET_NODE_STATUS_DESC = """Get detailed status information for a specific Proxmox node.
Parameters:
node* - Name/ID of node to query (e.g. 'pve1')
Example:
{"cpu": {"usage": 0.15}, "memory": {"used": "8GB", "total": "32GB"}}"""
# VM tool descriptions
GET_VMS_DESC = """List all virtual machines across the cluster with their status and resource usage.
Example:
{"vmid": "100", "name": "ubuntu", "status": "running", "cpu": 2, "memory": 4096}"""
EXECUTE_VM_COMMAND_DESC = """Execute commands in a VM via QEMU guest agent.
Parameters:
node* - Host node name (e.g. 'pve1')
vmid* - VM ID number (e.g. '100')
command* - Shell command to run (e.g. 'uname -a')
Example:
{"success": true, "output": "Linux vm1 5.4.0", "exit_code": 0}"""
# Container tool descriptions
GET_CONTAINERS_DESC = """List all LXC containers across the cluster with their status and configuration.
Example:
{"vmid": "200", "name": "nginx", "status": "running", "template": "ubuntu-20.04"}"""
# Storage tool descriptions
GET_STORAGE_DESC = """List storage pools across the cluster with their usage and configuration.
Example:
{"storage": "local-lvm", "type": "lvm", "used": "500GB", "total": "1TB"}"""
# Cluster tool descriptions
GET_CLUSTER_STATUS_DESC = """Get overall Proxmox cluster health and configuration status.
Example:
{"name": "proxmox", "quorum": "ok", "nodes": 3, "ha_status": "active"}"""

45
src/proxmox_mcp/tools/node.py Normal file
View File

@@ -0,0 +1,45 @@
"""
Node-related tools for Proxmox MCP.
"""
from typing import List
from mcp.types import TextContent as Content
from .base import ProxmoxTool
from .definitions import GET_NODES_DESC, GET_NODE_STATUS_DESC
class NodeTools(ProxmoxTool):
"""Tools for managing Proxmox nodes."""
def get_nodes(self) -> List[Content]:
"""List all nodes in the Proxmox cluster.
Returns:
List of Content objects containing node information
Raises:
RuntimeError: If the operation fails
"""
try:
result = self.proxmox.nodes.get()
nodes = [{"node": node["node"], "status": node["status"]} for node in result]
return self._format_response(nodes)
except Exception as e:
self._handle_error("get nodes", e)
def get_node_status(self, node: str) -> List[Content]:
"""Get detailed status information for a specific node.
Args:
node: Name/ID of node to query
Returns:
List of Content objects containing node status
Raises:
ValueError: If node is not found
RuntimeError: If the operation fails
"""
try:
result = self.proxmox.nodes(node).status.get()
return self._format_response(result)
except Exception as e:
self._handle_error(f"get status for node {node}", e)

31
src/proxmox_mcp/tools/storage.py Normal file
View File

@@ -0,0 +1,31 @@
"""
Storage-related tools for Proxmox MCP.
"""
from typing import List
from mcp.types import TextContent as Content
from .base import ProxmoxTool
from .definitions import GET_STORAGE_DESC
class StorageTools(ProxmoxTool):
"""Tools for managing Proxmox storage."""
def get_storage(self) -> List[Content]:
"""List storage pools across the cluster.
Returns:
List of Content objects containing storage information
Raises:
RuntimeError: If the operation fails
"""
try:
result = self.proxmox.storage.get()
storage = [{
"storage": storage["storage"],
"type": storage["type"],
"content": storage.get("content", []),
"enabled": storage.get("enabled", True)
} for storage in result]
return self._format_response(storage)
except Exception as e:
self._handle_error("get storage", e)

64
src/proxmox_mcp/tools/vm.py Normal file
View File

@@ -0,0 +1,64 @@
"""
VM-related tools for Proxmox MCP.
"""
from typing import List
from mcp.types import TextContent as Content
from .base import ProxmoxTool
from .definitions import GET_VMS_DESC, EXECUTE_VM_COMMAND_DESC
from .console.manager import VMConsoleManager
class VMTools(ProxmoxTool):
"""Tools for managing Proxmox VMs."""
def __init__(self, proxmox_api):
"""Initialize VM tools.
Args:
proxmox_api: Initialized ProxmoxAPI instance
"""
super().__init__(proxmox_api)
self.console_manager = VMConsoleManager(proxmox_api)
def get_vms(self) -> List[Content]:
"""List all virtual machines across the cluster.
Returns:
List of Content objects containing VM information
Raises:
RuntimeError: If the operation fails
"""
try:
result = []
for node in self.proxmox.nodes.get():
vms = self.proxmox.nodes(node["node"]).qemu.get()
result.extend([{
"vmid": vm["vmid"],
"name": vm["name"],
"status": vm["status"],
"node": node["node"]
} for vm in vms])
return self._format_response(result)
except Exception as e:
self._handle_error("get VMs", e)
async def execute_command(self, node: str, vmid: str, command: str) -> List[Content]:
"""Execute a command in a VM via QEMU guest agent.
Args:
node: Host node name
vmid: VM ID number
command: Shell command to run
Returns:
List of Content objects containing command output
Raises:
ValueError: If VM is not found or not running
RuntimeError: If command execution fails
"""
try:
result = await self.console_manager.execute_command(node, vmid, command)
return self._format_response(result)
except Exception as e:
self._handle_error(f"execute command on VM {vmid}", e)