diff --git a/Server/src/cli/commands/unity_hub.py b/Server/src/cli/commands/unity_hub.py new file mode 100644 index 000000000..f7d6ff2d0 --- /dev/null +++ b/Server/src/cli/commands/unity_hub.py @@ -0,0 +1,130 @@ +"""Unity Hub CLI commands — runs on host, does not require Unity Editor.""" + +import asyncio +import json + +import click + +from cli.utils.output import print_info, print_success +from services.unity_hub import ( + _INSTALL_TIMEOUT, + detect_hub_path, + parse_available_releases, + parse_installed_editors, + run_hub_command, +) + + +@click.group("hub") +def unity_hub(): + """Unity Hub operations - editors, releases, install path (host-side, no Unity needed).""" + pass + + +def _run_async(coro): + """Run an async function synchronously.""" + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + if loop and loop.is_running(): + import concurrent.futures + with concurrent.futures.ThreadPoolExecutor() as pool: + return pool.submit(asyncio.run, coro).result() + return asyncio.run(coro) + + +def _print_result(result: dict) -> None: + click.echo(json.dumps(result, indent=2)) + + +@unity_hub.command("info") +def info() -> None: + """Show detected Unity Hub and host information.""" + import platform as _platform + hub_path = detect_hub_path() + _print_result({ + "hub_detected": hub_path is not None, + "hub_path": hub_path, + "os": _platform.system(), + "architecture": _platform.machine(), + }) + + +@unity_hub.command("editors") +def editors() -> None: + """List locally installed Unity Editor versions.""" + result = _run_async(run_hub_command(["editors", "--installed"])) + if not result["success"]: + _print_result(result) + return + parsed = parse_installed_editors(result["raw_output"]) + for editor in parsed: + click.echo(f" {editor['version']} -> {editor['path']}") + if not parsed: + click.echo(" No editors installed.") + + +@unity_hub.command("releases") +@click.option("--limit", type=int, default=None, help="Maximum number of releases to return.") +def releases(limit: int | None) -> None: + """List Unity Editor releases available through Unity Hub.""" + result = _run_async(run_hub_command(["editors", "--releases"])) + if not result["success"]: + _print_result(result) + return + parsed = parse_available_releases(result["raw_output"], limit) + for release in parsed: + channel = f" ({release['channel']})" if "channel" in release else "" + click.echo(f" {release['version']}{channel}") + + +@unity_hub.command("install-path") +@click.option("--set", "new_path", type=str, default=None, help="Set the Unity Editor install path.") +def install_path(new_path: str | None) -> None: + """Get or set the Unity Editor install path.""" + if new_path is None: + result = _run_async(run_hub_command(["install-path", "--get"])) + if result["success"]: + click.echo(f" Install path: {result['raw_output']}") + else: + _print_result(result) + return + + click.confirm(f"Change Unity Editor install path to '{new_path}'?", abort=True) + result = _run_async(run_hub_command(["install-path", "--set", new_path])) + if result["success"]: + print_success(f"Install path changed to: {new_path}") + else: + _print_result(result) + + +@unity_hub.command("install") +@click.argument("version") +def install(version: str) -> None: + """Download and install a Unity Editor version via Unity Hub.""" + click.confirm(f"Install Unity Editor {version}?", abort=True) + result = _run_async(run_hub_command(["install", "--version", version], timeout=_INSTALL_TIMEOUT)) + if result["success"]: + print_success(f"Unity Editor {version} installation started.") + print_info("Unity Hub may continue the install in the background.") + else: + _print_result(result) + + +@unity_hub.command("install-modules") +@click.argument("version") +@click.option("--modules", "-m", multiple=True, required=True, help="Module to install (can repeat: -m android -m ios).") +def install_modules(version: str, modules: tuple[str, ...]) -> None: + """Install platform modules for an existing Unity Editor version.""" + module_list = list(modules) + click.confirm(f"Install modules [{', '.join(module_list)}] for Unity {version}?", abort=True) + args = ["install-modules", "--version", version] + for mod in module_list: + args.extend(["--module", mod]) + result = _run_async(run_hub_command(args, timeout=_INSTALL_TIMEOUT)) + if result["success"]: + print_success(f"Module installation started for Unity {version}.") + print_info("Unity Hub may continue the module install in the background.") + else: + _print_result(result) diff --git a/Server/src/cli/main.py b/Server/src/cli/main.py index 44afce32c..2ac91d09c 100644 --- a/Server/src/cli/main.py +++ b/Server/src/cli/main.py @@ -271,6 +271,7 @@ def register_optional_command(module_name: str, command_name: str) -> None: ("cli.commands.camera", "camera"), ("cli.commands.graphics", "graphics"), ("cli.commands.packages", "packages"), + ("cli.commands.unity_hub", "unity_hub"), ("cli.commands.reflect", "reflect"), ("cli.commands.docs", "docs"), ("cli.commands.physics", "physics"), diff --git a/Server/src/services/registry/tool_registry.py b/Server/src/services/registry/tool_registry.py index 069b44ab9..e12bb526d 100644 --- a/Server/src/services/registry/tool_registry.py +++ b/Server/src/services/registry/tool_registry.py @@ -18,6 +18,7 @@ TOOL_GROUPS: dict[str, str] = { "core": "Essential scene, script, asset & editor tools (always on by default)", "docs": "Unity API reflection and documentation lookup", + "unity_hub": "Host-side Unity Hub and editor installation management", "vfx": "Visual effects – VFX Graph, shaders, procedural textures", "animation": "Animator control & AnimationClip creation", "ui": "UI Toolkit (UXML, USS, UIDocument)", diff --git a/Server/src/services/tools/manage_unity_hub.py b/Server/src/services/tools/manage_unity_hub.py new file mode 100644 index 000000000..aeab7a89c --- /dev/null +++ b/Server/src/services/tools/manage_unity_hub.py @@ -0,0 +1,280 @@ +"""MCP tool for managing Unity Hub and Unity Editor installations on the host machine.""" + +from typing import Annotated, Any, Optional + +from mcp.types import ToolAnnotations + +from services.registry import mcp_for_unity_tool +from services.unity_hub import ( + _INSTALL_TIMEOUT, + detect_hub_path, + parse_available_releases, + parse_installed_editors, + run_hub_command, +) + + +ALL_ACTIONS = [ + "get_hub_info", + "list_installed_editors", + "list_available_releases", + "get_install_path", + "set_install_path", + "install_editor", + "install_modules", +] + +READ_ONLY_ACTIONS = { + "get_hub_info", + "list_installed_editors", + "list_available_releases", + "get_install_path", +} + + +@mcp_for_unity_tool( + group="unity_hub", + unity_target=None, + description=( + "Manage Unity Hub and Unity Editor installations on the host machine.\n\n" + "This tool interacts with the Unity Hub CLI directly on the host - " + "it does NOT require a running Unity Editor instance.\n\n" + "READ-ONLY:\n" + "- get_hub_info: Detect Hub installation, show path and OS info\n" + "- list_installed_editors: List all locally installed Unity Editor versions\n" + "- list_available_releases: List available Unity Editor versions for download\n" + "- get_install_path: Get the current Unity Editor install location\n\n" + "STATE-CHANGING (requires confirmation):\n" + "- set_install_path: Change where Unity Editors are installed\n" + "- install_editor: Download and install a Unity Editor version\n" + "- install_modules: Add platform modules (Android, iOS, etc.) to an installed editor" + ), + annotations=ToolAnnotations( + title="Manage Unity Hub", + destructiveHint=True, + readOnlyHint=False, + idempotentHint=False, + openWorldHint=True, + ), +) +async def manage_unity_hub( + action: Annotated[str, "The Hub action to perform."], + version: Annotated[ + Optional[str], + "Unity Editor version (e.g., '2022.3.0f1', '6000.0.0f1').", + ] = None, + modules: Annotated[ + Optional[list[str]], + "Platform modules to install (e.g., ['android', 'ios', 'webgl']).", + ] = None, + path: Annotated[ + Optional[str], + "File system path for set_install_path.", + ] = None, + limit: Annotated[ + Optional[int], + "Max number of releases to return for list_available_releases.", + ] = None, + confirm: Annotated[ + Optional[bool], + "Set to true to confirm state-changing actions.", + ] = None, +) -> dict[str, Any]: + action_lower = action.lower().strip() + + if action_lower not in ALL_ACTIONS: + return { + "success": False, + "message": f"Unknown action '{action}'. Valid actions: {', '.join(ALL_ACTIONS)}", + } + + if action_lower not in READ_ONLY_ACTIONS and not confirm: + hub_path = detect_hub_path() or "not found" + details = _build_confirmation_message( + action_lower, + hub_path, + version, + modules, + path, + ) + return { + "success": False, + "confirmation_required": True, + "message": details, + "hint": "Set confirm=true to proceed.", + } + + if action_lower == "get_hub_info": + return await _get_hub_info() + if action_lower == "list_installed_editors": + return await _list_installed_editors() + if action_lower == "list_available_releases": + return await _list_available_releases(limit) + if action_lower == "get_install_path": + return await _get_install_path() + if action_lower == "set_install_path": + return await _set_install_path(path) + if action_lower == "install_editor": + return await _install_editor(version) + if action_lower == "install_modules": + return await _install_modules(version, modules) + + return {"success": False, "message": "Action not implemented."} + + +def _build_confirmation_message( + action: str, + hub_path: str, + version: Optional[str], + modules: Optional[list[str]], + path: Optional[str], +) -> str: + if action == "install_editor": + return f"Install Unity Editor {version or '(version required)'} using Hub at '{hub_path}'?" + if action == "install_modules": + mods = ", ".join(modules) if modules else "(modules required)" + return f"Install modules [{mods}] for Unity {version or '(version required)'} using Hub at '{hub_path}'?" + if action == "set_install_path": + return f"Change Unity Editor install path to '{path or '(path required)'}' using Hub at '{hub_path}'?" + return f"Execute '{action}' on Hub at '{hub_path}'?" + + +async def _get_hub_info() -> dict[str, Any]: + import platform as _platform + + hub_path = detect_hub_path() + return { + "success": True, + "action": "get_hub_info", + "data": { + "hub_detected": hub_path is not None, + "hub_path": hub_path, + "os": _platform.system(), + "os_version": _platform.version(), + "architecture": _platform.machine(), + }, + } + + +async def _list_installed_editors() -> dict[str, Any]: + result = await run_hub_command(["editors", "--installed"]) + if not result["success"]: + return {**result, "action": "list_installed_editors"} + + editors = parse_installed_editors(result["raw_output"]) + return { + "success": True, + "action": "list_installed_editors", + "hub_path": result["hub_path"], + "data": editors, + "raw_output": result["raw_output"], + } + + +async def _list_available_releases(limit: Optional[int]) -> dict[str, Any]: + result = await run_hub_command(["editors", "--releases"]) + if not result["success"]: + return {**result, "action": "list_available_releases"} + + releases = parse_available_releases(result["raw_output"], limit) + return { + "success": True, + "action": "list_available_releases", + "hub_path": result["hub_path"], + "data": releases, + "raw_output": result["raw_output"], + } + + +async def _get_install_path() -> dict[str, Any]: + result = await run_hub_command(["install-path", "--get"]) + if not result["success"]: + return {**result, "action": "get_install_path"} + + return { + "success": True, + "action": "get_install_path", + "hub_path": result["hub_path"], + "data": {"install_path": result["raw_output"]}, + } + + +async def _set_install_path(path: Optional[str]) -> dict[str, Any]: + if not path: + return { + "success": False, + "action": "set_install_path", + "message": "path is required.", + } + + result = await run_hub_command(["install-path", "--set", path]) + if not result["success"]: + return {**result, "action": "set_install_path"} + + return { + "success": True, + "action": "set_install_path", + "hub_path": result["hub_path"], + "data": {"install_path": path}, + "message": f"Install path changed to: {path}", + } + + +async def _install_editor(version: Optional[str]) -> dict[str, Any]: + if not version: + return { + "success": False, + "action": "install_editor", + "message": "version is required.", + } + + result = await run_hub_command( + ["install", "--version", version], + timeout=_INSTALL_TIMEOUT, + ) + if not result["success"]: + return {**result, "action": "install_editor"} + + return { + "success": True, + "action": "install_editor", + "hub_path": result["hub_path"], + "data": {"version": version}, + "message": f"Unity Editor {version} installation started.", + "raw_output": result["raw_output"], + } + + +async def _install_modules( + version: Optional[str], + modules: Optional[list[str]], +) -> dict[str, Any]: + if not version: + return { + "success": False, + "action": "install_modules", + "message": "version is required.", + } + if not modules: + return { + "success": False, + "action": "install_modules", + "message": "modules list is required and must not be empty.", + } + + args = ["install-modules", "--version", version] + for module_name in modules: + args.extend(["--module", module_name]) + + result = await run_hub_command(args, timeout=_INSTALL_TIMEOUT) + if not result["success"]: + return {**result, "action": "install_modules"} + + return { + "success": True, + "action": "install_modules", + "hub_path": result["hub_path"], + "data": {"version": version, "modules": modules}, + "message": f"Modules {modules} installation started for Unity {version}.", + "raw_output": result["raw_output"], + } diff --git a/Server/src/services/unity_hub.py b/Server/src/services/unity_hub.py new file mode 100644 index 000000000..ac090931d --- /dev/null +++ b/Server/src/services/unity_hub.py @@ -0,0 +1,183 @@ +"""Unity Hub CLI integration - runs on the host machine, not inside Unity Editor.""" + +import asyncio +import os +import platform +import shutil +from typing import Any, Optional + + +_HUB_PATHS = { + "Darwin": ["/Applications/Unity Hub.app/Contents/MacOS/Unity Hub"], + "Windows": [ + r"C:\Program Files\Unity Hub\Unity Hub.exe", + r"C:\Program Files (x86)\Unity Hub\Unity Hub.exe", + ], + "Linux": ["/usr/bin/unityhub", "/snap/bin/unityhub"], +} + +_DEFAULT_TIMEOUT = 30 +_INSTALL_TIMEOUT = 600 +_SPAWN_TIMEOUT = 15 + + +def detect_hub_path() -> Optional[str]: + """Find the Unity Hub executable on the host machine.""" + env_path = os.environ.get("UNITY_HUB_PATH") + if env_path and os.path.isfile(env_path): + return env_path + + system = platform.system() + for path in _HUB_PATHS.get(system, []): + if os.path.isfile(path): + return path + + which = shutil.which("unityhub") or shutil.which("Unity Hub") + if which: + return which + + return None + + +async def run_hub_command( + args: list[str], + timeout: int = _DEFAULT_TIMEOUT, + hub_path: Optional[str] = None, +) -> dict[str, Any]: + """Run a Unity Hub CLI command and return a structured result.""" + hub = hub_path or detect_hub_path() + if not hub: + searched = _HUB_PATHS.get(platform.system(), []) + return { + "success": False, + "error": { + "type": "hub_not_found", + "message": ( + "Unity Hub executable not found. " + f"Searched: {searched}. Set UNITY_HUB_PATH env var to override." + ), + }, + } + + cmd = [hub, "--", "--headless", *args] + + try: + spawn_timeout = min(_SPAWN_TIMEOUT, timeout) + proc = await asyncio.wait_for( + asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ), + timeout=spawn_timeout, + ) + stdout_bytes, stderr_bytes = await asyncio.wait_for( + proc.communicate(), + timeout=timeout, + ) + except asyncio.TimeoutError: + return { + "success": False, + "error": { + "type": "timeout", + "message": f"Hub command timed out after {timeout}s", + "command": args, + }, + } + except FileNotFoundError: + return { + "success": False, + "error": { + "type": "hub_not_found", + "message": f"Hub executable not found at: {hub}", + }, + } + except Exception as exc: + return { + "success": False, + "error": { + "type": "subprocess_error", + "message": str(exc), + }, + } + + stdout = stdout_bytes.decode("utf-8", errors="replace").strip() + stderr = stderr_bytes.decode("utf-8", errors="replace").strip() + + if proc.returncode != 0: + return { + "success": False, + "hub_path": hub, + "error": { + "type": "hub_command_failed", + "message": stderr or stdout or f"Exit code {proc.returncode}", + "exit_code": proc.returncode, + "stderr": stderr, + "stdout": stdout, + }, + } + + return { + "success": True, + "hub_path": hub, + "raw_output": stdout, + "stderr": stderr if stderr else None, + } + + +def parse_installed_editors(raw_output: str) -> list[dict[str, str]]: + """Parse `editors --installed` output into a structured list.""" + editors: list[dict[str, str]] = [] + for line in raw_output.strip().splitlines(): + line = line.strip() + if not line: + continue + + # Format: "6000.3.9f1 (Apple silicon) installed at /path/to/Unity.app" + # or: "2022.3.0f1 , installed at /path/to/editor" + path = "" + installed_at_idx = line.lower().find("installed at") + if installed_at_idx >= 0: + path = line[installed_at_idx + len("installed at"):].strip() + version_part = line[:installed_at_idx].strip().rstrip(",").strip() + else: + parts = line.split(",", 1) + version_part = parts[0].strip() + if len(parts) > 1: + path = parts[1].strip() + + # Extract clean version (first token before any parenthetical) + version = version_part.split("(")[0].strip().split()[0] if version_part else "" + + if version: + editors.append({"version": version, "path": path}) + + return editors + + +def parse_available_releases( + raw_output: str, + limit: Optional[int] = None, +) -> list[dict[str, str]]: + """Parse `editors --releases` output into a structured list.""" + releases: list[dict[str, str]] = [] + for line in raw_output.strip().splitlines(): + line = line.strip() + if not line: + continue + + version = line.split(",", 1)[0].strip().split(" ")[0].strip() + if not version: + continue + + entry = {"version": version} + if "LTS" in line: + entry["channel"] = "LTS" + elif "Tech" in line: + entry["channel"] = "Tech" + releases.append(entry) + + if limit and limit > 0: + releases = releases[:limit] + + return releases