fix(mcp): require opt-in for TestQL mutations

packages/mcp2testql/README.md

@@ -1,3 +1,7 @@
 # mcp2testql
 
 MCP server for TestQL control (query, patch, validate, DSL).
+
+Query and validation tools remain read-only. Materialization, patching and DSL
+execution are disabled unless the server operator explicitly sets
+`TESTQL_MCP_ALLOW_MUTATION=1` for trusted MCP clients.

packages/mcp2testql/src/mcp2testql/server.py

@@ -2,9 +2,20 @@
 
 from __future__ import annotations
 
+import os
 from dataclasses import dataclass
 from typing import Any
 
+_MUTATION_ENV = "TESTQL_MCP_ALLOW_MUTATION"
+
+
+def _require_mutation(action: str) -> None:
+    enabled = os.getenv(_MUTATION_ENV, "").strip().lower() in {"1", "true", "yes", "on"}
+    if not enabled:
+        raise PermissionError(
+            f"MCP mutation '{action}' is disabled; start the server with {_MUTATION_ENV}=1"
+        )
+
 
 def _require_fastmcp():
     try:
@@ -44,6 +55,7 @@ def testql_query(uri: str, file: str = "", fmt: str = "json") -> dict[str, Any]:
         @self.app.tool()
         def testql_materialize(uri: str, dest: str = "") -> dict[str, Any]:
             """Materialize addressed TestQL content to a file."""
+            _require_mutation("testql_materialize")
             result = materialize_uri(uri, dest=dest or None)
             return result.to_dict()
 
@@ -57,30 +69,35 @@ def testql_validate(path: str) -> dict[str, Any]:
         @self.app.tool()
         def testql_run_dsl(script: str, default_file: str = "") -> list[dict[str, Any]]:
             """Execute TestQL control DSL commands (one per line)."""
+            _require_mutation("testql_run_dsl")
             results = execute_dsl(script, default_file=default_file or None)
             return [r.to_dict() for r in results]
 
         @self.app.tool()
         def testql_run_command(command: str, default_file: str = "") -> dict[str, Any]:
             """Execute a single TestQL control DSL command."""
+            _require_mutation("testql_run_command")
             result = execute_dsl_line(command, default_file=default_file or None)
             return result.to_dict()
 
         @self.app.tool()
         def testql_run_command_pb(envelope_bytes: bytes, default_file: str = "") -> bytes:
             """Execute protobuf DslEnvelope; returns DslResult protobuf."""
+            _require_mutation("testql_run_command_pb")
             result = dispatch(envelope_bytes, default_file=default_file or None)
             return encode_result_protobuf(result)
 
         @self.app.tool()
         def testql_patch(uri: str, content: str, file: str = "") -> dict[str, Any]:
             """Replace a TestQL block referenced by URI."""
+            _require_mutation("testql_patch")
             result = patch_uri(uri, content=content, file=file or None)
             return result.to_dict()
 
         @self.app.tool()
         def testql_update(uri: str, content: str, file: str = "") -> dict[str, Any]:
             """Update a TestQL block referenced by URI."""
+            _require_mutation("testql_update")
             result = update_uri(uri, content=content, file=file or None)
             return result.to_dict()
 
@@ -93,6 +110,7 @@ def testql_apply(
             file: str = "",
         ) -> dict[str, Any]:
             """Apply URI action: materialize, patch, append, update."""
+            _require_mutation("testql_apply")
             result = apply_uri(
                 uri,
                 dest=dest or None,

packages/mcp2testql/tests/test_mcp2testql.py

@@ -0,0 +1,14 @@
+"""Safety tests for mcp2testql."""
+
+import pytest
+
+from mcp2testql.server import _require_mutation
+
+
+def test_mcp_mutations_require_operator_capability(monkeypatch) -> None:
+    monkeypatch.delenv("TESTQL_MCP_ALLOW_MUTATION", raising=False)
+    with pytest.raises(PermissionError, match="TESTQL_MCP_ALLOW_MUTATION"):
+        _require_mutation("testql_patch")
+
+    monkeypatch.setenv("TESTQL_MCP_ALLOW_MUTATION", "true")
+    _require_mutation("testql_patch")