VMware Fusion MCP Server

Instructions

Implementation Reference

• vmware_fusion_mcp/server.py:41-46 (handler)

  Core handler logic for the power_vm tool: instantiates a VMwareClient using environment credentials and delegates to its power_vm method to execute the power action.

  async def _power_vm_impl(vm_id: str, action: str) -> Dict[str, Any]:
      """Perform a power action on a VM. Valid actions are 'on', 'off', 'shutdown', 'suspend', 'pause', 'unpause'."""
      async with VMwareClient(username=VMREST_USER, password=VMREST_PASS) as client:
          result = await client.power_vm(vm_id, action)
          return result  # type: ignore[no-any-return]

• vmware_fusion_mcp/server.py:48-51 (registration)

  Registers the power_vm tool in the FastMCP server with input parameters vm_id and action, and docstring describing usage.

  @mcp.tool
  async def power_vm(vm_id: str, action: str) -> Dict[str, Any]:
      """Perform a power action on a VM. Valid actions are 'on', 'off', 'shutdown', 'suspend', 'pause', 'unpause'."""
      return await _power_vm_impl(vm_id, action)

• vmware_fusion_mcp/vmware_client.py:112-159 (helper)

  Key helper method in VMwareClient that implements the power operation: validates the action, constructs and sends a PUT request to the Fusion REST API endpoint /api/vms/{vm_id}/power, handles authentication, parameters, responses, and errors.

  async def power_vm(
      self, vm_id: str, action: str, vm_password: Optional[str] = None
  ) -> Dict[str, Any]:
      """Perform a power action on a VM.
  
      Args:
          vm_id: The ID of the VM
          action: Power action (on, off, shutdown, suspend, pause, unpause)
          vm_password: The password for the VM (if required)
  
      Returns:
          Dictionary with the result of the power action
      """
      valid_actions = ["on", "off", "shutdown", "suspend", "pause", "unpause"]
      if action not in valid_actions:
          raise ValueError(
              f"Invalid action '{action}'. Valid actions: {valid_actions}"
          )
  
      try:
          url = f"{self.base_url}/api/vms/{vm_id}/power"
          params = {}
          if vm_password:
              params["vmPassword"] = vm_password
          headers = self._auth_header.copy()
          headers["Content-Type"] = "application/vnd.vmware.vmw.rest-v1+json"
          response = await self._client.put(
              url,
              headers=headers,
              params=params or None,
              content=action.encode(),
          )
          response.raise_for_status()
          return (
              response.json()
              if response.content
              else {"status": "success", "action": action}
          )
      except httpx.RequestError as e:
          raise Exception(f"Failed to connect to VMware Fusion API: {e}")
      except httpx.HTTPStatusError as e:
          if e.response.status_code == 404:
              raise Exception(f"VM with ID '{vm_id}' not found")
          raise Exception(
              f"VMware Fusion API error: {e.response.status_code} - "
              f"{e.response.text}"
          )

• vmware_fusion_mcp/vmware_client.py:125-129 (schema)

  Input schema validation for the action parameter, enforcing allowed power actions.

  valid_actions = ["on", "off", "shutdown", "suspend", "pause", "unpause"]
  if action not in valid_actions:
      raise ValueError(
          f"Invalid action '{action}'. Valid actions: {valid_actions}"
      )