import os import logging from mcp.server.fastmcp import FastMCP from proxmoxer import ProxmoxAPI, ProxmoxAPIException from dotenv import load_dotenv # Load environment variables load_dotenv() # Logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # --- Proxmox API Configuration --- PROXMOX_URL = os.getenv("PROXMOX_URL") PROXMOX_USER = os.getenv("PROXMOX_USER") PROXMOX_PASSWORD = os.getenv("PROXMOX_PASSWORD") PROXMOX_TOKEN_ID = os.getenv("PROXMOX_TOKEN_ID") PROXMOX_VERIFY_SSL = os.getenv("PROXMOX_VERIFY_SSL", "false").lower() == "true" # --- Initialize Proxmox Client --- proxmox = None if PROXMOX_URL and (PROXMOX_PASSWORD and PROXMOX_TOKEN_ID): try: proxmox = ProxmoxAPI(PROXMOX_URL, user=f"{PROXMOX_USER}@{PROXMOX_TOKEN_ID}", token_name=PROXMOX_TOKEN_ID, token_value=PROXMOX_PASSWORD, verify_ssl=PROXMOX_VERIFY_SSL) logger.info("Proxmox API client configured.") except Exception as e: logger.error(f"Failed to configure Proxmox API: {e}") elif PROXMOX_URL and PROXMOX_USER and PROXMOX_PASSWORD: try: proxmox = ProxmoxAPI(PROXMOX_URL, user=PROXMOX_USER, password=PROXMOX_PASSWORD, verify_ssl=PROXMOX_VERIFY_SSL) logger.info("Proxmox API client configured.") except Exception as e: logger.error(f"Failed to configure Proxmox API: {e}") else: logger.warning("Proxmox API credentials not fully set. Tools may fail.") # --- FastMCP Server --- mcp = FastMCP("Proxmox MCP") @mcp.tool() def list_nodes() -> dict: """Lists all Proxmox nodes.""" if not proxmox: return {"error": "Proxmox API not configured"} return {"nodes": proxmox.nodes.get()} @mcp.tool() def get_cluster_resources() -> dict: """Gets a summary of all cluster resources.""" if not proxmox: return {"error": "Proxmox API not configured"} return {"resources": proxmox.cluster.resources.get()} @mcp.tool() def proxmox_api_call(path: str, method: str = "GET", data: dict = {}, node: str = None, vmid: int = None, lxcid: int = None) -> dict: """Executes a raw Proxmox API call.""" if not proxmox: return {"error": "Proxmox API not configured"} try: # Build path api_path = proxmox for segment in path.strip('/').split('/'): if segment == "nodes" and node: api_path = api_path.nodes(node) elif segment == "qemu" and vmid: api_path = api_path.qemu(vmid) elif segment == "lxc" and lxcid: api_path = api_path.lxc(lxcid) elif segment: api_path = api_path(segment) # Execute method_func = getattr(api_path, method.lower()) if method.upper() in ["POST", "PUT"]: return {"result": method_func(**data)} else: return {"result": method_func()} except Exception as e: return {"error": str(e)} if __name__ == "__main__": # FastMCP uses uvicorn internally when transport='sse' is implied or configured? # Actually, standard FastMCP might default to stdio. # To run SSE with uvicorn, we usually do: mcp.run(transport="sse")