Compare commits

..

No commits in common. "13e4734ef115fdcaa590bb559c2d05d5f5bd9ef9" and "0e7942f51015e041f55f0e009ef6579a21c2aed1" have entirely different histories.

19 changed files with 416 additions and 1684 deletions

View File

@ -24,48 +24,6 @@ These are required to connect to your VMware infrastructure.
Only set `VCENTER_INSECURE=true` in development environments. Production deployments should use valid SSL certificates.
</Aside>
## Managing Multiple ESXi Hosts
mcvsphere can manage a list of ESXi hosts and route each tool call to whichever one you choose. Define the list with numbered `ESXI_*` variables:
| Variable | Description |
|----------|-------------|
| `ESXI_HOST` | First host — the **default** used when a call omits `host` |
| `ESXI_USER` / `ESXI_PASS` | Credentials for the default host |
| `ESXI_INSECURE` / `ESXI_NETWORK` | SSL skip / default network for the default host |
| `ESXI_HOST_1`, `ESXI_HOST_2`, … | Additional hosts |
| `ESXI_USER_1` / `ESXI_PASS_1` / … | Per-host credentials (optional) |
| `ESXI_INSECURE_1` / `ESXI_NETWORK_1` / … | Per-host SSL / network (optional) |
A suffixed host that omits its own `USER`/`PASS`/`INSECURE`/`NETWORK` inherits the unsuffixed value, so hosts that share credentials only need a `HOST` line:
```bash
ESXI_HOST=10.0.0.10
ESXI_USER=root
ESXI_PASS=secret
ESXI_NETWORK=VM Network
ESXI_HOST_1=10.0.0.11 # reuses root / secret / VM Network
ESXI_HOST_2=10.0.0.12 # reuses root / secret / VM Network
```
Hosts are identified by their `host` value. Connections are established **lazily** on first use, so an unreachable host never blocks startup or the other hosts.
### Selecting a host per call
Every tool except `list_servers` accepts an optional `host` argument:
```text
list_servers() → list managed hosts + connection status
list_vms(host="10.0.0.11") → VMs on that specific host
get_host_info() → omit host → the default host
create_vm(name="web", host="10.0.0.12")
```
:::note
The host list is produced by a single pluggable function, `load_servers()`. The `ESXI_*` env-var families are the current source; a future HTTP deployment can back it with an API with no tool changes — the per-call `host` selector is stateless by design.
:::
## MCP Transport
Control how mcvsphere communicates with MCP clients.

View File

@ -1,68 +0,0 @@
#!/usr/bin/env bash
# esxi-ssh.sh — on-demand ESXi SSH toggle via the vSphere API (govc + .env creds)
#
# The API account in .env (ESXI_USER, e.g. "claude") can toggle the TSM-SSH
# service but has no interactive host shell, so the `shell` subcommand logs in
# as root by default. Override the shell login with ESXI_SSH_USER.
#
# Usage:
# ./esxi-ssh.sh status # show current TSM-SSH state
# ./esxi-ssh.sh on # enable + start SSH (persists across reboot)
# ./esxi-ssh.sh off # stop + disable SSH (secure default)
# ./esxi-ssh.sh shell [cmd...] # start SSH, log in (root), restore prior state on exit
#
# Env overrides: ENV_FILE (default ../.env), ESXI_SSH_USER (default root)
set -euo pipefail
ENV_FILE="${ENV_FILE:-$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)/.env}"
[[ -r "$ENV_FILE" ]] || { echo "esxi-ssh: env file not readable: $ENV_FILE" >&2; exit 1; }
getenv() { grep -E "^$1=" "$ENV_FILE" | head -1 | cut -d= -f2-; }
HOST="$(getenv ESXI_HOST)"
export GOVC_URL="https://${HOST}"
export GOVC_USERNAME="$(getenv ESXI_USER)"
export GOVC_PASSWORD="$(getenv ESXI_PASS)"
export GOVC_INSECURE=1
SSH_USER="${ESXI_SSH_USER:-root}"
# Prints the TSM-SSH row's "policy status" (e.g. "off Running" / "on Stopped")
ssh_state() { govc host.service.ls 2>/dev/null | awk '/^TSM-SSH[[:space:]]/{print $2, $3}'; }
status() { govc host.service.ls | awk 'NR==1 || /^TSM-SSH[[:space:]]/'; }
on() {
govc host.service enable TSM-SSH >/dev/null 2>&1 || true
govc host.service start TSM-SSH >/dev/null 2>&1 || true
echo "esxi-ssh: SSH enabled on ${HOST}"
status
}
off() {
govc host.service stop TSM-SSH >/dev/null 2>&1 || true
govc host.service disable TSM-SSH >/dev/null 2>&1 || true
echo "esxi-ssh: SSH disabled on ${HOST}"
status
}
shell() {
read -r _ prev_status < <(ssh_state)
echo "esxi-ssh: prior TSM-SSH status=${prev_status:-unknown}"
govc host.service start TSM-SSH >/dev/null 2>&1 || true
# Leave it as we found it: only stop SSH on exit if it was NOT already running
if [[ "${prev_status:-}" != "Running" ]]; then
trap 'echo "esxi-ssh: restoring SSH to stopped"; govc host.service stop TSM-SSH >/dev/null 2>&1 || true' EXIT
fi
echo "esxi-ssh: connecting as ${SSH_USER}@${HOST} ..."
ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null "${SSH_USER}@${HOST}" "$@"
}
case "${1:-status}" in
on) on ;;
off) off ;;
status) status ;;
shell) shift; shell "$@" ;;
*) echo "usage: $(basename "$0") {on|off|status|shell [cmd...]}" >&2; exit 1 ;;
esac

View File

@ -1,73 +0,0 @@
"""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

@ -1,31 +0,0 @@
"""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,27 +6,27 @@ from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any
import requests
from fastmcp.contrib.mcp_mixin import mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class ConsoleMixin(VSphereMixin):
class ConsoleMixin(MCPMixin):
"""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.",
annotations=ToolAnnotations(readOnlyHint=True),
)
def wait_for_vm_tools(
self, name: str, timeout: int = 120, poll_interval: int = 5,
host: str | None = None
self, name: str, timeout: int = 120, poll_interval: int = 5
) -> dict[str, Any]:
"""Wait for VMware Tools to become available.
@ -34,13 +34,11 @@ class ConsoleMixin(VSphereMixin):
name: VM name
timeout: Maximum seconds to wait (default: 120)
poll_interval: Seconds between status checks (default: 5)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with tools status, version, and guest info when ready
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -79,18 +77,16 @@ class ConsoleMixin(VSphereMixin):
description="Get current VMware Tools status for a VM",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_vm_tools_status(self, name: str, host: str | None = None) -> dict[str, Any]:
def get_vm_tools_status(self, name: str) -> dict[str, Any]:
"""Get VMware Tools status without waiting.
Args:
name: VM name
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with current tools status and guest info
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -120,7 +116,6 @@ class ConsoleMixin(VSphereMixin):
name: str,
width: int | None = None,
height: int | None = None,
host: str | None = None,
) -> dict[str, Any]:
"""Capture VM console screenshot via vSphere HTTP API.
@ -128,21 +123,19 @@ class ConsoleMixin(VSphereMixin):
name: VM name
width: Optional width to scale the image
height: Optional height to scale the image
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with base64-encoded image data and metadata
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
# Build screenshot URL
# Format: https://{host}/screen?id={moid}
vcenter_host = conn.settings.vcenter_host
host = self.conn.settings.vcenter_host
moid = vm._moId
screenshot_url = f"https://{vcenter_host}/screen?id={moid}"
screenshot_url = f"https://{host}/screen?id={moid}"
# Add optional scaling parameters
params = []
@ -154,8 +147,8 @@ class ConsoleMixin(VSphereMixin):
screenshot_url += "&" + "&".join(params)
# Build auth header
username = conn.settings.vcenter_user
password = conn.settings.vcenter_password.get_secret_value()
username = self.conn.settings.vcenter_user
password = self.conn.settings.vcenter_password.get_secret_value()
auth = base64.b64encode(f"{username}:{password}".encode()).decode("ascii")
# Make request
@ -163,7 +156,7 @@ class ConsoleMixin(VSphereMixin):
response = requests.get(
screenshot_url,
headers={"Authorization": f"Basic {auth}"},
verify=not conn.settings.vcenter_insecure,
verify=not self.conn.settings.vcenter_insecure,
timeout=30,
)
response.raise_for_status()

View File

@ -2,19 +2,20 @@
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class DiskManagementMixin(VSphereMixin):
class DiskManagementMixin(MCPMixin):
"""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 = []
@ -64,104 +65,6 @@ class DiskManagementMixin(VSphereMixin):
return device
return None
def _find_free_ide_slot(
self, vm: vim.VirtualMachine
) -> tuple[int | None, int | None]:
"""Return (controllerKey, unitNumber) for a free IDE slot, or (None, None)."""
ide_controllers = [
d
for d in vm.config.hardware.device
if isinstance(d, vim.vm.device.VirtualIDEController)
]
used: dict[int, set[int]] = {}
for d in vm.config.hardware.device:
if hasattr(d, "controllerKey") and hasattr(d, "unitNumber"):
used.setdefault(d.controllerKey, set()).add(d.unitNumber)
for controller in ide_controllers:
for unit in (0, 1): # each IDE controller holds two devices
if unit not in used.get(controller.key, set()):
return controller.key, unit
return None, None
@mcp_tool(
name="add_cdrom",
description="Add a CD/DVD drive to a VM, optionally mounting an ISO and booting from it",
annotations=ToolAnnotations(destructiveHint=True),
)
def add_cdrom(
self,
vm_name: str,
iso_path: str | None = None,
iso_datastore: str | None = None,
boot_from_iso: bool = False,
host: str | None = None,
) -> dict[str, Any]:
"""Add a CD/DVD drive to an existing VM.
Useful for appliances deployed from an OVA (e.g. Cisco CUCM), which
ship without a CD/DVD drive and must install from a bootable ISO.
Args:
vm_name: Name of the virtual machine
iso_path: ISO path on a datastore (e.g. 'iso/installer.iso') to mount
iso_datastore: Datastore holding the ISO (default: the VM's datastore)
boot_from_iso: Put the CD/DVD first in the boot order (for installers)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with the CD/DVD drive details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
controller_key, unit = self._find_free_ide_slot(vm)
if controller_key is None:
raise ValueError("No free IDE slot available for a CD/DVD drive")
cdrom_spec = vim.vm.device.VirtualDeviceSpec()
cdrom_spec.operation = vim.vm.device.VirtualDeviceSpec.Operation.add
cdrom_spec.device = vim.vm.device.VirtualCdrom()
cdrom_spec.device.controllerKey = controller_key
cdrom_spec.device.unitNumber = unit
cdrom_spec.device.key = -1
connectable = vim.vm.device.VirtualDevice.ConnectInfo()
connectable.allowGuestControl = True
connectable.connected = False
mounted_iso = None
if iso_path:
ds_name = iso_datastore or vm.config.files.vmPathName.split("]")[0].strip("[ ")
mounted_iso = f"[{ds_name}] {iso_path}"
backing = vim.vm.device.VirtualCdrom.IsoBackingInfo()
backing.fileName = mounted_iso
connectable.startConnected = True
else:
backing = vim.vm.device.VirtualCdrom.RemotePassthroughBackingInfo()
backing.deviceName = ""
backing.exclusive = False
connectable.startConnected = False
cdrom_spec.device.backing = backing
cdrom_spec.device.connectable = connectable
config_spec = vim.vm.ConfigSpec(deviceChange=[cdrom_spec])
if iso_path and boot_from_iso:
config_spec.bootOptions = vim.vm.BootOptions(
bootOrder=[vim.vm.BootOptions.BootableCdromDevice()]
)
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
return {
"vm": vm_name,
"action": "cdrom_added",
"iso": mounted_iso,
"boot_from_iso": bool(iso_path and boot_from_iso),
}
@mcp_tool(
name="add_disk",
description="Add a new virtual disk to a VM",
@ -173,7 +76,6 @@ class DiskManagementMixin(VSphereMixin):
size_gb: int,
thin_provisioned: bool = True,
datastore: str | None = None,
host: str | None = None,
) -> dict[str, Any]:
"""Add a new virtual disk to a VM.
@ -182,13 +84,11 @@ class DiskManagementMixin(VSphereMixin):
size_gb: Size of the new disk in GB
thin_provisioned: Use thin provisioning (default True)
datastore: Datastore for the disk (default: same as VM)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with new disk details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -197,7 +97,7 @@ class DiskManagementMixin(VSphereMixin):
# Determine datastore
if datastore:
ds = conn.find_datastore(datastore)
ds = self.conn.find_datastore(datastore)
if not ds:
raise ValueError(f"Datastore '{datastore}' not found")
ds_name = datastore
@ -212,7 +112,7 @@ class DiskManagementMixin(VSphereMixin):
backing = vim.vm.device.VirtualDisk.FlatVer2BackingInfo()
backing.diskMode = "persistent"
backing.thinProvisioned = thin_provisioned
backing.datastore = conn.find_datastore(ds_name)
backing.datastore = self.conn.find_datastore(ds_name)
# Create the virtual disk
disk = vim.vm.device.VirtualDisk()
@ -233,7 +133,7 @@ class DiskManagementMixin(VSphereMixin):
# Reconfigure VM
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm": vm_name,
@ -255,7 +155,6 @@ class DiskManagementMixin(VSphereMixin):
vm_name: str,
disk_label: str,
delete_file: bool = False,
host: str | None = None,
) -> dict[str, Any]:
"""Remove a virtual disk from a VM.
@ -263,13 +162,11 @@ class DiskManagementMixin(VSphereMixin):
vm_name: Name of the virtual machine
disk_label: Label of disk to remove (e.g., 'Hard disk 2')
delete_file: Also delete the VMDK file (default False - keep file)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with removal details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -300,7 +197,7 @@ class DiskManagementMixin(VSphereMixin):
# Reconfigure VM
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm": vm_name,
@ -321,7 +218,6 @@ class DiskManagementMixin(VSphereMixin):
vm_name: str,
disk_label: str,
new_size_gb: int,
host: str | None = None,
) -> dict[str, Any]:
"""Extend a virtual disk to a larger size.
@ -329,13 +225,11 @@ class DiskManagementMixin(VSphereMixin):
vm_name: Name of the virtual machine
disk_label: Label of disk to extend (e.g., 'Hard disk 1')
new_size_gb: New total size in GB (must be larger than current)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with extension details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -369,7 +263,7 @@ class DiskManagementMixin(VSphereMixin):
# Reconfigure VM
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm": vm_name,
@ -384,18 +278,16 @@ class DiskManagementMixin(VSphereMixin):
description="List all virtual disks attached to a VM",
annotations=ToolAnnotations(readOnlyHint=True),
)
def list_disks(self, vm_name: str, host: str | None = None) -> list[dict[str, Any]]:
def list_disks(self, vm_name: str) -> list[dict[str, Any]]:
"""List all virtual disks attached to a VM.
Args:
vm_name: Name of the virtual machine
host: Managed ESXi host to target (default: the default host)
Returns:
List of disk details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -430,7 +322,6 @@ class DiskManagementMixin(VSphereMixin):
vm_name: str,
iso_path: str,
datastore: str | None = None,
host: str | None = None,
) -> dict[str, Any]:
"""Attach an ISO file to a VM's CD/DVD drive.
@ -438,13 +329,11 @@ class DiskManagementMixin(VSphereMixin):
vm_name: Name of the virtual machine
iso_path: Path to ISO file on datastore (e.g., 'iso/ubuntu.iso')
datastore: Datastore containing the ISO (default: first VM datastore)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with attachment details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -481,7 +370,7 @@ class DiskManagementMixin(VSphereMixin):
# Reconfigure VM
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm": vm_name,
@ -496,18 +385,16 @@ class DiskManagementMixin(VSphereMixin):
description="Detach/eject ISO from a VM's CD/DVD drive",
annotations=ToolAnnotations(destructiveHint=True),
)
def detach_iso(self, vm_name: str, host: str | None = None) -> dict[str, Any]:
def detach_iso(self, vm_name: str) -> dict[str, Any]:
"""Detach/eject ISO from a VM's CD/DVD drive.
Args:
vm_name: Name of the virtual machine
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with detachment details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -542,7 +429,7 @@ class DiskManagementMixin(VSphereMixin):
# Reconfigure VM
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm": vm_name,

