From 65c8fede362286c091a5ec4a0b9f2b6b5ee57cac Mon Sep 17 00:00:00 2001
From: David <3dgiordano@gmail.com>
Date: Fri, 14 Aug 2026 15:27:02 -0300
Subject: [PATCH 1/4] Better handling of ambiguity in arguments

---
 main.py                        | 19 +++++++++++
 tools/ai_scriptless_manager.py | 26 +++++++--------
 tools/device_manager.py        |  6 ++--
 tools/execution_manager.py     |  6 ++--
 tools/help_manager.py          |  6 ++--
 tools/user_manager.py          |  7 ++--
 tools/utils.py                 | 60 +++++++++++++++++++++++++++++++++-
 7 files changed, 103 insertions(+), 27 deletions(-)

diff --git a/main.py b/main.py
index 016e47b..e2b05d9 100644
--- a/main.py
+++ b/main.py
@@ -5,6 +5,25 @@
 import sys
 from typing import Literal, cast
 
+# Patch MCP ArgModelBase so tools with an "arguments" param receive the full payload
+# when the client sends {"action": "x", "key": "value"} instead of {"arguments": {...}}
+from mcp.server.fastmcp.utilities import func_metadata
+from pydantic import model_validator
+
+_OriginalArgModelBase = func_metadata.ArgModelBase
+
+
+class _PatchedArgModelBase(_OriginalArgModelBase):
+    @model_validator(mode="before")
+    @classmethod
+    def _wrap_root_as_arguments(cls, data: object) -> object:
+        if isinstance(data, dict) and "arguments" not in data:
+            return {"arguments": data}
+        return data
+
+
+func_metadata.ArgModelBase = _PatchedArgModelBase
+
 from mcp.server.fastmcp import FastMCP, Icon
 
 from config.perfecto import SECURITY_TOKEN_FILE_ENV_NAME, SECURITY_TOKEN_ENV_NAME, PERFECTO_CLOUD_NAME_ENV_NAME, \
diff --git a/tools/ai_scriptless_manager.py b/tools/ai_scriptless_manager.py
index 15629e5..e392db2 100644
--- a/tools/ai_scriptless_manager.py
+++ b/tools/ai_scriptless_manager.py
@@ -44,7 +44,7 @@
     format_test_ui_location,
     update_element_arguments,
 )
