PHXH
/
unity-mcp
51
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

server: centralize reload-aware retries and single-source retry_after_ms via config; increase default retry window (40 x 250ms); preserve structured reloading failures

main
David Sarno 2025-08-10 22:49:24 -07:00
parent b179ce1ed8
commit 1938756844
10 changed files with 104 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
UnityMcpBridge/UnityMcpServer~/src/config.py
View File

@ -25,6 +25,11 @@ class ServerConfig:
# Server settings
max_retries: int = 10
retry_delay: float = 0.25
# Backoff hint returned to clients when Unity is reloading (milliseconds)
reload_retry_ms: int = 250
# Number of polite retries when Unity reports reloading
# 40 × 250ms ≈ 10s default window
reload_max_retries: int = 40
# Create a global config instance
config = ServerConfig()

18
UnityMcpBridge/UnityMcpServer~/src/tools/execute_menu_item.py
View File

@ -3,7 +3,8 @@ Defines the execute_menu_item tool for running Unity Editor menu commands.
"""
from typing import Dict, Any
from mcp.server.fastmcp import FastMCP, Context
from unity_connection import get_unity_connection # Import unity_connection module
from unity_connection import get_unity_connection, send_command_with_retry # Import retry helper
from config import config
import time
def register_execute_menu_item_tools(mcp: FastMCP):
@ -43,15 +44,6 @@ def register_execute_menu_item_tools(mcp: FastMCP):
if "parameters" not in params_dict:
params_dict["parameters"] = {} # Ensure parameters dict exists
# Get Unity connection and send the command
# We use the unity_connection module to communicate with Unity
unity_conn = get_unity_connection()
# Send command to the ExecuteMenuItem C# handler
# The command type should match what the Unity side expects
resp = unity_conn.send_command("execute_menu_item", params_dict)
if isinstance(resp, dict) and not resp.get("success", True) and resp.get("state") == "reloading":
delay_ms = int(resp.get("retry_after_ms", 250))
time.sleep(max(0.0, delay_ms / 1000.0))
resp = unity_conn.send_command("execute_menu_item", params_dict)
return resp
# Use centralized retry helper
resp = send_command_with_retry("execute_menu_item", params_dict)
return resp if isinstance(resp, dict) else {"success": False, "message": str(resp)}

24
UnityMcpBridge/UnityMcpServer~/src/tools/manage_asset.py
View File

@ -5,7 +5,8 @@ import asyncio # Added: Import asyncio for running sync code in async
from typing import Dict, Any
from mcp.server.fastmcp import FastMCP, Context
# from ..unity_connection import get_unity_connection # Original line that caused error
from unity_connection import get_unity_connection # Use absolute import relative to Python dir
from unity_connection import get_unity_connection, async_send_command_with_retry # Use centralized retry helper
from config import config
import time
def register_manage_asset_tools(mcp: FastMCP):
@ -72,22 +73,7 @@ def register_manage_asset_tools(mcp: FastMCP):
# Get the Unity connection instance
connection = get_unity_connection()
# Run the synchronous send_command in the default executor (thread pool)
# This prevents blocking the main async event loop.
result = await loop.run_in_executor(
None, # Use default executor
connection.send_command, # The function to call
"manage_asset", # First argument for send_command
params_dict # Second argument for send_command
)
if isinstance(result, dict) and not result.get("success", True) and result.get("state") == "reloading":
delay_ms = int(result.get("retry_after_ms", 250))
await asyncio.sleep(max(0.0, delay_ms / 1000.0))
result = await loop.run_in_executor(
None,
connection.send_command,
"manage_asset",
params_dict
)
# Use centralized async retry helper to avoid blocking the event loop
result = await async_send_command_with_retry("manage_asset", params_dict, loop=loop)
# Return the result obtained from Unity
return result
return result if isinstance(result, dict) else {"success": False, "message": str(result)}

18
UnityMcpBridge/UnityMcpServer~/src/tools/manage_editor.py
View File

@ -1,7 +1,8 @@
from mcp.server.fastmcp import FastMCP, Context
import time
from typing import Dict, Any
from unity_connection import get_unity_connection
from unity_connection import get_unity_connection, send_command_with_retry
from config import config
def register_manage_editor_tools(mcp: FastMCP):
"""Register all editor management tools with the MCP server."""
@ -41,18 +42,13 @@ def register_manage_editor_tools(mcp: FastMCP):
}
params = {k: v for k, v in params.items() if v is not None}
# Send command to Unity (with a single polite retry if reloading)
response = get_unity_connection().send_command("manage_editor", params)
if isinstance(response, dict) and not response.get("success", True) and response.get("state") == "reloading":
delay_ms = int(response.get("retry_after_ms", 250))
time.sleep(max(0.0, delay_ms / 1000.0))
response = get_unity_connection().send_command("manage_editor", params)
# Send command using centralized retry helper
response = send_command_with_retry("manage_editor", params)
# Process response
if response.get("success"):
# Preserve structured failure data; unwrap success into a friendlier shape
if isinstance(response, dict) and response.get("success"):
return {"success": True, "message": response.get("message", "Editor operation successful."), "data": response.get("data")}
else:
return {"success": False, "message": response.get("error", "An unknown error occurred during editor management.")}
return response if isinstance(response, dict) else {"success": False, "message": str(response)}
except Exception as e:
return {"success": False, "message": f"Python error managing editor: {str(e)}"}

18
UnityMcpBridge/UnityMcpServer~/src/tools/manage_gameobject.py
View File

@ -1,6 +1,7 @@
from mcp.server.fastmcp import FastMCP, Context
from typing import Dict, Any, List
from unity_connection import get_unity_connection
from unity_connection import get_unity_connection, send_command_with_retry
from config import config
import time
def register_manage_gameobject_tools(mcp: FastMCP):
@ -123,21 +124,14 @@ def register_manage_gameobject_tools(mcp: FastMCP):
params.pop("prefab_folder", None)
# --------------------------------
# Send the command to Unity via the established connection
# Use the get_unity_connection function to retrieve the active connection instance
# Changed "MANAGE_GAMEOBJECT" to "manage_gameobject" to potentially match Unity expectation
response = get_unity_connection().send_command("manage_gameobject", params)
if isinstance(response, dict) and not response.get("success", True) and response.get("state") == "reloading":
delay_ms = int(response.get("retry_after_ms", 250))
time.sleep(max(0.0, delay_ms / 1000.0))
response = get_unity_connection().send_command("manage_gameobject", params)
# Use centralized retry helper
response = send_command_with_retry("manage_gameobject", params)
# Check if the response indicates success
# If the response is not successful, raise an exception with the error message
if response.get("success"):
if isinstance(response, dict) and response.get("success"):
return {"success": True, "message": response.get("message", "GameObject operation successful."), "data": response.get("data")}
else:
return {"success": False, "message": response.get("error", "An unknown error occurred during GameObject management.")}
return response if isinstance(response, dict) else {"success": False, "message": str(response)}
except Exception as e:
return {"success": False, "message": f"Python error managing GameObject: {str(e)}"}

18
UnityMcpBridge/UnityMcpServer~/src/tools/manage_scene.py
View File

@ -1,6 +1,7 @@
from mcp.server.fastmcp import FastMCP, Context
from typing import Dict, Any
from unity_connection import get_unity_connection
from unity_connection import get_unity_connection, send_command_with_retry
from config import config
import time
def register_manage_scene_tools(mcp: FastMCP):
@ -35,18 +36,13 @@ def register_manage_scene_tools(mcp: FastMCP):
}
params = {k: v for k, v in params.items() if v is not None}
# Send command to Unity (with a single polite retry if reloading)
response = get_unity_connection().send_command("manage_scene", params)
if isinstance(response, dict) and not response.get("success", True) and response.get("state") == "reloading":
delay_ms = int(response.get("retry_after_ms", 250))
time.sleep(max(0.0, delay_ms / 1000.0))
response = get_unity_connection().send_command("manage_scene", params)
# Use centralized retry helper
response = send_command_with_retry("manage_scene", params)
# Process response
if response.get("success"):
# Preserve structured failure data; unwrap success into a friendlier shape
if isinstance(response, dict) and response.get("success"):
return {"success": True, "message": response.get("message", "Scene operation successful."), "data": response.get("data")}
else:
return {"success": False, "message": response.get("error", "An unknown error occurred during scene management.")}
return response if isinstance(response, dict) else {"success": False, "message": str(response)}
except Exception as e:
return {"success": False, "message": f"Python error managing scene: {str(e)}"}

16
UnityMcpBridge/UnityMcpServer~/src/tools/manage_script.py
View File

@ -1,6 +1,7 @@
from mcp.server.fastmcp import FastMCP, Context
from typing import Dict, Any
from unity_connection import get_unity_connection
from unity_connection import get_unity_connection, send_command_with_retry
from config import config
import time
import os
import base64
@ -54,15 +55,11 @@ def register_manage_script_tools(mcp: FastMCP):
# Remove None values so they don't get sent as null
params = {k: v for k, v in params.items() if v is not None}
# Send command to Unity (with single polite retry if reloading)
response = get_unity_connection().send_command("manage_script", params)
if isinstance(response, dict) and not response.get("success", True) and response.get("state") == "reloading":
delay_ms = int(response.get("retry_after_ms", 250))
time.sleep(max(0.0, delay_ms / 1000.0))
response = get_unity_connection().send_command("manage_script", params)
# Send command via centralized retry helper
response = send_command_with_retry("manage_script", params)
# Process response from Unity
if response.get("success"):
if isinstance(response, dict) and response.get("success"):
# If the response contains base64 encoded content, decode it
if response.get("data", {}).get("contentsEncoded"):
decoded_contents = base64.b64decode(response["data"]["encodedContents"]).decode('utf-8')
@ -71,8 +68,7 @@ def register_manage_script_tools(mcp: FastMCP):
del response["data"]["contentsEncoded"]
return {"success": True, "message": response.get("message", "Operation successful."), "data": response.get("data")}
else:
return {"success": False, "message": response.get("error", "An unknown error occurred.")}
return response if isinstance(response, dict) else {"success": False, "message": str(response)}
except Exception as e:
# Handle Python-side errors (e.g., connection issues)

16
UnityMcpBridge/UnityMcpServer~/src/tools/manage_shader.py
View File

@ -1,6 +1,7 @@
from mcp.server.fastmcp import FastMCP, Context
from typing import Dict, Any
from unity_connection import get_unity_connection
from unity_connection import get_unity_connection, send_command_with_retry
from config import config
import time
import os
import base64
@ -47,15 +48,11 @@ def register_manage_shader_tools(mcp: FastMCP):
# Remove None values so they don't get sent as null
params = {k: v for k, v in params.items() if v is not None}
# Send command to Unity
response = get_unity_connection().send_command("manage_shader", params)
if isinstance(response, dict) and not response.get("success", True) and response.get("state") == "reloading":
delay_ms = int(response.get("retry_after_ms", 250))
time.sleep(max(0.0, delay_ms / 1000.0))
response = get_unity_connection().send_command("manage_shader", params)
# Send command via centralized retry helper
response = send_command_with_retry("manage_shader", params)
# Process response from Unity
if response.get("success"):
if isinstance(response, dict) and response.get("success"):
# If the response contains base64 encoded content, decode it
if response.get("data", {}).get("contentsEncoded"):
decoded_contents = base64.b64decode(response["data"]["encodedContents"]).decode('utf-8')
@ -64,8 +61,7 @@ def register_manage_shader_tools(mcp: FastMCP):
del response["data"]["contentsEncoded"]
return {"success": True, "message": response.get("message", "Operation successful."), "data": response.get("data")}
else:
return {"success": False, "message": response.get("error", "An unknown error occurred.")}
return response if isinstance(response, dict) else {"success": False, "message": str(response)}
except Exception as e:
# Handle Python-side errors (e.g., connection issues)

13
UnityMcpBridge/UnityMcpServer~/src/tools/read_console.py
View File

@ -4,7 +4,8 @@ Defines the read_console tool for accessing Unity Editor console messages.
from typing import List, Dict, Any
import time
from mcp.server.fastmcp import FastMCP, Context
from unity_connection import get_unity_connection
from unity_connection import get_unity_connection, send_command_with_retry
from config import config
def register_read_console_tools(mcp: FastMCP):
"""Registers the read_console tool with the MCP server."""
@ -67,10 +68,6 @@ def register_read_console_tools(mcp: FastMCP):
if 'count' not in params_dict:
params_dict['count'] = None
# Forward the command using the bridge's send_command method (with a single polite retry on reload)
resp = bridge.send_command("read_console", params_dict)
if isinstance(resp, dict) and not resp.get("success", True) and resp.get("state") == "reloading":
delay_ms = int(resp.get("retry_after_ms", 250))
time.sleep(max(0.0, delay_ms / 1000.0))
resp = bridge.send_command("read_console", params_dict)
return resp
# Use centralized retry helper
resp = send_command_with_retry("read_console", params_dict)
return resp if isinstance(resp, dict) else {"success": False, "message": str(resp)}

53
UnityMcpBridge/UnityMcpServer~/src/unity_connection.py
View File

@ -139,7 +139,7 @@ class UnityConnection:
return {
"success": False,
"state": "reloading",
"retry_after_ms": int(250),
"retry_after_ms": int(config.reload_retry_ms),
"error": "Unity domain reload in progress",
"message": "Unity is reloading scripts; please retry shortly"
}
@ -278,3 +278,54 @@ def get_unity_connection() -> UnityConnection:
pass
_unity_connection = None
raise ConnectionError(f"Could not establish valid Unity connection: {str(e)}")
# -----------------------------
# Centralized retry helpers
# -----------------------------
def _is_reloading_response(resp: dict) -> bool:
"""Return True if the Unity response indicates the editor is reloading."""
if not isinstance(resp, dict):
return False
if resp.get("state") == "reloading":
return True
message_text = (resp.get("message") or resp.get("error") or "").lower()
return "reload" in message_text
def send_command_with_retry(command_type: str, params: Dict[str, Any], *, max_retries: int | None = None, retry_ms: int | None = None) -> Dict[str, Any]:
"""Send a command via the shared connection, waiting politely through Unity reloads.
Uses config.reload_retry_ms and config.reload_max_retries by default. Preserves the
structured failure if retries are exhausted.
"""
conn = get_unity_connection()
if max_retries is None:
max_retries = getattr(config, "reload_max_retries", 40)
if retry_ms is None:
retry_ms = getattr(config, "reload_retry_ms", 250)
response = conn.send_command(command_type, params)
retries = 0
while _is_reloading_response(response) and retries < max_retries:
delay_ms = int(response.get("retry_after_ms", retry_ms)) if isinstance(response, dict) else retry_ms
time.sleep(max(0.0, delay_ms / 1000.0))
retries += 1
response = conn.send_command(command_type, params)
return response
async def async_send_command_with_retry(command_type: str, params: Dict[str, Any], *, loop=None, max_retries: int | None = None, retry_ms: int | None = None) -> Dict[str, Any]:
"""Async wrapper that runs the blocking retry helper in a thread pool."""
try:
import asyncio # local import to avoid mandatory asyncio dependency for sync callers
if loop is None:
loop = asyncio.get_running_loop()
return await loop.run_in_executor(
None,
lambda: send_command_with_retry(command_type, params, max_retries=max_retries, retry_ms=retry_ms),
)
except Exception as e:
# Return a structured error dict for consistency with other responses
return {"success": False, "error": f"Python async retry helper failed: {str(e)}"}