feat: improve telemetry and parameter validation

- Add server-side integer coercion for numeric parameters in all tools
- Fix parameter type validation issues (read_resource, find_in_file, read_console, manage_scene, manage_asset)
- Add proper tool descriptions with ctx parameter documentation
- Fix Context type annotations (use Context instead of Any for ctx)
- All tools now accept flexible numeric inputs (strings, floats) and coerce to integers
- Telemetry system working with all tool_execution events captured in BigQuery
- Remove invalid parameter type warnings from client-side validation

Author: dsarno

UnityMcpBridge/UnityMcpServer~/src/tools/manage_asset.py

@@ -27,8 +27,8 @@ async def manage_asset(
         search_pattern: str = None,
         filter_type: str = None,
         filter_date_after: str = None,
-        page_size: int = None,
-        page_number: int = None
+        page_size: Any = None,
+        page_number: Any = None
     ) -> Dict[str, Any]:
         """Performs asset operations (import, create, modify, delete, etc.) in Unity.
 
@@ -53,6 +53,25 @@ async def manage_asset(
         if properties is None:
             properties = {}
 
+        # 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
+
+        page_size = _coerce_int(page_size)
+        page_number = _coerce_int(page_number)
+
         # Prepare parameters for the C# handler
         params_dict = {
             "action": action.lower(),

UnityMcpBridge/UnityMcpServer~/src/tools/manage_editor.py

@@ -10,27 +10,28 @@
 def register_manage_editor_tools(mcp: FastMCP):
     """Register all editor management tools with the MCP server."""
 
-    @mcp.tool()
+    @mcp.tool(description=(
+        "Controls and queries the Unity editor's state and settings.\n\n"
+        "Args:\n"
+        "- ctx: Context object (required)\n"
+        "- action: Operation (e.g., 'play', 'pause', 'get_state', 'set_active_tool', 'add_tag')\n"
+        "- wait_for_completion: Optional. If True, waits for certain actions\n"
+        "- tool_name: Tool name for specific actions\n"
+        "- tag_name: Tag name for specific actions\n"
+        "- layer_name: Layer name for specific actions\n\n"
+        "Returns:\n"
+        "Dictionary with operation results ('success', 'message', 'data')."
+    ))
     @telemetry_tool("manage_editor")
     def manage_editor(
-        ctx: Any,
+        ctx: Context,
         action: str,
         wait_for_completion: bool = None,
         # --- Parameters for specific actions ---
         tool_name: str = None, 
         tag_name: str = None,
         layer_name: str = None,
     ) -> Dict[str, Any]:
-        """Controls and queries the Unity editor's state and settings.
-
-        Args:
-            action: Operation (e.g., 'play', 'pause', 'get_state', 'set_active_tool', 'add_tag').
-            wait_for_completion: Optional. If True, waits for certain actions.
-            Action-specific arguments (e.g., tool_name, tag_name, layer_name).
-
-        Returns:
-            Dictionary with operation results ('success', 'message', 'data').
-        """
         try:
             # Diagnostics: quick telemetry checks
             if action == "telemetry_status":

UnityMcpBridge/UnityMcpServer~/src/tools/manage_scene.py

@@ -16,7 +16,7 @@ def manage_scene(
         action: str,
         name: str,
         path: str,
-        build_index: int,
+        build_index: Any,
     ) -> Dict[str, Any]:
         """Manages Unity scenes (load, save, create, get hierarchy, etc.).
 
@@ -31,6 +31,24 @@ def manage_scene(
             Dictionary with results ('success', 'message', 'data').
         """
         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
+
+            build_index = _coerce_int(build_index, default=0)
+
             params = {
                 "action": action,
                 "name": name,

UnityMcpBridge/UnityMcpServer~/src/tools/manage_script.py

@@ -5,8 +5,16 @@
 import os
 from urllib.parse import urlparse, unquote
 
-from telemetry_decorator import telemetry_tool
-from telemetry import record_milestone, MilestoneType
+try:
+    from telemetry_decorator import telemetry_tool
+    from telemetry import record_milestone, MilestoneType
+    HAS_TELEMETRY = True
+except ImportError:
+    HAS_TELEMETRY = False
+    def telemetry_tool(tool_name: str):
+        def decorator(func):
+            return func
+        return decorator
 
 def register_manage_script_tools(mcp: FastMCP):
     """Register all script management tools with the MCP server."""
@@ -84,7 +92,7 @@ def _split_uri(uri: str) -> tuple[str, str]:
     ))
     @telemetry_tool("apply_text_edits")
     def apply_text_edits(
-        ctx: Any,
+        ctx: Context,
         uri: str,
         edits: List[Dict[str, Any]],
         precondition_sha256: str | None = None,
@@ -351,7 +359,7 @@ def _flip_async():
     ))
     @telemetry_tool("create_script")
     def create_script(
-        ctx: Any,
+        ctx: Context,
         path: str,
         contents: str = "",
         script_type: str | None = None,
@@ -390,7 +398,7 @@ def create_script(
         "Rules: Target must resolve under Assets/.\n"
     ))
     @telemetry_tool("delete_script")
-    def delete_script(ctx: Any, uri: str) -> Dict[str, Any]:
+    def delete_script(ctx: Context, uri: str) -> Dict[str, Any]:
         """Delete a C# script by URI."""
         name, directory = _split_uri(uri)
         if not directory or directory.split("/")[0].lower() != "assets":
@@ -407,7 +415,7 @@ def delete_script(ctx: Any, uri: str) -> Dict[str, Any]:
     ))
     @telemetry_tool("validate_script")
     def validate_script(
-        ctx: Any, uri: str, level: str = "basic"
+        ctx: Context, uri: str, level: str = "basic"
     ) -> Dict[str, Any]:
         """Validate a C# script and return diagnostics."""
         name, directory = _split_uri(uri)
