check_service_requirements
Verify device compatibility for service installation by checking system requirements against target host specifications.
Instructions
Check if a device meets the requirements for a service installation
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| service_name | Yes | Name of the service to check requirements for | |
| hostname | Yes | Hostname or IP address of the target device | |
| username | No | SSH username (use 'mcp_admin' for passwordless access after setup) | mcp_admin |
| password | No | SSH password (not needed for mcp_admin after setup) | |
| port | No | SSH port (default: 22) |
Implementation Reference
- MCP tool handler function that processes check_service_requirements requests. Creates a ServiceInstaller instance and calls check_service_requirements with the provided arguments.
async def handle_check_service_requirements( arguments: dict[str, Any], ) -> dict[str, Any]: """Handle check_service_requirements tool.""" installer = ServiceInstaller() requirements_result = await installer.check_service_requirements(**arguments) return {"content": [{"type": "text", "text": json.dumps(requirements_result, indent=2)}]} - Core implementation of check_service_requirements in ServiceInstaller class. Validates service exists, checks available ports, memory, and disk space on remote device via SSH commands.
async def check_service_requirements( self, service_name: str, hostname: str, username: str = "mcp_admin", password: str | None = None, ) -> dict[str, Any]: """Check if a device meets the requirements for a service.""" if service_name not in self.templates: return {"status": "error", "error": f"Unknown service: {service_name}"} service = self.templates[service_name] requirements = service.get("requirements", {}) results: dict[str, Any] = { "service": service_name, "hostname": hostname, "requirements_met": True, "checks": {}, } # Check available ports if "ports" in requirements: for port in requirements["ports"]: # Check if port is available cmd = f"ss -tlnp | grep :{port}" port_result = await ssh_execute_command( hostname=hostname, username=username, password=password, command=cmd ) port_data = json.loads(port_result) port_available = port_data.get("exit_code", 1) != 0 # Port is free if command fails checks_dict = results.get("checks", {}) if isinstance(checks_dict, dict): checks_dict[f"port_{port}"] = { "required": True, "available": port_available, "status": "pass" if port_available else "fail", } if not port_available: results["requirements_met"] = False # Check available memory if "memory_gb" in requirements: cmd = "free -m | grep '^Mem:' | awk '{print $2}'" mem_result = await ssh_execute_command(hostname=hostname, username=username, password=password, command=cmd) mem_data = json.loads(mem_result) if mem_data.get("exit_code") == 0: available_mb = int(mem_data.get("output", "0").split("Output:\n")[-1]) required_mb = requirements["memory_gb"] * 1024 memory_ok = available_mb >= required_mb checks_dict = results.get("checks", {}) if isinstance(checks_dict, dict): checks_dict["memory"] = { "required_mb": required_mb, "available_mb": available_mb, "status": "pass" if memory_ok else "fail", } if not memory_ok: results["requirements_met"] = False # Check disk space if "disk_gb" in requirements: cmd = "df / | tail -1 | awk '{print $4}'" disk_result = await ssh_execute_command( hostname=hostname, username=username, password=password, command=cmd ) disk_data = json.loads(disk_result) if disk_data.get("exit_code") == 0: available_kb = int(disk_data.get("output", "0").split("Output:\n")[-1]) required_kb = requirements["disk_gb"] * 1024 * 1024 disk_ok = available_kb >= required_kb checks_dict = results.get("checks", {}) if isinstance(checks_dict, dict): checks_dict["disk_space"] = { "required_gb": requirements["disk_gb"], "available_gb": round(available_kb / 1024 / 1024, 2), "status": "pass" if disk_ok else "fail", } if not disk_ok: results["requirements_met"] = False return results - Input schema definition for check_service_requirements tool. Defines required and optional parameters including service_name, hostname, username, password, and port.
"check_service_requirements": { "description": "Check if a device meets the requirements for a service installation", "inputSchema": { "type": "object", "properties": { "service_name": { "type": "string", "description": "Name of the service to check requirements for", }, "hostname": { "type": "string", "description": "Hostname or IP address of the target device", }, "username": { "type": "string", "description": "SSH username (use 'mcp_admin' for passwordless access after setup)", "default": "mcp_admin", }, "password": { "type": "string", "description": "SSH password (not needed for mcp_admin after setup)", }, "port": { "type": "integer", "description": "SSH port (default: 22)", "default": 22, }, }, "required": ["service_name", "hostname"], }, }, - src/homelab_mcp/tool_handlers/__init__.py:107-107 (registration)Tool registration mapping the check_service_requirements name to its handler function in the TOOL_HANDLERS dictionary.
"check_service_requirements": handle_check_service_requirements, - src/homelab_mcp/ssh_tools.py:604-660 (helper)SSH execution helper function used by check_service_requirements to run commands on remote devices. Handles SSH connection, credential resolution, and command execution with optional sudo support.
async def ssh_execute_command( hostname: str, username: str | None = None, command: str = "", password: str | None = None, sudo: bool = False, port: int = 22, **kwargs: Any, ) -> str: """Execute a command on a remote system via SSH.""" # Resolve credentials using priority order creds = resolve_ssh_credentials( hostname=hostname, username=username, password=password, port=port, ) # Prepare connection options connect_kwargs: dict[str, Any] = { "host": creds.hostname, "port": creds.port, "username": creds.username, "known_hosts": None, } if creds.key_path: connect_kwargs["client_keys"] = [creds.key_path] if creds.password: connect_kwargs["password"] = creds.password # If no credentials available, try to fall back to default mcp_admin key if "client_keys" not in connect_kwargs and "password" not in connect_kwargs: if creds.username == "mcp_admin": mcp_key_path = await ensure_mcp_ssh_key() if mcp_key_path: connect_kwargs["client_keys"] = [mcp_key_path] else: raise ValueError( f"No credentials available for {hostname}. " "Register the server first with register_server or provide password." ) async with asyncssh.connect(**connect_kwargs) as conn: # Prepare the command with sudo if requested if sudo: if creds.username == "mcp_admin": # mcp_admin has passwordless sudo full_command = f"sudo {command}" else: # Other users might need password for sudo full_command = f"echo '{creds.password}' | sudo -S {command}" if creds.password else f"sudo {command}" else: full_command = command # Execute the command