Ben
23ad267949
All checks were successful
Build and Push Proxmox MCP Docker Image / build (push) Successful in 8s
87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
import os
|
|
import logging
|
|
from mcp.server.fastmcp import FastMCP
|
|
from proxmoxer import ProxmoxAPI
|
|
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"}
|
|
try:
|
|
nodes = proxmox.nodes.get()
|
|
return {"nodes": nodes}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
@mcp.tool()
|
|
def get_cluster_resources() -> dict:
|
|
"""Gets a summary of all cluster resources."""
|
|
if not proxmox: return {"error": "Proxmox API not configured"}
|
|
try:
|
|
resources = proxmox.cluster.resources.get()
|
|
return {"resources": resources}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
@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__":
|
|
import uvicorn
|
|
# Use the exposed sse_app from FastMCP and run it with uvicorn
|
|
uvicorn.run(mcp.sse_app, host="0.0.0.0", port=8000) |