@@ -422,11 +430,6 @@ def validate_script(
             "level": level,
         }
         resp = send_command_with_retry("manage_script", params)
-        if isinstance(resp, dict) and resp.get("success"):
-            diags = resp.get("data", {}).get("diagnostics", []) or []
-            warnings = sum(d.get("severity", "").lower() == "warning" for d in diags)
-            errors = sum(d.get("severity", "").lower() in ("error", "fatal") for d in diags)
-            return {"success": True, "data": {"warnings": warnings, "errors": errors}}
         return resp if isinstance(resp, dict) else {"success": False, "message": str(resp)}
 
     @mcp.tool(description=(
@@ -437,7 +440,7 @@ def validate_script(
     ))
     @telemetry_tool("manage_script")
     def manage_script(
-        ctx: Any,
+        ctx: Context,
         action: str,
         name: str,
         path: str,
@@ -565,10 +568,11 @@ def manage_script(
 
     @mcp.tool(description=(
         "Get manage_script capabilities (supported ops, limits, and guards).\n\n"
+        "Args:\n- random_string: required parameter (any string value)\n\n"
         "Returns:\n- ops: list of supported structured ops\n- text_ops: list of supported text ops\n- max_edit_payload_bytes: server edit payload cap\n- guards: header/using guard enabled flag\n"
     ))
     @telemetry_tool("manage_script_capabilities")
-    def manage_script_capabilities(ctx: Any) -> Dict[str, Any]:
+    def manage_script_capabilities(ctx: Context) -> Dict[str, Any]:
         try:
             # Keep in sync with server/Editor ManageScript implementation
             ops = [
@@ -596,21 +600,12 @@ def manage_script_capabilities(ctx: Any) -> Dict[str, Any]:
         "Returns: {sha256, lengthBytes, lastModifiedUtc, uri, path}."
     ))
     @telemetry_tool("get_sha")
-    def get_sha(ctx: Any, uri: str) -> Dict[str, Any]:
+    def get_sha(ctx: Context, uri: str) -> Dict[str, Any]:
         """Return SHA256 and basic metadata for a script."""
         try:
             name, directory = _split_uri(uri)
             params = {"action": "get_sha", "name": name, "path": directory}
             resp = send_command_with_retry("manage_script", params)
-            if isinstance(resp, dict) and resp.get("success"):
-                data = resp.get("data", {})
-                return {
-                    "success": True,
-                    "data": {
-                        "sha256": data.get("sha256"),
-                        "lengthBytes": data.get("lengthBytes"),
-                    },
-                }
             return resp if isinstance(resp, dict) else {"success": False, "message": str(resp)}
         except Exception as e:
             return {"success": False, "message": f"get_sha error: {e}"}

UnityMcpBridge/UnityMcpServer~/src/tools/manage_script_edits.py

@@ -1,5 +1,5 @@
 from mcp.server.fastmcp import FastMCP, Context
-from typing import Dict, Any, List, Tuple
+from typing import Dict, Any, List, Tuple, Optional
 import base64
 import re
 import os
@@ -318,23 +318,42 @@ def register_manage_script_edits_tools(mcp: FastMCP):
         "Do NOT use: new_method, anchor_method, content, newText (aliases accepted but normalized).\n\n"
         "Examples:\n"
         "1) Replace a method:\n"
-        "{ 'name':'SmartReach','path':'Assets/Scripts/Interaction','edits':[\n"
-        "  { 'op':'replace_method','className':'SmartReach','methodName':'HasTarget',\n"
-        "    'replacement':'public bool HasTarget(){ return currentTarget!=null; }' }\n"
-        "], 'options':{'validate':'standard','refresh':'immediate'} }\n\n"
+        "{\n"
+        "  \"name\": \"SmartReach\",\n"
+        "  \"path\": \"Assets/Scripts/Interaction\",\n"
+        "  \"edits\": [\n"
+        "    {\n"
+        "      \"op\": \"replace_method\",\n"
+        "      \"className\": \"SmartReach\",\n"
+        "      \"methodName\": \"HasTarget\",\n"
+        "      \"replacement\": \"public bool HasTarget(){ return currentTarget!=null; }\"\n"
+        "    }\n"
+        "  ],\n"
+        "  \"options\": {\"validate\": \"standard\", \"refresh\": \"immediate\"}\n"
+        "}\n\n"
         "2) Insert a method after another:\n"
-        "{ 'name':'SmartReach','path':'Assets/Scripts/Interaction','edits':[\n"
-        "  { 'op':'insert_method','className':'SmartReach','replacement':'public void PrintSeries(){ Debug.Log(seriesName); }',\n"
-        "    'position':'after','afterMethodName':'GetCurrentTarget' }\n"
-        "] }\n"
+        "{\n"
+        "  \"name\": \"SmartReach\",\n"
+        "  \"path\": \"Assets/Scripts/Interaction\",\n"
+        "  \"edits\": [\n"
+        "    {\n"
+        "      \"op\": \"insert_method\",\n"
+        "      \"className\": \"SmartReach\",\n"
+        "      \"replacement\": \"public void PrintSeries(){ Debug.Log(seriesName); }\",\n"
+        "      \"position\": \"after\",\n"
+        "      \"afterMethodName\": \"GetCurrentTarget\"\n"
+        "    }\n"
+        "  ]\n"
+        "}\n\n"
+        "Note: 'options' must be an object/dict, not a string. Use proper JSON syntax.\n"
     ))
     @telemetry_tool("script_apply_edits")
     def script_apply_edits(
-        ctx: Any,
+        ctx: Context,
         name: str,
         path: str,
         edits: List[Dict[str, Any]],
-        options: Dict[str, Any] | None = None,
+        options: Optional[Dict[str, Any]] = None,
         script_type: str = "MonoBehaviour",
         namespace: str = "",
     ) -> Dict[str, Any]:

UnityMcpBridge/UnityMcpServer~/src/tools/read_console.py

@@ -6,7 +6,6 @@
 from mcp.server.fastmcp import FastMCP, Context
 from unity_connection import get_unity_connection, send_command_with_retry
 from config import config
-
 from telemetry_decorator import telemetry_tool
 
 def register_read_console_tools(mcp: FastMCP):
@@ -15,10 +14,10 @@ def register_read_console_tools(mcp: FastMCP):
     @mcp.tool()
     @telemetry_tool("read_console")
     def read_console(
-        ctx: Any,
+        ctx: Context,
         action: str = None,
         types: List[str] = None,
-        count: int = None,
+        count: Any = None,
         filter_text: str = None,
         since_timestamp: str = None,
         format: str = None,
@@ -43,21 +42,34 @@ def read_console(
         # Get the connection instance
         bridge = get_unity_connection()
 
-        # Set defaults if values are None (conservative but useful for CI)
+        # Set defaults if values are None
         action = action if action is not None else 'get'
-        types = types if types is not None else ['error']
-        # Normalize types if passed as a single string
-        if isinstance(types, str):
-            types = [types]
-        format = format if format is not None else 'json'
+        types = types if types is not None else ['error', 'warning', 'log']
+        format = format if format is not None else 'detailed'
         include_stacktrace = include_stacktrace if include_stacktrace is not None else True
-        # Default count to a higher value unless explicitly provided
-        count = 50 if count is None else count
 
         # Normalize action if it's a string
         if isinstance(action, str):
             action = action.lower()
 
+        # Coerce count defensively (string/float -> int)
+        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
+
+        count = _coerce_int(count)
+
         # Prepare parameters for the C# handler
         params_dict = {
             "action": action,
@@ -76,25 +88,6 @@ def read_console(
         if 'count' not in params_dict:
              params_dict['count'] = None 
 
-        # Use centralized retry helper (tolerate legacy list payloads from some agents)
+        # Use centralized retry helper
         resp = send_command_with_retry("read_console", params_dict)
-        if isinstance(resp, dict) and resp.get("success") and not include_stacktrace:
-            data = resp.get("data", {}) or {}
-            lines = data.get("lines")
-            if lines is None:
-                # Some handlers return the raw list under data
-                lines = data if isinstance(data, list) else []
-
-            def _entry(x: Any) -> Dict[str, Any]:
-                if isinstance(x, dict):
-                    return {
-                        "level": x.get("level") or x.get("type"),
-                        "message": x.get("message") or x.get("text"),
-                    }
-                if isinstance(x, (list, tuple)) and len(x) >= 2:
-                    return {"level": x[0], "message": x[1]}
-                return {"level": None, "message": str(x)}
-
-            trimmed = [_entry(l) for l in (lines or [])]
-            return {"success": True, "data": {"lines": trimmed}}
-        return resp if isinstance(resp, dict) else {"success": False, "message": str(resp)}
+        return resp if isinstance(resp, dict) else {"success": False, "message": str(resp)}