Validate operation value arity in propose_operations

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "vymcp"
-version = "0.1.1"
+version = "0.1.2"
 description = "MCP server for VyOS routers, powered by VyManager"
 readme = "README.md"
 requires-python = ">=3.10"

src/vymcp/__init__.py

@@ -1,3 +1,3 @@
 """VyMCP — Model Context Protocol server for VyOS routers, powered by VyManager."""
 
-__version__ = "0.1.0"
+__version__ = "0.1.2"

src/vymcp/writes.py

@@ -177,17 +177,33 @@ async def describe_feature_operations(feature: str) -> dict[str, Any]:
         operations = await discovery.get_feature_operations(feature)
         fields = await discovery.get_top_level_fields(feature)
         subject_field = await discovery.get_subject_field(feature)
+
+        # Spell out, per op, exactly what the *value* must contain: the args AFTER
+        # the subject (which is passed separately via fields). This is the contract
+        # callers most often get wrong — mixing the subject back into the value.
+        subject_args = 1 if subject_field else 0
+        described = [
+            {
+                **op,
+                "value_args": op.get("args", [])[subject_args:],
+                "value_arg_count": max(op.get("arg_count", 0) - subject_args, 0),
+            }
+            for op in operations
+        ]
+        note = (
+            f"Pass the subject via fields['{subject_field}']. Each op's 'value' supplies "
+            "ONLY its 'value_args', comma-joined in order — never include the subject in "
+            "the value. Ops with value_arg_count 0 take no value."
+            if subject_field
+            else "Each op's 'value' supplies its 'value_args', comma-joined in order. "
+            "Ops with value_arg_count 0 take no value."
+        )
         return {
             "feature": feature,
             "subject_field": subject_field,
             "top_level_fields": fields,
-            "operations": operations,
-            "note": (
-                f"This feature requires the '{subject_field}' field (the subject, e.g. the "
-                f"interface name); each op's value supplies the args after it."
-                if subject_field
-                else None
-            ),
+            "operations": described,
+            "note": note,
         }
 
     @mcp.tool()
@@ -233,18 +249,32 @@ async def propose_operations(
                     f"describe_feature_operations('{feature}') for the vocabulary."
                 )
             value = op.get("value")
-            if vocab[name]["arg_count"] - subject_args >= 1 and not value:
+            expected = vocab[name]["arg_count"] - subject_args
+            if expected >= 1 and not value:
                 raise ValueError(f"Operation '{name}' requires a value.")
+            if expected >= 1 and value:
+                parts = value.split(",")
+                if len(parts) != expected:
+                    wanted = ", ".join(vocab[name].get("args", [])[subject_args:])
+                    hint = (
+                        f" The subject '{subject_field}' is supplied via fields — do NOT "
+                        "include it in the value."
+                        if subject_field
+                        else ""
+                    )
+                    raise ValueError(
+                        f"Operation '{name}' expects {expected} comma-separated value(s) "
+                        f"({wanted}), but got {len(parts)}: {value!r}.{hint}"
+                    )
             entry: dict[str, Any] = {"op": name}
             if value is not None:
                 entry["value"] = value
             normalized.append(entry)
 
         body: dict[str, Any] = dict(fields or {})
         body["operations"] = normalized
-        summary = (
-            f"{len(normalized)} operation(s) on '{feature}' (instance {instance_id})"
-            + (f"; fields {fields}" if fields else "")
+        summary = f"{len(normalized)} operation(s) on '{feature}' (instance {instance_id})" + (
+            f"; fields {fields}" if fields else ""
         )
         plan = plan_store.create(
             owner=current_owner(),

tests/test_generic_driver.py

@@ -5,8 +5,12 @@
     "feature": "nat",
     "operations": [
         {"op": "set_source_rule", "args": ["rule"], "arg_count": 1, "description": "Rule."},
-        {"op": "set_source_rule_translation_address", "args": ["rule", "addr"],
-         "arg_count": 2, "description": "Translation address."},
+        {
+            "op": "set_source_rule_translation_address",
+            "args": ["rule", "addr"],
+            "arg_count": 2,
+            "description": "Translation address.",
+        },
         {"op": "delete_all", "args": [], "arg_count": 0, "description": "Delete all."},
     ],
 }
@@ -51,27 +55,32 @@ def handler(request: httpx.Request) -> httpx.Response:
             if r is not None:
                 return r
         return httpx.Response(404, json={})
+
     return handler
 
 
 async def test_describe_merges_ops_and_fields(write_tools, install_client):
     install_client(_discovery_handler())
     out = await write_tools["describe_feature_operations"]("nat")
     assert {o["op"] for o in out["operations"]} == {
-        "set_source_rule", "set_source_rule_translation_address", "delete_all"
+        "set_source_rule",
+        "set_source_rule_translation_address",
+        "delete_all",
     }
     assert out["top_level_fields"]["nat_type"]["required"] is True
 
 
 async def test_propose_builds_body_with_fields(write_tools, install_client):
     install_client(_discovery_handler())
     plan = await write_tools["propose_operations"](
-        "nat", "i1",
+        "nat",
+        "i1",
         [{"op": "set_source_rule", "value": "100"}],
         fields={"nat_type": "source"},
     )
     assert plan["applied"] is False
     from vymcp.changes import plan_store
+
     body = plan_store.get(plan["plan_id"], "local").body
     assert body == {"nat_type": "source", "operations": [{"op": "set_source_rule", "value": "100"}]}
 
@@ -92,6 +101,7 @@ async def test_propose_zero_arg_op_needs_no_value(write_tools, install_client):
     install_client(_discovery_handler())
     plan = await write_tools["propose_operations"]("nat", "i1", [{"op": "delete_all"}])
     from vymcp.changes import plan_store
+
     assert plan_store.get(plan["plan_id"], "local").body["operations"] == [{"op": "delete_all"}]
 
 
@@ -106,8 +116,12 @@ async def test_propose_rejects_empty(write_tools, install_client):
     "subject_field": "interface_name",
     "operations": [
         {"op": "set_interface_disable", "args": ["interface"], "arg_count": 1, "description": "x"},
-        {"op": "set_interface_address", "args": ["interface", "addr"],
-         "arg_count": 2, "description": "x"},
+        {
+            "op": "set_interface_address",
+            "args": ["interface", "addr"],
+            "arg_count": 2,
+            "description": "x",
+        },
     ],
 }
 
