diff --git a/src/proxmox_mcp/config/__init__.py b/src/proxmox_mcp/config/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/proxmox_mcp/config/loader.py b/src/proxmox_mcp/config/loader.py new file mode 100644 index 0000000..fbdae83 --- /dev/null +++ b/src/proxmox_mcp/config/loader.py @@ -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}") diff --git a/src/proxmox_mcp/config/models.py b/src/proxmox_mcp/config/models.py new file mode 100644 index 0000000..80028e5 --- /dev/null +++ b/src/proxmox_mcp/config/models.py @@ -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 diff --git a/src/proxmox_mcp/core/__init__.py b/src/proxmox_mcp/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/proxmox_mcp/core/logging.py b/src/proxmox_mcp/core/logging.py new file mode 100644 index 0000000..708fb5f --- /dev/null +++ b/src/proxmox_mcp/core/logging.py @@ -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 diff --git a/src/proxmox_mcp/core/proxmox.py b/src/proxmox_mcp/core/proxmox.py new file mode 100644 index 0000000..496ce20 --- /dev/null +++ b/src/proxmox_mcp/core/proxmox.py @@ -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 diff --git a/src/proxmox_mcp/server.py b/src/proxmox_mcp/server.py index 593eec8..9b1e397 100644 --- a/src/proxmox_mcp/server.py +++ b/src/proxmox_mcp/server.py @@ -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.""" diff --git a/src/proxmox_mcp/tools/base.py b/src/proxmox_mcp/tools/base.py new file mode 100644 index 0000000..f714e63 --- /dev/null +++ b/src/proxmox_mcp/tools/base.py @@ -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}") diff --git a/src/proxmox_mcp/tools/cluster.py b/src/proxmox_mcp/tools/cluster.py new file mode 100644 index 0000000..9a7537a --- /dev/null +++ b/src/proxmox_mcp/tools/cluster.py @@ -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) diff --git a/src/proxmox_mcp/tools/console/__init__.py b/src/proxmox_mcp/tools/console/__init__.py new file mode 100644 index 0000000..2b63a9d --- /dev/null +++ b/src/proxmox_mcp/tools/console/__init__.py @@ -0,0 +1,6 @@ +""" +Console management package for Proxmox MCP. +""" +from .manager import VMConsoleManager + +__all__ = ['VMConsoleManager'] diff --git a/src/proxmox_mcp/tools/vm_console.py b/src/proxmox_mcp/tools/console/manager.py similarity index 100% rename from src/proxmox_mcp/tools/vm_console.py rename to src/proxmox_mcp/tools/console/manager.py diff --git a/src/proxmox_mcp/tools/definitions.py b/src/proxmox_mcp/tools/definitions.py new file mode 100644 index 0000000..4f2ed77 --- /dev/null +++ b/src/proxmox_mcp/tools/definitions.py @@ -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"}""" diff --git a/src/proxmox_mcp/tools/node.py b/src/proxmox_mcp/tools/node.py new file mode 100644 index 0000000..ac2da71 --- /dev/null +++ b/src/proxmox_mcp/tools/node.py @@ -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) diff --git a/src/proxmox_mcp/tools/storage.py b/src/proxmox_mcp/tools/storage.py new file mode 100644 index 0000000..9fb42d3 --- /dev/null +++ b/src/proxmox_mcp/tools/storage.py @@ -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) diff --git a/src/proxmox_mcp/tools/vm.py b/src/proxmox_mcp/tools/vm.py new file mode 100644 index 0000000..d398d68 --- /dev/null +++ b/src/proxmox_mcp/tools/vm.py @@ -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)