deploy_vm
Deploy virtual machines or containers on homelab devices using Docker or LXD platforms. Configure images, ports, volumes, and environment variables for automated infrastructure management.
Instructions
Deploy a new VM/container on a specific device
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| device_id | Yes | Database ID of the target device | |
| platform | Yes | VM platform to use (docker or lxd) | |
| vm_name | Yes | Name for the new VM/container | |
| vm_config | No | VM configuration |
Implementation Reference
- src/homelab_mcp/vm_operations.py:31-61 (handler)Core implementation of deploy_vm that handles device lookup, SSH connection, provider selection, and VM deployment execution with error handling
async def deploy_vm(device_id: int, platform: str, vm_name: str, vm_config: dict[str, Any]) -> str: """Deploy a new VM/container on a specific device.""" try: manager = VMManager() connection_info = await manager.get_device_connection_info(device_id) if not connection_info: return json.dumps( { "status": "error", "message": f"Device with ID {device_id} not found in sitemap", } ) provider = get_vm_provider(platform) async with asyncssh.connect( connection_info["hostname"], username=connection_info["username"], known_hosts=None, ) as conn: result = await provider.deploy_vm(conn, vm_name, vm_config) result["device_id"] = device_id result["platform"] = platform return json.dumps(result, indent=2) except ValueError as e: return json.dumps({"status": "error", "message": str(e)}) except Exception as e: return json.dumps({"status": "error", "message": f"VM deployment failed: {str(e)}"}) - MCP tool handler function that wraps the core deploy_vm implementation and formats the response for the MCP protocol
async def handle_deploy_vm(arguments: dict[str, Any]) -> dict[str, Any]: """Handle deploy_vm tool.""" result = await deploy_vm( device_id=arguments["device_id"], platform=arguments["platform"], vm_name=arguments["vm_name"], vm_config=arguments.get("vm_config", {}), ) return {"content": [{"type": "text", "text": result}]} - Input schema definition for deploy_vm tool specifying required parameters (device_id, platform, vm_name) and optional vm_config with image, ports, volumes, environment, and command options
"deploy_vm": { "description": "Deploy a new VM/container on a specific device", "inputSchema": { "type": "object", "properties": { "device_id": { "type": "integer", "description": "Database ID of the target device", }, "platform": { "type": "string", "enum": ["docker", "lxd"], "description": "VM platform to use (docker or lxd)", }, "vm_name": { "type": "string", "description": "Name for the new VM/container", }, "vm_config": { "type": "object", "description": "VM configuration", "properties": { "image": { "type": "string", "description": "Container/VM image to use", }, "ports": { "type": "array", "items": {"type": "string"}, "description": "Port mappings (e.g., '80:80')", }, "volumes": { "type": "array", "items": {"type": "string"}, "description": "Volume mounts (e.g., '/host/path:/container/path')", }, "environment": { "type": "object", "description": "Environment variables", }, "command": { "type": "string", "description": "Command to run in container", }, }, }, }, "required": ["device_id", "platform", "vm_name"], }, }, - src/homelab_mcp/tool_handlers/__init__.py:61-103 (registration)Registration of deploy_vm handler in the TOOL_HANDLERS dictionary, mapping the tool name 'deploy_vm' to handle_deploy_vm function
from .vm_handlers import ( handle_control_vm, handle_deploy_vm, handle_get_vm_logs, handle_get_vm_status, handle_list_vms, handle_remove_vm, ) # Type alias for handler functions ToolHandler = Callable[[dict[str, Any]], Awaitable[dict[str, Any]]] # Tool handler registry mapping tool names to their handler functions TOOL_HANDLERS: dict[str, ToolHandler] = { # SSH tools "ssh_discover": handle_ssh_discover, "setup_mcp_admin": handle_setup_mcp_admin, "verify_mcp_admin": handle_verify_mcp_admin, "ssh_execute_command": handle_ssh_execute_command, "start_interactive_shell": handle_start_interactive_shell, "update_mcp_admin_groups": handle_update_mcp_admin_groups, # Network tools "discover_and_map": handle_discover_and_map, "bulk_discover_and_map": handle_bulk_discover_and_map, "get_network_sitemap": handle_get_network_sitemap, "analyze_network_topology": handle_analyze_network_topology, "suggest_deployments": handle_suggest_deployments, "get_device_changes": handle_get_device_changes, # Infrastructure tools "deploy_infrastructure": handle_deploy_infrastructure, "update_device_config": handle_update_device_config, "decommission_device": handle_decommission_device, "scale_services": handle_scale_services, "validate_infrastructure_changes": handle_validate_infrastructure_changes, "create_infrastructure_backup": handle_create_infrastructure_backup, "rollback_infrastructure_changes": handle_rollback_infrastructure_changes, # VM tools "deploy_vm": handle_deploy_vm, "control_vm": handle_control_vm, "get_vm_status": handle_get_vm_status, "list_vms": handle_list_vms, "get_vm_logs": handle_get_vm_logs, "remove_vm": handle_remove_vm, - Docker provider implementation of deploy_vm that checks for existing containers, builds docker run command with configuration options (ports, volumes, environment), and executes the deployment
async def deploy_vm(self, conn: Any, vm_name: str, vm_config: dict[str, Any]) -> dict[str, Any]: """Deploy a new Docker container.""" try: # Check if container already exists check_result = await self._run_command(conn, f"docker inspect {vm_name}") if check_result["exit_status"] == 0: return self._format_error("deploy", vm_name, "Container already exists") # Extract configuration image = vm_config.get("image", "ubuntu:22.04") ports = vm_config.get("ports", []) volumes = vm_config.get("volumes", []) environment = vm_config.get("environment", {}) network = vm_config.get("network", "bridge") restart_policy = vm_config.get("restart_policy", "unless-stopped") # Build docker run command cmd_parts = ["docker", "run", "-d"] cmd_parts.extend(["--name", vm_name]) cmd_parts.extend(["--restart", restart_policy]) cmd_parts.extend(["--network", network]) # Add port mappings for port in ports: cmd_parts.extend(["-p", port]) # Add volume mounts for volume in volumes: cmd_parts.extend(["-v", volume]) # Add environment variables for key, value in environment.items(): cmd_parts.extend(["-e", f"{key}={value}"]) # Add the image cmd_parts.append(image) # Add command if specified if "command" in vm_config: cmd_parts.extend(vm_config["command"].split()) # Execute deployment deploy_result = await self._run_command(conn, " ".join(cmd_parts)) if deploy_result["exit_status"] == 0: container_id = deploy_result["stdout"].strip() # Get container details inspect_result = await self._run_command(conn, f"docker inspect {vm_name}") if inspect_result["exit_status"] == 0: container_info = json.loads(inspect_result["stdout"])[0] return self._format_success( "deploy", vm_name, { "container_id": container_id, "image": image, "network": network, "ports": container_info.get("NetworkSettings", {}).get("Ports", {}), "container_status": container_info.get("State", {}).get("Status", "unknown"), }, ) else: return self._format_success("deploy", vm_name, {"container_id": container_id}) else: return self._format_error("deploy", vm_name, deploy_result["stderr"]) except Exception as e: return self._format_error("deploy", vm_name, str(e))