@@ -123,9 +137,7 @@ def _bonding_handler(request: httpx.Request) -> httpx.Response:
 async def test_subject_feature_requires_subject(write_tools, install_client):
     install_client(_bonding_handler)
     with pytest.raises(ValueError, match="interface_name"):
-        await write_tools["propose_operations"](
-            "bonding", "i1", [{"op": "set_interface_disable"}]
-        )
+        await write_tools["propose_operations"]("bonding", "i1", [{"op": "set_interface_disable"}])
 
 
 async def test_subject_feature_zero_value_op_ok(write_tools, install_client):
@@ -135,6 +147,7 @@ async def test_subject_feature_zero_value_op_ok(write_tools, install_client):
         "bonding", "i1", [{"op": "set_interface_disable"}], fields={"interface_name": "bond0"}
     )
     from vymcp.changes import plan_store
+
     body = plan_store.get(plan["plan_id"], "local").body
     assert body == {"interface_name": "bond0", "operations": [{"op": "set_interface_disable"}]}
 
@@ -148,6 +161,54 @@ async def test_subject_feature_value_op_needs_value(write_tools, install_client)
         )
 
 
+async def test_subject_included_in_value_is_rejected(write_tools, install_client):
+    install_client(_bonding_handler)
+    # The classic mistake: subject re-jammed into the value. set_interface_address
+    # wants 1 value arg (the address); passing "bond0,10.0.0.1/32" is 2 -> reject
+    # here instead of letting the device answer an opaque 400.
+    with pytest.raises(ValueError, match="expects 1 comma-separated value"):
+        await write_tools["propose_operations"](
+            "bonding",
+            "i1",
+            [{"op": "set_interface_address", "value": "bond0,10.0.0.1/32"}],
+            fields={"interface_name": "bond0"},
+        )
+
+
+async def test_correct_single_value_accepted(write_tools, install_client):
+    install_client(_bonding_handler)
+    plan = await write_tools["propose_operations"](
+        "bonding",
+        "i1",
+        [{"op": "set_interface_address", "value": "10.0.0.1/32"}],
+        fields={"interface_name": "bond0"},
+    )
+    from vymcp.changes import plan_store
+
+    body = plan_store.get(plan["plan_id"], "local").body
+    assert body["operations"] == [{"op": "set_interface_address", "value": "10.0.0.1/32"}]
+
+
+async def test_too_many_args_rejected_non_subject(write_tools, install_client):
+    install_client(_discovery_handler())
+    # set_source_rule takes 1 arg; passing 2 comma parts is rejected.
+    with pytest.raises(ValueError, match="expects 1 comma-separated value"):
+        await write_tools["propose_operations"](
+            "nat", "i1", [{"op": "set_source_rule", "value": "100,extra"}]
+        )
+
+
+async def test_describe_exposes_value_args(write_tools, install_client):
+    install_client(_bonding_handler)
+    out = await write_tools["describe_feature_operations"]("bonding")
+    ops = {o["op"]: o for o in out["operations"]}
+    # The subject is stripped from what the value must supply.
+    assert ops["set_interface_address"]["value_args"] == ["addr"]
+    assert ops["set_interface_address"]["value_arg_count"] == 1
+    assert ops["set_interface_disable"]["value_arg_count"] == 0
+    assert "never include the subject" in out["note"]
+
+
 async def test_generic_plan_applies(write_tools, install_client):
     def extra(request):
         p = request.url.path