Gate destructive tools behind elicitation confirmation

Add MCPServer.request_elicitation that fires a server-initiated
elicitation/create when JE_AUTOCONTROL_MCP_CONFIRM_DESTRUCTIVE=1
and the client advertised the elicitation capability. Tools whose
annotations mark them destructive (and not read-only) are gated:
the user sees a confirmation prompt before the action runs, and
declining returns a clean -32000 error to the model. Servers that
talk to non-elicitation clients fall through with a logged warning
so the feature is opt-in and never blocks unexpectedly.

Author: JE-Chen

je_auto_control/utils/mcp_server/server.py

@@ -510,6 +510,7 @@ def _handle_tools_call(self, msg_id: Any,
             raise _MCPError(-32602, f"Invalid arguments for {name}: {violation}")
         if self._rate_limiter is not None and not self._rate_limiter.try_acquire():
             raise _MCPError(-32000, f"Rate limit exceeded for tool {name!r}")
+        self._maybe_confirm_destructive(name, tool, arguments)
         ctx = self._build_call_context(msg_id, params)
         with self._calls_lock:
             self._active_calls[msg_id] = ctx
@@ -551,6 +552,21 @@ def _handle_tools_call(self, msg_id: Any,
             "isError": False,
         }
 
+    def request_elicitation(self, message: str,
+                            requested_schema: Optional[Dict[str, Any]] = None,
+                            timeout: float = 60.0) -> Dict[str, Any]:
+        """Ask the connected client to elicit a response from the user.
+
+        Returns the raw payload (typically ``{"action": "accept" | "decline" | "cancel", ...}``).
+        Requires the client to advertise the ``elicitation`` capability.
+        """
+        params: Dict[str, Any] = {"message": str(message)}
+        if requested_schema is not None:
+            params["requestedSchema"] = requested_schema
+        return self._send_outbound_request(
+            "elicitation/create", params=params, timeout=timeout,
+        )
+
     def request_sampling(self, messages: List[Dict[str, Any]],
                          system_prompt: Optional[str] = None,
                          max_tokens: int = 1024,
@@ -600,6 +616,42 @@ def request_sampling(self, messages: List[Dict[str, Any]],
             raise RuntimeError(f"sampling failed: {slot['error']}")
         return slot.get("result") or {}
 
+    def _maybe_confirm_destructive(self, name: str, tool: MCPTool,
+                                    arguments: Dict[str, Any]) -> None:
+        """Ask the client to confirm before running a destructive tool."""
+        if not _confirm_destructive_enabled():
+            return
+        annotations = tool.annotations
+        if annotations.read_only or not annotations.destructive:
+            return
+        if "elicitation" not in self._client_capabilities:
+            autocontrol_logger.info(
+                "MCP confirmation requested for %s but client lacks "
+                "elicitation capability — proceeding without prompt", name,
+            )
+            return
+        if self._writer is None:
+            return
+        prompt = (f"AutoControl is about to run a destructive tool "
+                  f"'{name}'. Continue?")
+        try:
+            response = self.request_elicitation(
+                message=prompt, requested_schema={"type": "object",
+                                                    "properties": {}},
+                timeout=60.0,
+            )
+        except (RuntimeError, TimeoutError) as error:
+            autocontrol_logger.info(
+                "MCP elicitation for %s failed (%r) — refusing call",
+                name, error,
+            )
+            raise _MCPError(-32000,
+                             f"User confirmation unavailable for {name}")
+        action = response.get("action") if isinstance(response, dict) else None
+        if action != "accept":
+            raise _MCPError(-32000, f"User declined to run {name}: action={action!r}")
+        del arguments  # available for future per-arg confirmation policies
+
     def _build_call_context(self, msg_id: Any,
                             params: Dict[str, Any]) -> ToolCallContext:
         meta = params.get("_meta") if isinstance(params.get("_meta"),
@@ -631,6 +683,12 @@ def _stringify_result(value: Any) -> str:
         return repr(value)
 
 
+def _confirm_destructive_enabled() -> bool:
+    """Return True when the operator wants destructive tools gated on user OK."""
+    raw = os.environ.get("JE_AUTOCONTROL_MCP_CONFIRM_DESTRUCTIVE", "")
+    return raw.strip().lower() in {"1", "true", "yes", "on"}
+
+
 def _capture_error_screenshot(tool_name: str) -> Optional[str]:
     """Save a debug screenshot when JE_AUTOCONTROL_MCP_ERROR_SHOTS is set."""
     debug_dir = os.environ.get("JE_AUTOCONTROL_MCP_ERROR_SHOTS")

test/unit_test/headless/test_mcp_server.py

@@ -1575,6 +1575,116 @@ def test_resources_subscribe_rejects_unknown_uri():
     assert response["error"]["code"] == -32602
 
 
+def test_destructive_confirmation_blocks_when_user_declines(monkeypatch):
+    monkeypatch.setenv("JE_AUTOCONTROL_MCP_CONFIRM_DESTRUCTIVE", "1")
+    captured_lines = []
+    tool = MCPTool(
+        name="zap", description="zap",
+        input_schema={"type": "object", "properties": {}},
+        handler=lambda: "should not run",
+    )
+    server = MCPServer(tools=[tool], concurrent_tools=True)
+    server.set_writer(captured_lines.append)
+    server._client_capabilities = {"elicitation": {}}
+
+    def run_call():
+        server.handle_line(_request("tools/call", msg_id=11, params={
+            "name": "zap", "arguments": {},
+        }))
+
+    t = threading.Thread(target=run_call)
+    t.start()
+    deadline = threading.Event()
+    for _ in range(200):
+        if any('"elicitation/create"' in line for line in captured_lines):
+            break
+        deadline.wait(0.01)
+    eli_lines = [line for line in captured_lines
+                  if '"elicitation/create"' in line]
+    assert eli_lines, "expected elicitation/create"
+    eli_id = json.loads(eli_lines[-1])["id"]
+
+    server.handle_line(json.dumps({
+        "jsonrpc": "2.0", "id": eli_id,
+        "result": {"action": "decline"},
+    }))
+    t.join(timeout=2.0)
+    assert not t.is_alive()
+    final_lines = [line for line in captured_lines if '"id": 11' in line]
+    assert final_lines
+    final = json.loads(final_lines[-1])
+    assert final["error"]["code"] == -32000
+    assert "declined" in final["error"]["message"]
+
+
+def test_destructive_confirmation_allows_when_user_accepts(monkeypatch):
+    monkeypatch.setenv("JE_AUTOCONTROL_MCP_CONFIRM_DESTRUCTIVE", "1")
+    captured_lines = []
+    tool = MCPTool(
+        name="zap2", description="zap2",
+        input_schema={"type": "object", "properties": {}},
+        handler=lambda: "ran",
+    )
+    server = MCPServer(tools=[tool], concurrent_tools=True)
+    server.set_writer(captured_lines.append)
+    server._client_capabilities = {"elicitation": {}}
+
+    def run_call():
+        server.handle_line(_request("tools/call", msg_id=12, params={
+            "name": "zap2", "arguments": {},
+        }))
+
+    t = threading.Thread(target=run_call)
+    t.start()
+    deadline = threading.Event()
+    for _ in range(200):
+        if any('"elicitation/create"' in line for line in captured_lines):
+            break
+        deadline.wait(0.01)
+    eli_id = json.loads([line for line in captured_lines
+                          if '"elicitation/create"' in line][-1])["id"]
+
+    server.handle_line(json.dumps({
+        "jsonrpc": "2.0", "id": eli_id,
+        "result": {"action": "accept", "content": {}},
+    }))
+    t.join(timeout=2.0)
+    final = json.loads([line for line in captured_lines
+                          if '"id": 12' in line][-1])
+    assert final["result"]["isError"] is False
+    assert final["result"]["content"][0]["text"] == "ran"
+
+
+def test_destructive_confirmation_skipped_when_client_lacks_capability(monkeypatch):
+    monkeypatch.setenv("JE_AUTOCONTROL_MCP_CONFIRM_DESTRUCTIVE", "1")
+    tool = MCPTool(
+        name="zap3", description="zap3",
+        input_schema={"type": "object", "properties": {}},
+        handler=lambda: "ok",
+    )
+    server = MCPServer(tools=[tool])
+    # Client did not advertise elicitation — server proceeds without asking.
+    response = _decode(server.handle_line(_request("tools/call", params={
+        "name": "zap3", "arguments": {},
+    })))
+    assert response["result"]["isError"] is False
+
+
+def test_destructive_confirmation_skipped_when_env_unset(monkeypatch):
+    monkeypatch.delenv("JE_AUTOCONTROL_MCP_CONFIRM_DESTRUCTIVE", raising=False)
+    tool = MCPTool(
+        name="zap4", description="zap4",
+        input_schema={"type": "object", "properties": {}},
+        handler=lambda: "ok",
+    )
+    server = MCPServer(tools=[tool])
+    server._client_capabilities = {"elicitation": {}}
+    response = _decode(server.handle_line(_request("tools/call", params={
+        "name": "zap4", "arguments": {},
+    })))
+    assert response["result"]["isError"] is False
+
+
 def test_default_registry_lists_core_automation_tools():
     names = {tool.name for tool in build_default_tool_registry()}
     expected = {