View File

@ -4,19 +4,20 @@ import base64
import time
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class GuestOpsMixin(VSphereMixin):
class GuestOpsMixin(MCPMixin):
"""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:
@ -53,7 +54,6 @@ class GuestOpsMixin(VSphereMixin):
working_directory: str = "",
wait_for_completion: bool = True,
timeout_seconds: int = 300,
host: str | None = None,
) -> dict[str, Any]:
"""Run a command in the guest OS.
@ -66,16 +66,14 @@ class GuestOpsMixin(VSphereMixin):
working_directory: Working directory for the command
wait_for_completion: Wait for command to complete
timeout_seconds: Timeout in seconds (only if waiting)
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
self._check_tools_running(vm)
guest_ops = conn.content.guestOperationsManager
guest_ops = self.conn.content.guestOperationsManager
process_manager = guest_ops.processManager
auth = self._get_guest_auth(username, password)
@ -121,17 +119,16 @@ class GuestOpsMixin(VSphereMixin):
annotations=ToolAnnotations(readOnlyHint=True),
)
def list_guest_processes(
self, name: str, username: str, password: str, host: str | None = None
self, name: str, username: str, password: str
) -> list[dict[str, Any]]:
"""List processes running in the guest OS."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
self._check_tools_running(vm)
guest_ops = conn.content.guestOperationsManager
guest_ops = self.conn.content.guestOperationsManager
process_manager = guest_ops.processManager
auth = self._get_guest_auth(username, password)
@ -154,8 +151,7 @@ class GuestOpsMixin(VSphereMixin):
annotations=ToolAnnotations(readOnlyHint=True),
)
def read_guest_file(
self, name: str, username: str, password: str, guest_path: str,
host: str | None = None
self, name: str, username: str, password: str, guest_path: str
) -> dict[str, Any]:
"""Read a file from the guest OS.
@ -164,16 +160,14 @@ class GuestOpsMixin(VSphereMixin):
username: Guest OS username
password: Guest OS password
guest_path: Path to file in guest (e.g., /etc/hosts, C:\\Windows\\System32\\hosts)
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
self._check_tools_running(vm)
guest_ops = conn.content.guestOperationsManager
guest_ops = self.conn.content.guestOperationsManager
file_manager = guest_ops.fileManager
auth = self._get_guest_auth(username, password)
@ -234,7 +228,6 @@ class GuestOpsMixin(VSphereMixin):
guest_path: str,
content: str,
overwrite: bool = True,
host: str | None = None,
) -> str:
"""Write a file to the guest OS.
@ -245,16 +238,14 @@ class GuestOpsMixin(VSphereMixin):
guest_path: Destination path in guest
content: File content (text)
overwrite: Overwrite if exists
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
self._check_tools_running(vm)
guest_ops = conn.content.guestOperationsManager
guest_ops = self.conn.content.guestOperationsManager
file_manager = guest_ops.fileManager
auth = self._get_guest_auth(username, password)
@ -297,18 +288,16 @@ class GuestOpsMixin(VSphereMixin):
annotations=ToolAnnotations(readOnlyHint=True),
)
def list_guest_directory(
self, name: str, username: str, password: str, guest_path: str,
host: str | None = None
self, name: str, username: str, password: str, guest_path: str
) -> list[dict[str, Any]]:
"""List files in a guest directory."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
self._check_tools_running(vm)
guest_ops = conn.content.guestOperationsManager
guest_ops = self.conn.content.guestOperationsManager
file_manager = guest_ops.fileManager
auth = self._get_guest_auth(username, password)
@ -343,17 +332,15 @@ class GuestOpsMixin(VSphereMixin):
password: str,
guest_path: str,
create_parents: bool = True,
host: str | None = None,
) -> str:
"""Create a directory in the guest OS."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
self._check_tools_running(vm)
guest_ops = conn.content.guestOperationsManager
guest_ops = self.conn.content.guestOperationsManager
file_manager = guest_ops.fileManager
auth = self._get_guest_auth(username, password)
@ -369,18 +356,16 @@ class GuestOpsMixin(VSphereMixin):
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=True),
)
def delete_guest_file(
self, name: str, username: str, password: str, guest_path: str,
host: str | None = None
self, name: str, username: str, password: str, guest_path: str
) -> str:
"""Delete a file or directory from the guest OS."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
self._check_tools_running(vm)
guest_ops = conn.content.guestOperationsManager
guest_ops = self.conn.content.guestOperationsManager
file_manager = guest_ops.fileManager
auth = self._get_guest_auth(username, password)

View File

@ -2,22 +2,23 @@
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class HostManagementMixin(VSphereMixin):
class HostManagementMixin(MCPMixin):
"""ESXi host management tools."""
def _get_host(self, conn) -> vim.HostSystem:
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
def _get_host(self) -> vim.HostSystem:
"""Get the ESXi host system."""
for entity in conn.datacenter.hostFolder.childEntity:
for entity in self.conn.datacenter.hostFolder.childEntity:
if isinstance(entity, vim.ComputeResource):
if entity.host:
return entity.host[0]
@ -30,17 +31,13 @@ class HostManagementMixin(VSphereMixin):
description="Get detailed information about the ESXi host",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_host_info(self, host: str | None = None) -> dict[str, Any]:
def get_host_info(self) -> dict[str, Any]:
"""Get detailed ESXi host information.
Args:
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with host details including hardware, software, and status
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
summary = host.summary
hardware = summary.hardware
config = summary.config
@ -84,20 +81,17 @@ class HostManagementMixin(VSphereMixin):
self,
evacuate_vms: bool = True,
timeout_seconds: int = 300,
host: str | None = None,
) -> dict[str, Any]:
"""Put ESXi host into maintenance mode.
Args:
evacuate_vms: Evacuate/suspend VMs before entering (default True)
timeout_seconds: Timeout for the operation (default 300)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with operation result
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
if host.runtime.inMaintenanceMode:
return {
@ -111,7 +105,7 @@ class HostManagementMixin(VSphereMixin):
timeout=timeout_seconds,
evacuatePoweredOffVms=evacuate_vms,
)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"host": host.name,
@ -128,19 +122,16 @@ class HostManagementMixin(VSphereMixin):
def exit_maintenance_mode(
self,
timeout_seconds: int = 300,
host: str | None = None,
) -> dict[str, Any]:
"""Exit ESXi host from maintenance mode.
Args:
timeout_seconds: Timeout for the operation (default 300)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with operation result
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
if not host.runtime.inMaintenanceMode:
return {
@ -150,7 +141,7 @@ class HostManagementMixin(VSphereMixin):
}
task = host.ExitMaintenanceMode_Task(timeout=timeout_seconds)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"host": host.name,
@ -163,17 +154,13 @@ class HostManagementMixin(VSphereMixin):
description="List all services on the ESXi host",
annotations=ToolAnnotations(readOnlyHint=True),
)
def list_services(self, host: str | None = None) -> list[dict[str, Any]]:
def list_services(self) -> list[dict[str, Any]]:
"""List all services on the ESXi host.
Args:
host: Managed ESXi host to target (default: the default host)
Returns:
List of service details
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
service_system = host.configManager.serviceSystem
services = []
@ -194,18 +181,16 @@ class HostManagementMixin(VSphereMixin):
description="Start a service on the ESXi host",
annotations=ToolAnnotations(destructiveHint=True),
)
def start_service(self, service_key: str, host: str | None = None) -> dict[str, Any]:
def start_service(self, service_key: str) -> dict[str, Any]:
"""Start a service on the ESXi host.
Args:
service_key: Service key (e.g., 'TSM-SSH', 'ntpd', 'sfcbd')
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with operation result
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
service_system = host.configManager.serviceSystem
# Verify service exists
@ -241,18 +226,16 @@ class HostManagementMixin(VSphereMixin):
description="Stop a service on the ESXi host",
annotations=ToolAnnotations(destructiveHint=True),
)
def stop_service(self, service_key: str, host: str | None = None) -> dict[str, Any]:
def stop_service(self, service_key: str) -> dict[str, Any]:
"""Stop a service on the ESXi host.
Args:
service_key: Service key (e.g., 'TSM-SSH', 'ntpd')
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with operation result
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
service_system = host.configManager.serviceSystem
# Verify service exists
@ -292,20 +275,17 @@ class HostManagementMixin(VSphereMixin):
self,
service_key: str,
policy: str,
host: str | None = None,
) -> dict[str, Any]:
"""Set the startup policy for a service.
Args:
service_key: Service key (e.g., 'TSM-SSH', 'ntpd')
policy: Startup policy - 'on' (auto), 'off' (manual), 'automatic'
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with operation result
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
service_system = host.configManager.serviceSystem
valid_policies = ["on", "off", "automatic"]
@ -339,17 +319,13 @@ class HostManagementMixin(VSphereMixin):
description="Get NTP configuration for the ESXi host",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_ntp_config(self, host: str | None = None) -> dict[str, Any]:
def get_ntp_config(self) -> dict[str, Any]:
"""Get NTP configuration for the ESXi host.
Args:
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with NTP configuration
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
datetime_system = host.configManager.dateTimeSystem
ntp_config = datetime_system.dateTimeInfo.ntpConfig
@ -382,20 +358,17 @@ class HostManagementMixin(VSphereMixin):
self,
ntp_servers: list[str],
start_service: bool = True,
host: str | None = None,
) -> dict[str, Any]:
"""Configure NTP servers for the ESXi host.
Args:
ntp_servers: List of NTP server addresses
start_service: Start ntpd service after configuring (default True)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with configuration result
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
datetime_system = host.configManager.dateTimeSystem
# Create NTP config
@ -434,18 +407,16 @@ class HostManagementMixin(VSphereMixin):
description="Reboot the ESXi host (requires maintenance mode)",
annotations=ToolAnnotations(destructiveHint=True),
)
def reboot_host(self, force: bool = False, host: str | None = None) -> dict[str, Any]:
def reboot_host(self, force: bool = False) -> dict[str, Any]:
"""Reboot the ESXi host.
Args:
force: Force reboot even if VMs are running (dangerous!)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with operation result
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
if not host.runtime.inMaintenanceMode and not force:
raise ValueError(
@ -468,18 +439,16 @@ class HostManagementMixin(VSphereMixin):
description="Shutdown the ESXi host (requires maintenance mode)",
annotations=ToolAnnotations(destructiveHint=True),
)
def shutdown_host(self, force: bool = False, host: str | None = None) -> dict[str, Any]:
def shutdown_host(self, force: bool = False) -> dict[str, Any]:
"""Shutdown the ESXi host.
Args:
force: Force shutdown even if VMs are running (dangerous!)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with operation result
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
if not host.runtime.inMaintenanceMode and not force:
raise ValueError(
@ -502,17 +471,13 @@ class HostManagementMixin(VSphereMixin):
description="Get detailed hardware information for the ESXi host",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_host_hardware(self, host: str | None = None) -> dict[str, Any]:
def get_host_hardware(self) -> dict[str, Any]:
"""Get detailed hardware information.
Args:
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with hardware details
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
hardware = host.hardware
# CPU info
@ -568,17 +533,13 @@ class HostManagementMixin(VSphereMixin):
description="Get network configuration for the ESXi host",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_host_networking(self, host: str | None = None) -> dict[str, Any]:
def get_host_networking(self) -> dict[str, Any]:
"""Get network configuration for the ESXi host.
Args:
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with networking details
"""
conn = self._conn(host)
host = self._get_host(conn)
host = self._get_host()
network_config = host.config.network
# Virtual switches

