Ben
9c7e2d41dc
All checks were successful
Build and Push Proxmox MCP Docker Image / build (push) Successful in 7s
114 lines
4.2 KiB
Python
114 lines
4.2 KiB
Python
import os
|
|
import logging
|
|
from mcp.server.fastmcp import FastMCP
|
|
from mcp.server.transport_security import TransportSecuritySettings
|
|
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"
|
|
|
|
# --- MCP Transport Security ---
|
|
# Allow external access via reverse proxy and local development
|
|
MCP_ALLOWED_HOSTS = os.getenv("MCP_ALLOWED_HOSTS", "proxmox-mcp.ext.ben.io,localhost:*,127.0.0.1:*")
|
|
|
|
# --- Initialize Proxmox Client ---
|
|
proxmox = None
|
|
if PROXMOX_URL and PROXMOX_USER:
|
|
try:
|
|
if PROXMOX_TOKEN_ID and PROXMOX_PASSWORD:
|
|
# Token-based authentication
|
|
# PROXMOX_USER should be like "user@pam" or "user@pve"
|
|
# PROXMOX_TOKEN_ID is the token name (e.g., "mcp-token")
|
|
# PROXMOX_PASSWORD is the token secret value
|
|
proxmox = ProxmoxAPI(
|
|
PROXMOX_URL,
|
|
user=PROXMOX_USER,
|
|
token_name=PROXMOX_TOKEN_ID,
|
|
token_value=PROXMOX_PASSWORD,
|
|
verify_ssl=PROXMOX_VERIFY_SSL
|
|
)
|
|
logger.info(f"Proxmox API client configured with token auth for {PROXMOX_USER}")
|
|
elif PROXMOX_PASSWORD:
|
|
# Password-based authentication
|
|
proxmox = ProxmoxAPI(
|
|
PROXMOX_URL,
|
|
user=PROXMOX_USER,
|
|
password=PROXMOX_PASSWORD,
|
|
verify_ssl=PROXMOX_VERIFY_SSL
|
|
)
|
|
logger.info(f"Proxmox API client configured with password auth for {PROXMOX_USER}")
|
|
else:
|
|
logger.warning("PROXMOX_PASSWORD (or token secret) not set. Tools may fail.")
|
|
except Exception as e:
|
|
logger.error(f"Failed to configure Proxmox API: {e}")
|
|
else:
|
|
logger.warning("PROXMOX_URL or PROXMOX_USER not set. Tools may fail.")
|
|
|
|
# --- FastMCP Server ---
|
|
# Configure transport security to allow external access via reverse proxy
|
|
transport_security = TransportSecuritySettings(
|
|
enable_dns_rebinding_protection=True,
|
|
allowed_hosts=[h.strip() for h in MCP_ALLOWED_HOSTS.split(",")],
|
|
allowed_origins=[], # Empty = allow any origin (or none for same-origin)
|
|
)
|
|
mcp = FastMCP("Proxmox MCP", transport_security=transport_security)
|
|
|
|
@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) |