Multi-host support: manage a list of ESXi servers

Adds a pluggable server inventory and per-host connection routing:

- servers.py: load_servers() returns the managed host list. Today it reads
  ESXI_HOST[_N] env vars (a suffixed host inherits the unsuffixed
  USER/PASS/INSECURE/NETWORK); this one function is the seam an API-backed
  source replaces later. Servers are identified by host.
- connection_manager.py: ConnectionManager holds the inventory and lazily
  connects per host (one dead host doesn't block startup or the others),
  keyed by host, with a default.
- mixins/_base.py: VSphereMixin base gives every mixin self.conn (default
  host) and self._conn(host) for per-call routing. All 13 mixins now take
  the manager instead of a single connection.
- server.py builds the manager from load_servers() (falling back to the
  single VCENTER_* server) and eager-connects the default.
- New list_servers tool; list_vms/get_vm_info take an optional host arg as
  the per-call routing pattern.

Verified across two live hosts: default + host='10.22.22.222' route to the
right box; unknown host gives a clear error.
This commit is contained in:
Ryan Malloy 2026-06-08 05:33:22 -06:00
parent b743da7666
commit 2930318125
17 changed files with 347 additions and 100 deletions

View File

@ -0,0 +1,73 @@
"""Manages VMware connections to multiple ESXi hosts, keyed by host.
Connections are established lazily on first use so one unreachable host does
not block startup or the other hosts. Tools select a host per call via the
``host`` argument; omitting it uses the default (first) server.
"""
import logging
from typing import TYPE_CHECKING, Any
from mcvsphere.connection import VMwareConnection
if TYPE_CHECKING:
from mcvsphere.config import Settings
from mcvsphere.servers import ServerConfig
logger = logging.getLogger(__name__)
class ConnectionManager:
"""Holds the managed server inventory and their lazy connections."""
def __init__(self, servers: list["ServerConfig"], settings: "Settings"):
if not servers:
raise ValueError("No ESXi servers configured (set ESXI_HOST[_N] vars)")
self._servers: dict[str, ServerConfig] = {s.host: s for s in servers}
self._default_host: str = servers[0].host
self._settings = settings
self._connections: dict[str, VMwareConnection] = {}
@property
def hosts(self) -> list[str]:
"""All managed host identifiers."""
return list(self._servers)
@property
def default_host(self) -> str:
return self._default_host
def get(self, host: str | None = None) -> VMwareConnection:
"""Return the connection for ``host`` (default if None), connecting lazily."""
target = host or self._default_host
if target not in self._servers:
known = ", ".join(self._servers) or "(none)"
raise ValueError(f"Unknown server '{target}'. Managed hosts: {known}")
if target not in self._connections:
logger.info("Connecting to ESXi host %s", target)
self._connections[target] = VMwareConnection(
self._servers[target].to_settings(self._settings)
)
return self._connections[target]
def describe(self) -> list[dict[str, Any]]:
"""Read-only summary of managed servers (no connection attempts)."""
return [
{
"host": host,
"user": cfg.user,
"network": cfg.network,
"insecure": cfg.insecure,
"default": host == self._default_host,
"connected": host in self._connections,
}
for host, cfg in self._servers.items()
]
def disconnect_all(self) -> None:
for conn in self._connections.values():
try:
conn.disconnect()
except Exception:
logger.warning("Error disconnecting", exc_info=True)
self._connections.clear()

View File

@ -0,0 +1,31 @@
"""Shared base for vSphere tool mixins — routes to the target ESXi host.
Tools that expose multi-host selection take a ``host`` argument and call
``self._conn(host)``. Tools that don't use ``self.conn``, which resolves to the
default (first) managed host. Both go through the ConnectionManager, so
connections are shared and established lazily.
"""
from typing import TYPE_CHECKING
from fastmcp.contrib.mcp_mixin import MCPMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
from mcvsphere.connection_manager import ConnectionManager
class VSphereMixin(MCPMixin):
"""Base class giving every mixin host-aware connection access."""
def __init__(self, manager: "ConnectionManager"):
self.manager = manager
@property
def conn(self) -> "VMwareConnection":
"""The default host's connection (back-compat for single-host tools)."""
return self.manager.get()
def _conn(self, host: str | None = None) -> "VMwareConnection":
"""Resolve the connection for a specific managed host (default if None)."""
return self.manager.get(host)

View File

@ -6,20 +6,19 @@ from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any
import requests
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class ConsoleMixin(MCPMixin):
class ConsoleMixin(VSphereMixin):
"""VM console operations - screenshots and VMware Tools monitoring."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
@mcp_tool(
name="wait_for_vm_tools",
description="Wait for VMware Tools to become available on a VM. Useful after powering on a VM.",

View File

@ -2,20 +2,19 @@
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class DiskManagementMixin(MCPMixin):
class DiskManagementMixin(VSphereMixin):
"""Virtual disk and ISO management tools."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
def _get_next_disk_unit_number(self, vm: vim.VirtualMachine) -> tuple[int, vim.vm.device.VirtualSCSIController]:
"""Find the next available SCSI unit number and controller."""
scsi_controllers = []

View File

@ -4,20 +4,19 @@ import base64
import time
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class GuestOpsMixin(MCPMixin):
class GuestOpsMixin(VSphereMixin):
"""Guest OS operations (requires VMware Tools running in the VM)."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
def _get_guest_auth(
self, username: str, password: str
) -> vim.vm.guest.NamePasswordAuthentication:

View File

@ -2,20 +2,19 @@
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class HostManagementMixin(MCPMixin):
class HostManagementMixin(VSphereMixin):
"""ESXi host management tools."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
def _get_host(self) -> vim.HostSystem:
"""Get the ESXi host system."""
for entity in self.conn.datacenter.hostFolder.childEntity:

View File

@ -3,20 +3,19 @@
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class MonitoringMixin(MCPMixin):
class MonitoringMixin(VSphereMixin):
"""VM and host monitoring tools."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
@mcp_tool(
name="get_vm_stats",
description="Get current performance statistics for a virtual machine",

View File

@ -2,20 +2,19 @@
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class NICManagementMixin(MCPMixin):
class NICManagementMixin(VSphereMixin):
"""Virtual network adapter management tools."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
def _find_nic_by_label(
self, vm: vim.VirtualMachine, label: str
) -> vim.vm.device.VirtualEthernetCard | None:

View File

@ -14,20 +14,19 @@ from pathlib import Path
from typing import TYPE_CHECKING, Any
from urllib.parse import urlparse
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class OVFManagementMixin(MCPMixin):
class OVFManagementMixin(VSphereMixin):
"""OVF/OVA deployment and export tools."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
def _extract_ova(self, ova_path: str) -> tuple[str, str, list[str]]:
"""Extract OVA file and return (temp_dir, ovf_path, disk_files)."""
temp_dir = tempfile.mkdtemp(prefix="ovf_")

View File

@ -2,20 +2,19 @@
from typing import TYPE_CHECKING
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class PowerOpsMixin(MCPMixin):
class PowerOpsMixin(VSphereMixin):
"""VM power management tools."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
@mcp_tool(
name="power_on",
description="Power on a virtual machine",

View File

@ -2,20 +2,19 @@
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_resource, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_resource, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class ResourcesMixin(MCPMixin):
class ResourcesMixin(VSphereMixin):
"""MCP Resources for vSphere infrastructure."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
# ─────────────────────────────────────────────────────────────────────────────
# Datastore File Browser (templated resource)
# ─────────────────────────────────────────────────────────────────────────────

View File

@ -5,15 +5,17 @@ import socket
import time
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class SerialPortMixin(MCPMixin):
class SerialPortMixin(VSphereMixin):
"""Serial port management for VM network console access.
Network serial ports allow telnet/TCP connections to VM consoles,
@ -26,9 +28,6 @@ class SerialPortMixin(MCPMixin):
- tcp+ssl: Encrypted SSL over TCP
"""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
def _get_serial_port(self, vm: vim.VirtualMachine) -> vim.vm.device.VirtualSerialPort | None:
"""Find existing serial port with URI backing on a VM."""
if not vm.config:

View File

@ -2,20 +2,19 @@
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class SnapshotsMixin(MCPMixin):
class SnapshotsMixin(VSphereMixin):
"""VM snapshot management tools."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
def _get_snapshot_tree(
self, snapshots: list, parent_path: str = ""
) -> list[dict[str, Any]]:

View File

@ -3,19 +3,31 @@
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class VCenterOpsMixin(MCPMixin):
class VCenterOpsMixin(VSphereMixin):
"""vCenter-specific operations (require vCenter, not just ESXi)."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
@mcp_tool(
name="list_servers",
description="List the ESXi hosts this server manages. Pass a host's value as the 'host' argument to other tools to target it.",
annotations=ToolAnnotations(readOnlyHint=True),
)
def list_servers(self) -> list[dict[str, Any]]:
"""List managed ESXi hosts and their connection status.
The ``host`` field is the identifier to pass as the ``host`` argument on
host-aware tools; omitting it targets the ``default`` host.
"""
return self.manager.describe()
# ─────────────────────────────────────────────────────────────────────────────
# Storage vMotion (works even on single-host vCenter)

View File

@ -2,29 +2,34 @@
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from fastmcp.contrib.mcp_mixin import mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
from mcvsphere.connection import VMwareConnection
pass
class VMLifecycleMixin(MCPMixin):
class VMLifecycleMixin(VSphereMixin):
"""VM lifecycle management tools - CRUD operations for virtual machines."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
@mcp_tool(
name="list_vms",
description="List all virtual machines in the vSphere inventory",
annotations=ToolAnnotations(readOnlyHint=True),
)
def list_vms(self) -> list[dict[str, Any]]:
"""List all virtual machines with basic info."""
def list_vms(self, host: str | None = None) -> list[dict[str, Any]]:
"""List all virtual machines with basic info.
Args:
host: Managed ESXi host to query (default: the default host).
Use list_servers to see available hosts.
"""
conn = self._conn(host)
vms = []
for vm in self.conn.get_all_vms():
for vm in conn.get_all_vms():
vms.append(
{
"name": vm.name,
@ -41,9 +46,15 @@ class VMLifecycleMixin(MCPMixin):
description="Get detailed information about a specific virtual machine",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_vm_info(self, name: str) -> dict[str, Any]:
"""Get detailed VM information including hardware, network, and storage."""
vm = self.conn.find_vm(name)
def get_vm_info(self, name: str, host: str | None = None) -> dict[str, Any]:
"""Get detailed VM information including hardware, network, and storage.
Args:
name: VM name
host: Managed ESXi host to query (default: the default host)
"""
conn = self._conn(host)
vm = conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")

View File

@ -8,7 +8,7 @@ from fastmcp import FastMCP
from mcvsphere.auth import create_auth_provider
from mcvsphere.config import Settings, get_settings
from mcvsphere.connection import VMwareConnection
from mcvsphere.connection_manager import ConnectionManager
from mcvsphere.middleware import RBACMiddleware
from mcvsphere.mixins import (
ConsoleMixin,
@ -25,6 +25,7 @@ from mcvsphere.mixins import (
VCenterOpsMixin,
VMLifecycleMixin,
)
from mcvsphere.servers import ServerConfig, load_servers
logger = logging.getLogger(__name__)
@ -74,25 +75,43 @@ def create_server(settings: Settings | None = None) -> FastMCP:
mcp.add_middleware(RBACMiddleware())
logger.info("RBAC middleware enabled - permissions enforced via OAuth groups")
# Create shared VMware connection
logger.info("Connecting to VMware vCenter/ESXi...")
conn = VMwareConnection(settings)
# Build the managed-server inventory (pluggable source: ESXI_HOST[_N] env
# now, an API later). Fall back to the single VCENTER_* server if no
# ESXI_* hosts are configured.
servers = load_servers()
if not servers:
servers = [
ServerConfig(
host=settings.vcenter_host,
user=settings.vcenter_user,
password=settings.vcenter_password,
insecure=settings.vcenter_insecure,
network=settings.vcenter_network,
)
]
manager = ConnectionManager(servers, settings)
logger.info(
"Managing %d ESXi host(s); connecting to default %s...",
len(manager.hosts),
manager.default_host,
)
manager.get() # eager-connect the default host (fail fast); others are lazy
# Create and register all mixins
mixins = [
VMLifecycleMixin(conn),
PowerOpsMixin(conn),
SnapshotsMixin(conn),
MonitoringMixin(conn),
GuestOpsMixin(conn),
ResourcesMixin(conn),
DiskManagementMixin(conn),
NICManagementMixin(conn),
OVFManagementMixin(conn),
HostManagementMixin(conn),
VCenterOpsMixin(conn),
ConsoleMixin(conn),
SerialPortMixin(conn),
VMLifecycleMixin(manager),
PowerOpsMixin(manager),
SnapshotsMixin(manager),
MonitoringMixin(manager),
GuestOpsMixin(manager),
ResourcesMixin(manager),
DiskManagementMixin(manager),
NICManagementMixin(manager),
OVFManagementMixin(manager),
HostManagementMixin(manager),
VCenterOpsMixin(manager),
ConsoleMixin(manager),
SerialPortMixin(manager),
]
tool_count = 0

112
src/mcvsphere/servers.py Normal file
View File

@ -0,0 +1,112 @@
"""ESXi server inventory — the pluggable source of hosts to manage.
This module is the single seam between "where the server list comes from" and
the rest of the app. Today ``load_servers`` reads ``ESXI_HOST[_N]`` environment
variables (from ``.env`` or the process environment). When mcvsphere becomes an
HTTP service, replace ``load_servers`` with an API-backed implementation
nothing downstream depends on the source, only on the ``ServerConfig`` shape.
"""
import os
from typing import TYPE_CHECKING
from dotenv import dotenv_values
from pydantic import BaseModel, SecretStr
if TYPE_CHECKING:
from mcvsphere.config import Settings
class ServerConfig(BaseModel):
"""A single ESXi host to manage. Servers are identified by ``host``."""
host: str
user: str
password: SecretStr
insecure: bool = True
network: str = "VM Network"
def to_settings(self, base: "Settings") -> "Settings":
"""Derive a per-server Settings from the base settings.
Connection fields are overridden with this server's values; everything
else (transport, OAuth, logging) is inherited so a single VMwareConnection
can be built per host without duplicating global config.
"""
return base.model_copy(
update={
"vcenter_host": self.host,
"vcenter_user": self.user,
"vcenter_password": self.password,
"vcenter_insecure": self.insecure,
"vcenter_network": self.network,
}
)
def _as_bool(value: str | None, default: bool = True) -> bool:
if value is None or value == "":
return default
return str(value).lower() in ("true", "1", "yes", "on")
def load_servers(env_file: str = ".env") -> list[ServerConfig]:
"""Return the list of ESXi servers to manage.
Source seam currently the ``ESXI_HOST[_N]`` env-var families:
ESXI_HOST / ESXI_USER / ESXI_PASS [/ ESXI_INSECURE / ESXI_NETWORK]
ESXI_HOST_1 / ESXI_USER_1 / ESXI_PASS_1 [/ ...]
ESXI_HOST_2 / ...
Rules:
- Servers are keyed by ``host``; the first (unsuffixed ``ESXI_HOST``) is the
default used when a tool call omits ``host``.
- A suffixed host that omits its own USER/PASS/INSECURE/NETWORK inherits the
unsuffixed ``ESXI_USER``/``ESXI_PASS``/``ESXI_INSECURE``/``ESXI_NETWORK``.
- ``.env`` and the real environment are merged (environment wins), so the
list works whether vars are exported or only present in ``.env``.
"""
env: dict[str, str | None] = {**dotenv_values(env_file), **os.environ}
base_user = env.get("ESXI_USER")
base_pass = env.get("ESXI_PASS")
base_insecure = env.get("ESXI_INSECURE")
base_network = env.get("ESXI_NETWORK")
# Discover suffixes in order: "" (base), then _1, _2, ... while present.
suffixes: list[str] = []
if env.get("ESXI_HOST"):
suffixes.append("")
index = 1
while env.get(f"ESXI_HOST_{index}"):
suffixes.append(f"_{index}")
index += 1
servers: list[ServerConfig] = []
seen: set[str] = set()
for suffix in suffixes:
host = env.get(f"ESXI_HOST{suffix}")
if not host or host in seen:
continue
user = env.get(f"ESXI_USER{suffix}") or base_user
password = env.get(f"ESXI_PASS{suffix}") or base_pass
if not user or not password:
# Incomplete credentials and no base fallback — skip this host.
continue
insecure = env.get(f"ESXI_INSECURE{suffix}")
if insecure is None:
insecure = base_insecure
network = env.get(f"ESXI_NETWORK{suffix}") or base_network or "VM Network"
seen.add(host)
servers.append(
ServerConfig(
host=host,
user=user,
password=SecretStr(password),
insecure=_as_bool(insecure),
network=network,
)
)
return servers