deploy_infrastructure
Deploy infrastructure services like Docker containers, LXD instances, and network configurations to homelab devices based on AI recommendations or user specifications.
Instructions
Deploy new infrastructure based on AI recommendations or user specifications
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| deployment_plan | Yes | Infrastructure deployment plan | |
| validate_only | No | Only validate the plan without executing |
Implementation Reference
- Tool handler function that processes deploy_infrastructure requests, extracts arguments and calls the core implementation
async def handle_deploy_infrastructure(arguments: dict[str, Any]) -> dict[str, Any]: """Handle deploy_infrastructure tool.""" result = await deploy_infrastructure_plan( deployment_plan=arguments["deployment_plan"], validate_only=arguments.get("validate_only", False), ) return {"content": [{"type": "text", "text": result}]} - JSON schema definition for deploy_infrastructure tool input, including deployment_plan structure with services and network_changes arrays, plus validate_only option
"deploy_infrastructure": { "description": "Deploy new infrastructure based on AI recommendations or user specifications", "inputSchema": { "type": "object", "properties": { "deployment_plan": { "type": "object", "description": "Infrastructure deployment plan", "properties": { "services": { "type": "array", "items": { "type": "object", "properties": { "name": {"type": "string"}, "type": { "type": "string", "enum": ["docker", "lxd", "service"], }, "target_device_id": {"type": "integer"}, "config": {"type": "object"}, }, "required": ["name", "type", "target_device_id"], }, }, "network_changes": { "type": "array", "items": { "type": "object", "properties": { "action": { "type": "string", "enum": [ "create_vlan", "configure_firewall", "setup_routing", ], }, "target_device_id": {"type": "integer"}, "config": {"type": "object"}, }, }, }, }, }, "validate_only": { "type": "boolean", "default": False, "description": "Only validate the plan without executing", }, }, "required": ["deployment_plan"], }, }, - Core implementation function that validates deployment plans, optionally executes deployments for Docker/LXD/systemd services and network changes, and returns JSON-formatted results
async def deploy_infrastructure_plan(deployment_plan: dict[str, Any], validate_only: bool = False) -> str: """Deploy new infrastructure based on AI recommendations or user specifications.""" try: manager = InfrastructureManager() # Validate the deployment plan validation_result = await _validate_deployment_plan(deployment_plan) if not validation_result["valid"]: return json.dumps( { "status": "error", "message": f"Deployment plan validation failed: {validation_result['errors']}", } ) if validate_only: return json.dumps( { "status": "success", "message": "Deployment plan validation passed", "validation_result": validation_result, "estimated_duration": "15-30 minutes", "affected_devices": len( {service.get("target_device_id") for service in deployment_plan.get("services", [])} ), } ) # Execute deployment plan deployment_results = [] # Deploy services for service in deployment_plan.get("services", []): result = await _deploy_service(manager, service) deployment_results.append(result) # Apply network changes for network_change in deployment_plan.get("network_changes", []): result = await _apply_network_change(manager, network_change) deployment_results.append(result) # Update sitemap with new infrastructure await _update_sitemap_after_deployment(manager, deployment_results) successful_deployments = [r for r in deployment_results if r.get("status") == "success"] failed_deployments = [r for r in deployment_results if r.get("status") == "error"] return json.dumps( { "status": "success" if len(failed_deployments) == 0 else "partial_success", "message": f"Deployed {len(successful_deployments)} components successfully", "successful_deployments": len(successful_deployments), "failed_deployments": len(failed_deployments), "deployment_results": deployment_results, "next_steps": [ "Verify services are running correctly", "Update DNS/load balancer configurations if needed", "Monitor resource usage for optimization opportunities", ], }, indent=2, ) except Exception as e: return json.dumps( { "status": "error", "message": f"Infrastructure deployment failed: {str(e)}", } ) - src/homelab_mcp/tool_handlers/__init__.py:90-90 (registration)Tool registration mapping the 'deploy_infrastructure' tool name to its handler function in the TOOL_HANDLERS registry
"deploy_infrastructure": handle_deploy_infrastructure, - Helper function that deploys individual services (Docker containers, LXD containers, or systemd services) to target devices via SSH
async def _deploy_service(manager: InfrastructureManager, service: dict[str, Any]) -> dict[str, Any]: """Deploy a single service.""" try: device_id = service["target_device_id"] connection_info = await manager.get_device_connection_info(device_id) if not connection_info: return { "status": "error", "service": service["name"], "error": f"Device {device_id} not found", } async with asyncssh.connect( connection_info["hostname"], username=connection_info["username"], known_hosts=None, ) as conn: service_type = service["type"] service_name = service["name"] config = service.get("config", {}) if service_type == "docker": # Deploy Docker container docker_image = config.get("image", "nginx:latest") ports = config.get("ports", []) volumes = config.get("volumes", []) env_vars = config.get("environment", {}) # Build docker run command cmd_parts = ["docker", "run", "-d", "--name", service_name] # Add port mappings for port_mapping in ports: cmd_parts.extend(["-p", port_mapping]) # Add volume mounts for volume in volumes: cmd_parts.extend(["-v", volume]) # Add environment variables for key, value in env_vars.items(): cmd_parts.extend(["-e", f"{key}={value}"]) cmd_parts.append(docker_image) result = await conn.run(" ".join(cmd_parts)) if result.exit_status == 0: stdout_text = ( result.stdout.decode() if isinstance(result.stdout, bytes) else str(result.stdout) if result.stdout else "" ) return { "status": "success", "service": service_name, "container_id": stdout_text.strip(), } else: return { "status": "error", "service": service_name, "error": result.stderr, } elif service_type == "lxd": # Deploy LXD container lxd_image = config.get("image", "ubuntu:22.04") # Launch LXD container result = await conn.run(f"lxc launch {lxd_image} {service_name}") if result.exit_status == 0: return { "status": "success", "service": service_name, "container": service_name, } else: return { "status": "error", "service": service_name, "error": result.stderr, } elif service_type == "service": # Deploy systemd service service_file = config.get("service_file", "") if not service_file: return { "status": "error", "service": service_name, "error": "Service file content required", } # Write service file await conn.run(f'echo "{service_file}" | sudo tee /etc/systemd/system/{service_name}.service') await conn.run("sudo systemctl daemon-reload") await conn.run(f"sudo systemctl enable {service_name}") result = await conn.run(f"sudo systemctl start {service_name}") if result.exit_status == 0: return { "status": "success", "service": service_name, "systemd_service": service_name, } else: return { "status": "error", "service": service_name, "error": result.stderr, } else: return { "status": "error", "service": service_name, "error": f"Unknown service type: {service_type}", } except Exception as e: return { "status": "error", "service": service.get("name", "unknown"), "error": str(e), }