Initial commit

src/proxmox_mcp/__init__.py

@@ -0,0 +1,8 @@
+"""
+Proxmox MCP Server - A Model Context Protocol server for interacting with Proxmox hypervisors.
+"""
+
+from .server import ProxmoxMCPServer
+
+__version__ = "0.1.0"
+__all__ = ["ProxmoxMCPServer"]

src/proxmox_mcp/server.py

@@ -0,0 +1,223 @@
+#!/usr/bin/env python3
+import json
+import logging
+import os
+from pathlib import Path
+from typing import Dict, Any, Optional, List
+
+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 pydantic import BaseModel
+
+from .tools.vm_console import VMConsoleManager
+
+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
+
+class ProxmoxMCPServer:
+    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)
+        self.mcp = FastMCP("ProxmoxMCP")
+        self._setup_tools()
+
+    def _load_config(self, config_path: Optional[str]) -> Config:
+        """Load configuration from file or environment variables."""
+        if config_path:
+            with open(config_path) as f:
+                config_data = json.load(f)
+        else:
+            # Load from environment variables
+            config_data = {
+                "proxmox": {
+                    "host": os.getenv("PROXMOX_HOST", ""),
+                    "port": int(os.getenv("PROXMOX_PORT", "8006")),
+                    "verify_ssl": os.getenv("PROXMOX_VERIFY_SSL", "true").lower() == "true",
+                    "service": os.getenv("PROXMOX_SERVICE", "PVE"),
+                },
+                "auth": {
+                    "user": os.getenv("PROXMOX_USER", ""),
+                    "token_name": os.getenv("PROXMOX_TOKEN_NAME", ""),
+                    "token_value": os.getenv("PROXMOX_TOKEN_VALUE", ""),
+                },
+                "logging": {
+                    "level": os.getenv("LOG_LEVEL", "INFO"),
+                    "format": os.getenv("LOG_FORMAT", "%(asctime)s - %(name)s - %(levelname)s - %(message)s"),
+                    "file": os.getenv("LOG_FILE"),
+                },
+            }
+
+        return Config(**config_data)
+
+    def _setup_logging(self) -> None:
+        """Configure logging based on settings."""
+        logging.basicConfig(
+            level=getattr(logging, self.config.logging.level.upper()),
+            format=self.config.logging.format,
+            filename=self.config.logging.file,
+        )
+        self.logger = logging.getLogger("proxmox-mcp")
+
+    def _setup_proxmox(self) -> ProxmoxAPI:
+        """Initialize Proxmox API connection."""
+        try:
+            return ProxmoxAPI(
+                host=self.config.proxmox.host,
+                port=self.config.proxmox.port,
+                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=self.config.proxmox.service,
+            )
+        except Exception as e:
+            self.logger.error(f"Failed to connect to Proxmox: {e}")
+            raise
+
+    def _setup_tools(self) -> None:
+        """Register MCP tools."""
+
+        @self.mcp.tool()
+        def get_nodes() -> List[Content]:
+            """List all nodes in the Proxmox cluster."""
+            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()
+        def get_node_status(node: str) -> List[Content]:
+            """Get detailed status of a specific node.
+
+            Args:
+                node: Name of the node to get status for
+            """
+            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
+
+        @self.mcp.tool()
+        def get_vms() -> List[Content]:
+            """List all VMs across the cluster."""
+            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()
+        def get_containers() -> List[Content]:
+            """List all LXC containers."""
+            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()
+        def get_storage() -> List[Content]:
+            """List available storage."""
+            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()
+        def get_cluster_status() -> List[Content]:
+            """Get overall cluster status."""
+            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()
+        async def execute_vm_command(node: str, vmid: str, command: str) -> List[Content]:
+            """Execute a command in a VM's console.
+
+            Args:
+                node: Name of the node where VM is running
+                vmid: ID of the VM
+                command: Command to execute
+            """
+            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
+
+    async def run(self) -> None:
+        """Start the MCP server."""
+        try:
+            await self.mcp.run()
+            self.logger.info("Proxmox MCP server running")
+        except Exception as e:
+            self.logger.error(f"Server error: {e}")
+            raise
+
+def main():
+    """Entry point for the MCP server."""
+    import asyncio
+
+    config_path = os.getenv("PROXMOX_MCP_CONFIG")
+    server = ProxmoxMCPServer(config_path)
+
+    try:
+        asyncio.run(server.run())
+    except KeyboardInterrupt:
+        pass
+
+if __name__ == "__main__":
+    main()

src/proxmox_mcp/tools/__init__.py

@@ -0,0 +1,5 @@
+"""
+MCP tools for interacting with Proxmox hypervisors.
+"""
+
+__all__ = []

src/proxmox_mcp/tools/vm_console.py

