#!/usr/bin/env python3 """Simple MCP client to test the ESXi MCP server.""" import asyncio import json import os from pathlib import Path from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client def load_env_file(path: str = ".env") -> dict[str, str]: """Load environment variables from a .env file.""" env = {} env_path = Path(path) if env_path.exists(): with open(env_path) as f: for line in f: line = line.strip() if line and not line.startswith("#") and "=" in line: key, _, value = line.partition("=") env[key.strip()] = value.strip() return env async def main(): """Test the ESXi MCP server.""" print("🔌 Connecting to ESXi MCP server...") # Load from .env file dotenv = load_env_file() server_params = StdioServerParameters( command="uv", args=["run", "esxi-mcp-server"], env={ **os.environ, "VCENTER_HOST": dotenv.get("VCENTER_HOST", os.environ.get("VCENTER_HOST", "")), "VCENTER_USER": dotenv.get("VCENTER_USER", os.environ.get("VCENTER_USER", "")), "VCENTER_PASSWORD": dotenv.get("VCENTER_PASSWORD", os.environ.get("VCENTER_PASSWORD", "")), "VCENTER_INSECURE": dotenv.get("VCENTER_INSECURE", os.environ.get("VCENTER_INSECURE", "true")), "MCP_TRANSPORT": "stdio", # Force stdio transport } ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # Initialize await session.initialize() print("✓ Connected!\n") # List available tools print("=== Available Tools ===") tools_result = await session.list_tools() print(f"Total: {len(tools_result.tools)} tools") # Group by category (based on tool name patterns) categories = { "VM Lifecycle": [], "Power Ops": [], "Snapshots": [], "Guest Ops": [], "Monitoring": [], "Datastore": [], "Disk Mgmt": [], "NIC Mgmt": [], "Host Mgmt": [], "OVF/OVA": [], "Other": [], } for tool in tools_result.tools: name = tool.name if name in ["create_vm", "clone_vm", "delete_vm", "reconfigure_vm", "get_vm_info"]: categories["VM Lifecycle"].append(name) elif "power" in name or name in ["reset_vm", "suspend_vm"]: categories["Power Ops"].append(name) elif "snapshot" in name: categories["Snapshots"].append(name) elif "guest" in name or "execute" in name: categories["Guest Ops"].append(name) elif "metrics" in name or "events" in name or "alarms" in name: categories["Monitoring"].append(name) elif "datastore" in name or "browse" in name or "upload" in name or "download" in name: categories["Datastore"].append(name) elif "disk" in name or "iso" in name: categories["Disk Mgmt"].append(name) elif "nic" in name or "mac" in name: categories["NIC Mgmt"].append(name) elif "host" in name or "maintenance" in name or "service" in name or "ntp" in name: categories["Host Mgmt"].append(name) elif "ovf" in name: categories["OVF/OVA"].append(name) else: categories["Other"].append(name) for cat, tools in categories.items(): if tools: print(f" {cat}: {len(tools)} tools") print() # List available resources print("=== Available Resources ===") resources_result = await session.list_resources() for r in resources_result.resources: print(f" • {r.uri} ({r.name})") print() # Test 1: Browse datastore print("=== Test 1: Browse Datastore ===") result = await session.call_tool("browse_datastore", { "datastore": "c1_ds-02", "path": "iso" }) files = json.loads(result.content[0].text) print(f"Found {len(files)} files in iso/") for f in files[:5]: print(f" {f['name']}: {f['size_human']}") print() # Test 2: Read VMs resource print("=== Test 2: Read Resource (esxi://vms) ===") try: result = await session.read_resource("esxi://vms") vms = json.loads(result.contents[0].text) print(f"Found {len(vms)} VMs:") for vm in vms[:5]: print(f" • {vm['name']}: {vm['power_state']}") except Exception as e: print(f"Resource read error: {e}") print() # Test 3: Get Host Info (NEW!) print("=== Test 3: Get Host Info (NEW!) ===") try: result = await session.call_tool("get_host_info", {}) host_info = json.loads(result.content[0].text) print(f" Host: {host_info.get('name', 'N/A')}") prod = host_info.get('product', {}) print(f" ESXi: {prod.get('version', 'N/A')} (build {prod.get('build', 'N/A')})") hw = host_info.get('hardware', {}) print(f" CPU: {hw.get('cpu_cores', 'N/A')} cores @ {hw.get('cpu_mhz', 'N/A')} MHz") print(f" RAM: {hw.get('memory_gb', 'N/A')} GB") status = host_info.get('status', {}) print(f" Maintenance: {status.get('maintenance_mode', 'N/A')}") except Exception as e: print(f"Error: {e}") print() # Test 4: List Services (NEW!) print("=== Test 4: List Services (NEW!) ===") try: result = await session.call_tool("list_services", {}) services = json.loads(result.content[0].text) running = [s for s in services if s.get('running')] print(f"Found {len(services)} services ({len(running)} running):") for s in running[:8]: print(f" ✓ {s['key']}: {s['label']}") except Exception as e: print(f"Error: {e}") print() # Test 5: Get NTP Config (NEW!) print("=== Test 5: Get NTP Config (NEW!) ===") try: result = await session.call_tool("get_ntp_config", {}) ntp = json.loads(result.content[0].text) print(f" NTP Servers: {ntp.get('ntp_servers', [])}") print(f" Service Running: {ntp.get('service_running', 'N/A')}") print(f" Timezone: {ntp.get('timezone', 'N/A')}") except Exception as e: print(f"Error: {e}") print() # Test 6: List VM Disks (if there are VMs) print("=== Test 6: List VM Disks (NEW!) ===") try: # First get a VM name result = await session.read_resource("esxi://vms") vms = json.loads(result.contents[0].text) if vms: vm_name = vms[0]['name'] print(f" Checking disks on: {vm_name}") result = await session.call_tool("list_disks", {"vm_name": vm_name}) disks = json.loads(result.content[0].text) for d in disks: print(f" • {d['label']}: {d.get('size_gb', 'N/A')} GB") else: print(" No VMs found to check") except Exception as e: print(f"Error: {e}") print() # Test 7: List VM NICs (NEW!) print("=== Test 7: List VM NICs (NEW!) ===") try: if vms: vm_name = vms[0]['name'] print(f" Checking NICs on: {vm_name}") result = await session.call_tool("list_nics", {"vm_name": vm_name}) nics = json.loads(result.content[0].text) for n in nics: print(f" • {n['label']}: {n.get('type', 'N/A')} - {n.get('network', 'N/A')}") print(f" MAC: {n.get('mac_address', 'N/A')}, Connected: {n.get('connected', 'N/A')}") except Exception as e: print(f"Error: {e}") print() # ───────────────────────────────────────────────────────────────────────── # vCenter-Specific Tests (NEW!) # ───────────────────────────────────────────────────────────────────────── # Test 8: List Folders print("=== Test 8: List Folders (vCenter) ===") try: result = await session.call_tool("list_folders", {}) folders = json.loads(result.content[0].text) print(f"Found {len(folders)} folders:") for f in folders[:10]: print(f" 📁 {f['path']} ({f.get('children', 0)} children)") except Exception as e: print(f"Error: {e}") print() # Test 9: List Clusters print("=== Test 9: List Clusters (vCenter) ===") try: result = await session.call_tool("list_clusters", {}) clusters = json.loads(result.content[0].text) # Check if it's an empty-result message if clusters and 'message' in clusters[0]: print(f" {clusters[0]['message']}") elif clusters: print(f"Found {len(clusters)} clusters:") for c in clusters: print(f" 🖥️ {c['name']}: {c.get('host_count', 0)} hosts") print(f" DRS: {c.get('drs', {}).get('enabled', 'N/A')}, HA: {c.get('ha', {}).get('enabled', 'N/A')}") else: print(" No clusters found (standalone host mode)") except Exception as e: print(f"Error: {e}") print() # Test 10: List Recent Tasks print("=== Test 10: Recent Tasks (vCenter) ===") try: result = await session.call_tool("list_recent_tasks", {"max_count": 5}) tasks = json.loads(result.content[0].text) # Check if it's an empty-result message if tasks and 'message' in tasks[0]: print(f" {tasks[0]['message']}") else: print(f"Found {len(tasks)} recent tasks:") for t in tasks[:5]: entity = t.get('entity', 'N/A') print(f" 📋 {t.get('name', 'N/A')} - {t.get('state', 'N/A')}") print(f" Entity: {entity} | Started: {t.get('start_time', 'N/A')}") except Exception as e: print(f"Error: {e}") print() # Test 11: Get Alarms print("=== Test 11: Get Alarms (vCenter) ===") try: result = await session.call_tool("get_alarms", {}) alarms = json.loads(result.content[0].text) if alarms: print(f"Found {len(alarms)} active alarms:") for a in alarms[:5]: print(f" ⚠️ {a.get('alarm', 'N/A')} on {a.get('entity', 'N/A')}") print(f" Status: {a.get('status', 'N/A')}, Acknowledged: {a.get('acknowledged', 'N/A')}") else: print(" No active alarms (all clear!)") except Exception as e: print(f"Error: {e}") print() # Test 12: List Recent Events print("=== Test 12: Recent Events (vCenter) ===") try: result = await session.call_tool("list_recent_events", {"max_count": 5, "hours_back": 24}) events = json.loads(result.content[0].text) print(f"Found {len(events)} events in last 24h:") for e in events[:5]: vm_info = f" (VM: {e['vm']})" if e.get('vm') else "" print(f" 📌 {e.get('type', 'N/A')}{vm_info}") msg = e.get('message', 'N/A') if len(msg) > 80: msg = msg[:77] + "..." print(f" {msg}") except Exception as e: print(f"Error: {e}") print() print("✅ All tests completed!") if __name__ == "__main__": asyncio.run(main())