import argparse import asyncio import logging from contextlib import asynccontextmanager import os import threading import time from typing import AsyncIterator, Any from urllib.parse import urlparse # Workaround for environments where tool signature evaluation runs with a globals # dict that does not include common `typing` names (e.g. when annotations are strings # and evaluated via `eval()` during schema generation). # Making these names available in builtins avoids `NameError: Annotated/Literal/... is not defined`. try: # pragma: no cover - startup safety guard import builtins import typing as _typing _typing_names = ( "Annotated", "Literal", "Any", "Union", "Optional", "Dict", "List", "Tuple", "Set", "FrozenSet", ) for _name in _typing_names: if not hasattr(builtins, _name) and hasattr(_typing, _name): # type: ignore[attr-defined] setattr(builtins, _name, getattr(_typing, _name)) except Exception: pass from fastmcp import FastMCP from logging.handlers import RotatingFileHandler from starlette.requests import Request from starlette.responses import JSONResponse from starlette.routing import WebSocketRoute from core.config import config from services.custom_tool_service import CustomToolService from transport.plugin_hub import PluginHub from transport.plugin_registry import PluginRegistry from services.resources import register_all_resources from core.telemetry import record_milestone, record_telemetry, MilestoneType, RecordType, get_package_version from services.tools import register_all_tools from transport.legacy.unity_connection import get_unity_connection_pool, UnityConnectionPool from transport.unity_instance_middleware import ( UnityInstanceMiddleware, get_unity_instance_middleware ) # Configure logging using settings from config logging.basicConfig( level=getattr(logging, config.log_level), format=config.log_format, stream=None, # None -> defaults to sys.stderr; avoid stdout used by MCP stdio force=True # Ensure our handler replaces any prior stdout handlers ) logger = logging.getLogger("mcp-for-unity-server") # Also write logs to a rotating file so logs are available when launched via stdio try: _log_dir = os.path.join(os.path.expanduser( "~/Library/Application Support/UnityMCP"), "Logs") os.makedirs(_log_dir, exist_ok=True) _file_path = os.path.join(_log_dir, "unity_mcp_server.log") _fh = RotatingFileHandler( _file_path, maxBytes=512*1024, backupCount=2, encoding="utf-8") _fh.setFormatter(logging.Formatter(config.log_format)) _fh.setLevel(getattr(logging, config.log_level)) logger.addHandler(_fh) logger.propagate = False # Prevent double logging to root logger # Also route telemetry logger to the same rotating file and normal level try: tlog = logging.getLogger("unity-mcp-telemetry") tlog.setLevel(getattr(logging, config.log_level)) tlog.addHandler(_fh) tlog.propagate = False # Prevent double logging for telemetry too except Exception as exc: # Never let logging setup break startup logger.debug("Failed to configure telemetry logger", exc_info=exc) except Exception as exc: # Never let logging setup break startup logger.debug("Failed to configure main logger file handler", exc_info=exc) # Quieten noisy third-party loggers to avoid clutter during stdio handshake for noisy in ("httpx", "urllib3", "mcp.server.lowlevel.server"): try: logging.getLogger(noisy).setLevel( max(logging.WARNING, getattr(logging, config.log_level))) logging.getLogger(noisy).propagate = False except Exception: pass # Import telemetry only after logging is configured to ensure its logs use stderr and proper levels # Ensure a slightly higher telemetry timeout unless explicitly overridden by env try: # Ensure generous timeout unless explicitly overridden by env if not os.environ.get("UNITY_MCP_TELEMETRY_TIMEOUT"): os.environ["UNITY_MCP_TELEMETRY_TIMEOUT"] = "5.0" except Exception: pass # Global connection pool _unity_connection_pool: UnityConnectionPool | None = None _plugin_registry: PluginRegistry | None = None # In-memory custom tool service initialized after MCP construction custom_tool_service: CustomToolService | None = None @asynccontextmanager async def server_lifespan(server: FastMCP) -> AsyncIterator[dict[str, Any]]: """Handle server startup and shutdown.""" global _unity_connection_pool logger.info("MCP for Unity Server starting up") # Register custom tool management endpoints with FastMCP # Routes are declared globally below after FastMCP initialization # Note: When using HTTP transport, FastMCP handles the HTTP server # Tool registration will be handled through FastMCP endpoints enable_http_server = os.environ.get( "UNITY_MCP_ENABLE_HTTP_SERVER", "").lower() in ("1", "true", "yes", "on") if enable_http_server: http_host = os.environ.get("UNITY_MCP_HTTP_HOST", "localhost") http_port = int(os.environ.get("UNITY_MCP_HTTP_PORT", "8080")) logger.info( f"HTTP tool registry will be available on http://{http_host}:{http_port}") global _plugin_registry if _plugin_registry is None: _plugin_registry = PluginRegistry() loop = asyncio.get_running_loop() PluginHub.configure(_plugin_registry, loop) # Record server startup telemetry start_time = time.time() start_clk = time.perf_counter() server_version = get_package_version() # Defer initial telemetry by 1s to avoid stdio handshake interference def _emit_startup(): try: record_telemetry(RecordType.STARTUP, { "server_version": server_version, "startup_time": start_time, }) record_milestone(MilestoneType.FIRST_STARTUP) except Exception: logger.debug("Deferred startup telemetry failed", exc_info=True) threading.Timer(1.0, _emit_startup).start() try: skip_connect = os.environ.get( "UNITY_MCP_SKIP_STARTUP_CONNECT", "").lower() in ("1", "true", "yes", "on") if skip_connect: logger.info( "Skipping Unity connection on startup (UNITY_MCP_SKIP_STARTUP_CONNECT=1)") else: # Initialize connection pool and discover instances _unity_connection_pool = get_unity_connection_pool() instances = _unity_connection_pool.discover_all_instances() if instances: logger.info( f"Discovered {len(instances)} Unity instance(s): {[i.id for i in instances]}") # Try to connect to default instance try: _unity_connection_pool.get_connection() logger.info( "Connected to default Unity instance on startup") # Record successful Unity connection (deferred) threading.Timer(1.0, lambda: record_telemetry( RecordType.UNITY_CONNECTION, { "status": "connected", "connection_time_ms": (time.perf_counter() - start_clk) * 1000, "instance_count": len(instances) } )).start() except Exception as e: logger.warning( f"Could not connect to default Unity instance: {e}") else: logger.warning("No Unity instances found on startup") except ConnectionError as e: logger.warning(f"Could not connect to Unity on startup: {e}") # Record connection failure (deferred) _err_msg = str(e)[:200] threading.Timer(1.0, lambda: record_telemetry( RecordType.UNITY_CONNECTION, { "status": "failed", "error": _err_msg, "connection_time_ms": (time.perf_counter() - start_clk) * 1000, } )).start() except Exception as e: logger.warning(f"Unexpected error connecting to Unity on startup: {e}") _err_msg = str(e)[:200] threading.Timer(1.0, lambda: record_telemetry( RecordType.UNITY_CONNECTION, { "status": "failed", "error": _err_msg, "connection_time_ms": (time.perf_counter() - start_clk) * 1000, } )).start() try: # Yield shared state for lifespan consumers (e.g., middleware) yield { "pool": _unity_connection_pool, "plugin_registry": _plugin_registry, } finally: if _unity_connection_pool: _unity_connection_pool.disconnect_all() logger.info("MCP for Unity Server shut down") # Initialize MCP server mcp = FastMCP( name="mcp-for-unity-server", lifespan=server_lifespan, instructions=""" This server provides tools to interact with the Unity Game Engine Editor. I have a dynamic tool system. Always check the mcpforunity://custom-tools resource first to see what special capabilities are available for the current project. Targeting Unity instances: - Use the resource mcpforunity://instances to list active Unity sessions (Name@hash). - When multiple instances are connected, call set_active_instance with the exact Name@hash before using tools/resources. The server will error if multiple are connected and no active instance is set. Important Workflows: Resources vs Tools: - Use RESOURCES to read editor state (editor_state, project_info, project_tags, tests, etc) - Use TOOLS to perform actions and mutations (manage_editor for play mode control, tag/layer management, etc) - Always check related resources before modifying the engine state with tools Script Management: - After creating or modifying scripts (by your own tools or the `manage_script` tool) use `read_console` to check for compilation errors before proceeding - Only after successful compilation can new components/types be used - You can poll the `editor_state` resource's `isCompiling` field to check if the domain reload is complete Scene Setup: - Always include a Camera and main Light (Directional Light) in new scenes - Create prefabs with `manage_asset` for reusable GameObjects - Use `manage_scene` to load, save, and query scene information Path Conventions: - Unless specified otherwise, all paths are relative to the project's `Assets/` folder - Use forward slashes (/) in paths for cross-platform compatibility Console Monitoring: - Check `read_console` regularly to catch errors, warnings, and compilation status - Filter by log type (Error, Warning, Log) to focus on specific issues Menu Items: - Use `execute_menu_item` when you have read the menu items resource - This lets you interact with Unity's menu system and third-party tools Payload sizing & paging (important): - Many Unity queries can return very large JSON. Prefer **paged + summary-first** calls. - `manage_scene(action="get_hierarchy")`: - Use `page_size` + `cursor` and follow `next_cursor` until null. - `page_size` is **items per page**; recommended starting point: **50**. - `manage_gameobject(action="get_components")`: - Start with `include_properties=false` (metadata-only) and small `page_size` (e.g. **10-25**). - Only request `include_properties=true` when needed; keep `page_size` small (e.g. **3-10**) to bound payloads. - `manage_asset(action="search")`: - Use paging (`page_size`, `page_number`) and keep `page_size` modest (e.g. **25-50**) to avoid token-heavy responses. - Keep `generate_preview=false` unless you explicitly need thumbnails (previews may include large base64 payloads). """ ) custom_tool_service = CustomToolService(mcp) @mcp.custom_route("/health", methods=["GET"]) async def health_http(_: Request) -> JSONResponse: return JSONResponse({ "status": "healthy", "timestamp": time.time(), "message": "MCP for Unity server is running" }) @mcp.custom_route("/plugin/sessions", methods=["GET"]) async def plugin_sessions_route(_: Request) -> JSONResponse: data = await PluginHub.get_sessions() return JSONResponse(data.model_dump()) # Initialize and register middleware for session-based Unity instance routing # Using the singleton getter ensures we use the same instance everywhere unity_middleware = get_unity_instance_middleware() mcp.add_middleware(unity_middleware) logger.info("Registered Unity instance middleware for session-based routing") # Mount plugin websocket hub at /hub/plugin when HTTP transport is active existing_routes = [ route for route in mcp._get_additional_http_routes() if isinstance(route, WebSocketRoute) and route.path == "/hub/plugin" ] if not existing_routes: mcp._additional_http_routes.append( WebSocketRoute("/hub/plugin", PluginHub)) # Register all tools register_all_tools(mcp) # Register all resources register_all_resources(mcp) def main(): """Entry point for uvx and console scripts.""" parser = argparse.ArgumentParser( description="MCP for Unity Server", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Environment Variables: UNITY_MCP_DEFAULT_INSTANCE Default Unity instance to target (project name, hash, or 'Name@hash') UNITY_MCP_SKIP_STARTUP_CONNECT Skip initial Unity connection attempt (set to 1/true/yes/on) UNITY_MCP_TELEMETRY_ENABLED Enable telemetry (set to 1/true/yes/on) UNITY_MCP_TRANSPORT Transport protocol: stdio or http (default: stdio) UNITY_MCP_HTTP_URL HTTP server URL (default: http://localhost:8080) UNITY_MCP_HTTP_HOST HTTP server host (overrides URL host) UNITY_MCP_HTTP_PORT HTTP server port (overrides URL port) Examples: # Use specific Unity project as default python -m src.server --default-instance "MyProject" # Start with HTTP transport python -m src.server --transport http --http-url http://localhost:8080 # Start with stdio transport (default) python -m src.server --transport stdio # Use environment variable for transport UNITY_MCP_TRANSPORT=http UNITY_MCP_HTTP_URL=http://localhost:9000 python -m src.server """ ) parser.add_argument( "--default-instance", type=str, metavar="INSTANCE", help="Default Unity instance to target (project name, hash, or 'Name@hash'). " "Overrides UNITY_MCP_DEFAULT_INSTANCE environment variable." ) parser.add_argument( "--transport", type=str, choices=["stdio", "http"], default="stdio", help="Transport protocol to use: stdio or http (default: stdio). " "Overrides UNITY_MCP_TRANSPORT environment variable." ) parser.add_argument( "--http-url", type=str, default="http://localhost:8080", metavar="URL", help="HTTP server URL (default: http://localhost:8080). " "Can also set via UNITY_MCP_HTTP_URL environment variable." ) parser.add_argument( "--http-host", type=str, default=None, metavar="HOST", help="HTTP server host (overrides URL host). " "Overrides UNITY_MCP_HTTP_HOST environment variable." ) parser.add_argument( "--http-port", type=int, default=None, metavar="PORT", help="HTTP server port (overrides URL port). " "Overrides UNITY_MCP_HTTP_PORT environment variable." ) parser.add_argument( "--unity-instance-token", type=str, default=None, metavar="TOKEN", help="Optional per-launch token set by Unity for deterministic lifecycle management. " "Used by Unity to validate it is stopping the correct process." ) parser.add_argument( "--pidfile", type=str, default=None, metavar="PATH", help="Optional path where the server will write its PID on startup. " "Used by Unity to stop the exact process it launched when running in a terminal." ) args = parser.parse_args() # Set environment variables from command line args if args.default_instance: os.environ["UNITY_MCP_DEFAULT_INSTANCE"] = args.default_instance logger.info( f"Using default Unity instance from command-line: {args.default_instance}") # Set transport mode transport_mode = args.transport or os.environ.get( "UNITY_MCP_TRANSPORT", "stdio") os.environ["UNITY_MCP_TRANSPORT"] = transport_mode logger.info(f"Transport mode: {transport_mode}") http_url = os.environ.get("UNITY_MCP_HTTP_URL", args.http_url) parsed_url = urlparse(http_url) # Allow individual host/port to override URL components http_host = args.http_host or os.environ.get( "UNITY_MCP_HTTP_HOST") or parsed_url.hostname or "localhost" http_port = args.http_port or (int(os.environ.get("UNITY_MCP_HTTP_PORT")) if os.environ.get( "UNITY_MCP_HTTP_PORT") else None) or parsed_url.port or 8080 os.environ["UNITY_MCP_HTTP_HOST"] = http_host os.environ["UNITY_MCP_HTTP_PORT"] = str(http_port) # Optional lifecycle handshake for Unity-managed terminal launches if args.unity_instance_token: os.environ["UNITY_MCP_INSTANCE_TOKEN"] = args.unity_instance_token if args.pidfile: try: pid_dir = os.path.dirname(args.pidfile) if pid_dir: os.makedirs(pid_dir, exist_ok=True) with open(args.pidfile, "w", encoding="ascii") as f: f.write(str(os.getpid())) except Exception as exc: logger.warning( "Failed to write pidfile '%s': %s", args.pidfile, exc) if args.http_url != "http://localhost:8080": logger.info(f"HTTP URL set to: {http_url}") if args.http_host: logger.info(f"HTTP host override: {http_host}") if args.http_port: logger.info(f"HTTP port override: {http_port}") # Determine transport mode if transport_mode == 'http': # Use HTTP transport for FastMCP transport = 'http' # Use the parsed host and port from URL/args http_url = os.environ.get("UNITY_MCP_HTTP_URL", args.http_url) parsed_url = urlparse(http_url) host = args.http_host or os.environ.get( "UNITY_MCP_HTTP_HOST") or parsed_url.hostname or "localhost" port = args.http_port or (int(os.environ.get("UNITY_MCP_HTTP_PORT")) if os.environ.get( "UNITY_MCP_HTTP_PORT") else None) or parsed_url.port or 8080 logger.info(f"Starting FastMCP with HTTP transport on {host}:{port}") mcp.run(transport=transport, host=host, port=port) else: # Use stdio transport for traditional MCP logger.info("Starting FastMCP with stdio transport") mcp.run(transport='stdio') # Run the server if __name__ == "__main__": main()