@@ -0,0 +1,63 @@
+"""
+Module for managing VM console operations.
+"""
+
+import logging
+from typing import Dict, Any
+
+class VMConsoleManager:
+    """Manager class for VM console operations."""
+
+    def __init__(self, proxmox_api):
+        """Initialize the VM console manager.
+
+        Args:
+            proxmox_api: Initialized ProxmoxAPI instance
+        """
+        self.proxmox = proxmox_api
+        self.logger = logging.getLogger("proxmox-mcp.vm-console")
+
+    async def execute_command(self, node: str, vmid: str, command: str) -> Dict[str, Any]:
+        """Execute a command in a VM's console.
+
+        Args:
+            node: Name of the node where VM is running
+            vmid: ID of the VM
+            command: Command to execute
+
+        Returns:
+            Dictionary containing command output and status
+
+        Raises:
+            ValueError: If VM is not found or not running
+            RuntimeError: If command execution fails
+        """
+        try:
+            # Verify VM exists and is running
+            vm_status = self.proxmox.nodes(node).qemu(vmid).status.current.get()
+            if vm_status["status"] != "running":
+                self.logger.error(f"Failed to execute command on VM {vmid}: VM is not running")
+                raise ValueError(f"VM {vmid} on node {node} is not running")
+
+            # Get VM's console
+            console = self.proxmox.nodes(node).qemu(vmid).agent.exec.post(
+                command=command
+            )
+
+            self.logger.debug(f"Executed command '{command}' on VM {vmid} (node: {node})")
+
+            return {
+                "success": True,
+                "output": console.get("out", ""),
+                "error": console.get("err", ""),
+                "exit_code": console.get("exitcode", 0)
+            }
+
+        except ValueError:
+            # Re-raise ValueError for VM not running
+            raise
+        except Exception as e:
+            self.logger.error(f"Failed to execute command on VM {vmid}: {str(e)}")
+            if "not found" in str(e).lower():
+                raise ValueError(f"VM {vmid} not found on node {node}")
+            raise RuntimeError(f"Failed to execute command: {str(e)}")

src/proxmox_mcp/utils/__init__.py

@@ -0,0 +1,5 @@
+"""
+Utility functions and helpers for the Proxmox MCP server.
+"""
+
+__all__ = []

src/proxmox_mcp/utils/auth.py

@@ -0,0 +1,86 @@
+"""
+Authentication utilities for the Proxmox MCP server.
+"""
+
+import os
+from typing import Dict, Optional, Tuple
+
+from pydantic import BaseModel
+
+class ProxmoxAuth(BaseModel):
+    """Proxmox authentication configuration."""
+    user: str
+    token_name: str
+    token_value: str
+
+def load_auth_from_env() -> ProxmoxAuth:
+    """
+    Load Proxmox authentication details from environment variables.
+
+    Environment Variables:
+        PROXMOX_USER: Username with realm (e.g., 'root@pam' or 'user@pve')
+        PROXMOX_TOKEN_NAME: API token name
+        PROXMOX_TOKEN_VALUE: API token value
+
+    Returns:
+        ProxmoxAuth: Authentication configuration
+
+    Raises:
+        ValueError: If required environment variables are missing
+    """
+    user = os.getenv("PROXMOX_USER")
+    token_name = os.getenv("PROXMOX_TOKEN_NAME")
+    token_value = os.getenv("PROXMOX_TOKEN_VALUE")
+
+    if not all([user, token_name, token_value]):
+        missing = []
+        if not user:
+            missing.append("PROXMOX_USER")
+        if not token_name:
+            missing.append("PROXMOX_TOKEN_NAME")
+        if not token_value:
+            missing.append("PROXMOX_TOKEN_VALUE")
+        raise ValueError(f"Missing required environment variables: {', '.join(missing)}")
+
+    return ProxmoxAuth(
+        user=user,
+        token_name=token_name,
+        token_value=token_value,
+    )
+
+def parse_user(user: str) -> Tuple[str, str]:
+    """
+    Parse a Proxmox user string into username and realm.
+
+    Args:
+        user: User string in format 'username@realm'
+
+    Returns:
+        Tuple[str, str]: (username, realm)
+
+    Raises:
+        ValueError: If user string is not in correct format
+    """
+    try:
+        username, realm = user.split("@")
+        return username, realm
+    except ValueError:
+        raise ValueError(
+            "Invalid user format. Expected 'username@realm' (e.g., 'root@pam' or 'user@pve')"
+        )
+
+def get_auth_dict(auth: ProxmoxAuth) -> Dict[str, str]:
+    """
+    Convert ProxmoxAuth model to dictionary for Proxmoxer API.
+
+    Args:
+        auth: ProxmoxAuth configuration
+
+    Returns:
+        Dict[str, str]: Authentication dictionary for Proxmoxer
+    """
+    return {
+        "user": auth.user,
+        "token_name": auth.token_name,
+        "token_value": auth.token_value,
+    }

src/proxmox_mcp/utils/logging.py

@@ -0,0 +1,51 @@
+"""
+Logging configuration for the Proxmox MCP server.
+"""
+
+import logging
+import sys
+from typing import Optional
+
+def setup_logging(
+    level: str = "INFO",
+    format_str: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    log_file: Optional[str] = None,
+) -> logging.Logger:
+    """
+    Configure logging for the Proxmox MCP server.
+
+    Args:
+        level: The logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
+        format_str: The format string for log messages
+        log_file: Optional file path to write logs to
+
+    Returns:
+        logging.Logger: Configured logger instance
+    """
+    # Create logger
+    logger = logging.getLogger("proxmox-mcp")
+    logger.setLevel(getattr(logging, level.upper()))
+
+    # Create handlers
+    handlers = []
+
+    # Console handler
+    console_handler = logging.StreamHandler(sys.stderr)
+    console_handler.setLevel(getattr(logging, level.upper()))
+    handlers.append(console_handler)
+
+    # File handler if log_file is specified
+    if log_file:
+        file_handler = logging.FileHandler(log_file)
+        file_handler.setLevel(getattr(logging, level.upper()))
+        handlers.append(file_handler)
+
+    # Create formatter
+    formatter = logging.Formatter(format_str)
+
+    # Add formatter to handlers and handlers to logger
+    for handler in handlers:
+        handler.setFormatter(formatter)
+        logger.addHandler(handler)
+
+    return logger