View File

@ -3,32 +3,28 @@
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class MonitoringMixin(VSphereMixin):
class MonitoringMixin(MCPMixin):
"""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",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_vm_stats(self, name: str, host: str | None = None) -> dict[str, Any]:
"""Get VM performance statistics.
Args:
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
vm = conn.find_vm(name)
def get_vm_stats(self, name: str) -> dict[str, Any]:
"""Get VM performance statistics."""
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -63,25 +59,19 @@ class MonitoringMixin(VSphereMixin):
description="Get performance statistics for an ESXi host",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_host_stats(
self, host_name: str | None = None, host: str | None = None
) -> dict[str, Any]:
def get_host_stats(self, host_name: str | None = None) -> dict[str, Any]:
"""Get ESXi host performance statistics.
If host_name is not provided, returns stats for the first host.
Args:
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
if host_name:
host = conn.find_host(host_name)
host = self.conn.find_host(host_name)
if not host:
raise ValueError(f"Host '{host_name}' not found")
else:
# Get first host
container = conn.content.viewManager.CreateContainerView(
conn.content.rootFolder, [vim.HostSystem], True
container = self.conn.content.viewManager.CreateContainerView(
self.conn.content.rootFolder, [vim.HostSystem], True
)
try:
hosts = list(container.view)
@ -128,15 +118,10 @@ class MonitoringMixin(VSphereMixin):
description="List all ESXi hosts in the datacenter",
annotations=ToolAnnotations(readOnlyHint=True),
)
def list_hosts(self, host: str | None = None) -> list[dict[str, Any]]:
"""List all ESXi hosts with basic info.
Args:
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
container = conn.content.viewManager.CreateContainerView(
conn.content.rootFolder, [vim.HostSystem], True
def list_hosts(self) -> list[dict[str, Any]]:
"""List all ESXi hosts with basic info."""
container = self.conn.content.viewManager.CreateContainerView(
self.conn.content.rootFolder, [vim.HostSystem], True
)
try:
hosts = []
@ -163,16 +148,9 @@ class MonitoringMixin(VSphereMixin):
description="Get recent vSphere tasks (VM operations, etc.)",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_recent_tasks(
self, count: int = 20, host: str | None = None
) -> list[dict[str, Any]]:
"""Get recent vSphere tasks.
Args:
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
task_manager = conn.content.taskManager
def get_recent_tasks(self, count: int = 20) -> list[dict[str, Any]]:
"""Get recent vSphere tasks."""
task_manager = self.conn.content.taskManager
recent_tasks = task_manager.recentTask[:count] if task_manager.recentTask else []
tasks = []
@ -210,15 +188,10 @@ class MonitoringMixin(VSphereMixin):
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_recent_events(
self, count: int = 50, hours: int = 24, host: str | None = None
self, count: int = 50, hours: int = 24
) -> list[dict[str, Any]]:
"""Get recent vSphere events.
Args:
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
event_manager = conn.content.eventManager
"""Get recent vSphere events."""
event_manager = self.conn.content.eventManager
# Create time filter
time_filter = vim.event.EventFilterSpec.ByTime()
@ -262,22 +235,17 @@ class MonitoringMixin(VSphereMixin):
description="Get triggered alarms in the datacenter",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_alarms(self, host: str | None = None) -> list[dict[str, Any]]:
"""Get all triggered alarms.
Args:
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
def get_alarms(self) -> list[dict[str, Any]]:
"""Get all triggered alarms."""
alarms = []
# Check datacenter alarms
if conn.datacenter.triggeredAlarmState:
for alarm_state in conn.datacenter.triggeredAlarmState:
if self.conn.datacenter.triggeredAlarmState:
for alarm_state in self.conn.datacenter.triggeredAlarmState:
alarms.append(self._format_alarm(alarm_state))
# Check VM alarms
for vm in conn.get_all_vms():
for vm in self.conn.get_all_vms():
if vm.triggeredAlarmState:
for alarm_state in vm.triggeredAlarmState:
alarms.append(self._format_alarm(alarm_state, vm.name))

View File

@ -2,19 +2,20 @@
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class NICManagementMixin(VSphereMixin):
class NICManagementMixin(MCPMixin):
"""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:
@ -28,10 +29,10 @@ class NICManagementMixin(VSphereMixin):
return None
def _get_network_backing(
self, conn, network_name: str
self, network_name: str
) -> vim.vm.device.VirtualEthernetCard.NetworkBackingInfo:
"""Get the appropriate backing info for a network."""
network = conn.find_network(network_name)
network = self.conn.find_network(network_name)
if not network:
raise ValueError(f"Network '{network_name}' not found")
@ -54,18 +55,16 @@ class NICManagementMixin(VSphereMixin):
description="List all network adapters attached to a VM",
annotations=ToolAnnotations(readOnlyHint=True),
)
def list_nics(self, vm_name: str, host: str | None = None) -> list[dict[str, Any]]:
def list_nics(self, vm_name: str) -> list[dict[str, Any]]:
"""List all virtual network adapters on a VM.
Args:
vm_name: Name of the virtual machine
host: Managed ESXi host to target (default: the default host)
Returns:
List of NIC details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -89,7 +88,7 @@ class NICManagementMixin(VSphereMixin):
# For distributed switch, look up the portgroup name
nic_info["network"] = f"DVS:{backing.port.portgroupKey}"
# Try to get actual name
for net in conn.datacenter.networkFolder.childEntity:
for net in self.conn.datacenter.networkFolder.childEntity:
if hasattr(net, "key") and net.key == backing.port.portgroupKey:
nic_info["network"] = net.name
break
@ -109,7 +108,6 @@ class NICManagementMixin(VSphereMixin):
network: str,
nic_type: str = "vmxnet3",
start_connected: bool = True,
host: str | None = None,
) -> dict[str, Any]:
"""Add a new network adapter to a VM.
@ -118,13 +116,11 @@ class NICManagementMixin(VSphereMixin):
network: Network/portgroup name to connect to
nic_type: Adapter type - vmxnet3 (default), e1000, e1000e
start_connected: Connect adapter when VM powers on (default True)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with new NIC details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -142,7 +138,7 @@ class NICManagementMixin(VSphereMixin):
# Create the NIC
nic = nic_class()
nic.backing = self._get_network_backing(conn, network)
nic.backing = self._get_network_backing(network)
nic.connectable = vim.vm.device.VirtualDevice.ConnectInfo()
nic.connectable.startConnected = start_connected
nic.connectable.connected = False # Can't connect until powered on
@ -160,7 +156,7 @@ class NICManagementMixin(VSphereMixin):
# Reconfigure VM
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
# Get the MAC address that was assigned
vm.Reload()
@ -195,20 +191,17 @@ class NICManagementMixin(VSphereMixin):
self,
vm_name: str,
nic_label: str,
host: str | None = None,
) -> dict[str, Any]:
"""Remove a network adapter from a VM.
Args:
vm_name: Name of the virtual machine
nic_label: Label of NIC to remove (e.g., 'Network adapter 1')
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with removal details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -238,7 +231,7 @@ class NICManagementMixin(VSphereMixin):
# Reconfigure VM
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm": vm_name,
@ -258,7 +251,6 @@ class NICManagementMixin(VSphereMixin):
vm_name: str,
nic_label: str,
new_network: str,
host: str | None = None,
) -> dict[str, Any]:
"""Change which network a NIC is connected to.
@ -266,13 +258,11 @@ class NICManagementMixin(VSphereMixin):
vm_name: Name of the virtual machine
nic_label: Label of NIC to modify (e.g., 'Network adapter 1')
new_network: New network/portgroup name
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with change details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -291,7 +281,7 @@ class NICManagementMixin(VSphereMixin):
old_network = nic.backing.deviceName
# Update backing to new network
nic.backing = self._get_network_backing(conn, new_network)
nic.backing = self._get_network_backing(new_network)
# Create device edit spec
nic_spec = vim.vm.device.VirtualDeviceSpec()
@ -304,7 +294,7 @@ class NICManagementMixin(VSphereMixin):
# Reconfigure VM
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm": vm_name,
@ -325,7 +315,6 @@ class NICManagementMixin(VSphereMixin):
vm_name: str,
nic_label: str,
connected: bool = True,
host: str | None = None,
) -> dict[str, Any]:
"""Connect or disconnect a NIC on a running VM.
@ -333,13 +322,11 @@ class NICManagementMixin(VSphereMixin):
vm_name: Name of the virtual machine
nic_label: Label of NIC (e.g., 'Network adapter 1')
connected: True to connect, False to disconnect
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with connection status
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -369,7 +356,7 @@ class NICManagementMixin(VSphereMixin):
# Reconfigure VM
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm": vm_name,
@ -388,7 +375,6 @@ class NICManagementMixin(VSphereMixin):
vm_name: str,
nic_label: str,
mac_address: str,
host: str | None = None,
) -> dict[str, Any]:
"""Set a custom MAC address for a NIC.
@ -396,13 +382,11 @@ class NICManagementMixin(VSphereMixin):
vm_name: Name of the virtual machine
nic_label: Label of NIC (e.g., 'Network adapter 1')
mac_address: MAC address in format XX:XX:XX:XX:XX:XX
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with MAC address change details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -437,7 +421,7 @@ class NICManagementMixin(VSphereMixin):
# Reconfigure VM
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm": vm_name,

View File

