Replace session-global Unity instance selection with explicit routing

Server/src/main.py

@@ -15,10 +15,11 @@
     resolve_project_id_for_unity_instance,
 )
 from core.config import config
-from starlette.routing import WebSocketRoute
+from starlette.routing import Route, WebSocketRoute
 from starlette.responses import JSONResponse
 import argparse
 import asyncio
+import uvicorn
 
 # Fix to IPV4 Connection Issue #853
 # Will disable features in ProactorEventLoop including subprocess pipes and named pipes
@@ -304,8 +305,10 @@ def _build_instructions(project_scoped_tools: bool) -> str:
 
 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 to pin routing for the whole session. The server will error if multiple are connected and no active instance is set.
-- Alternatively, pass unity_instance as a parameter on any individual tool call to route just that call (e.g. unity_instance="MyGame@abc123", unity_instance="abc" for a hash prefix, or unity_instance="6401" for a port number in stdio mode). This does not change the session default.
+- The preferred flow is to pass unity_instance on any tool call or resource read to route that request explicitly (e.g. unity_instance="MyGame@abc123", unity_instance="abc" for a hash prefix, or unity_instance="6401" for a port number in stdio mode).
+- HTTP clients can bind a whole MCP session to one Unity editor by connecting to /mcp/instance/{{Name@hash}} instead of the unbound /mcp endpoint.
+- set_active_instance remains available as a compatibility path for simple single-agent workflows, but it should not be the default multi-agent HTTP pattern.
+- When multiple instances are connected and no explicit target or compatibility default is set, Unity-backed calls will return a selection error instead of guessing.
 
 Important Workflows:
 
@@ -366,6 +369,62 @@ def _normalize_instance_token(instance_token: str | None) -> tuple[str | None, s
     return None, instance_token
 
 
+def _build_bound_mcp_path(base_path: str) -> str:
+    normalized = "/" + base_path.strip("/")
+    if normalized == "//":
+        normalized = "/"
+    return f"{normalized.rstrip('/')}/instance/{{instance:path}}"
+
+
+def _attach_bound_mcp_route(app) -> str | None:
+    """Expose an HTTP MCP endpoint that binds a session to one Unity instance."""
+    base_path = getattr(app.state, "path", None)
+    if not isinstance(base_path, str) or not base_path:
+        return None
+
+    bound_path = _build_bound_mcp_path(base_path)
+    existing_paths = {
+        getattr(route, "path", None)
+        for route in app.router.routes
+    }
+    if bound_path in existing_paths:
+        return bound_path
+
+    base_route = next(
+        (
+            route for route in app.router.routes
+            if isinstance(route, Route) and route.path == base_path
+        ),
+        None,
+    )
+    if base_route is None:
+        return None
+
+    app.router.routes.append(
+        Route(
+            bound_path,
+            endpoint=base_route.endpoint,
+            methods=base_route.methods,
+            name=f"{base_route.name or 'mcp'}_bound_instance",
+            include_in_schema=False,
+        )
+    )
+    return bound_path
+
+
+def create_http_app(
+    mcp: FastMCP,
+    *,
+    transport: str = "http",
+    path: str | None = None,
+):
+    app = mcp.http_app(path=path, transport=transport)
+    bound_path = _attach_bound_mcp_route(app)
+    if bound_path:
+        logger.info("Registered instance-bound MCP HTTP endpoint at %s", bound_path)
+    return app
+
+
 def create_mcp_server(project_scoped_tools: bool) -> FastMCP:
     mcp = FastMCP(
         name="mcp-for-unity-server",
@@ -891,16 +950,21 @@ def main():
 
     # Determine transport mode
     if config.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 "127.0.0.1"
         port = args.http_port or _env_port 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)
+        app = create_http_app(mcp, transport="http")
+        uvicorn.run(
+            app,
+            host=host,
+            port=port,
+            timeout_graceful_shutdown=0,
+            lifespan="on",
+            ws="websockets-sansio",
+        )
     else:
         # Use stdio transport for traditional MCP
         logger.info("Starting FastMCP with stdio transport")

Server/src/services/custom_tool_service.py

@@ -374,7 +374,7 @@ async def _handler(ctx: Context, **kwargs) -> MCPResponse:
             if not unity_instance:
                 return MCPResponse(
                     success=False,
-                    message="No active Unity instance. Call set_active_instance with Name@hash from mcpforunity://instances.",
+                    message="No active Unity instance. Pass unity_instance explicitly or call set_active_instance with Name@hash from mcpforunity://instances.",
                 )
 
             project_id = resolve_project_id_for_unity_instance(unity_instance)

Server/src/services/registry/resource_registry.py

@@ -3,6 +3,11 @@
 """
 from typing import Callable, Any
 
+from .unity_targeting import (
+    add_optional_unity_instance_parameter,
+    append_unity_instance_query_template,
+)
+
 # Global registry to collect decorated resources
 _resource_registry: list[dict[str, Any]] = []
 
@@ -11,6 +16,7 @@ def mcp_for_unity_resource(
     uri: str,
     name: str | None = None,
     description: str | None = None,
+    unity_target: bool = True,
     **kwargs
 ) -> Callable:
     """
@@ -30,15 +36,21 @@ async def my_custom_resource(ctx: Context, ...):
     """
     def decorator(func: Callable) -> Callable:
         resource_name = name if name is not None else func.__name__
+        registered_func = func
+        registered_uri = uri
+        if unity_target:
+            registered_func = add_optional_unity_instance_parameter(func)
+            registered_uri = append_unity_instance_query_template(uri)
         _resource_registry.append({
-            'func': func,
-            'uri': uri,
+            'func': registered_func,
+            'uri': registered_uri,
             'name': resource_name,
             'description': description,
+            'unity_target': unity_target,
             'kwargs': kwargs
         })
 
-        return func
+        return registered_func
 
     return decorator
 

Server/src/services/registry/tool_registry.py

@@ -11,6 +11,8 @@
 """
 from typing import Callable, Any
 
+from .unity_targeting import add_optional_unity_instance_parameter
+
 # Global registry to collect decorated tools
 _tool_registry: list[dict[str, Any]] = []
 
@@ -94,16 +96,20 @@ def decorator(func: Callable) -> Callable:
                 "Expected None or a non-empty string."
             )
 
