from typing import Annotated, Literal, Any
from fastmcp import Context
from services.registry import mcp_for_unity_tool
from services.tools import get_unity_instance_from_context
from transport.unity_transport import send_with_unity_instance
from transport.legacy.unity_connection import async_send_command_with_retry
@mcp_for_unity_tool(
description="Performs CRUD operations on Unity scenes."
)
async def manage_scene(
ctx: Context,
action: Annotated[Literal[
"create",
"load",
"save",
"get_hierarchy",
"get_active",
"get_build_settings",
"screenshot",
], "Perform CRUD operations on Unity scenes, and capture a screenshot."],
name: Annotated[str, "Scene name."] | None = None,
path: Annotated[str, "Scene path."] | None = None,
build_index: Annotated[int | str,
"Unity build index (quote as string, e.g., '0')."] | None = None,
screenshot_file_name: Annotated[str, "Screenshot file name (optional). Defaults to timestamp when omitted."] | None = None,
screenshot_super_size: Annotated[int | str, "Screenshot supersize multiplier (integer ≥1). Optional." ] | None = None,
) -> dict[str, Any]:
# Get active instance from session state
# Removed session_state import
unity_instance = get_unity_instance_from_context(ctx)
try:
# Coerce numeric inputs defensively
def _coerce_int(value, default=None):
if value is None:
return default
try:
if isinstance(value, bool):
return default
if isinstance(value, int):
return int(value)
s = str(value).strip()
if s.lower() in ("", "none", "null"):
return default
return int(float(s))
except Exception:
return default
coerced_build_index = _coerce_int(build_index, default=None)
coerced_super_size = _coerce_int(screenshot_super_size, default=None)
params: dict[str, Any] = {"action": action}
if name:
params["name"] = name
if path:
params["path"] = path
if coerced_build_index is not None:
params["buildIndex"] = coerced_build_index
if screenshot_file_name:
params["fileName"] = screenshot_file_name
if coerced_super_size is not None:
params["superSize"] = coerced_super_size
# Use centralized retry helper with instance routing
response = await send_with_unity_instance(async_send_command_with_retry, unity_instance, "manage_scene", params)
# 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")}
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)}"}