@ -1,32 +1,26 @@
"""OVF/OVA Management - deploy and export virtual appliances."""
import contextlib
import http.client
import shutil
import ssl
import tarfile
import tempfile
import threading
import time
import urllib.request
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import TYPE_CHECKING, Any
from urllib.parse import urlparse
from fastmcp.contrib.mcp_mixin import mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class OVFManagementMixin(VSphereMixin):
class OVFManagementMixin(MCPMixin):
"""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_")
@ -56,12 +50,11 @@ class OVFManagementMixin(VSphereMixin):
_lease: vim.HttpNfcLease,
disk_path: str,
device_url: str,
conn,
) -> None:
"""Upload a disk file via NFC lease."""
# Create SSL context
context = ssl.create_default_context()
if conn.settings.vcenter_insecure:
if self.conn.settings.vcenter_insecure:
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
@ -75,8 +68,8 @@ class OVFManagementMixin(VSphereMixin):
request.add_header("Connection", "Keep-Alive")
# Add session cookie
if hasattr(conn.service_instance, "_stub"):
cookie = conn.service_instance._stub.cookie
if hasattr(self.conn.service_instance, "_stub"):
cookie = self.conn.service_instance._stub.cookie
if cookie:
request.add_header("Cookie", cookie)
@ -97,7 +90,6 @@ class OVFManagementMixin(VSphereMixin):
datastore: str,
network: str | None = None,
power_on: bool = False,
host: str | None = None,
) -> dict[str, Any]:
"""Deploy a virtual machine from an OVF or OVA file.
@ -109,23 +101,21 @@ class OVFManagementMixin(VSphereMixin):
datastore: Target datastore for VM files
network: Network to connect VM to (optional)
power_on: Power on VM after deployment (default False)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with deployment details
"""
conn = self._conn(host)
# Get OVF Manager
ovf_manager = conn.content.ovfManager
ovf_manager = self.conn.content.ovfManager
# Find target datastore
ds = conn.find_datastore(datastore)
ds = self.conn.find_datastore(datastore)
if not ds:
raise ValueError(f"Datastore '{datastore}' not found")
# Get resource pool and folder
host = None
for h in conn.datacenter.hostFolder.childEntity:
for h in self.conn.datacenter.hostFolder.childEntity:
if hasattr(h, "host"):
host = h.host[0] if h.host else None
break
@ -143,7 +133,7 @@ class OVFManagementMixin(VSphereMixin):
resource_pool = host.parent.resourcePool
# Get VM folder
vm_folder = conn.datacenter.vmFolder
vm_folder = self.conn.datacenter.vmFolder
# Read OVF descriptor from datastore
# For OVA, we need to extract first
@ -158,7 +148,7 @@ class OVFManagementMixin(VSphereMixin):
)
# Read OVF descriptor via datastore browser
ovf_descriptor = self._read_datastore_file(datastore, ovf_path, conn)
ovf_descriptor = self._read_datastore_file(datastore, ovf_path)
# Create import spec params
import_spec_params = vim.OvfManager.CreateImportSpecParams(
@ -168,7 +158,7 @@ class OVFManagementMixin(VSphereMixin):
# If network specified, add network mapping
if network:
net = conn.find_network(network)
net = self.conn.find_network(network)
if net:
network_mapping = vim.OvfManager.NetworkMapping(
name="VM Network", # Common default in OVF
@ -217,7 +207,7 @@ class OVFManagementMixin(VSphereMixin):
lease.Complete()
# Find the newly created VM
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
result = {
"vm": vm_name,
@ -230,35 +220,35 @@ class OVFManagementMixin(VSphereMixin):
result["uuid"] = vm.config.uuid
if power_on:
task = vm.PowerOnVM_Task()
conn.wait_for_task(task)
self.conn.wait_for_task(task)
result["power_state"] = "poweredOn"
else:
result["power_state"] = "poweredOff"
return result
def _read_datastore_file(self, datastore: str, path: str, conn) -> str:
def _read_datastore_file(self, datastore: str, path: str) -> str:
"""Read a text file from datastore."""
ds = conn.find_datastore(datastore)
ds = self.conn.find_datastore(datastore)
if not ds:
raise ValueError(f"Datastore '{datastore}' not found")
# Build HTTP URL
dc_name = conn.datacenter.name
dc_name = self.conn.datacenter.name
url = (
f"https://{conn.settings.vcenter_host}/folder/{path}"
f"https://{self.conn.settings.vcenter_host}/folder/{path}"
f"?dcPath={dc_name}&dsName={datastore}"
)
# Setup request
context = ssl.create_default_context()
if conn.settings.vcenter_insecure:
if self.conn.settings.vcenter_insecure:
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
request = urllib.request.Request(url)
if hasattr(conn.service_instance, "_stub"):
cookie = conn.service_instance._stub.cookie
if hasattr(self.conn.service_instance, "_stub"):
cookie = self.conn.service_instance._stub.cookie
if cookie:
request.add_header("Cookie", cookie)
@ -273,452 +263,6 @@ class OVFManagementMixin(VSphereMixin):
# For now, document this limitation
pass
def _resolve_ova_file(self, ova_path: str, conn) -> tuple[str, bool]:
"""Return (local_path, is_temp); downloads the OVA first if a URL is given."""
if ova_path.startswith(("http://", "https://")):
context = ssl.create_default_context()
if conn.settings.vcenter_insecure:
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
fd, tmp = tempfile.mkstemp(suffix=".ova", prefix="ova_dl_")
with open(fd, "wb") as out:
req = urllib.request.Request(ova_path)
with urllib.request.urlopen(req, context=context) as resp:
shutil.copyfileobj(resp, out)
return tmp, True
if not Path(ova_path).is_file():
raise ValueError(f"OVA file not found: {ova_path}")
return ova_path, False
@staticmethod
def _parse_ovf_network_names(ovf_xml: str) -> list[str]:
"""Extract declared network names from an OVF descriptor's NetworkSection."""
names: list[str] = []
try:
root = ET.fromstring(ovf_xml)
except ET.ParseError:
return names
for el in root.iter():
if el.tag.rsplit("}", 1)[-1] == "Network":
for key, val in el.attrib.items():
if key.rsplit("}", 1)[-1] == "name":
names.append(val)
return names
def _put_stream_to_lease(
self, device_url: str, fileobj: Any, size: int, conn
) -> None:
"""Stream a file object to an NFC lease device URL via chunked HTTP PUT."""
parsed = urlparse(device_url)
host = parsed.hostname
# ESXi returns '*' (meaning "the host you connected to") or its own
# mgmt name. For a direct host connection the only address we know is
# reachable is the one we connected on, so use that.
if (
not host
or host == "*"
or conn.content.about.apiType == "HostAgent"
):
host = conn.settings.vcenter_host
port = parsed.port or 443
path = parsed.path + (f"?{parsed.query}" if parsed.query else "")
context = ssl.create_default_context()
if conn.settings.vcenter_insecure:
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
http_conn = http.client.HTTPSConnection(
host, port, context=context, timeout=900
)
try:
http_conn.putrequest(
"PUT", path, skip_host=True, skip_accept_encoding=True
)
http_conn.putheader("Host", host)
http_conn.putheader("Content-Length", str(size))
http_conn.putheader("Content-Type", "application/x-vnd.vmware-streamVmdk")
# ImportVApp pre-creates the disk file, so the NFC stream must
# explicitly overwrite it (else ESXi returns 403 "File exists").
http_conn.putheader("Overwrite", "t")
cookie = getattr(conn.si._stub, "cookie", None)
if cookie:
http_conn.putheader("Cookie", cookie)
http_conn.endheaders()
while True:
chunk = fileobj.read(1024 * 1024)
if not chunk:
break
http_conn.send(chunk)
resp = http_conn.getresponse()
resp.read()
if resp.status not in (200, 201):
raise ValueError(
f"Disk upload failed: HTTP {resp.status} {resp.reason}"
)
finally:
http_conn.close()
@mcp_tool(
name="deploy_ova",
description="Deploy a VM from an OVA file (local path or URL) by streaming it to the host over an NFC lease",
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=False),
)
def deploy_ova(
self,
ova_path: str,
vm_name: str,
datastore: str | None = None,
network: str | None = None,
power_on: bool = False,
disk_provisioning: str = "thin",
deployment_option: str | None = None,
iso_path: str | None = None,
iso_datastore: str | None = None,
boot_from_iso: bool = True,
host: str | None = None,
) -> dict[str, Any]:
"""Deploy a virtual machine from an OVA file.
The OVA is read from the MCP server host (local filesystem path) or
downloaded if an http(s) URL is given, then streamed to the ESXi host
over an NFC lease. Unlike datastore-based deploy, the OVA does not need
to be pre-staged on a datastore.
Args:
ova_path: Local path to the .ova file, or an http(s) URL to download
vm_name: Name for the new VM
datastore: Target datastore (default: largest available)
network: Target port group; every network in the OVF is mapped to it
power_on: Power on the VM after deployment
disk_provisioning: 'thin', 'thick', or 'eagerZeroedThick'
deployment_option: OVF configuration id for multi-config templates
(e.g. Cisco 'S'/'M'/'L'); defaults to the OVF's default config.
Use inspect_ova to list available options.
iso_path: ISO on a datastore to mount in a new CD/DVD drive after
deploy (for diskless appliance OVAs like CUCM that install from
a bootable ISO). A CD/DVD drive is added automatically.
iso_datastore: Datastore holding the ISO (default: the VM's datastore)
boot_from_iso: When an ISO is given, put the CD/DVD first in boot order
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with deployment details
"""
conn = self._conn(host)
ds = conn.find_datastore(datastore) if datastore else conn.datastore
if not ds:
raise ValueError(f"Datastore '{datastore}' not found")
resource_pool = conn.resource_pool
vm_folder = conn.datacenter.vmFolder
# An ESXi host is required as the import target
view = conn.content.viewManager.CreateContainerView(
conn.content.rootFolder, [vim.HostSystem], True
)
hosts = list(view.view)
view.Destroy()
host_system = hosts[0] if hosts else None
local_ova, is_temp = self._resolve_ova_file(ova_path, conn)
import_spec = None
try:
with tarfile.open(local_ova) as tar:
ovf_member = next(
(m for m in tar.getmembers() if m.name.lower().endswith(".ovf")),
None,
)
if not ovf_member:
raise ValueError("No .ovf descriptor found in OVA")
ovf_xml = tar.extractfile(ovf_member).read().decode("utf-8")
# Build the import spec, mapping every OVF network to the target
spec_params = vim.OvfManager.CreateImportSpecParams(
entityName=vm_name,
diskProvisioning=disk_provisioning,
)
if deployment_option:
spec_params.deploymentOption = deployment_option
if network:
net = conn.find_network(network)
if not net:
raise ValueError(f"Network '{network}' not found")
ovf_nets = self._parse_ovf_network_names(ovf_xml) or ["VM Network"]
spec_params.networkMapping = [
vim.OvfManager.NetworkMapping(name=n, network=net)
for n in ovf_nets
]
ovf_manager = conn.content.ovfManager
import_spec = ovf_manager.CreateImportSpec(
ovfDescriptor=ovf_xml,
resourcePool=resource_pool,
datastore=ds,
cisp=spec_params,
)
if import_spec.error:
raise ValueError(
"OVF import errors: "
+ "; ".join(str(e.msg) for e in import_spec.error)
)
lease = resource_pool.ImportVApp(
spec=import_spec.importSpec, folder=vm_folder, host=host_system
)
while lease.state == vim.HttpNfcLease.State.initializing:
time.sleep(0.1)
if lease.state == vim.HttpNfcLease.State.error:
raise ValueError(f"NFC lease error: {lease.error}")
# Stream each disk to its lease URL, keeping the lease alive
total = sum(max(fi.size, 1) for fi in import_spec.fileItem) or 1
uploaded = [0]
stop = threading.Event()
def _keepalive() -> None:
while not stop.wait(20):
try:
lease.HttpNfcLeaseProgress(
min(99, int(uploaded[0] * 100 / total))
)
except Exception:
return
keeper = threading.Thread(target=_keepalive, daemon=True)
keeper.start()
try:
url_by_key = {du.importKey: du.url for du in lease.info.deviceUrl}
with tarfile.open(local_ova) as tar:
by_base = {m.name.rsplit("/", 1)[-1]: m for m in tar.getmembers()}
for fi in import_spec.fileItem:
device_url = url_by_key.get(fi.deviceId)
if not device_url:
raise ValueError(f"No lease URL for device {fi.deviceId}")
member = by_base.get(fi.path.rsplit("/", 1)[-1])
if not member:
raise ValueError(f"Disk '{fi.path}' missing from OVA")
src = tar.extractfile(member)
self._put_stream_to_lease(
device_url, src, member.size, conn
)
uploaded[0] += member.size
lease.HttpNfcLeaseProgress(
min(99, int(uploaded[0] * 100 / total))
)
lease.HttpNfcLeaseProgress(100)
lease.Complete()
except Exception:
with contextlib.suppress(Exception):
lease.Abort()
raise
finally:
stop.set()
finally:
if is_temp:
Path(local_ova).unlink(missing_ok=True)
vm = conn.find_vm(vm_name)
result: dict[str, Any] = {
"vm": vm_name,
"action": "ova_deployed",
"datastore": ds.name,
"source": ova_path,
"disks": len(import_spec.fileItem) if import_spec else 0,
}
if vm:
result["uuid"] = vm.config.uuid
if iso_path:
result["iso"] = self._add_cdrom_with_iso(
vm, iso_path, iso_datastore, boot_from_iso, conn
)
if power_on:
task = vm.PowerOnVM_Task()
conn.wait_for_task(task)
result["power_state"] = "poweredOn"
else:
result["power_state"] = "poweredOff"
return result
def _add_cdrom_with_iso(
self,
vm: vim.VirtualMachine,
iso_path: str,
iso_datastore: str | None,
boot_from_iso: bool,
conn,
) -> str:
"""Mount an ISO in the VM's CD/DVD drive (reusing the first existing
drive, or adding one if none exist) and optionally boot from it.
Reusing the existing primary drive matters: many appliance OVAs (e.g.
Cisco CUCM) already define empty CD/DVD drives, and the BIOS boots the
first one, so the ISO must land there rather than on a newly-added
secondary drive.
Returns the mounted datastore ISO path.
"""
ds_name = iso_datastore or vm.config.files.vmPathName.split("]")[0].strip("[ ")
mounted = f"[{ds_name}] {iso_path}"
backing = vim.vm.device.VirtualCdrom.IsoBackingInfo()
backing.fileName = mounted
connectable = vim.vm.device.VirtualDevice.ConnectInfo()
connectable.allowGuestControl = True
connectable.startConnected = True
connectable.connected = (
vm.runtime.powerState == vim.VirtualMachinePowerState.poweredOn
)
existing = next(
(
d
for d in vm.config.hardware.device
if isinstance(d, vim.vm.device.VirtualCdrom)
),
None,
)
cdrom_spec = vim.vm.device.VirtualDeviceSpec()
if existing is not None:
existing.backing = backing
existing.connectable = connectable
cdrom_spec.operation = vim.vm.device.VirtualDeviceSpec.Operation.edit
cdrom_spec.device = existing
else:
# No drive present — add one on a free IDE slot
used: dict[int, set[int]] = {}
ide_controllers = []
for d in vm.config.hardware.device:
if isinstance(d, vim.vm.device.VirtualIDEController):
ide_controllers.append(d)
if hasattr(d, "controllerKey") and hasattr(d, "unitNumber"):
used.setdefault(d.controllerKey, set()).add(d.unitNumber)
controller_key = unit = None
for controller in ide_controllers:
for candidate in (0, 1):
if candidate not in used.get(controller.key, set()):
controller_key, unit = controller.key, candidate
break
if controller_key is not None:
break
if controller_key is None:
raise ValueError("No free IDE slot for a CD/DVD drive")
cdrom_spec.operation = vim.vm.device.VirtualDeviceSpec.Operation.add
cdrom_spec.device = vim.vm.device.VirtualCdrom()
cdrom_spec.device.controllerKey = controller_key
cdrom_spec.device.unitNumber = unit
cdrom_spec.device.key = -1
cdrom_spec.device.backing = backing
cdrom_spec.device.connectable = connectable
config_spec = vim.vm.ConfigSpec(deviceChange=[cdrom_spec])
if boot_from_iso:
config_spec.bootOptions = vim.vm.BootOptions(
bootOrder=[vim.vm.BootOptions.BootableCdromDevice()]
)
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
return mounted
@mcp_tool(
name="inspect_ova",
description="Inspect an OVA (local path or URL): product, deployment configurations, networks, and disks",
annotations=ToolAnnotations(readOnlyHint=True),
)
def inspect_ova(self, ova_path: str, host: str | None = None) -> dict[str, Any]:
"""Inspect an OVA without deploying it.
Reads the OVF descriptor and reports the product, guest OS, hardware
version, selectable deployment configurations (e.g. Cisco S/M/L), the
networks the OVF expects, and its disks. Useful for picking a
``deployment_option`` and ``network`` before calling ``deploy_ova``.
Args:
ova_path: Local path to the .ova file, or an http(s) URL
host: Managed ESXi host to target (default: the default host)
Returns:
Dict describing the OVA
"""
conn = self._conn(host)
local_ova, is_temp = self._resolve_ova_file(ova_path, conn)
try:
with tarfile.open(local_ova) as tar:
ovf_member = next(
(m for m in tar.getmembers() if m.name.lower().endswith(".ovf")),
None,
)
if not ovf_member:
raise ValueError("No .ovf descriptor found in OVA")
ovf_xml = tar.extractfile(ovf_member).read().decode("utf-8")
disk_files = [
m.name for m in tar.getmembers() if m.name.lower().endswith(".vmdk")
]
finally:
if is_temp:
Path(local_ova).unlink(missing_ok=True)
def lname(tag: str) -> str:
return tag.rsplit("}", 1)[-1]
root = ET.fromstring(ovf_xml)
product: str | None = None
os_type: str | None = None
hw_version: str | None = None
configs: list[dict[str, Any]] = []
networks: list[str] = []
disks: list[dict[str, Any]] = []
for el in root.iter():
tag = lname(el.tag)
if tag == "Product" and product is None:
product = (el.text or "").strip()
elif tag == "OperatingSystemSection":
for key, val in el.attrib.items():
if lname(key) == "osType":
os_type = val
elif tag == "VirtualSystemType" and hw_version is None:
hw_version = (el.text or "").strip()
elif tag == "Configuration":
cid = None
default = False
for key, val in el.attrib.items():
if lname(key) == "id":
cid = val
elif lname(key) == "default":
default = val == "true"
label = ""
for child in el:
if lname(child.tag) == "Label":
label = (child.text or "").strip()
configs.append({"id": cid, "label": label, "default": default})
elif tag == "Network":
for key, val in el.attrib.items():
if lname(key) == "name":
networks.append(val)
elif tag == "Disk":
disk: dict[str, Any] = {}
for key, val in el.attrib.items():
name = lname(key)
if name == "diskId":
disk["disk_id"] = val
elif name == "capacity":
disk["capacity"] = val
elif name == "capacityAllocationUnits":
disk["units"] = val
disks.append(disk)
return {
"source": ova_path,
"product": product,
"os_type": os_type,
"hardware_version": hw_version,
"deployment_options": configs,
"networks": networks,
"disks": disks,
"disk_files": disk_files,
}
@mcp_tool(
name="export_vm_ovf",
description="Export a VM to OVF format on a datastore",
@ -729,7 +273,6 @@ class OVFManagementMixin(VSphereMixin):
vm_name: str,
target_path: str,
datastore: str | None = None,
host: str | None = None,
) -> dict[str, Any]:
"""Export a virtual machine to OVF format.
@ -737,13 +280,11 @@ class OVFManagementMixin(VSphereMixin):
vm_name: Name of the VM to export
target_path: Target directory path on datastore
datastore: Target datastore (default: VM's datastore)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with export details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -753,7 +294,7 @@ class OVFManagementMixin(VSphereMixin):
# Determine target datastore
if datastore:
ds = conn.find_datastore(datastore)
ds = self.conn.find_datastore(datastore)
if not ds:
raise ValueError(f"Datastore '{datastore}' not found")
ds_name = datastore
@ -771,7 +312,7 @@ class OVFManagementMixin(VSphereMixin):
raise ValueError(f"Export lease error: {lease.error}")
# Get OVF descriptor
ovf_manager = conn.content.ovfManager
ovf_manager = self.conn.content.ovfManager
ovf_descriptor = ovf_manager.CreateDescriptor(
obj=vm,
cdp=vim.OvfManager.CreateDescriptorParams(
@ -805,9 +346,7 @@ class OVFManagementMixin(VSphereMixin):
ovf_output_path = f"{target_path}/{ovf_filename}"
# Upload OVF descriptor to datastore
self._write_datastore_file(
ds_name, ovf_output_path, ovf_descriptor.ovfDescriptor, conn
)
self._write_datastore_file(ds_name, ovf_output_path, ovf_descriptor.ovfDescriptor)
exported_files.append(ovf_output_path)
@ -823,19 +362,17 @@ class OVFManagementMixin(VSphereMixin):
"ovf_descriptor": ovf_filename,
}
def _write_datastore_file(
self, datastore: str, path: str, content: str, conn
) -> None:
def _write_datastore_file(self, datastore: str, path: str, content: str) -> None:
"""Write a text file to datastore."""
dc_name = conn.datacenter.name
dc_name = self.conn.datacenter.name
url = (
f"https://{conn.settings.vcenter_host}/folder/{path}"
f"https://{self.conn.settings.vcenter_host}/folder/{path}"
f"?dcPath={dc_name}&dsName={datastore}"
)
# Setup request
context = ssl.create_default_context()
if conn.settings.vcenter_insecure:
if self.conn.settings.vcenter_insecure:
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
@ -844,8 +381,8 @@ class OVFManagementMixin(VSphereMixin):
request.add_header("Content-Type", "application/xml")
request.add_header("Content-Length", str(len(data)))
if hasattr(conn.service_instance, "_stub"):
cookie = conn.service_instance._stub.cookie
if hasattr(self.conn.service_instance, "_stub"):
cookie = self.conn.service_instance._stub.cookie
if cookie:
request.add_header("Cookie", cookie)
@ -860,28 +397,25 @@ class OVFManagementMixin(VSphereMixin):
self,
ovf_path: str,
datastore: str,
host: str | None = None,
) -> list[dict[str, str]]:
"""List networks defined in an OVF descriptor.
Args:
ovf_path: Path to OVF file on datastore
datastore: Datastore containing the OVF
host: Managed ESXi host to target (default: the default host)
Returns:
List of network definitions
"""
conn = self._conn(host)
# Read OVF descriptor
ovf_descriptor = self._read_datastore_file(datastore, ovf_path, conn)
ovf_descriptor = self._read_datastore_file(datastore, ovf_path)
# Parse network references
ovf_manager = conn.content.ovfManager
ovf_manager = self.conn.content.ovfManager
# Get resource pool for parsing
host = None
for h in conn.datacenter.hostFolder.childEntity:
for h in self.conn.datacenter.hostFolder.childEntity:
if hasattr(h, "host"):
host = h.host[0] if h.host else None
break
@ -890,7 +424,7 @@ class OVFManagementMixin(VSphereMixin):
raise ValueError("No ESXi host found")
resource_pool = host.parent.resourcePool if hasattr(host, "parent") else None
ds = conn.find_datastore(datastore)
ds = self.conn.find_datastore(datastore)
# Create parse params to extract network info
import_spec_params = vim.OvfManager.CreateImportSpecParams()

View File

@ -2,28 +2,28 @@
from typing import TYPE_CHECKING
from fastmcp.contrib.mcp_mixin import mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class PowerOpsMixin(VSphereMixin):
class PowerOpsMixin(MCPMixin):
"""VM power management tools."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
@mcp_tool(
name="power_on",
description="Power on a virtual machine",
annotations=ToolAnnotations(destructiveHint=False, idempotentHint=True),
)
def power_on(self, name: str, host: str | None = None) -> str:
def power_on(self, name: str) -> str:
"""Power on a virtual machine."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -31,7 +31,7 @@ class PowerOpsMixin(VSphereMixin):
return f"VM '{name}' is already powered on"
task = vm.PowerOnVM_Task()
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"VM '{name}' powered on"
@ -40,10 +40,9 @@ class PowerOpsMixin(VSphereMixin):
description="Power off a virtual machine (hard shutdown, like pulling the power cord)",
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=True),
)
def power_off(self, name: str, host: str | None = None) -> str:
def power_off(self, name: str) -> str:
"""Power off a virtual machine (hard shutdown)."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -51,7 +50,7 @@ class PowerOpsMixin(VSphereMixin):
return f"VM '{name}' is already powered off"
task = vm.PowerOffVM_Task()
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"VM '{name}' powered off"
@ -60,10 +59,9 @@ class PowerOpsMixin(VSphereMixin):
description="Gracefully shutdown the guest OS (requires VMware Tools installed and running)",
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=True),
)
def shutdown_guest(self, name: str, host: str | None = None) -> str:
def shutdown_guest(self, name: str) -> str:
"""Gracefully shutdown the guest OS."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -84,10 +82,9 @@ class PowerOpsMixin(VSphereMixin):
description="Gracefully reboot the guest OS (requires VMware Tools)",
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=False),
)
def reboot_guest(self, name: str, host: str | None = None) -> str:
def reboot_guest(self, name: str) -> str:
"""Gracefully reboot the guest OS."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -108,15 +105,14 @@ class PowerOpsMixin(VSphereMixin):
description="Reset (hard reboot) a virtual machine - like pressing the reset button",
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=False),
)
def reset_vm(self, name: str, host: str | None = None) -> str:
def reset_vm(self, name: str) -> str:
"""Reset (hard reboot) a virtual machine."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
task = vm.ResetVM_Task()
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"VM '{name}' reset"
@ -125,10 +121,9 @@ class PowerOpsMixin(VSphereMixin):
description="Suspend a virtual machine (save state to disk)",
annotations=ToolAnnotations(destructiveHint=False, idempotentHint=True),
)
def suspend_vm(self, name: str, host: str | None = None) -> str:
def suspend_vm(self, name: str) -> str:
"""Suspend a virtual machine."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -139,7 +134,7 @@ class PowerOpsMixin(VSphereMixin):
return f"VM '{name}' is powered off, cannot suspend"
task = vm.SuspendVM_Task()
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"VM '{name}' suspended"
@ -148,10 +143,9 @@ class PowerOpsMixin(VSphereMixin):
description="Put guest OS into standby mode (requires VMware Tools)",
annotations=ToolAnnotations(destructiveHint=False, idempotentHint=True),
)
def standby_guest(self, name: str, host: str | None = None) -> str:
def standby_guest(self, name: str) -> str:
"""Put guest OS into standby mode."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")

View File

@ -2,19 +2,20 @@
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import mcp_resource, mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_resource, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class ResourcesMixin(VSphereMixin):
class ResourcesMixin(MCPMixin):
"""MCP Resources for vSphere infrastructure."""
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
# ─────────────────────────────────────────────────────────────────────────────
# Datastore File Browser (templated resource)
# ─────────────────────────────────────────────────────────────────────────────
@ -26,13 +27,13 @@ class ResourcesMixin(VSphereMixin):
)
def browse_datastore_root(self, datastore_name: str) -> list[dict[str, Any]]:
"""Browse files and folders at the root of a datastore."""
return self._browse_datastore_path(self.conn, datastore_name, "")
return self._browse_datastore_path(datastore_name, "")
def _browse_datastore_path(
self, conn, datastore_name: str, path: str
self, datastore_name: str, path: str
) -> list[dict[str, Any]]:
"""Browse files and folders on a datastore at a given path."""
ds = conn.find_datastore(datastore_name)
ds = self.conn.find_datastore(datastore_name)
if not ds:
raise ValueError(f"Datastore '{datastore_name}' not found")
@ -56,7 +57,7 @@ class ResourcesMixin(VSphereMixin):
# Search for files
task = browser.SearchDatastore_Task(ds_path, search_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
results = []
if task.info.result and task.info.result.file:
@ -85,7 +86,7 @@ class ResourcesMixin(VSphereMixin):
return sorted(results, key=lambda x: (x["type"] != "Folder", x["name"]))
def _stream_from_esxi(self, conn, datastore: str, path: str, chunk_size: int = 1024 * 1024):
def _stream_from_esxi(self, datastore: str, path: str, chunk_size: int = 1024 * 1024):
"""Generator that streams file content from ESXi datastore.
Yields raw bytes chunks as they arrive from ESXi HTTP API.
@ -95,13 +96,13 @@ class ResourcesMixin(VSphereMixin):
import urllib.request
from urllib.parse import quote
ds = conn.find_datastore(datastore)
ds = self.conn.find_datastore(datastore)
if not ds:
raise ValueError(f"Datastore '{datastore}' not found")
# Build download URL
dc_name = conn.datacenter.name
host = conn.settings.vcenter_host
dc_name = self.conn.datacenter.name
host = self.conn.settings.vcenter_host
encoded_path = quote(path, safe="")
url = (
@ -111,12 +112,12 @@ class ResourcesMixin(VSphereMixin):
# Create SSL context
context = ssl.create_default_context()
if conn.settings.vcenter_insecure:
if self.conn.settings.vcenter_insecure:
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
# Get session cookie
stub = conn.si._stub
stub = self.conn.si._stub
cookie = stub.cookie
request = urllib.request.Request(url, method="GET")
@ -146,17 +147,15 @@ class ResourcesMixin(VSphereMixin):
annotations=ToolAnnotations(readOnlyHint=True),
)
def browse_datastore_tool(
self, datastore: str, path: str = "", host: str | None = None
self, datastore: str, path: str = ""
) -> list[dict[str, Any]]:
"""Browse files at a specific path on a datastore.
Args:
datastore: Datastore name (e.g., c1_ds-02)
path: Path within datastore (e.g., "rpm-desktop-1/" or "" for root)
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
return self._browse_datastore_path(conn, datastore, path)
return self._browse_datastore_path(datastore, path)
@mcp_tool(
name="download_from_datastore",
@ -169,7 +168,6 @@ class ResourcesMixin(VSphereMixin):
path: str,
save_to: str | None = None,
max_memory_mb: int = 50,
host: str | None = None,
) -> dict[str, Any]:
"""Download a file from a datastore using streaming.
@ -182,16 +180,14 @@ class ResourcesMixin(VSphereMixin):
path: Path to file on datastore (e.g., "iso/readme.txt")
save_to: Local path to save file (recommended for large files)
max_memory_mb: Max file size in MB to return in response (default 50MB)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with file content or save confirmation
"""
import base64
conn = self._conn(host)
max_bytes = max_memory_mb * 1024 * 1024
stream = self._stream_from_esxi(conn, datastore, path)
stream = self._stream_from_esxi(datastore, path)
# First yield is total size (or None)
total_size = next(stream)
@ -265,7 +261,6 @@ class ResourcesMixin(VSphereMixin):
local_path: str | None = None,
content_base64: str | None = None,
chunk_size: int = 8 * 1024 * 1024, # 8MB chunks
host: str | None = None,
) -> dict[str, Any]:
"""Upload a file to a datastore.
@ -278,7 +273,6 @@ class ResourcesMixin(VSphereMixin):
local_path: Local file path to upload - streams from disk (preferred for large files)
content_base64: Base64-encoded file content (for small files only)
chunk_size: Chunk size for streaming uploads (default 8MB)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with upload details including size and whether streaming was used
@ -288,37 +282,35 @@ class ResourcesMixin(VSphereMixin):
import ssl
import urllib.request
conn = self._conn(host)
if not local_path and not content_base64:
raise ValueError("Either local_path or content_base64 must be provided")
if local_path and content_base64:
raise ValueError("Only one of local_path or content_base64 can be provided")
ds = conn.find_datastore(datastore)
ds = self.conn.find_datastore(datastore)
if not ds:
raise ValueError(f"Datastore '{datastore}' not found")
# Build upload URL
dc_name = conn.datacenter.name
vc_host = conn.settings.vcenter_host
dc_name = self.conn.datacenter.name
host = self.conn.settings.vcenter_host
from urllib.parse import quote
encoded_path = quote(remote_path, safe="")
url = (
f"https://{vc_host}/folder/{encoded_path}"
f"https://{host}/folder/{encoded_path}"
f"?dcPath={quote(dc_name)}&dsName={quote(datastore)}"
)
# Create SSL context
context = ssl.create_default_context()
if conn.settings.vcenter_insecure:
if self.conn.settings.vcenter_insecure:
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
# Get session cookie from existing connection
stub = conn.si._stub
stub = self.conn.si._stub
cookie = stub.cookie
if local_path:
@ -407,21 +399,17 @@ class ResourcesMixin(VSphereMixin):
description="Delete a file or folder from a datastore",
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=True),
)
def delete_datastore_file(
self, datastore: str, path: str, host: str | None = None
) -> str:
def delete_datastore_file(self, datastore: str, path: str) -> str:
"""Delete a file or folder from a datastore.
Args:
datastore: Datastore name
path: Path to file or folder to delete (e.g., "iso/old-file.iso")
host: Managed ESXi host to target (default: the default host)
Returns:
Success message
"""
conn = self._conn(host)
ds = conn.find_datastore(datastore)
ds = self.conn.find_datastore(datastore)
if not ds:
raise ValueError(f"Datastore '{datastore}' not found")
@ -429,11 +417,11 @@ class ResourcesMixin(VSphereMixin):
ds_path = f"[{datastore}] {path}"
# Use FileManager to delete
file_manager = conn.content.fileManager
dc = conn.datacenter
file_manager = self.conn.content.fileManager
dc = self.conn.datacenter
task = file_manager.DeleteDatastoreFile_Task(name=ds_path, datacenter=dc)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"Deleted [{datastore}] {path}"
@ -442,21 +430,17 @@ class ResourcesMixin(VSphereMixin):
description="Create a folder on a datastore",
annotations=ToolAnnotations(destructiveHint=False, idempotentHint=True),
)
def create_datastore_folder(
self, datastore: str, path: str, host: str | None = None
) -> str:
def create_datastore_folder(self, datastore: str, path: str) -> str:
"""Create a folder on a datastore.
Args:
datastore: Datastore name
path: Folder path to create (e.g., "iso/new-folder")
host: Managed ESXi host to target (default: the default host)
Returns:
Success message
"""
conn = self._conn(host)
ds = conn.find_datastore(datastore)
ds = self.conn.find_datastore(datastore)
if not ds:
raise ValueError(f"Datastore '{datastore}' not found")
@ -464,8 +448,8 @@ class ResourcesMixin(VSphereMixin):
ds_path = f"[{datastore}] {path}"
# Use FileManager to create directory
file_manager = conn.content.fileManager
file_manager.MakeDirectory(name=ds_path, datacenter=conn.datacenter)
file_manager = self.conn.content.fileManager
file_manager.MakeDirectory(name=ds_path, datacenter=self.conn.datacenter)
return f"Created folder [{datastore}] {path}"
@ -490,7 +474,6 @@ class ResourcesMixin(VSphereMixin):
source_path: str,
dest_datastore: str | None = None,
dest_path: str | None = None,
host: str | None = None,
) -> dict[str, Any]:
"""Move or rename a file or folder on a datastore.
@ -499,13 +482,11 @@ class ResourcesMixin(VSphereMixin):
source_path: Source path (e.g., "iso/old-name.iso")
dest_datastore: Destination datastore (default: same as source)
dest_path: Destination path (default: same as source with new name)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with move operation details
"""
conn = self._conn(host)
ds = conn.find_datastore(source_datastore)
ds = self.conn.find_datastore(source_datastore)
if not ds:
raise ValueError(f"Datastore '{source_datastore}' not found")
@ -513,7 +494,7 @@ class ResourcesMixin(VSphereMixin):
if not dest_datastore:
dest_datastore = source_datastore
else:
dest_ds = conn.find_datastore(dest_datastore)
dest_ds = self.conn.find_datastore(dest_datastore)
if not dest_ds:
raise ValueError(f"Destination datastore '{dest_datastore}' not found")
@ -525,8 +506,8 @@ class ResourcesMixin(VSphereMixin):
dest_ds_path = f"[{dest_datastore}] {dest_path}"
# Use FileManager to move
file_manager = conn.content.fileManager
dc = conn.datacenter
file_manager = self.conn.content.fileManager
dc = self.conn.datacenter
task = file_manager.MoveDatastoreFile_Task(
sourceName=source_ds_path,
@ -535,7 +516,7 @@ class ResourcesMixin(VSphereMixin):
destinationDatacenter=dc,
force=False,
)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"action": "moved",
@ -555,7 +536,6 @@ class ResourcesMixin(VSphereMixin):
dest_datastore: str | None = None,
dest_path: str | None = None,
force: bool = False,
host: str | None = None,
) -> dict[str, Any]:
"""Copy a file or folder on a datastore.
@ -565,13 +545,11 @@ class ResourcesMixin(VSphereMixin):
dest_datastore: Destination datastore (default: same as source)
dest_path: Destination path (required)
force: Overwrite destination if exists (default False)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with copy operation details
"""
conn = self._conn(host)
ds = conn.find_datastore(source_datastore)
ds = self.conn.find_datastore(source_datastore)
if not ds:
raise ValueError(f"Datastore '{source_datastore}' not found")
@ -579,7 +557,7 @@ class ResourcesMixin(VSphereMixin):
if not dest_datastore:
dest_datastore = source_datastore
else:
dest_ds = conn.find_datastore(dest_datastore)
dest_ds = self.conn.find_datastore(dest_datastore)
if not dest_ds:
raise ValueError(f"Destination datastore '{dest_datastore}' not found")
@ -591,8 +569,8 @@ class ResourcesMixin(VSphereMixin):
dest_ds_path = f"[{dest_datastore}] {dest_path}"
# Use FileManager to copy
file_manager = conn.content.fileManager
dc = conn.datacenter
file_manager = self.conn.content.fileManager
dc = self.conn.datacenter
task = file_manager.CopyDatastoreFile_Task(
sourceName=source_ds_path,
@ -601,7 +579,7 @@ class ResourcesMixin(VSphereMixin):
destinationDatacenter=dc,
force=force,
)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"action": "copied",
@ -741,15 +719,9 @@ class ResourcesMixin(VSphereMixin):
description="Get detailed information about a specific datastore",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_datastore_info(self, name: str, host: str | None = None) -> dict[str, Any]:
"""Get detailed datastore information.
Args:
name: Datastore name
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
ds = conn.find_datastore(name)
def get_datastore_info(self, name: str) -> dict[str, Any]:
"""Get detailed datastore information."""
ds = self.conn.find_datastore(name)
if not ds:
raise ValueError(f"Datastore '{name}' not found")
@ -781,15 +753,9 @@ class ResourcesMixin(VSphereMixin):
description="Get detailed information about a specific network",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_network_info(self, name: str, host: str | None = None) -> dict[str, Any]:
"""Get detailed network information.
Args:
name: Network name
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
net = conn.find_network(name)
def get_network_info(self, name: str) -> dict[str, Any]:
"""Get detailed network information."""
net = self.conn.find_network(name)
if not net:
raise ValueError(f"Network '{name}' not found")
@ -818,21 +784,14 @@ class ResourcesMixin(VSphereMixin):
description="Get information about resource pools",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_resource_pool_info(
self, name: str | None = None, host: str | None = None
) -> dict[str, Any]:
def get_resource_pool_info(self, name: str | None = None) -> dict[str, Any]:
"""Get resource pool information.
If name is not provided, returns info for the default resource pool.
Args:
name: Resource pool name (default: the default resource pool)
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
if name:
container = conn.content.viewManager.CreateContainerView(
conn.content.rootFolder, [vim.ResourcePool], True
container = self.conn.content.viewManager.CreateContainerView(
self.conn.content.rootFolder, [vim.ResourcePool], True
)
try:
pool = next((p for p in container.view if p.name == name), None)
@ -841,7 +800,7 @@ class ResourcesMixin(VSphereMixin):
if not pool:
raise ValueError(f"Resource pool '{name}' not found")
else:
pool = conn.resource_pool
pool = self.conn.resource_pool
runtime = pool.summary.runtime
config = pool.summary.config
@ -864,15 +823,10 @@ class ResourcesMixin(VSphereMixin):
description="List all VM templates in the inventory",
annotations=ToolAnnotations(readOnlyHint=True),
)
def list_templates(self, host: str | None = None) -> list[dict[str, Any]]:
"""List all VM templates.
Args:
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
def list_templates(self) -> list[dict[str, Any]]:
"""List all VM templates."""
templates = []
for vm in conn.get_all_vms():
for vm in self.conn.get_all_vms():
if vm.config and vm.config.template:
templates.append(
{
@ -889,14 +843,9 @@ class ResourcesMixin(VSphereMixin):
description="Get vCenter/ESXi server information",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_vcenter_info(self, host: str | None = None) -> dict[str, Any]:
"""Get vCenter/ESXi server information.
Args:
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
about = conn.content.about
def get_vcenter_info(self) -> dict[str, Any]:
"""Get vCenter/ESXi server information."""
about = self.conn.content.about
return {
"name": about.name,
"full_name": about.fullName,

View File

@ -5,17 +5,15 @@ import socket
import time
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class SerialPortMixin(VSphereMixin):
class SerialPortMixin(MCPMixin):
"""Serial port management for VM network console access.
Network serial ports allow telnet/TCP connections to VM consoles,
@ -28,6 +26,9 @@ class SerialPortMixin(VSphereMixin):
- 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:
@ -66,18 +67,16 @@ class SerialPortMixin(VSphereMixin):
description="Get current serial port configuration for a VM",
annotations=ToolAnnotations(readOnlyHint=True),
)
def get_serial_port(self, name: str, host: str | None = None) -> dict[str, Any]:
def get_serial_port(self, name: str) -> dict[str, Any]:
"""Get serial port configuration.
Args:
name: VM name
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with serial port details or message if not configured
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -111,7 +110,6 @@ class SerialPortMixin(VSphereMixin):
port: int | None = None,
direction: str = "server",
yield_on_poll: bool = True,
host: str | None = None,
) -> dict[str, Any]:
"""Setup or update network serial port.
@ -121,13 +119,11 @@ class SerialPortMixin(VSphereMixin):
port: TCP port number. If not specified, auto-assigns unused port.
direction: 'server' (VM listens) or 'client' (VM connects). Default: server
yield_on_poll: Enable CPU yield behavior. Default: True
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with configured serial port URI and details
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -146,8 +142,8 @@ class SerialPortMixin(VSphereMixin):
# Find or assign port
if port is None:
vm_host = vm.runtime.host
host_ip = vm_host.name if vm_host else conn.settings.vcenter_host
host = vm.runtime.host
host_ip = host.name if host else self.conn.settings.vcenter_host
port = self._find_unused_port(host_ip)
# Build service URI
@ -182,11 +178,11 @@ class SerialPortMixin(VSphereMixin):
spec = vim.vm.ConfigSpec()
spec.deviceChange = [serial_spec]
task = vm.ReconfigVM_Task(spec=spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
# Get ESXi host info for connection string
vm_host = vm.runtime.host
host_ip = vm_host.name if vm_host else conn.settings.vcenter_host
host = vm.runtime.host
host_ip = host.name if host else self.conn.settings.vcenter_host
return {
"vm_name": name,
@ -204,19 +200,17 @@ class SerialPortMixin(VSphereMixin):
description="Connect or disconnect an existing serial port on a VM",
annotations=ToolAnnotations(destructiveHint=False, idempotentHint=True),
)
def connect_serial_port(self, name: str, connected: bool = True, host: str | None = None) -> dict[str, Any]:
def connect_serial_port(self, name: str, connected: bool = True) -> dict[str, Any]:
"""Connect or disconnect serial port.
Args:
name: VM name
connected: True to connect, False to disconnect. Default: True
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with result
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -233,7 +227,7 @@ class SerialPortMixin(VSphereMixin):
spec = vim.vm.ConfigSpec()
spec.deviceChange = [serial_spec]
task = vm.ReconfigVM_Task(spec=spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm_name": name,
@ -246,20 +240,18 @@ class SerialPortMixin(VSphereMixin):
description="Reset serial port by disconnecting and reconnecting (clears stuck connections)",
annotations=ToolAnnotations(destructiveHint=False, idempotentHint=True),
)
def clear_serial_port(self, name: str, host: str | None = None) -> dict[str, Any]:
def clear_serial_port(self, name: str) -> dict[str, Any]:
"""Clear serial port by cycling connection state.
Useful for clearing stuck or stale connections.
Args:
name: VM name
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with result
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -268,11 +260,11 @@ class SerialPortMixin(VSphereMixin):
raise ValueError(f"No network serial port configured on VM '{name}'")
# Disconnect
self.connect_serial_port(name, connected=False, host=host)
self.connect_serial_port(name, connected=False)
time.sleep(1)
# Reconnect
self.connect_serial_port(name, connected=True, host=host)
self.connect_serial_port(name, connected=True)
return {
"vm_name": name,
@ -286,18 +278,16 @@ class SerialPortMixin(VSphereMixin):
description="Remove the network serial port from a VM. VM must be powered off.",
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=True),
)
def remove_serial_port(self, name: str, host: str | None = None) -> str:
def remove_serial_port(self, name: str) -> str:
"""Remove serial port from VM.
Args:
name: VM name
host: Managed ESXi host to target (default: the default host)
Returns:
Success message
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -317,6 +307,6 @@ class SerialPortMixin(VSphereMixin):
spec = vim.vm.ConfigSpec()
spec.deviceChange = [serial_spec]
task = vm.ReconfigVM_Task(spec=spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"Serial port removed from VM '{name}'"

View File

@ -2,19 +2,20 @@
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class SnapshotsMixin(VSphereMixin):
class SnapshotsMixin(MCPMixin):
"""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]]:
@ -55,15 +56,9 @@ class SnapshotsMixin(VSphereMixin):
description="List all snapshots for a virtual machine",
annotations=ToolAnnotations(readOnlyHint=True),
)
def list_snapshots(self, name: str, host: str | None = None) -> list[dict[str, Any]]:
"""List all snapshots for a VM.
Args:
name: VM name
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
vm = conn.find_vm(name)
def list_snapshots(self, name: str) -> list[dict[str, Any]]:
"""List all snapshots for a VM."""
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -101,7 +96,6 @@ class SnapshotsMixin(VSphereMixin):
description: str = "",
memory: bool = True,
quiesce: bool = False,
host: str | None = None,
) -> str:
"""Create a VM snapshot.
@ -111,10 +105,8 @@ class SnapshotsMixin(VSphereMixin):
description: Optional description
memory: Include memory state (allows instant restore to running state)
quiesce: Quiesce guest filesystem (requires VMware Tools, ensures consistent state)
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -135,7 +127,7 @@ class SnapshotsMixin(VSphereMixin):
memory=memory,
quiesce=quiesce,
)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"Snapshot '{snapshot_name}' created for VM '{name}'"
@ -144,18 +136,9 @@ class SnapshotsMixin(VSphereMixin):
description="Revert a VM to a specific snapshot",
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=False),
)
def revert_to_snapshot(
self, name: str, snapshot_name: str, host: str | None = None
) -> str:
"""Revert VM to a specific snapshot.
Args:
name: VM name
snapshot_name: Name of snapshot to revert to
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
vm = conn.find_vm(name)
def revert_to_snapshot(self, name: str, snapshot_name: str) -> str:
"""Revert VM to a specific snapshot."""
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -169,7 +152,7 @@ class SnapshotsMixin(VSphereMixin):
raise ValueError(f"Snapshot '{snapshot_name}' not found on VM '{name}'")
task = snapshot.RevertToSnapshot_Task()
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"VM '{name}' reverted to snapshot '{snapshot_name}'"
@ -178,15 +161,9 @@ class SnapshotsMixin(VSphereMixin):
description="Revert a VM to its current (most recent) snapshot",
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=False),
)
def revert_to_current_snapshot(self, name: str, host: str | None = None) -> str:
"""Revert VM to its current snapshot.
Args:
name: VM name
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
vm = conn.find_vm(name)
def revert_to_current_snapshot(self, name: str) -> str:
"""Revert VM to its current snapshot."""
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -194,7 +171,7 @@ class SnapshotsMixin(VSphereMixin):
raise ValueError(f"VM '{name}' has no current snapshot")
task = vm.RevertToCurrentSnapshot_Task()
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"VM '{name}' reverted to current snapshot"
@ -204,11 +181,7 @@ class SnapshotsMixin(VSphereMixin):
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=True),
)
def delete_snapshot(
self,
name: str,
snapshot_name: str,
remove_children: bool = False,
host: str | None = None,
self, name: str, snapshot_name: str, remove_children: bool = False
) -> str:
"""Delete a VM snapshot.
@ -216,10 +189,8 @@ class SnapshotsMixin(VSphereMixin):
name: VM name
snapshot_name: Name of snapshot to delete
remove_children: If True, also delete child snapshots
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -233,7 +204,7 @@ class SnapshotsMixin(VSphereMixin):
raise ValueError(f"Snapshot '{snapshot_name}' not found on VM '{name}'")
task = snapshot.RemoveSnapshot_Task(removeChildren=remove_children)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
msg = f"Snapshot '{snapshot_name}' deleted from VM '{name}'"
if remove_children:
@ -245,15 +216,9 @@ class SnapshotsMixin(VSphereMixin):
description="Delete ALL snapshots from a VM (consolidates disk)",
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=True),
)
def delete_all_snapshots(self, name: str, host: str | None = None) -> str:
"""Delete all snapshots from a VM.
Args:
name: VM name
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
vm = conn.find_vm(name)
def delete_all_snapshots(self, name: str) -> str:
"""Delete all snapshots from a VM."""
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -261,7 +226,7 @@ class SnapshotsMixin(VSphereMixin):
return f"VM '{name}' has no snapshots to delete"
task = vm.RemoveAllSnapshots_Task()
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"All snapshots deleted from VM '{name}'"
@ -276,19 +241,9 @@ class SnapshotsMixin(VSphereMixin):
snapshot_name: str,
new_name: str | None = None,
new_description: str | None = None,
host: str | None = None,
) -> str:
"""Rename a snapshot or update its description.
Args:
name: VM name
snapshot_name: Name of snapshot to rename
new_name: New name for the snapshot
new_description: New description for the snapshot
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
vm = conn.find_vm(name)
"""Rename a snapshot or update its description."""
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")

View File

@ -3,31 +3,19 @@
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class VCenterOpsMixin(VSphereMixin):
class VCenterOpsMixin(MCPMixin):
"""vCenter-specific operations (require vCenter, not just ESXi)."""
@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()
def __init__(self, conn: "VMwareConnection"):
self.conn = conn
# ─────────────────────────────────────────────────────────────────────────────
# Storage vMotion (works even on single-host vCenter)
@ -43,7 +31,6 @@ class VCenterOpsMixin(VSphereMixin):
vm_name: str,
target_datastore: str,
thin_provision: bool | None = None,
host: str | None = None,
) -> dict[str, Any]:
"""Move a VM's storage to a different datastore.
@ -54,17 +41,15 @@ class VCenterOpsMixin(VSphereMixin):
vm_name: Name of the VM to migrate
target_datastore: Target datastore name
thin_provision: Convert to thin provisioning (None = keep current)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with migration details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
ds = conn.find_datastore(target_datastore)
ds = self.conn.find_datastore(target_datastore)
if not ds:
raise ValueError(f"Datastore '{target_datastore}' not found")
@ -91,7 +76,7 @@ class VCenterOpsMixin(VSphereMixin):
# Perform the relocation
task = vm.RelocateVM_Task(spec=relocate_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm": vm_name,
@ -111,7 +96,6 @@ class VCenterOpsMixin(VSphereMixin):
vm_name: str,
disk_label: str,
target_datastore: str,
host: str | None = None,
) -> dict[str, Any]:
"""Move a specific VM disk to a different datastore.
@ -119,17 +103,15 @@ class VCenterOpsMixin(VSphereMixin):
vm_name: Name of the VM
disk_label: Label of the disk (e.g., 'Hard disk 1')
target_datastore: Target datastore name
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with migration details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
ds = conn.find_datastore(target_datastore)
ds = self.conn.find_datastore(target_datastore)
if not ds:
raise ValueError(f"Datastore '{target_datastore}' not found")
@ -163,7 +145,7 @@ class VCenterOpsMixin(VSphereMixin):
# Perform the relocation
task = vm.RelocateVM_Task(spec=relocate_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm": vm_name,
@ -182,7 +164,7 @@ class VCenterOpsMixin(VSphereMixin):
description="Convert a VM to a template (idempotent - safe to call on existing template)",
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=True),
)
def convert_to_template(self, vm_name: str, host: str | None = None) -> dict[str, Any]:
def convert_to_template(self, vm_name: str) -> dict[str, Any]:
"""Convert a VM to a template.
The VM must be powered off. Once converted, it cannot be powered on
@ -190,13 +172,11 @@ class VCenterOpsMixin(VSphereMixin):
Args:
vm_name: Name of the VM to convert
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with conversion details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
@ -227,20 +207,17 @@ class VCenterOpsMixin(VSphereMixin):
self,
template_name: str,
resource_pool: str | None = None,
host: str | None = None,
) -> dict[str, Any]:
"""Convert a template back to a regular VM.
Args:
template_name: Name of the template
resource_pool: Resource pool for the VM (optional)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with conversion details
"""
conn = self._conn(host)
vm = conn.find_vm(template_name)
vm = self.conn.find_vm(template_name)
if not vm:
raise ValueError(f"Template '{template_name}' not found")
@ -253,20 +230,20 @@ class VCenterOpsMixin(VSphereMixin):
# Get resource pool
if resource_pool:
pool = self._find_resource_pool(resource_pool, conn)
pool = self._find_resource_pool(resource_pool)
if not pool:
raise ValueError(f"Resource pool '{resource_pool}' not found")
else:
pool = conn.resource_pool
pool = self.conn.resource_pool
# Get a host from the resource pool
target_host = None
host = None
if hasattr(pool, "owner") and hasattr(pool.owner, "host"):
hosts = pool.owner.host
if hosts:
target_host = hosts[0]
host = hosts[0]
vm.MarkAsVirtualMachine(pool=pool, host=target_host)
vm.MarkAsVirtualMachine(pool=pool, host=host)
return {
"vm": template_name,
@ -274,10 +251,10 @@ class VCenterOpsMixin(VSphereMixin):
"is_template": False,
}
def _find_resource_pool(self, name: str, conn) -> vim.ResourcePool | None:
def _find_resource_pool(self, name: str) -> vim.ResourcePool | None:
"""Find a resource pool by name."""
container = conn.content.viewManager.CreateContainerView(
conn.content.rootFolder, [vim.ResourcePool], True
container = self.conn.content.viewManager.CreateContainerView(
self.conn.content.rootFolder, [vim.ResourcePool], True
)
try:
for pool in container.view:
@ -298,7 +275,6 @@ class VCenterOpsMixin(VSphereMixin):
new_vm_name: str,
datastore: str | None = None,
power_on: bool = False,
host: str | None = None,
) -> dict[str, Any]:
"""Deploy a new VM from a template.
@ -307,13 +283,11 @@ class VCenterOpsMixin(VSphereMixin):
new_vm_name: Name for the new VM
datastore: Target datastore (default: same as template)
power_on: Power on after deployment (default False)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with deployment details
"""
conn = self._conn(host)
template = conn.find_vm(template_name)
template = self.conn.find_vm(template_name)
if not template:
raise ValueError(f"Template '{template_name}' not found")
@ -321,15 +295,15 @@ class VCenterOpsMixin(VSphereMixin):
raise ValueError(f"'{template_name}' is not a template")
# Check if target VM already exists
if conn.find_vm(new_vm_name):
if self.conn.find_vm(new_vm_name):
raise ValueError(f"VM '{new_vm_name}' already exists")
# Build clone spec
relocate_spec = vim.vm.RelocateSpec()
relocate_spec.pool = conn.resource_pool
relocate_spec.pool = self.conn.resource_pool
if datastore:
ds = conn.find_datastore(datastore)
ds = self.conn.find_datastore(datastore)
if not ds:
raise ValueError(f"Datastore '{datastore}' not found")
relocate_spec.datastore = ds
@ -340,14 +314,14 @@ class VCenterOpsMixin(VSphereMixin):
clone_spec.template = False # Create VM, not another template
# Get target folder
folder = conn.datacenter.vmFolder
folder = self.conn.datacenter.vmFolder
# Clone the template
task = template.Clone(folder=folder, name=new_vm_name, spec=clone_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
# Get the new VM info
new_vm = conn.find_vm(new_vm_name)
new_vm = self.conn.find_vm(new_vm_name)
return {
"vm": new_vm_name,
@ -366,16 +340,12 @@ class VCenterOpsMixin(VSphereMixin):
description="List VM folders in the datacenter",
annotations=ToolAnnotations(readOnlyHint=True),
)
def list_folders(self, host: str | None = None) -> list[dict[str, Any]]:
def list_folders(self) -> list[dict[str, Any]]:
"""List all VM folders in the datacenter.
Args:
host: Managed ESXi host to target (default: the default host)
Returns:
List of folder details
"""
conn = self._conn(host)
folders = []
def _collect_folders(folder: vim.Folder, path: str = ""):
@ -393,7 +363,7 @@ class VCenterOpsMixin(VSphereMixin):
_collect_folders(child, current_path)
# Start from VM folder
vm_folder = conn.datacenter.vmFolder
vm_folder = self.conn.datacenter.vmFolder
_collect_folders(vm_folder)
return folders
@ -407,25 +377,22 @@ class VCenterOpsMixin(VSphereMixin):
self,
folder_name: str,
parent_path: str | None = None,
host: str | None = None,
) -> dict[str, Any]:
"""Create a new VM folder.
Args:
folder_name: Name for the new folder
parent_path: Path to parent folder (None = root vm folder)
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with folder details
"""
conn = self._conn(host)
if parent_path:
parent = self._find_folder_by_path(parent_path, conn)
parent = self._find_folder_by_path(parent_path)
if not parent:
raise ValueError(f"Parent folder '{parent_path}' not found")
else:
parent = conn.datacenter.vmFolder
parent = self.conn.datacenter.vmFolder
parent.CreateFolder(name=folder_name)
@ -436,11 +403,11 @@ class VCenterOpsMixin(VSphereMixin):
"path": f"{parent_path}/{folder_name}" if parent_path else f"vm/{folder_name}",
}
def _find_folder_by_path(self, path: str, conn) -> vim.Folder | None:
def _find_folder_by_path(self, path: str) -> vim.Folder | None:
"""Find a folder by its path (e.g., 'vm/Production/WebServers')."""
parts = [p for p in path.split("/") if p and p != "vm"]
current = conn.datacenter.vmFolder
current = self.conn.datacenter.vmFolder
for part in parts:
found = None
if hasattr(current, "childEntity"):
@ -463,24 +430,21 @@ class VCenterOpsMixin(VSphereMixin):
self,
vm_name: str,
folder_path: str,
host: str | None = None,
) -> dict[str, Any]:
"""Move a VM to a different folder.
Args:
vm_name: Name of the VM to move
folder_path: Path to target folder
host: Managed ESXi host to target (default: the default host)
Returns:
Dict with move details
"""
conn = self._conn(host)
vm = conn.find_vm(vm_name)
vm = self.conn.find_vm(vm_name)
if not vm:
raise ValueError(f"VM '{vm_name}' not found")
folder = self._find_folder_by_path(folder_path, conn)
folder = self._find_folder_by_path(folder_path)
if not folder:
raise ValueError(f"Folder '{folder_path}' not found")
@ -489,7 +453,7 @@ class VCenterOpsMixin(VSphereMixin):
# Move the VM
task = folder.MoveIntoFolder_Task([vm])
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return {
"vm": vm_name,
@ -511,20 +475,17 @@ class VCenterOpsMixin(VSphereMixin):
self,
max_count: int = 20,
entity_name: str | None = None,
host: str | None = None,
) -> list[dict[str, Any]]:
"""List recent tasks from vCenter.
Args:
max_count: Maximum number of tasks to return (default 20)
entity_name: Filter by entity name (optional)
host: Managed ESXi host to target (default: the default host)
Returns:
List of task details
"""
conn = self._conn(host)
task_manager = conn.content.taskManager
task_manager = self.conn.content.taskManager
recent_tasks = task_manager.recentTask
tasks = []
@ -570,7 +531,6 @@ class VCenterOpsMixin(VSphereMixin):
max_count: int = 50,
event_types: list[str] | None = None,
hours_back: int = 24,
host: str | None = None,
) -> list[dict[str, Any]]:
"""List recent events from vCenter.
@ -578,13 +538,11 @@ class VCenterOpsMixin(VSphereMixin):
max_count: Maximum number of events (default 50)
event_types: Filter by event type names (optional)
hours_back: How many hours back to look (default 24)
host: Managed ESXi host to target (default: the default host)
Returns:
List of event details
"""
conn = self._conn(host)
event_manager = conn.content.eventManager
event_manager = self.conn.content.eventManager
# Create filter spec
filter_spec = vim.event.EventFilterSpec()
@ -635,19 +593,15 @@ class VCenterOpsMixin(VSphereMixin):
description="List all clusters in the datacenter",
annotations=ToolAnnotations(readOnlyHint=True),
)
def list_clusters(self, host: str | None = None) -> list[dict[str, Any]]:
def list_clusters(self) -> list[dict[str, Any]]:
"""List all clusters in the datacenter.
Args:
host: Managed ESXi host to target (default: the default host)
Returns:
List of cluster details with DRS/HA status
"""
conn = self._conn(host)
clusters = []
for entity in conn.datacenter.hostFolder.childEntity:
for entity in self.conn.datacenter.hostFolder.childEntity:
if isinstance(entity, vim.ClusterComputeResource):
drs_config = entity.configuration.drsConfig
ha_config = entity.configuration.dasConfig
@ -686,19 +640,16 @@ class VCenterOpsMixin(VSphereMixin):
def get_drs_recommendations(
self,
cluster_name: str,
host: str | None = None,
) -> list[dict[str, Any]]:
"""Get DRS recommendations for a cluster.
Args:
cluster_name: Name of the cluster
host: Managed ESXi host to target (default: the default host)
Returns:
List of DRS recommendations
"""
conn = self._conn(host)
cluster = self._find_cluster(cluster_name, conn)
cluster = self._find_cluster(cluster_name)
if not cluster:
raise ValueError(f"Cluster '{cluster_name}' not found")
@ -738,9 +689,9 @@ class VCenterOpsMixin(VSphereMixin):
return recommendations
def _find_cluster(self, name: str, conn) -> vim.ClusterComputeResource | None:
def _find_cluster(self, name: str) -> vim.ClusterComputeResource | None:
"""Find a cluster by name."""
for entity in conn.datacenter.hostFolder.childEntity:
for entity in self.conn.datacenter.hostFolder.childEntity:
if isinstance(entity, vim.ClusterComputeResource) and entity.name == name:
return entity
return None