+        registered_func = func
+        if normalized_unity_target is not None:
+            registered_func = add_optional_unity_instance_parameter(func)
+
         _tool_registry.append({
-            'func': func,
+            'func': registered_func,
             'name': tool_name,
             'description': description,
             'unity_target': normalized_unity_target,
             'group': resolved_group,
             'kwargs': tool_kwargs,
         })
 
-        return func
+        return registered_func
 
     return decorator
 

Server/src/services/registry/unity_targeting.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import functools
+import inspect
+from typing import Annotated, Any, Callable
+
+UnityInstanceParameter = Annotated[
+    str | None,
+    "Target Unity instance (Name@hash, hash prefix, or port number in stdio mode).",
+]
+
+
+def add_optional_unity_instance_parameter(
+    func: Callable[..., Any],
+) -> Callable[..., Any]:
+    """Expose an optional unity_instance kwarg without forwarding it downstream."""
+    signature = inspect.signature(func)
+    if "unity_instance" in signature.parameters:
+        return func
+
+    parameters = list(signature.parameters.values())
+    unity_parameter = inspect.Parameter(
+        "unity_instance",
+        inspect.Parameter.KEYWORD_ONLY,
+        default=None,
+        annotation=UnityInstanceParameter,
+    )
+
+    insert_at = len(parameters)
+    for index, parameter in enumerate(parameters):
+        if parameter.kind == inspect.Parameter.VAR_KEYWORD:
+            insert_at = index
+            break
+    parameters.insert(insert_at, unity_parameter)
+
+    wrapped_signature = signature.replace(parameters=parameters)
+
+    if inspect.iscoroutinefunction(func):
+        @functools.wraps(func)
+        async def wrapper(*args, **kwargs):
+            kwargs.pop("unity_instance", None)
+            return await func(*args, **kwargs)
+    else:
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            kwargs.pop("unity_instance", None)
+            return func(*args, **kwargs)
+
+    wrapper.__signature__ = wrapped_signature
+    wrapper.__annotations__ = {
+        **getattr(func, "__annotations__", {}),
+        "unity_instance": UnityInstanceParameter,
+    }
+    return wrapper
+
+
+def append_unity_instance_query_template(uri: str) -> str:
+    """Append unity_instance to a resource URI template if it is not present."""
+    if "unity_instance" in uri:
+        return uri
+
+    if "{?" not in uri:
+        return f"{uri}{{?unity_instance}}"
+
+    prefix, _, suffix = uri.partition("{?")
+    query_suffix = suffix[:-1] if suffix.endswith("}") else suffix
+    query_names = [name for name in query_suffix.split(",") if name]
+    if "unity_instance" not in query_names:
+        query_names.append("unity_instance")
+    return f"{prefix}{{?{','.join(query_names)}}}"

Server/src/services/resources/custom_tools.py

@@ -32,7 +32,7 @@ async def get_custom_tools(ctx: Context) -> CustomToolsResourceResponse | MCPRes
     if not unity_instance:
         return MCPResponse(
             success=False,
-            message="No active Unity instance. Call set_active_instance with Name@hash from mcpforunity://instances.",
+            message="No active Unity instance. Pass unity_instance explicitly or call set_active_instance with Name@hash from mcpforunity://instances.",
         )
 
     project_id = resolve_project_id_for_unity_instance(unity_instance)

Server/src/services/resources/gameobject.py