-from tools.utils import api_request, format_sanitized_traceback
+from tools.utils import api_request, format_sanitized_traceback, normalize_action_args
 
 STEP_PATH_REFRESH_NOTES = [
     "step_path values are dot-separated positional paths (e.g. 0, 2.0, 5.b0.1); Perfecto does not persist them.",
@@ -247,7 +247,7 @@ async def add_command(
             self,
             test_id: str,
             command_id: str,
-            arguments: Optional[dict[str, Any]] = None,
+            cmd_arguments: Optional[dict[str, Any]] = None,
             after_path: Optional[str] = None,
             parent_path: Optional[str] = None,
     ) -> BaseResult:
@@ -256,7 +256,7 @@ async def add_command(
         if not command_id:
             return BaseResult(error="command_id is required (from list_commands)")
 
-        element = build_flow_element(command_id, arguments)
+        element = build_flow_element(command_id, cmd_arguments)
         inserted_path: dict[str, Optional[str]] = {"step_path": None}
 
         def mutator(script: dict[str, Any]) -> None:
@@ -271,20 +271,20 @@ def mutator(script: dict[str, Any]) -> None:
         return _append_step_path_refresh_notes(result)
 
     @token_verify
-    async def modify_command(self, test_id: str, step_path: str, arguments: dict[str, Any]) -> BaseResult:
+    async def modify_command(self, test_id: str, step_path: str, cmd_arguments: dict[str, Any]) -> BaseResult:
         if not test_id:
             return BaseResult(error="test_id is required")
         if not step_path:
             return BaseResult(error="step_path is required (from view_test_structure)")
-        if not arguments:
-            return BaseResult(error="arguments is required")
+        if not cmd_arguments:
+            return BaseResult(error="cmd_arguments is required")
 
         def mutator(script: dict[str, Any]) -> None:
             located = find_element_by_path(script, step_path)
             if located is None:
                 raise ValueError(f"step_path not found: {step_path}")
             _, _, element = located
-            update_element_arguments(element, arguments)
+            update_element_arguments(element, cmd_arguments)
 
         return _append_step_path_refresh_notes(
             await load_and_mutate(self.token, test_id, mutator)
@@ -681,14 +681,14 @@ def register(mcp, token: Optional[PerfectoToken]):
     args(dict): Dictionary with the following parameters:
         test_id (str, required): Test itemKey from list_tests.
         command_id (str, required): Command ID from list_commands.
-        arguments (dict, optional): Command argument names to values.
+        cmd_arguments (dict, optional): Command argument names to values.
         after_path (str, optional): Insert after this step (step_path from view_test_structure).
         parent_path (str, optional): Insert inside a container (step_path of LogicalStep, Loop, or Branch).
 - modify_command: Update command arguments and persist.
     args(dict): Dictionary with the following required parameters:
         test_id (str): Test itemKey from list_tests.
         step_path (str): Step path from view_test_structure (e.g. 0, 2.0, 5.b0.1).
-        arguments (dict): Argument names to new values.
+        cmd_arguments (dict): Argument names to new values.
 - delete_command: Remove a command from a test and persist.
     args(dict): Dictionary with the following required parameters:
         test_id (str): Test itemKey from list_tests.
@@ -811,10 +811,10 @@ def register(mcp, token: Optional[PerfectoToken]):
 """
     )
     async def ai_scriptless(
-            action: str = Field(description="The action id to execute"),
-            args: Dict[str, Any] = Field(description="Dictionary with parameters", default=None),
+            arguments: Dict[str, Any] = Field(description="Dictionary with arguments", default=None),
             ctx: Context = Field(description="Context object providing access to MCP capabilities")
     ) -> BaseResult:
+        action, args = normalize_action_args(arguments)
         if args is None:
             args = {}
         ai_scriptless_manager = AiScriptlessManager(token, ctx)
@@ -839,7 +839,7 @@ async def _dispatch():
                     return await ai_scriptless_manager.add_command(
                         args.get("test_id", ""),
                         args.get("command_id", ""),
-                        args.get("arguments"),
+                        args.get("cmd_arguments"),
                         args.get("after_path"),
                         args.get("parent_path"),
                     )
@@ -847,7 +847,7 @@ async def _dispatch():
                     return await ai_scriptless_manager.modify_command(
                         args.get("test_id", ""),
                         args.get("step_path", ""),
-                        args.get("arguments", {}),
+                        args.get("cmd_arguments", {}),
                     )
                 case "delete_command":
                     return await ai_scriptless_manager.delete_command(
diff --git a/tools/device_manager.py b/tools/device_manager.py
index 23c84b9..89b9975 100644
--- a/tools/device_manager.py
+++ b/tools/device_manager.py
@@ -12,7 +12,7 @@
 from models.manager import Manager
 from models.result import BaseResult
 from telemetry import run_tool
-from tools.utils import api_request, format_sanitized_traceback
+from tools.utils import api_request, format_sanitized_traceback, normalize_action_args
 
 
 class DeviceManager(Manager):
@@ -76,10 +76,10 @@ def register(mcp, token: Optional[PerfectoToken]):
 """
     )
     async def devices(
-            action: str = Field(description="The action id to execute"),
-            args: Dict[str, Any] = Field(description="Dictionary with parameters", default=None),
+            arguments: Dict[str, Any] = Field(description="Dictionary with arguments", default=None),
             ctx: Context = Field(description="Context object providing access to MCP capabilities")
     ) -> BaseResult:
+        action, args = normalize_action_args(arguments)
         if args is None:
             args = {}
         device_manager = DeviceManager(token, ctx)
diff --git a/tools/execution_manager.py b/tools/execution_manager.py
index 7206966..5f9dd8c 100644
--- a/tools/execution_manager.py
+++ b/tools/execution_manager.py
@@ -12,7 +12,7 @@
 from models.manager import Manager
 from models.result import BaseResult, PaginationResult
 from telemetry import run_tool
-from tools.utils import api_request, format_sanitized_traceback
+from tools.utils import api_request, format_sanitized_traceback, normalize_action_args
 
 
 class ExecutionManager(Manager):
@@ -255,10 +255,10 @@ def register(mcp, token: Optional[PerfectoToken]):
 """
     )
     async def execution(
-            action: str = Field(description="The action id to execute"),
-            args: Dict[str, Any] = Field(description="Dictionary with parameters", default=None),
+            arguments: Dict[str, Any] = Field(description="Dictionary with arguments", default=None),
             ctx: Context = Field(description="Context object providing access to MCP capabilities")
     ) -> BaseResult:
+        action, args = normalize_action_args(arguments)
         if args is None:
             args = {}
         execution_manager = ExecutionManager(token, ctx)
diff --git a/tools/help_manager.py b/tools/help_manager.py
index 7322311..bf98437 100644
--- a/tools/help_manager.py
+++ b/tools/help_manager.py
@@ -16,7 +16,7 @@
 from models.result import BaseResult
 from telemetry import run_tool
 from tools.help_utils import convert_js_to_py_dict
-from tools.utils import http_request, format_sanitized_traceback
+from tools.utils import http_request, format_sanitized_traceback, normalize_action_args
 
 
 class HelpManager(Manager):
@@ -266,10 +266,10 @@ def register(mcp, token: Optional[PerfectoToken]):
 """
     )
     async def help_main(
-            action: str = Field(description="The action id to execute"),
-            args: Dict[str, Any] = Field(description="Dictionary with parameters", default=None),
+            arguments: Dict[str, Any] = Field(description="Dictionary with arguments", default=None),
             ctx: Context = Field(description="Context object providing access to MCP capabilities")
     ) -> BaseResult:
+        action, args = normalize_action_args(arguments)
         if args is None:
             args = {}
         help_manager = HelpManager(token, ctx)
diff --git a/tools/user_manager.py b/tools/user_manager.py
index f7b40d5..50125a2 100644
--- a/tools/user_manager.py
+++ b/tools/user_manager.py
@@ -11,8 +11,7 @@
 from models.manager import Manager
 from models.result import BaseResult
 from telemetry import run_tool
-from tools.utils import api_request, format_sanitized_traceback
-
+from tools.utils import api_request, format_sanitized_traceback, normalize_action_args
 
 class UserManager(Manager):
     def __init__(self, token: Optional[PerfectoToken], ctx: Context):
@@ -47,10 +46,10 @@ def register(mcp, token: Optional[PerfectoToken]):
 """
     )
     async def user(
-            action: str = Field(description="The action id to execute"),
-            args: Dict[str, Any] = Field(description="Dictionary with parameters", default=None),
+            arguments: Dict[str, Any] = Field(description="Dictionary with arguments", default=None),
             ctx: Context = Field(description="Context object providing access to MCP capabilities")
     ) -> BaseResult:
+        action, args = normalize_action_args(arguments)
         if args is None:
             args = {}
         user_manager = UserManager(token, ctx)
diff --git a/tools/utils.py b/tools/utils.py
index cc2e7e0..141e95d 100644
--- a/tools/utils.py
+++ b/tools/utils.py
@@ -11,7 +11,7 @@
 from datetime import datetime, timezone
 from importlib import resources
 from pathlib import Path
-from typing import Optional, Callable
+from typing import Optional, Callable, Any, Dict
 
 import httpx
 
@@ -211,3 +211,61 @@ def get_mcp_icon_uri():
     icon_path = get_resources_path().joinpath(name)
     icon_data = base64.standard_b64encode(icon_path.read_bytes()).decode()
     return f"data:image/png;base64,{icon_data}"
+
+def normalize_action_args(arguments: Optional[Dict[str, Any]] = None) -> tuple[str, Dict[str, Any]]:
+    """
+    Normalize tool arguments to (action, args) format.
+    Supports:
+      - {"action": "x", "args": {"key": "value"}}
+      - {"action": "x", "key": "value"}  (params at top level, merged into args)
+      - {"arguments": {"action": "x", "args": {...}}}  (double-wrapped by client)
+    Top-level keys other than 'action' and 'args' are merged into args.
+    Use a single 'arguments' param so the full MCP tool call payload is received
+    (avoids Pydantic dropping extra fields when using action/args separately).
+    """
+    arguments = arguments or {}
+    # Unwrap double-nested format: {"arguments": {"action": "x", "args": {...}}}
+    inner = arguments.get("arguments")
+    if (
+            isinstance(inner, dict)
+            and len(arguments) == 1
+            and ("action" in inner or "args" in inner)
+    ):
+        arguments = inner
+    action = str(arguments.get("action") or "").strip() or ""
+    args = dict(arguments.get("args") or {})
+    for key, value in arguments.items():
+        if key not in ("action", "args"):
+            args[key] = value
+    return action, args
+
+
+def validate_required_args(action: str, args: Optional[Dict[str, Any]], required: list[str]) -> Optional[BaseResult]:
+    args = args or {}
+    missing = [key for key in required if key not in args or args[key] is None]
+    if not missing:
+        return None
+    missing_str = ", ".join(missing)
+    required_str = ", ".join(required)
+    return BaseResult(
+        error=(
+            f"Missing required args for action '{action}': {missing_str} not found within 'args'. "
+            f"Required args: {required_str}. Ensure parameters are passed inside the 'args' argument."
+        )
+    )
+
+
+def validate_non_empty_str_arg(
+        action: str, args: Optional[Dict[str, Any]], key: str
+) -> Optional[BaseResult]:
+    """Return BaseResult error if args[key] is missing, not a str, or only whitespace."""
+    args = args or {}
+    value = args.get(key)
+    if not isinstance(value, str) or not value.strip():
+        return BaseResult(
+            error=(
+                f"Missing required args for action '{action}': {key} must be a non-empty string "
+                f"within 'args'. Required args: {key}."
+            )
+        )
+    return None
\ No newline at end of file

From 369ee793bb3e2c7ba8110aaefb3acf39e412d70f Mon Sep 17 00:00:00 2001
From: David <3dgiordano@gmail.com>
Date: Fri, 14 Aug 2026 15:42:50 -0300
Subject: [PATCH 2/4] Update tests

---
 tests/test_ai_scriptless_manager.py | 132 +++++++++++-----------------
 1 file changed, 49 insertions(+), 83 deletions(-)

diff --git a/tests/test_ai_scriptless_manager.py b/tests/test_ai_scriptless_manager.py
index d2e19f7..88310b1 100644
--- a/tests/test_ai_scriptless_manager.py
+++ b/tests/test_ai_scriptless_manager.py
@@ -489,10 +489,10 @@ def test_view_snapshot_rejects_current_marker(self, perfecto_token):
         result = asyncio.run(manager.view_snapshot("<current>"))
         assert "not a historical snapshot" in result.error
 
-    def test_modify_command_requires_arguments(self, perfecto_token):
+    def test_modify_command_requires_cmd_arguments(self, perfecto_token):
         manager = AiScriptlessManager(perfecto_token, ctx=None)
         result = asyncio.run(manager.modify_command(TEST_ID, "0", {}))
-        assert result.error == "arguments is required"
+        assert result.error == "cmd_arguments is required"
 
     def test_save_test_requires_test_id(self, perfecto_token):
         manager = AiScriptlessManager(perfecto_token, ctx=None)
@@ -519,7 +519,7 @@ def test_add_command_inserts_and_returns_step_path(self, perfecto_token, monkeyp
         result = asyncio.run(manager.add_command(
             TEST_ID,
             "wait",
-            arguments={"duration": "3"},
+            cmd_arguments={"duration": "3"},
         ))
 
         _assert_step_path_notes(result)
@@ -600,7 +600,7 @@ def test_add_command_after_path(self, perfecto_token, monkeypatch):
         result = asyncio.run(manager.add_command(
             TEST_ID,
             "ai_user-action",
-            arguments={"action": "Tap"},
+            cmd_arguments={"action": "Tap"},
             after_path="0",
         ))
 
@@ -618,7 +618,7 @@ def test_add_command_inside_logical_step(self, perfecto_token, monkeypatch):
         result = asyncio.run(manager.add_command(
             TEST_ID,
             "comment",
-            arguments={"text": "inside"},
+            cmd_arguments={"text": "inside"},
             parent_path="0",
         ))
 
@@ -1043,8 +1043,8 @@ def _dispatcher_action_cases() -> list[tuple[str, dict]]:
         ("view_test_structure", {"test_id": test_id}),
         ("list_commands", {"checkpoint": True}),
         ("get_command_definitions", {"command_ids": ["wait"]}),
-        ("add_command", {"test_id": test_id, "command_id": "comment", "arguments": {"text": "hi"}}),
-        ("modify_command", {"test_id": test_id, "step_path": "0", "arguments": {"duration": "2"}}),
+        ("add_command", {"test_id": test_id, "command_id": "comment", "cmd_arguments": {"text": "hi"}}),
+        ("modify_command", {"test_id": test_id, "step_path": "0", "cmd_arguments": {"duration": "2"}}),
         ("delete_command", {"test_id": test_id, "step_path": "0"}),
         ("set_command_enabled", {"test_id": test_id, "step_path": "0", "enabled": False}),
         ("save_test", {"test_id": test_id, "comment": "saved"}),
@@ -1156,7 +1156,7 @@ async def fake_load_and_mutate(_token, test_id, mutator, snapshot_comment=None):
 class TestAiScriptlessDispatcher:
     def test_unknown_action_returns_error(self, perfecto_token):
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(action="not_a_real_action", args={}, ctx=None))
+        result = asyncio.run(_call_tool(tool, "not_a_real_action", {}))
         assert "not found in AI Scriptless manager tool" in result.error
 
     def test_routes_add_command(self, perfecto_token, monkeypatch):
@@ -1164,15 +1164,11 @@ def test_routes_add_command(self, perfecto_token, monkeypatch):
         _mock_load_and_mutate(monkeypatch, captured=captured)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(
-            action="add_command",
-            args={
-                "test_id": TEST_ID,
-                "command_id": "wait",
-                "arguments": {"duration": "1"},
-            },
-            ctx=None,
-        ))
+        result = asyncio.run(_call_tool(tool, "add_command", {
+            "test_id": TEST_ID,
+            "command_id": "wait",
+            "cmd_arguments": {"duration": "1"},
+        }))
 
         assert result.error is None
         assert result.result["command_id"] == "wait"
@@ -1198,11 +1194,7 @@ async def fake_fetch(_token, _test_id):
         monkeypatch.setattr(ai_scriptless_manager, "fetch_script_payload", fake_fetch)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(
-            action="list_test_variables",
-            args={"test_id": TEST_ID},
-            ctx=None,
-        ))
+        result = asyncio.run(_call_tool(tool, "list_test_variables", {"test_id": TEST_ID}))
 
         assert result.error is None
         assert result.result[0].name == "flag"
@@ -1214,7 +1206,7 @@ async def fake_api_request(*_args, **_kwargs):
         monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(action="list_tests", args=None, ctx=None))
+        result = asyncio.run(_call_tool(tool, "list_tests", None))
         assert result.error == "tree unavailable"
 
     def test_routes_move_command(self, perfecto_token, monkeypatch):
@@ -1222,10 +1214,8 @@ def test_routes_move_command(self, perfecto_token, monkeypatch):
         _mock_load_and_mutate(monkeypatch, _script_with_steps("wait", "comment"), captured)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(
-            action="move_command",
-            args={"test_id": TEST_ID, "step_path": "0", "after_path": "0"},
-            ctx=None,
+        result = asyncio.run(_call_tool(
+            tool, "move_command", {"test_id": TEST_ID, "step_path": "0", "after_path": "0"},
         ))
 
         assert result.error is None
@@ -1237,11 +1227,7 @@ def test_routes_delete_command(self, perfecto_token, monkeypatch):
         _mock_load_and_mutate(monkeypatch, _script_with_steps("wait", "comment"), captured)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(
-            action="delete_command",
-            args={"test_id": TEST_ID, "step_path": "0"},
-            ctx=None,
-        ))
+        result = asyncio.run(_call_tool(tool, "delete_command", {"test_id": TEST_ID, "step_path": "0"}))
 
         assert result.error is None
         assert len(captured["script"]["flowElements"]) == 1
@@ -1256,15 +1242,11 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs):
         monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(
-            action="execute_test",
-            args={
-                "test_id": TEST_ID,
-                "device_type": "real",
-                "device_under_test": {"device_id": "DEV-1"},
-            },
-            ctx=None,
-        ))
+        result = asyncio.run(_call_tool(tool, "execute_test", {
+            "test_id": TEST_ID,
+            "device_type": "real",
+            "device_under_test": {"device_id": "DEV-1"},
+        }))
 
         assert result.error is None
         assert captured["json"]["params"]["DUT"] == "DEV-1"
@@ -1279,23 +1261,19 @@ async def fake_persist(_token, item_key, script, saved_script=None, snapshot_com
         monkeypatch.setattr(ai_scriptless_manager, "persist_script", fake_persist)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(
-            action="create_test",
-            args={"name": "Smoke", "folder": "QA"},
-            ctx=None,
-        ))
+        result = asyncio.run(_call_tool(tool, "create_test", {"name": "Smoke", "folder": "QA"}))
 
         assert result.error is None
         assert persisted["item_key"] == "PRIVATE:QA/Smoke.xml"
 
     @pytest.mark.parametrize("action,args,expected_error", [
-        ("modify_command", {"test_id": TEST_ID, "step_path": "0"}, "arguments is required"),
+        ("modify_command", {"test_id": TEST_ID, "step_path": "0"}, "cmd_arguments is required"),
         ("delete_test", {"test_id": ""}, "test_id is required"),
         ("move_test", {"test_id": "", "folder": "Archive"}, "test_id is required"),
     ])
     def test_dispatcher_validation_errors(self, perfecto_token, action, args, expected_error):
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(action=action, args=args, ctx=None))
+        result = asyncio.run(_call_tool(tool, action, args))
         if expected_error:
             assert expected_error in result.error
 
@@ -1308,7 +1286,7 @@ async def raise_http_error(*_args, **_kwargs):
         monkeypatch.setattr(ai_scriptless_manager, "api_request", raise_http_error)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(action="list_tests", args={}, ctx=None))
+        result = asyncio.run(_call_tool(tool, "list_tests", {}))
 
         assert result.error is not None
         assert result.error.startswith("Error:")
@@ -1321,7 +1299,7 @@ async def raise_runtime_error(*_args, **_kwargs):
         monkeypatch.setattr(ai_scriptless_manager, "api_request", raise_runtime_error)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(action="list_tests", args={}, ctx=None))
+        result = asyncio.run(_call_tool(tool, "list_tests", {}))
 
         assert "boom" in result.error
         assert SUPPORT_MESSAGE in result.error
@@ -1345,11 +1323,7 @@ async def fake_api_request(
         monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(
-            action="view_test_structure",
-            args={"test_id": TEST_ID},
-            ctx=None,
-        ))
+        result = asyncio.run(_call_tool(tool, "view_test_structure", {"test_id": TEST_ID}))
 
         assert result.error is None
         assert result.result.item_key == TEST_ID
@@ -1360,11 +1334,7 @@ def test_routes_list_filter_values(self, perfecto_token, monkeypatch):
         _setup_dispatcher_mocks(monkeypatch)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(
-            action="list_filter_values",
-            args={"filter_names": ["test_name", "owner_list"]},
-            ctx=None,
-        ))
+        result = asyncio.run(_call_tool(tool, "list_filter_values", {"filter_names": ["test_name", "owner_list"]}))
 
         assert result.error is None
         assert "Login" in result.result["test_name"]
@@ -1374,17 +1344,13 @@ def test_routes_save_test_as(self, perfecto_token, monkeypatch):
         _setup_dispatcher_mocks(monkeypatch, persisted=persisted)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(
-            action="save_test_as",
-            args={
-                "test_id": TEST_ID,
-                "name": "Branch",
-                "folder": "Copies",
-                "visibility": "PUBLIC",
-                "comment": "v2",
-            },
-            ctx=None,
-        ))
+        result = asyncio.run(_call_tool(tool, "save_test_as", {
+            "test_id": TEST_ID,
+            "name": "Branch",
+            "folder": "Copies",
+            "visibility": "PUBLIC",
+            "comment": "v2",
+        }))
 
         assert result.error is None
         assert persisted["item_key"] == "PUBLIC:Copies/Branch.xml"
@@ -1395,17 +1361,13 @@ def test_routes_add_test_variable(self, perfecto_token, monkeypatch):
         _setup_dispatcher_mocks(monkeypatch, captured=captured)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(
-            action="add_test_variable",
-            args={
-                "test_id": TEST_ID,
-                "name": "retry",
-                "variable_type": "number",
-                "value": 3,
-                "set_at_runtime": True,
-            },
-            ctx=None,
-        ))
+        result = asyncio.run(_call_tool(tool, "add_test_variable", {
+            "test_id": TEST_ID,
+            "name": "retry",
+            "variable_type": "number",
+            "value": 3,
+            "set_at_runtime": True,
+        }))
 
         assert result.error is None
         assert result.result["name"] == "retry"
@@ -1417,11 +1379,15 @@ def test_dispatcher_routes_registered_action(self, perfecto_token, monkeypatch,
         _setup_dispatcher_mocks(monkeypatch)
 
         tool = _register_tool(perfecto_token)
-        result = asyncio.run(tool(action=action, args=args, ctx=None))
+        result = asyncio.run(_call_tool(tool, action, args))
 
         assert "not found in AI Scriptless manager tool" not in (result.error or "")
 
 
+def _call_tool(tool, action, args, ctx=None):
+    return tool(arguments={"action": action, "args": args}, ctx=ctx)
+
+
 def _register_tool(token):
     class _McpStub:
         def __init__(self):

From f8317fe992034308b00937cdf051ebdc076852b0 Mon Sep 17 00:00:00 2001
From: David <3dgiordano@gmail.com>
Date: Fri, 14 Aug 2026 15:43:04 -0300
Subject: [PATCH 3/4] Update uv.lock version

---
 uv.lock | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/uv.lock b/uv.lock
index c233035..1bea189 100644
--- a/uv.lock
+++ b/uv.lock
@@ -805,7 +805,7 @@ wheels = [
 
 [[package]]
 name = "perfecto-mcp"
-version = "1.1.1"
+version = "1.1.2"
 source = { virtual = "." }
 dependencies = [
     { name = "httpx", extra = ["http2"] },

From aee94c776b9620b93a183c03f34f08b91c538a31 Mon Sep 17 00:00:00 2001
From: David <3dgiordano@gmail.com>
Date: Tue, 18 Aug 2026 15:54:51 -0300
Subject: [PATCH 4/4] Better handling of ambiguity in arguments

---
 formatters/ai_scriptless.py             |   7 ++
 tests/conftest.py                       |  38 ++++++
 tests/test_ai_scriptless_definitions.py | 151 +++++++++++++++++++++++
 tests/test_ai_scriptless_manager.py     | 135 +++++++++++++++++++-
 tests/test_ai_scriptless_persistence.py |  43 +++++++
 tools/ai_scriptless/__init__.py         |  10 ++
 tools/ai_scriptless/definitions.py      | 156 ++++++++++++++++++++++++
 tools/ai_scriptless/persistence.py      |   6 +-
 tools/ai_scriptless_manager.py          |  57 ++++++++-
 tools/tools_manager.py                  |   6 +-
 10 files changed, 597 insertions(+), 12 deletions(-)
 create mode 100644 tests/test_ai_scriptless_definitions.py
 create mode 100644 tools/ai_scriptless/definitions.py

diff --git a/formatters/ai_scriptless.py b/formatters/ai_scriptless.py
index 90bd271..3054a80 100644
--- a/formatters/ai_scriptless.py
+++ b/formatters/ai_scriptless.py
@@ -37,6 +37,13 @@ def command_selection_policy_info() -> List[str]:
         "Structural helpers (add_logical_step, add_loop, add_condition, comment, wait) are OK; "
         "keep observable steps AI-driven when possible.",
         "Call get_command_definitions only for the AI command_ids you will use.",
+        "cmd_arguments keys are the parameter names returned by get_command_definitions "
+        "(mandatory_parameters / optional_parameters); undeclared names are rejected.",
+        "Keep command arguments nested inside cmd_arguments: the 'action' parameter of ai_user-action "
+        "collides with the tool's own action key if flattened into args.",
+        "Values are constants by default; pass {\"data_source\": \"VARIABLE\", \"value\": \"<variable name>\"} "
+        "to bind an argument to a script variable.",
+        "modify_command merges: only the arguments sent are replaced, the others keep their current value.",
     ]
 
 def format_ai_scriptless_tests_filter_values(tests: dict[str, Any], params: Optional[dict] = None) -> dict[str, Any]:
diff --git a/tests/conftest.py b/tests/conftest.py
index f1d39c6..7a3e4f0 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -17,8 +17,46 @@
 import pytest
 
 from config.token import PerfectoToken
+from models.result import BaseResult
+from tools.ai_scriptless import definitions
 
 
 @pytest.fixture
 def perfecto_token() -> PerfectoToken:
     return PerfectoToken("test-token", "demo")
+
+
+@pytest.fixture(autouse=True)
+def offline_command_definitions(monkeypatch):
+    """Keep cmd_arguments validation offline.
+
+    Validation resolves declared parameters over HTTP and memoizes them, so the
+    request is stubbed out (no declared parameters = validation fails open) and the
+    cache is reset on both ends. Use declare_command_parameters to opt into validation.
+    """
+    definitions.reset_declared_parameters_cache()
+
+    async def offline_api_request(*_args, **_kwargs):
+        return BaseResult(error="command definitions are not fetched in tests")
+
+    monkeypatch.setattr(definitions, "api_request", offline_api_request)
+    yield
+    definitions.reset_declared_parameters_cache()
+
+
+@pytest.fixture
+def declare_command_parameters(monkeypatch, offline_command_definitions):
+    """Declare parameters per command_id: {command_id: (mandatory, optional)}."""
+
+    def declare(declarations: dict[str, tuple[list[str], list[str]]]) -> None:
+        async def fake_fetch(_token, command_id):
+            declaration = declarations.get(command_id)
+            if declaration is None:
+                return None
+            mandatory, optional = declaration
+            return frozenset(mandatory), frozenset(optional)
+
+        definitions.reset_declared_parameters_cache()
+        monkeypatch.setattr(definitions, "_fetch_declared_parameters", fake_fetch)
+
+    return declare
diff --git a/tests/test_ai_scriptless_definitions.py b/tests/test_ai_scriptless_definitions.py
new file mode 100644
index 0000000..4268bd5
--- /dev/null
+++ b/tests/test_ai_scriptless_definitions.py
@@ -0,0 +1,151 @@
+"""
+Copyright 2025 Perforce Software, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import asyncio
+
+import httpx
+
+from models.result import BaseResult
+from tools.ai_scriptless import definitions
+from tools.ai_scriptless.definitions import (
+    declared_parameters,
+    empty_mandatory_note,
+    validate_argument_names,
+)
+
+USER_ACTION = (frozenset({"action"}), frozenset({"handsetId"}))
+
+
+class TestValidateArgumentNames:
+    def test_accepts_declared_names(self):
+        assert validate_argument_names(
+            "ai_user-action", {"action": "Tap", "handsetId": "DUT"}, USER_ACTION
+        ) is None
+
+    def test_rejects_undeclared_name_with_suggestion(self):
+        error = validate_argument_names("ai_user-action", {"actions": "Tap"}, USER_ACTION)
+        assert "'actions' (did you mean 'action'?)" in error
+        assert "Declared parameter names: action, handsetId" in error
+        assert "mandatory: action" in error
+
+    def test_reports_undeclared_name_without_close_match(self):
+        error = validate_argument_names("ai_user-action", {"xyz": "Tap"}, USER_ACTION)
+        assert "'xyz'" in error
+        assert "did you mean" not in error
+
+    def test_accepts_alias_in_either_direction(self):
+        # The spec canonicalizes waitDuration to duration; either name may be declared.
+        assert validate_argument_names("wait", {"duration": "3"}, (frozenset({"waitDuration"}), frozenset())) is None
+        assert validate_argument_names("wait", {"waitDuration": "3"}, (frozenset({"duration"}), frozenset())) is None
+
+    def test_fails_open_without_declared_parameters(self):
+        assert validate_argument_names("ai_user-action", {"anything": "value"}, None) is None
+
+    def test_accepts_variable_data_source_form(self):
+        assert validate_argument_names(
+            "ai_user-action",
+            {"action": {"data_source": "VARIABLE", "value": "loginStep"}},
+            USER_ACTION,
+        ) is None
+
+
+class TestEmptyMandatoryNote:
+    def test_notes_mandatory_left_empty_by_spec_default(self):
+        note = empty_mandatory_note("ai_user-action", None, USER_ACTION)
+        assert "Mandatory parameter(s) left empty on 'ai_user-action': action" in note
+
+    def test_no_note_when_mandatory_is_provided(self):
+        assert empty_mandatory_note("ai_user-action", {"action": "Tap"}, USER_ACTION) is None
+
+    def test_blank_string_counts_as_empty(self):
+        note = empty_mandatory_note("ai_user-action", {"action": "   "}, USER_ACTION)
+        assert "action" in note
+
+    def test_variable_binding_counts_as_provided(self):
+        assert empty_mandatory_note(
+            "ai_user-action",
+            {"action": {"data_source": "VARIABLE", "value": "loginStep"}},
+            USER_ACTION,
+        ) is None
+
+    def test_no_note_without_declared_parameters(self):
+        assert empty_mandatory_note("ai_user-action", None, None) is None
+
+
+class TestDeclaredParameters:
+    def test_parses_and_memoizes_definitions(self, perfecto_token, monkeypatch):
+        calls: list = []
+
+        async def fake_api_request(_token, _method, endpoint=None, result_formatter=None, **kwargs):
+            calls.append(kwargs.get("json"))
+            return BaseResult(result=result_formatter({
+                "definitions": [{
+                    "commandId": "ai_validation",
+                    "data": {
+                        "display": {"name": "AI Validation"},
+                        "mandatoryParameters": [{"name": "validation"}],
+                        "optionalParameters": [{"name": "handsetId"}],
+                    },
+                }],
+            }, None))
+
+        definitions.reset_declared_parameters_cache()
+        monkeypatch.setattr(definitions, "api_request", fake_api_request)
+
+        first = asyncio.run(declared_parameters(perfecto_token, "ai_validation"))
+        second = asyncio.run(declared_parameters(perfecto_token, "ai_validation"))
+
+        assert first == (frozenset({"validation"}), frozenset({"handsetId"}))
+        assert second == first
+        assert calls == [{"commandIds": ["ai_validation"]}]
+
+    def test_definition_without_parameters_is_treated_as_unknown(self, perfecto_token, monkeypatch):
+        async def fake_api_request(_token, _method, endpoint=None, result_formatter=None, **kwargs):
+            return BaseResult(result=result_formatter({
+                "definitions": [{
+                    "commandId": "wait",
+                    "data": {"display": {"name": "Wait"}, "mandatoryParameters": [], "optionalParameters": []},
+                }],
+            }, None))
+
+        definitions.reset_declared_parameters_cache()
+        monkeypatch.setattr(definitions, "api_request", fake_api_request)
+
+        assert asyncio.run(declared_parameters(perfecto_token, "wait")) is None
+
+    def test_fails_open_on_api_error(self, perfecto_token, monkeypatch):
+        async def fake_api_request(*_args, **_kwargs):
+            return BaseResult(error="Invalid credentials")
+
+        definitions.reset_declared_parameters_cache()
+        monkeypatch.setattr(definitions, "api_request", fake_api_request)
+
+        assert asyncio.run(declared_parameters(perfecto_token, "ai_validation")) is None
+
+    def test_fails_open_on_http_exception(self, perfecto_token, monkeypatch):
+        async def fake_api_request(*_args, **_kwargs):
+            request = httpx.Request("POST", "https://demo.perfectomobile.com/definitions")
+            raise httpx.HTTPStatusError(
+                "not found", request=request, response=httpx.Response(404, request=request)
+            )
+
+        definitions.reset_declared_parameters_cache()
+        monkeypatch.setattr(definitions, "api_request", fake_api_request)
+
+        assert asyncio.run(declared_parameters(perfecto_token, "ai_validation")) is None
+
+    def test_no_token_returns_none(self):
+        assert asyncio.run(declared_parameters(None, "ai_validation")) is None
diff --git a/tests/test_ai_scriptless_manager.py b/tests/test_ai_scriptless_manager.py
index 88310b1..e8d173e 100644
--- a/tests/test_ai_scriptless_manager.py
+++ b/tests/test_ai_scriptless_manager.py
@@ -16,6 +16,7 @@
 
 import asyncio
 import copy
+import inspect
 import json
 
 import httpx
@@ -24,6 +25,7 @@
 from config.perfecto import SUPPORT_MESSAGE
 from models.result import BaseResult
 from tools import ai_scriptless_manager
+from tools.ai_scriptless import definitions
 from tools.ai_scriptless.elements import (
     build_flow_element,
     build_if_statement,
@@ -89,7 +91,9 @@ def _mock_load_and_mutate(monkeypatch, initial_script: dict | None = None, captu
     async def fake_load_and_mutate(_token, test_id, mutator, snapshot_comment=None):
         script = copy.deepcopy(base_script)
         try:
-            mutator(script)
+            outcome = mutator(script)
+            if inspect.isawaitable(outcome):
+                await outcome
         except ValueError as exc:
             return BaseResult(error=str(exc))
         if captured is not None:
@@ -1139,7 +1143,9 @@ async def fake_persist(_token, item_key, payload, saved_script=None, snapshot_co
     async def fake_load_and_mutate(_token, test_id, mutator, snapshot_comment=None):
         payload = copy.deepcopy(script)
         try:
-            mutator(payload)
+            outcome = mutator(payload)
+            if inspect.isawaitable(outcome):
+                await outcome
         except ValueError as exc:
             return BaseResult(error=str(exc))
         if captured is not None:
@@ -1402,3 +1408,128 @@ def decorator(fn):
     mcp = _McpStub()
     ai_scriptless_manager.register(mcp, token)
     return mcp.tools["perfecto_ai_scriptless"]
+
+
+class TestCmdArgumentsValidation:
+    def test_add_command_rejects_undeclared_argument_name(
+            self, perfecto_token, monkeypatch, declare_command_parameters):
+        declare_command_parameters({"ai_user-action": (["action"], ["handsetId"])})
+        _mock_load_and_mutate(monkeypatch)
+
+        manager = AiScriptlessManager(perfecto_token, ctx=None)
+        result = asyncio.run(manager.add_command(
+            TEST_ID,
+            "ai_user-action",
+            cmd_arguments={"actions": "Tap on Login"},
+        ))
+
+        assert "Unknown cmd_arguments for command 'ai_user-action'" in result.error
+        assert "did you mean 'action'" in result.error
+        assert "get_command_definitions" in result.error
+
+    def test_add_command_accepts_declared_argument_name(
+            self, perfecto_token, monkeypatch, declare_command_parameters):
+        captured: dict = {}
+        declare_command_parameters({"ai_user-action": (["action"], ["handsetId"])})
+        _mock_load_and_mutate(monkeypatch, captured=captured)
+
+        manager = AiScriptlessManager(perfecto_token, ctx=None)
+        result = asyncio.run(manager.add_command(
+            TEST_ID,
+            "ai_user-action",
+            cmd_arguments={"action": "Tap on Login"},
+        ))
+
+        assert result.error is None
+        arguments = captured["script"]["flowElements"][0]["arguments"]
+        assert {argument["name"] for argument in arguments} == {"action", "handsetId"}
+
+    def test_add_command_accepts_canonical_name_when_alias_is_declared(
+            self, perfecto_token, monkeypatch, declare_command_parameters):
+        # The repository declares waitDuration; the spec canonicalizes it to duration.
+        declare_command_parameters({"wait": (["waitDuration"], [])})
+        _mock_load_and_mutate(monkeypatch)
+
+        manager = AiScriptlessManager(perfecto_token, ctx=None)
+        result = asyncio.run(manager.add_command(TEST_ID, "wait", cmd_arguments={"duration": "3"}))
+
+        assert result.error is None
+
+    def test_add_command_fails_open_without_definitions(self, perfecto_token, monkeypatch):
+        _mock_load_and_mutate(monkeypatch)
+
+        manager = AiScriptlessManager(perfecto_token, ctx=None)
+        result = asyncio.run(manager.add_command(
+            TEST_ID,
+            "ai_user-action",
+            cmd_arguments={"whatever": "value"},
+        ))
+
+        assert result.error is None
+
+    def test_add_command_notes_empty_mandatory_parameter(
+            self, perfecto_token, monkeypatch, declare_command_parameters):
+        declare_command_parameters({"ai_user-action": (["action"], ["handsetId"])})
+        _mock_load_and_mutate(monkeypatch)
+
+        manager = AiScriptlessManager(perfecto_token, ctx=None)
+        result = asyncio.run(manager.add_command(TEST_ID, "ai_user-action"))
+
+        assert result.error is None
+        assert any("Mandatory parameter(s) left empty" in note for note in result.result["notes"])
+
+    def test_modify_command_rejects_undeclared_argument_name(
+            self, perfecto_token, monkeypatch, declare_command_parameters):
+        declare_command_parameters({"wait": ([], ["duration"])})
+        _mock_load_and_mutate(monkeypatch, _script_with_steps("wait"))
+
+        manager = AiScriptlessManager(perfecto_token, ctx=None)
+        result = asyncio.run(manager.modify_command(TEST_ID, "0", {"timeout": "5"}))
+
+        assert "Unknown cmd_arguments for command 'wait'" in result.error
+
+    def test_modify_command_accepts_declared_argument_name(
+            self, perfecto_token, monkeypatch, declare_command_parameters):
+        captured: dict = {}
+        declare_command_parameters({"wait": ([], ["duration"])})
+        _mock_load_and_mutate(monkeypatch, _script_with_steps("wait"), captured)
+
+        manager = AiScriptlessManager(perfecto_token, ctx=None)
+        result = asyncio.run(manager.modify_command(TEST_ID, "0", {"duration": "5"}))
+
+        assert result.error is None
+        arguments = captured["script"]["flowElements"][0]["arguments"]
+        assert {"name": "duration", "value": "5"} in [
+            {"name": argument["name"], "value": argument["data"]["value"]} for argument in arguments
+        ]
+
+    def test_definitions_are_fetched_once_per_command(self, perfecto_token, monkeypatch):
+        calls: list[str] = []
+
+        async def counting_fetch(_token, command_id):
+            calls.append(command_id)
+            return frozenset({"action"}), frozenset({"handsetId"})
+
+        definitions.reset_declared_parameters_cache()
+        monkeypatch.setattr(definitions, "_fetch_declared_parameters", counting_fetch)
+        _mock_load_and_mutate(monkeypatch)
+
+        manager = AiScriptlessManager(perfecto_token, ctx=None)
+        for _ in range(3):
+            asyncio.run(manager.add_command(
+                TEST_ID,
+                "ai_user-action",
+                cmd_arguments={"action": "Tap"},
+            ))
+
+        assert calls == ["ai_user-action"]
+
+    def test_unknown_action_hints_cmd_arguments_collision(self, perfecto_token):
+        tool = _register_tool(perfecto_token)
+        result = asyncio.run(_call_tool(tool, "Tap on the Login button", {
+            "test_id": TEST_ID,
+            "command_id": "ai_user-action",
+        }))
+
+        assert "not found in AI Scriptless manager tool" in result.error
+        assert "must stay nested inside 'cmd_arguments'" in result.error
diff --git a/tests/test_ai_scriptless_persistence.py b/tests/test_ai_scriptless_persistence.py
index cdd7382..38cefbe 100644
--- a/tests/test_ai_scriptless_persistence.py
+++ b/tests/test_ai_scriptless_persistence.py
@@ -223,3 +223,46 @@ async def fake_fetch(_token, _test_id):
         )
 
         assert result.error == "not found"
+
+
+class TestLoadAndMutateAsyncMutator:
+    """Mutators may be async when they need the API before mutating (command definitions)."""
+
+    @staticmethod
+    def _mock_script_io(monkeypatch, persisted: dict):
+        async def fake_fetch(_token, _test_id):
+            return BaseResult(result={"script": new_empty_script()})
+
+        async def fake_persist(_token, item_key, mutated_script, saved_script, snapshot_comment=None):
+            persisted["item_key"] = item_key
+            persisted["flow_count"] = len(mutated_script.get("flowElements", []))
+            return BaseResult(result={"status": "ok"})
+
+        monkeypatch.setattr(persistence, "fetch_script_payload", fake_fetch)
+        monkeypatch.setattr(persistence, "_persist_script", fake_persist)
+
+    def test_awaits_async_mutator_before_persisting(self, perfecto_token, monkeypatch):
+        persisted: dict = {}
+        self._mock_script_io(monkeypatch, persisted)
+
+        async def mutator(current_script: dict) -> None:
+            await asyncio.sleep(0)
+            current_script.setdefault("flowElements", []).append(build_flow_element("wait"))
+
+        result = asyncio.run(load_and_mutate(perfecto_token, "PRIVATE:Folder/Test.xml", mutator))
+
+        assert result.error is None
+        assert persisted["flow_count"] == 1
+
+    def test_async_mutator_validation_error_skips_persist(self, perfecto_token, monkeypatch):
+        persisted: dict = {}
+        self._mock_script_io(monkeypatch, persisted)
+
+        async def mutator(_current_script: dict) -> None:
+            await asyncio.sleep(0)
+            raise ValueError("Unknown cmd_arguments for command 'wait'")
+
+        result = asyncio.run(load_and_mutate(perfecto_token, "PRIVATE:Folder/Test.xml", mutator))
+
+        assert result.error == "Unknown cmd_arguments for command 'wait'"
+        assert persisted == {}
diff --git a/tools/ai_scriptless/__init__.py b/tools/ai_scriptless/__init__.py
index 7f960c0..6c30ffc 100644
--- a/tools/ai_scriptless/__init__.py
+++ b/tools/ai_scriptless/__init__.py
@@ -5,6 +5,12 @@
     get_command_spec,
     parse_command_id,
 )
+from tools.ai_scriptless.definitions import (
+    declared_parameters,
+    empty_mandatory_note,
+    reset_declared_parameters_cache,
+    validate_argument_names,
+)
 from tools.ai_scriptless.elements import (
     build_arguments,
     build_branch,
@@ -94,8 +100,10 @@
     "coerce_script_dict",
     "coerce_step_path",
     "command_id_from_element",
+    "declared_parameters",
     "delete_element_by_path",
     "delete_script_variable",
+    "empty_mandatory_note",
     "fetch_script_payload",
     "find_container_by_path",
     "find_element_by_path",
@@ -113,6 +121,7 @@
     "new_empty_script",
     "persist_script",
     "parse_command_id",
+    "reset_declared_parameters_cache",
     "script_write_lock",
     "set_condition_expression",
     "set_element_enabled",
@@ -120,6 +129,7 @@
     "strip_non_api_script_fields",
     "update_element_arguments",
     "update_flow_element_counts",
+    "validate_argument_names",
     "validate_step_path",
     "validate_variable_name",
 ]
diff --git a/tools/ai_scriptless/definitions.py b/tools/ai_scriptless/definitions.py
new file mode 100644
index 0000000..0fc1f9a
--- /dev/null
+++ b/tools/ai_scriptless/definitions.py
@@ -0,0 +1,156 @@
+"""Cross-check between command parameters (command repository API) and command
+arguments (script model).
+
+The command repository declares ``mandatoryParameters`` / ``optionalParameters``;
+the script persists those same names as ``FunctionArgument`` entries. The names
+match, the vocabulary does not: a parameter is the declaration, an argument is
+the assigned value. Perfecto silently ignores arguments whose name is not
+declared, so an unknown name only shows up as a step that does nothing at
+execution time. These helpers turn that into an error at authoring time.
+
+Validation fails open on purpose: when the definitions API is unreachable or
+returns no declared parameters, authoring keeps working unvalidated rather than
+becoming unavailable.
+"""
+
+import asyncio
+from difflib import get_close_matches
+from typing import Any, Optional
+
+from config import perfecto
+from config.token import PerfectoToken
+from tools.ai_scriptless.commands import get_command_spec
+from tools.utils import api_request
+
+# (cloud_name, command_id) -> (mandatory, optional) names, or None when unknown.
+DeclaredParameters = tuple[frozenset[str], frozenset[str]]
+
+_declared_parameters_cache: dict[tuple[str, str], Optional[DeclaredParameters]] = {}
+_cache_guard = asyncio.Lock()
+
+
+def reset_declared_parameters_cache() -> None:
+    """Drop memoized definitions (used by tests and after a cloud switch)."""
+    _declared_parameters_cache.clear()
+
+
+async def _fetch_declared_parameters(
+        token: PerfectoToken,
+        command_id: str,
+) -> Optional[DeclaredParameters]:
+    # Local import: formatters.ai_scriptless imports tools.ai_scriptless.elements.
+    from formatters.ai_scriptless import format_command_definitions
+
+    definitions_url = perfecto.get_ai_scriptless_command_repository_url(token.cloud_name)
+    definitions_url = definitions_url + "/commands/definitions"
+    try:
+        result = await api_request(
+            token,
+            "POST",
+            endpoint=definitions_url,
+            json={"commandIds": [command_id]},
+            result_formatter=format_command_definitions,
+        )
+    except Exception:  # noqa: BLE001 - never block authoring on a definitions failure
+        return None
+    if result.error or not isinstance(result.result, list):
+        return None
+    for definition in result.result:
+        if definition.command_id != command_id:
+            continue
+        mandatory = frozenset(definition.mandatory_parameters)
+        optional = frozenset(definition.optional_parameters)
+        # A definition with no declared parameter carries no usable contract.
+        if not mandatory and not optional:
+            return None
+        return mandatory, optional
+    return None
+
+
+async def declared_parameters(
+        token: Optional[PerfectoToken],
+        command_id: str,
+) -> Optional[DeclaredParameters]:
+    """Declared parameter names for command_id, or None when unavailable."""
+    if not token or not command_id:
+        return None
+    cache_key = (token.cloud_name, command_id)
+    async with _cache_guard:
+        if cache_key in _declared_parameters_cache:
+            return _declared_parameters_cache[cache_key]
+    declared = await _fetch_declared_parameters(token, command_id)
+    async with _cache_guard:
+        _declared_parameters_cache[cache_key] = declared
+    return declared
+
+
+def _accepted_names(command_id: str, name: str) -> set[str]:
+    """The name plus its known aliases; the repository may declare any of them."""
+    spec = get_command_spec(command_id)
+    names = {name, spec.argument_aliases.get(name, name)}
+    names |= {alias for alias, canonical in spec.argument_aliases.items() if canonical == name}
+    return names
+
+
+def _argument_value(value: Any) -> Any:
+    if isinstance(value, dict) and "data_source" in value:
+        return value.get("value")
+    return value
+
+
+def validate_argument_names(
+        command_id: str,
+        cmd_arguments: Optional[dict[str, Any]],
+        declared: Optional[DeclaredParameters],
+) -> Optional[str]:
+    """Error message when cmd_arguments carries names the command does not declare."""
+    if not cmd_arguments or not declared:
+        return None
+    mandatory, optional = declared
+    known = mandatory | optional
+    unknown = [name for name in cmd_arguments if not (_accepted_names(command_id, name) & known)]
+    if not unknown:
+        return None
+
+    reported = []
+    for name in unknown:
+        closest = get_close_matches(name, sorted(known), n=1, cutoff=0.6)
+        reported.append(f"'{name}'" + (f" (did you mean '{closest[0]}'?)" if closest else ""))
+    return (
+        f"Unknown cmd_arguments for command '{command_id}': {', '.join(reported)}. "
+        f"Declared parameter names: {', '.join(sorted(known))}"
+        f" (mandatory: {', '.join(sorted(mandatory)) or 'none'}). "
+        "Keys of cmd_arguments are the parameter names from get_command_definitions; "
+        "Perfecto ignores undeclared argument names instead of failing."
+    )
+
+
+def empty_mandatory_note(
+        command_id: str,
+        cmd_arguments: Optional[dict[str, Any]],
+        declared: Optional[DeclaredParameters],
+) -> Optional[str]:
+    """Note when a mandatory parameter is left empty (the step persists but does nothing)."""
+    if not declared:
+        return None
+    mandatory, _optional = declared
+    if not mandatory:
+        return None
+
+    spec = get_command_spec(command_id)
+    values: dict[str, Any] = {
+        name: value for name, (_source, value) in spec.default_arguments_merged().items()
+    }
+    for name, value in spec.normalize_argument_names(cmd_arguments or {}).items():
+        values[name] = _argument_value(value)
+
+    empty = sorted(
+        name for name in mandatory
+        if values.get(name) is None or (isinstance(values[name], str) and not values[name].strip())
+    )
+    if not empty:
+        return None
+    return (
+        f"Mandatory parameter(s) left empty on '{command_id}': {', '.join(empty)}. "
+        "The step is persisted but will not do anything until set with modify_command."
+    )
diff --git a/tools/ai_scriptless/persistence.py b/tools/ai_scriptless/persistence.py
index ff8259c..1137696 100644
--- a/tools/ai_scriptless/persistence.py
+++ b/tools/ai_scriptless/persistence.py
@@ -1,5 +1,6 @@
 import asyncio
 import copy
+import inspect
 import json
 from contextlib import asynccontextmanager
 from typing import Any, Optional
@@ -147,7 +148,10 @@ async def load_and_mutate(
         saved_script = copy.deepcopy(payload.get("script", {}))
         normalize_if_statement_aliases(script)
         try:
-            mutator(script)
+            outcome = mutator(script)
+            # Mutators may be async when they need the API (e.g. command definitions).
+            if inspect.isawaitable(outcome):
+                await outcome
         except ValueError as exc:
             return BaseResult(error=str(exc))
         return await _persist_script(
diff --git a/tools/ai_scriptless_manager.py b/tools/ai_scriptless_manager.py
index e392db2..feeb641 100644
--- a/tools/ai_scriptless_manager.py
+++ b/tools/ai_scriptless_manager.py
@@ -25,8 +25,11 @@
     build_loop,
     build_move_test_body,
     build_snapshot_search_body,
+    command_id_from_element,
+    declared_parameters,
     delete_script_variable,
     delete_element_by_path,
+    empty_mandatory_note,
     fetch_script_payload,
     find_element_by_path,
     find_step_path_for_element,
@@ -43,6 +46,7 @@
     item_key_file_name,
     format_test_ui_location,
     update_element_arguments,
+    validate_argument_names,
 )
 from tools.utils import api_request, format_sanitized_traceback, normalize_action_args
 
@@ -52,6 +56,22 @@
     "do not reuse step_path values from this response.",
 ]
 
+CMD_ARGUMENTS_COLLISION_HINT = (
+    "The ai_user-action command declares a parameter named 'action', which collides with the action key "
+    "of this tool: command arguments must stay nested inside 'cmd_arguments', never flattened into args. "
+    "Example: {\"action\": \"add_command\", \"args\": {\"test_id\": \"...\", "
+    "\"command_id\": \"ai_user-action\", \"cmd_arguments\": {\"action\": \"Tap on the Login button\"}}}."
+)
+
+
+def _unknown_action_error(action: str, args: Dict[str, Any]) -> str:
+    error = f"Action {action} not found in AI Scriptless manager tool"
+    # A command argument flattened to the top level overwrites the dispatcher action with free text.
+    if args.get("command_id") or " " in action:
+        error = f"{error}. {CMD_ARGUMENTS_COLLISION_HINT}"
+    return error
+
+
 def _append_step_path_refresh_notes(result: BaseResult) -> BaseResult:
     if result.error or not isinstance(result.result, dict):
         return result
@@ -256,6 +276,11 @@ async def add_command(
         if not command_id:
             return BaseResult(error="command_id is required (from list_commands)")
 
+        declared = await declared_parameters(self.token, command_id)
+        names_error = validate_argument_names(command_id, cmd_arguments, declared)
+        if names_error:
+            return BaseResult(error=names_error)
+
         element = build_flow_element(command_id, cmd_arguments)
         inserted_path: dict[str, Optional[str]] = {"step_path": None}
 
@@ -268,6 +293,9 @@ def mutator(script: dict[str, Any]) -> None:
             return result
         result.result["step_path"] = inserted_path["step_path"]
         result.result["command_id"] = command_id
+        empty_note = empty_mandatory_note(command_id, cmd_arguments, declared)
+        if empty_note:
+            result.result.setdefault("notes", []).append(empty_note)
         return _append_step_path_refresh_notes(result)
 
     @token_verify
@@ -279,11 +307,17 @@ async def modify_command(self, test_id: str, step_path: str, cmd_arguments: dict
         if not cmd_arguments:
             return BaseResult(error="cmd_arguments is required")
 
-        def mutator(script: dict[str, Any]) -> None:
+        async def mutator(script: dict[str, Any]) -> None:
             located = find_element_by_path(script, step_path)
             if located is None:
                 raise ValueError(f"step_path not found: {step_path}")
             _, _, element = located
+            # command_id is only known after locating the step, so validation happens here.
+            command_id = command_id_from_element(element)
+            declared = await declared_parameters(self.token, command_id)
+            names_error = validate_argument_names(command_id, cmd_arguments, declared)
+            if names_error:
+                raise ValueError(names_error)
             update_element_arguments(element, cmd_arguments)
 
         return _append_step_path_refresh_notes(
@@ -675,20 +709,31 @@ def register(mcp, token: Optional[PerfectoToken]):
     args(dict): Dictionary with the following optional parameters:
         checkpoint (bool, default=false): If true, list checkpoint commands only.
 - get_command_definitions: Get parameter definitions for one or more commands.
+    Returns mandatory_parameters and optional_parameters: these names are exactly the keys to use
+    in cmd_arguments on add_command and modify_command (parameter = the declaration, argument = the value you set).
     args(dict): Dictionary with the following required parameters:
         command_ids (list[str]): Command IDs from list_commands (typically ai_user-action, ai_validation, ai_visual-comparison).
 - add_command: Add a command to a test and persist it.
     args(dict): Dictionary with the following parameters:
         test_id (str, required): Test itemKey from list_tests.
         command_id (str, required): Command ID from list_commands.
-        cmd_arguments (dict, optional): Command argument names to values.
+        cmd_arguments (dict, optional): Command argument names to values. Keys must be parameter names from
+            get_command_definitions (mandatory_parameters / optional_parameters); undeclared keys are rejected.
+            Values are constants by default. To point an argument at a script variable instead of a constant,
+            pass {"data_source": "VARIABLE", "value": "<variable name>"} (see list_test_variables).
+            Never flatten these keys to the top level of args: ai_user-action declares a parameter named
+            'action', which would collide with the action key of this tool. Always nest them in cmd_arguments,
+            e.g. {"action": "add_command", "args": {"test_id": "...", "command_id": "ai_user-action",
+            "cmd_arguments": {"action": "Tap on the Login button"}}}.
         after_path (str, optional): Insert after this step (step_path from view_test_structure).
         parent_path (str, optional): Insert inside a container (step_path of LogicalStep, Loop, or Branch).
 - modify_command: Update command arguments and persist.
     args(dict): Dictionary with the following required parameters:
         test_id (str): Test itemKey from list_tests.
         step_path (str): Step path from view_test_structure (e.g. 0, 2.0, 5.b0.1).
-        cmd_arguments (dict): Argument names to new values.
+        cmd_arguments (dict): Argument names to new values. Merge semantics: only the arguments you send are
+            replaced, the rest keep their current value, and arguments cannot be removed (delete_command removes
+            the whole step). Same key rules as add_command (declared parameter names, optional data_source form).
 - delete_command: Remove a command from a test and persist.
     args(dict): Dictionary with the following required parameters:
         test_id (str): Test itemKey from list_tests.
@@ -787,6 +832,8 @@ def register(mcp, token: Optional[PerfectoToken]):
 - HELP: For product behavior and workarounds, use the perfecto_help tool: Filter by category_id='perfecto', subcategory_id_list=['ide'].
 - UI_ACCESS: No per-test URL exists. Only UI entry: cloud_url/lab/scriptless-mobile/ (cloud_url from perfecto_user read_user). For debugging or unsupported MCP tasks, link the lab URL and tell the user to open the test via Tests → Open or Manage tests using the folder tree and test name from list_tests (itemKey is MCP-only; the UI shows folders and names, not itemKey). Never invent other scriptless URLs.
 - When authoring or editing test steps, call list_commands first and follow the command selection policy in the info field.
+- cmd_arguments keys are validated against the command definitions before saving: an undeclared name is rejected with
+  the list of valid parameter names instead of being persisted as a step argument that Perfecto ignores at runtime.
 - step_path is a dot-separated positional path without spaces (0-based indices; b0=Then branch, b1=Else). Example: root step 3 is "3"; first step inside Then of condition at 5 is "5.b0.0". Perfecto does not persist paths; they change when steps are inserted, moved, or deleted. Always call view_test_structure before the next structure edit; do not reuse step_path from a previous mutation response.
 - Use parent_path on add_command with the step_path of a LogicalStep, Loop, or Branch from view_test_structure.
 - Use add_logical_step, add_loop, and add_condition to build control-flow structures matching the UI toolbar Group, Loop, and Condition actions.
@@ -950,9 +997,7 @@ async def _dispatch():
                         args.get("name", ""),
                     )
                 case _:
-                    return BaseResult(
-                        error=f"Action {action} not found in AI Scriptless manager tool"
-                    )
+                    return BaseResult(error=_unknown_action_error(action, args))
 
         try:
             return await run_tool(f"{TOOLS_PREFIX}_ai_scriptless", action, ctx, _dispatch)
diff --git a/tools/tools_manager.py b/tools/tools_manager.py
index 6bb7ced..f99887d 100644
--- a/tools/tools_manager.py
+++ b/tools/tools_manager.py
@@ -21,7 +21,7 @@
 from models.manager import Manager
 from models.result import BaseResult
 from telemetry import run_tool
-from tools.utils import timeout, user_agent
+from tools.utils import normalize_action_args, timeout, user_agent
 
 
 def _normalize_system(system: str) -> str:
@@ -335,10 +335,10 @@ def register(mcp, token: Optional[PerfectoToken]):
 """
     )
     async def tools(
-            action: str = Field(description="The action id to execute"),
-            args: Dict[str, Any] = Field(description="Dictionary with parameters", default=None),
+            arguments: Dict[str, Any] = Field(description="Dictionary with arguments", default=None),
             ctx: Context = Field(description="Context object providing access to MCP capabilities")
     ) -> BaseResult:
+        action, args = normalize_action_args(arguments)
         if args is None:
             args = {}
         tools_manager = ToolsManager(token, ctx)