View File

@ -2,34 +2,29 @@
from typing import TYPE_CHECKING, Any
from fastmcp.contrib.mcp_mixin import mcp_tool
from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool
from mcp.types import ToolAnnotations
from pyVmomi import vim
from mcvsphere.mixins._base import VSphereMixin
if TYPE_CHECKING:
pass
from mcvsphere.connection import VMwareConnection
class VMLifecycleMixin(VSphereMixin):
class VMLifecycleMixin(MCPMixin):
"""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, 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)
def list_vms(self) -> list[dict[str, Any]]:
"""List all virtual machines with basic info."""
vms = []
for vm in conn.get_all_vms():
for vm in self.conn.get_all_vms():
vms.append(
{
"name": vm.name,
@ -46,15 +41,9 @@ class VMLifecycleMixin(VSphereMixin):
description="Get detailed information about a specific virtual machine",
annotations=ToolAnnotations(readOnlyHint=True),
)
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)
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)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -110,7 +99,7 @@ class VMLifecycleMixin(VSphereMixin):
@mcp_tool(
name="create_vm",
description="Create a new virtual machine with specified resources and a CD/DVD drive (optionally booting from an ISO)",
description="Create a new virtual machine with specified resources",
annotations=ToolAnnotations(destructiveHint=False, idempotentHint=False),
)
def create_vm(
@ -122,42 +111,19 @@ class VMLifecycleMixin(VSphereMixin):
datastore: str | None = None,
network: str | None = None,
guest_id: str = "otherGuest64",
iso_path: str | None = None,
iso_datastore: str | None = None,
boot_from_iso: bool = True,
host: str | None = None,
) -> str:
"""Create a new virtual machine with specified configuration.
The VM always gets a CD/DVD drive. If ``iso_path`` is given, the drive
is backed by that ISO and connected at power-on; otherwise it is an
empty client device (so ``attach_iso`` can mount media later).
Args:
name: VM name
cpu: vCPU count
memory_mb: Memory in MB
disk_gb: Primary disk size in GB
datastore: Datastore for the VM (default: largest available)
network: Port group for the NIC (default: configured network)
guest_id: vSphere guest OS identifier
iso_path: ISO path on a datastore (e.g. 'iso/ubuntu.iso') to mount in the CD/DVD drive
iso_datastore: Datastore holding the ISO (default: the VM's datastore)
boot_from_iso: When an ISO is given, put the CD/DVD first in the boot order
host: Managed ESXi host to target (default: the default host)
"""
conn = self._conn(host)
"""Create a new virtual machine with specified configuration."""
# Resolve datastore
datastore_obj = conn.datastore
datastore_obj = self.conn.datastore
if datastore:
datastore_obj = conn.find_datastore(datastore)
datastore_obj = self.conn.find_datastore(datastore)
if not datastore_obj:
raise ValueError(f"Datastore '{datastore}' not found")
# Resolve network
network_obj = conn.network
network_obj = self.conn.network
if network:
network_obj = conn.find_network(network)
network_obj = self.conn.find_network(network)
if not network_obj:
raise ValueError(f"Network '{network}' not found")
@ -229,47 +195,13 @@ class VMLifecycleMixin(VSphereMixin):
)
device_specs.append(nic_spec)
# Add a CD/DVD drive on the default IDE controller (vSphere auto-creates
# IDE controller key 200). Backed by an ISO if one was requested,
# otherwise an empty client device so media can be mounted later.
cdrom_spec = vim.vm.device.VirtualDeviceSpec()
cdrom_spec.operation = vim.vm.device.VirtualDeviceSpec.Operation.add
cdrom_spec.device = vim.vm.device.VirtualCdrom()
cdrom_spec.device.controllerKey = 200
cdrom_spec.device.unitNumber = 0
cdrom_spec.device.key = -2
connectable = vim.vm.device.VirtualDevice.ConnectInfo()
connectable.allowGuestControl = True
connectable.connected = False
if iso_path:
iso_ds = iso_datastore or datastore_obj.name
backing = vim.vm.device.VirtualCdrom.IsoBackingInfo()
backing.fileName = f"[{iso_ds}] {iso_path}"
connectable.startConnected = True
else:
backing = vim.vm.device.VirtualCdrom.RemotePassthroughBackingInfo()
backing.deviceName = ""
backing.exclusive = False
connectable.startConnected = False
cdrom_spec.device.backing = backing
cdrom_spec.device.connectable = connectable
device_specs.append(cdrom_spec)
# Boot from the CD/DVD first when installing from an ISO
if iso_path and boot_from_iso:
vm_spec.bootOptions = vim.vm.BootOptions(
bootOrder=[vim.vm.BootOptions.BootableCdromDevice()]
)
vm_spec.deviceChange = device_specs
# Create VM
task = conn.datacenter.vmFolder.CreateVM_Task(
config=vm_spec, pool=conn.resource_pool
task = self.conn.datacenter.vmFolder.CreateVM_Task(
config=vm_spec, pool=self.conn.resource_pool
)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"VM '{name}' created successfully"
@ -284,33 +216,31 @@ class VMLifecycleMixin(VSphereMixin):
new_name: str,
power_on: bool = False,
datastore: str | None = None,
host: str | None = None,
) -> str:
"""Clone a VM from a template or existing VM."""
conn = self._conn(host)
template_vm = conn.find_vm(template_name)
template_vm = self.conn.find_vm(template_name)
if not template_vm:
raise ValueError(f"Template VM '{template_name}' not found")
vm_folder = template_vm.parent
if not isinstance(vm_folder, vim.Folder):
vm_folder = conn.datacenter.vmFolder
vm_folder = self.conn.datacenter.vmFolder
# Resolve datastore
datastore_obj = conn.datastore
datastore_obj = self.conn.datastore
if datastore:
datastore_obj = conn.find_datastore(datastore)
datastore_obj = self.conn.find_datastore(datastore)
if not datastore_obj:
raise ValueError(f"Datastore '{datastore}' not found")
resource_pool = template_vm.resourcePool or conn.resource_pool
resource_pool = template_vm.resourcePool or self.conn.resource_pool
relocate_spec = vim.vm.RelocateSpec(pool=resource_pool, datastore=datastore_obj)
clone_spec = vim.vm.CloneSpec(
powerOn=power_on, template=False, location=relocate_spec
)
task = template_vm.Clone(folder=vm_folder, name=new_name, spec=clone_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"VM '{new_name}' cloned from '{template_name}'"
@ -319,20 +249,19 @@ class VMLifecycleMixin(VSphereMixin):
description="Delete a virtual machine permanently (powers off if running)",
annotations=ToolAnnotations(destructiveHint=True, idempotentHint=True),
)
def delete_vm(self, name: str, host: str | None = None) -> str:
def delete_vm(self, name: str) -> str:
"""Delete a virtual machine permanently."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
# Power off if running
if vm.runtime.powerState == vim.VirtualMachine.PowerState.poweredOn:
task = vm.PowerOffVM_Task()
conn.wait_for_task(task)
self.conn.wait_for_task(task)
task = vm.Destroy_Task()
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"VM '{name}' deleted"
@ -347,11 +276,9 @@ class VMLifecycleMixin(VSphereMixin):
cpu: int | None = None,
memory_mb: int | None = None,
annotation: str | None = None,
host: str | None = None,
) -> str:
"""Reconfigure VM hardware settings."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
@ -374,7 +301,7 @@ class VMLifecycleMixin(VSphereMixin):
return f"No changes specified for VM '{name}'"
task = vm.ReconfigVM_Task(spec=config_spec)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"VM '{name}' reconfigured: {', '.join(changes)}"
@ -383,14 +310,13 @@ class VMLifecycleMixin(VSphereMixin):
description="Rename a virtual machine",
annotations=ToolAnnotations(destructiveHint=False, idempotentHint=True),
)
def rename_vm(self, name: str, new_name: str, host: str | None = None) -> str:
def rename_vm(self, name: str, new_name: str) -> str:
"""Rename a virtual machine."""
conn = self._conn(host)
vm = conn.find_vm(name)
vm = self.conn.find_vm(name)
if not vm:
raise ValueError(f"VM '{name}' not found")
task = vm.Rename_Task(newName=new_name)
conn.wait_for_task(task)
self.conn.wait_for_task(task)
return f"VM renamed from '{name}' to '{new_name}'"

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_manager import ConnectionManager
from mcvsphere.connection import VMwareConnection
from mcvsphere.middleware import RBACMiddleware
from mcvsphere.mixins import (
ConsoleMixin,
@ -25,7 +25,6 @@ from mcvsphere.mixins import (
VCenterOpsMixin,
VMLifecycleMixin,
)
from mcvsphere.servers import ServerConfig, load_servers
logger = logging.getLogger(__name__)
@ -75,43 +74,25 @@ def create_server(settings: Settings | None = None) -> FastMCP:
mcp.add_middleware(RBACMiddleware())
logger.info("RBAC middleware enabled - permissions enforced via OAuth groups")
# 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 shared VMware connection
logger.info("Connecting to VMware vCenter/ESXi...")
conn = VMwareConnection(settings)
# Create and register all mixins
mixins = [
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),
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),
]
tool_count = 0

View File

@ -1,112 +0,0 @@
"""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