@@ -42,6 +42,7 @@ def _validate_instance_id(instance_id: str) -> tuple[int | None, MCPResponse | N
 @mcp_for_unity_resource(
     uri="mcpforunity://scene/gameobject-api",
     name="gameobject_api",
+    unity_target=False,
     description="Documentation for GameObject resources. Use find_gameobjects tool to get instance IDs, then access resources below.\n\nURI: mcpforunity://scene/gameobject-api"
 )
 async def get_gameobject_api_docs(_ctx: Context) -> MCPResponse:

Server/src/services/resources/prefab.py

@@ -43,6 +43,7 @@ def _decode_prefab_path(encoded_path: str) -> str:
 @mcp_for_unity_resource(
     uri="mcpforunity://prefab-api",
     name="prefab_api",
+    unity_target=False,
     description="Documentation for Prefab resources. Use manage_asset action=search filterType=Prefab to find prefabs, then access resources below.\n\nURI: mcpforunity://prefab-api"
 )
 async def get_prefab_api_docs(_ctx: Context) -> MCPResponse:

Server/src/services/resources/tool_groups.py

@@ -18,6 +18,7 @@
 @mcp_for_unity_resource(
     uri="mcpforunity://tool-groups",
     name="tool_groups",
+    unity_target=False,
     description=(
         "Available tool groups and their tools. "
         "Use manage_tools to activate/deactivate groups per session.\n\n"

Server/src/services/resources/unity_instances.py

@@ -13,6 +13,7 @@
 @mcp_for_unity_resource(
     uri="mcpforunity://instances",
     name="unity_instances",
+    unity_target=False,
     description="Lists all running Unity Editor instances with their details.\n\nURI: mcpforunity://instances"
 )
 async def unity_instances(ctx: Context) -> dict[str, Any]:

Server/src/services/tools/execute_custom_tool.py

@@ -26,7 +26,7 @@ async def execute_custom_tool(ctx: Context, tool_name: str, parameters: dict | N
     if not unity_instance:
         return MCPResponse(
             success=False,
-            message="No active Unity instance. Call set_active_instance with Name@hash from mcpforunity://instances.",
+            message="No active Unity instance. Pass unity_instance explicitly or call set_active_instance with Name@hash from mcpforunity://instances.",
         )
 
     project_id = resolve_project_id_for_unity_instance(unity_instance)

Server/src/services/tools/set_active_instance.py

@@ -23,6 +23,17 @@ async def set_active_instance(
         ctx: Context,
         instance: Annotated[str, "Target instance (Name@hash, hash prefix, or port number in stdio mode)"]
 ) -> dict[str, Any]:
+    get_state = getattr(ctx, "get_state", None)
+    bound_instance = await get_state("bound_unity_instance") if callable(get_state) else None
+    if bound_instance:
+        return {
+            "success": False,
+            "error": (
+                "set_active_instance is not available on an instance-bound MCP endpoint. "
+                "Use the bound endpoint target or switch back to the unbound /mcp endpoint."
+            ),
+        }
+
     transport = (config.transport_mode or "stdio").lower()
 
     # Port number shorthand (stdio only) — resolve to Name@hash via pool discovery

Server/src/transport/legacy/unity_connection.py

@@ -549,16 +549,27 @@ def _resolve_instance_id(self, instance_identifier: str | None, instances: list[
                 instance_identifier = self._default_instance_id
                 logger.debug(f"Using default instance: {instance_identifier}")
             else:
-                # Use the most recently active instance
-                # Instances with no heartbeat (None) should be sorted last (use 0 as sentinel)
-                sorted_instances = sorted(
-                    instances,
-                    key=lambda inst: inst.last_heartbeat.timestamp() if inst.last_heartbeat else 0.0,
-                    reverse=True,
+                if len(instances) == 1:
+                    logger.info(
+                        "No instance specified, auto-selecting sole instance: %s",
+                        instances[0].id,
+                    )
+                    return instances[0]
+
+                suggestions = [
+                    {
+                        "id": inst.id,
+                        "path": inst.path,
+                        "port": inst.port,
+                        "suggest": f"Use unity_instance='{inst.id}'",
+                    }
+                    for inst in instances
+                ]
+                raise ConnectionError(
+                    "Multiple Unity Editor instances are running. "
+                    "Pass unity_instance explicitly or call set_active_instance first. "
+                    f"Available instances: {suggestions}"
                 )
-                logger.info(
-                    f"No instance specified, using most recent: {sorted_instances[0].id}")
-                return sorted_instances[0]
 
         identifier = instance_identifier.strip()
 

Server/src/transport/plugin_hub.py

@@ -76,11 +76,11 @@ class InstanceSelectionRequiredError(RuntimeError):
 
     _SELECTION_REQUIRED = (
         "Unity instance selection is required. "
-        "Call set_active_instance with Name@hash from mcpforunity://instances."
+        "Pass unity_instance explicitly or call set_active_instance with Name@hash from mcpforunity://instances."
     )
     _MULTIPLE_INSTANCES = (
         "Multiple Unity instances are connected. "
-        "Call set_active_instance with Name@hash from mcpforunity://instances."
+        "Pass unity_instance explicitly or call set_active_instance with Name@hash from mcpforunity://instances."
     )
 
     def __init__(self, message: str | None = None):