add instance params for resources

Author: sakurachan

MCPForUnity/UnityMcpServer~/src/pyproject.toml

@@ -6,7 +6,7 @@ readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
  "httpx>=0.27.2",
- "fastmcp>=2.12.5",
+ "fastmcp>=2.13.0",
  "mcp>=1.16.0",
  "pydantic>=2.12.0",
  "tomli>=2.3.0",

MCPForUnity/UnityMcpServer~/src/resources/__init__.py

@@ -2,7 +2,9 @@
 MCP Resources package - Auto-discovers and registers all resources in this directory.
 """
 import logging
+import inspect
 from pathlib import Path
+from typing import get_type_hints
 
 from fastmcp import FastMCP
 from telemetry_decorator import telemetry_resource
@@ -16,6 +18,23 @@
 __all__ = ['register_all_resources']
 
 
+def _create_fixed_wrapper(original_func, has_other_params):
+    """
+    Factory function to create a wrapper that calls original_func with unity_instance=None.
+    This avoids closure issues in loops.
+    """
+    if has_other_params:
+        # Has other parameters (like mode)
+        async def fixed_wrapper(*args, **kwargs):
+            return await original_func(*args, **kwargs, unity_instance=None)
+    else:
+        # No other parameters, just unity_instance
+        async def fixed_wrapper():
+            return await original_func(unity_instance=None)
+
+    return fixed_wrapper
+
+
 def register_all_resources(mcp: FastMCP):
     """
     Auto-discover and register all resources in the resources/ directory.
@@ -36,18 +55,76 @@ def register_all_resources(mcp: FastMCP):
         logger.warning("No MCP resources registered!")
         return
 
+    registered_count = 0
     for resource_info in resources:
         func = resource_info['func']
         uri = resource_info['uri']
         resource_name = resource_info['name']
         description = resource_info['description']
         kwargs = resource_info['kwargs']
 
-        # Apply the @mcp.resource decorator and telemetry
-        wrapped = telemetry_resource(resource_name)(func)
-        wrapped = mcp.resource(uri=uri, name=resource_name,
-                               description=description, **kwargs)(wrapped)
-        resource_info['func'] = wrapped
-        logger.debug(f"Registered resource: {resource_name} - {description}")
+        # Check if URI contains query parameters (e.g., {?unity_instance})
+        has_query_params = '{?' in uri
+
+        if has_query_params:
+            # Register two versions for backward compatibility:
+            # 1. Template version with query parameters (for multi-instance)
+            wrapped_template = telemetry_resource(resource_name)(func)
+            wrapped_template = mcp.resource(uri=uri, name=resource_name,
+                                           description=description, **kwargs)(wrapped_template)
+            logger.debug(f"Registered resource template: {resource_name} - {uri}")
+            registered_count += 1
+
+            # 2. Fixed version without query parameters (for single-instance/default)
+            # Remove query parameters from URI
+            fixed_uri = uri.split('{?')[0]
+            fixed_name = f"{resource_name}_default"
+            fixed_description = f"{description} (default instance)"
+
+            # Create a wrapper function that doesn't accept unity_instance parameter
+            # This wrapper will call the original function with unity_instance=None
+            sig = inspect.signature(func)
+            params = list(sig.parameters.values())
+
+            # Filter out unity_instance parameter
+            fixed_params = [p for p in params if p.name != 'unity_instance']
+
+            # Create wrapper using factory function to avoid closure issues
+            has_other_params = len(fixed_params) > 0
+            fixed_wrapper = _create_fixed_wrapper(func, has_other_params)
+
+            # Update signature to match filtered parameters
+            if has_other_params:
+                fixed_wrapper.__signature__ = sig.replace(parameters=fixed_params)
+                fixed_wrapper.__annotations__ = {
+                    k: v for k, v in func.__annotations__.items()
+                    if k != 'unity_instance'
+                }
+            else:
+                fixed_wrapper.__signature__ = inspect.Signature(parameters=[])
+                fixed_wrapper.__annotations__ = {
+                    k: v for k, v in func.__annotations__.items()
+                    if k == 'return'
+                }
+
+            # Preserve function metadata
+            fixed_wrapper.__name__ = fixed_name
+            fixed_wrapper.__doc__ = func.__doc__
+
+            wrapped_fixed = telemetry_resource(fixed_name)(fixed_wrapper)
+            wrapped_fixed = mcp.resource(uri=fixed_uri, name=fixed_name,
+                                        description=fixed_description, **kwargs)(wrapped_fixed)
+            logger.debug(f"Registered resource (fixed): {fixed_name} - {fixed_uri}")
+            registered_count += 1
+
+            resource_info['func'] = wrapped_template
+        else:
+            # No query parameters, register as-is
+            wrapped = telemetry_resource(resource_name)(func)
+            wrapped = mcp.resource(uri=uri, name=resource_name,
+                                   description=description, **kwargs)(wrapped)
+            resource_info['func'] = wrapped
+            logger.debug(f"Registered resource: {resource_name} - {description}")
+            registered_count += 1
 
