Homelab MCP Server

Instructions

Implementation Reference

• src/homelab_mcp/vm_operations.py:64-94 (handler)

  Core implementation function control_vm_state that executes the VM control logic - retrieves device connection info, establishes SSH connection, calls provider.control_vm(), and returns formatted JSON response

  async def control_vm_state(device_id: int, platform: str, vm_name: str, action: str) -> str:
      """Control VM state (start, stop, restart)."""
      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.control_vm(conn, vm_name, action)
              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 control failed: {str(e)}"})

• src/homelab_mcp/tool_schemas/vm_tools_schema.py:56-82 (schema)

  Schema definition for control_vm tool - defines input parameters (device_id, platform, vm_name, action) with types, enums, and descriptions

  "control_vm": {
      "description": "Control VM state (start, stop, restart)",
      "inputSchema": {
          "type": "object",
          "properties": {
              "device_id": {
                  "type": "integer",
                  "description": "Database ID of the target device",
              },
              "platform": {
                  "type": "string",
                  "enum": ["docker", "lxd"],
                  "description": "VM platform",
              },
              "vm_name": {
                  "type": "string",
                  "description": "Name of the VM/container",
              },
              "action": {
                  "type": "string",
                  "enum": ["start", "stop", "restart"],
                  "description": "Action to perform",
              },
          },
          "required": ["device_id", "platform", "vm_name", "action"],
      },
  },

• src/homelab_mcp/tool_handlers/vm_handlers.py:26-34 (handler)

  Handler wrapper function handle_control_vm that extracts arguments from the tool request and calls control_vm_state with proper parameters

  async def handle_control_vm(arguments: dict[str, Any]) -> dict[str, Any]:
      """Handle control_vm tool."""
      result = await control_vm_state(
          device_id=arguments["device_id"],
          platform=arguments["platform"],
          vm_name=arguments["vm_name"],
          action=arguments["action"],
      )
      return {"content": [{"type": "text", "text": result}]}

• src/homelab_mcp/tool_handlers/__init__.py:99-99 (registration)

  Tool registration in TOOL_HANDLERS dictionary mapping 'control_vm' tool name to handle_control_vm handler function

  "control_vm": handle_control_vm,

• src/homelab_mcp/vm_providers/base.py:50-64 (helper)

  Provider interface control_vm method that dispatches to appropriate platform-specific methods (start_vm, stop_vm, restart_vm) based on the action parameter

  async def control_vm(self, conn: Any, vm_name: str, action: str) -> dict[str, Any]:
      """Control VM state with the specified action."""
      action_lower = action.lower()
  
      if action_lower == "start":
          return await self.start_vm(conn, vm_name)
      elif action_lower == "stop":
          return await self.stop_vm(conn, vm_name)
      elif action_lower == "restart":
          return await self.restart_vm(conn, vm_name)
      else:
          return {
              "status": "error",
              "message": f"Unknown action: {action}. Supported actions: start, stop, restart",
          }