-    logger.info(f"Registered {len(resources)} MCP resources")
+    logger.info(f"Registered {registered_count} MCP resources ({len(resources)} unique)")

MCPForUnity/UnityMcpServer~/src/resources/menu_items.py

@@ -8,18 +8,21 @@ class GetMenuItemsResponse(MCPResponse):
 
 
 @mcp_for_unity_resource(
-    uri="mcpforunity://menu-items",
+    uri="mcpforunity://menu-items{?unity_instance}",
     name="get_menu_items",
     description="Provides a list of all menu items."
 )
-async def get_menu_items() -> GetMenuItemsResponse:
-    """Provides a list of all menu items."""
-    # Later versions of FastMCP support these as query parameters
-    # See: https://gofastmcp.com/servers/resources#query-parameters
+async def get_menu_items(unity_instance: str | None = None) -> GetMenuItemsResponse:
+    """Provides a list of all menu items.
+
+    Args:
+        unity_instance: Target Unity instance (project name, hash, or 'Name@hash').
+                       If not specified, uses default instance.
+    """
     params = {
         "refresh": True,
         "search": "",
     }
 
-    response = await async_send_command_with_retry("get_menu_items", params)
+    response = await async_send_command_with_retry("get_menu_items", params, instance_id=unity_instance)
     return GetMenuItemsResponse(**response) if isinstance(response, dict) else response

MCPForUnity/UnityMcpServer~/src/resources/tests.py

@@ -17,15 +17,29 @@ class GetTestsResponse(MCPResponse):
     data: list[TestItem] = []
 
 
-@mcp_for_unity_resource(uri="mcpforunity://tests", name="get_tests", description="Provides a list of all tests.")
-async def get_tests() -> GetTestsResponse:
-    """Provides a list of all tests."""
-    response = await async_send_command_with_retry("get_tests", {})
+@mcp_for_unity_resource(uri="mcpforunity://tests{?unity_instance}", name="get_tests", description="Provides a list of all tests.")
+async def get_tests(unity_instance: str | None = None) -> GetTestsResponse:
+    """Provides a list of all tests.
+
+    Args:
+        unity_instance: Target Unity instance (project name, hash, or 'Name@hash').
+                       If not specified, uses default instance.
+    """
+    response = await async_send_command_with_retry("get_tests", {}, instance_id=unity_instance)
     return GetTestsResponse(**response) if isinstance(response, dict) else response
 
 
-@mcp_for_unity_resource(uri="mcpforunity://tests/{mode}", name="get_tests_for_mode", description="Provides a list of tests for a specific mode.")
-async def get_tests_for_mode(mode: Annotated[Literal["EditMode", "PlayMode"], Field(description="The mode to filter tests by.")]) -> GetTestsResponse:
-    """Provides a list of tests for a specific mode."""
-    response = await async_send_command_with_retry("get_tests_for_mode", {"mode": mode})
+@mcp_for_unity_resource(uri="mcpforunity://tests/{mode}{?unity_instance}", name="get_tests_for_mode", description="Provides a list of tests for a specific mode.")
+async def get_tests_for_mode(
+    mode: Annotated[Literal["EditMode", "PlayMode"], Field(description="The mode to filter tests by.")],
+    unity_instance: str | None = None
+) -> GetTestsResponse:
+    """Provides a list of tests for a specific mode.
+
+    Args:
+        mode: The test mode to filter by (EditMode or PlayMode).
+        unity_instance: Target Unity instance (project name, hash, or 'Name@hash').
+                       If not specified, uses default instance.
+    """
+    response = await async_send_command_with_retry("get_tests_for_mode", {"mode": mode}, instance_id=unity_instance)
     return GetTestsResponse(**response) if isinstance(response, dict) else response