feat: Tool Security Policies — per-tool approval gating (#966)

README.md

@@ -312,6 +312,7 @@ This project is licensed under the MIT License - see the [LICENSE](LICENSE) file
 - **[FastMCP](https://github.com/jlowin/fastmcp)**: Excellent MCP server framework
 - **[Model Context Protocol](https://modelcontextprotocol.io/)**: Standardized AI-application communication
 - **[Claude Code](https://github.com/anthropics/claude-code)**: AI-powered coding assistant
+- **[PolicyLayer](https://policylayer.com/)**: Argument-path predicate DSL shape (`args.domain in [...]` with `eq`/`in`/`regex`/`contains`/`exists`/...) inspired the per-tool approval rule schema (#966).
 
 ## 👥 Contributors
 
@@ -360,6 +361,7 @@ This project is licensed under the MIT License - see the [LICENSE](LICENSE) file
 - **[@drseanwing](https://github.com/drseanwing)** — Progress emission via FastMCP `Context` in long-running tools (#1124); tool-discovery / categorized-search docs (#1123).
 - **[@fnordpig](https://github.com/fnordpig)** — Config subentry support (#1393) and Assist pipeline management tool (#1392).
 - **[@paul43210](https://github.com/paul43210)** — `array_patch` mode in `ha_manage_addon` for atomic GET-modify-POST (#1063).
+- **[@L1AD](https://github.com/L1AD)** — Filed #966 proposing per-tool approval gating; pointed to PolicyLayer's MCP-security work as prior art that inspired the predicate DSL shape.
 
 ---
 

homeassistant-addon-dev/DOCS.md

@@ -15,6 +15,7 @@ The dev add-on uses the same configuration as the stable version. See the main a
 | `backup_hint` | Backup strength preference | `normal` |
 | `secret_path` | Custom secret path (optional) | auto-generated |
 | `enable_tool_search` | Replace full tool catalog with search-based discovery (~46K → ~5K tokens). ⚠️ Do NOT enable for Claude Sonnet/Opus — their built-in tool search conflicts with ha-mcp's. Disable one or the other. | `false` |
+| `enable_per_tool_approval` *(beta)* | Gate high-stakes tool calls (lock/alarm control, automation writes, etc.) behind user approval. Guarded calls return an approval URL the user clicks to allow. Per-tool rules with optional argument predicates are configured in the Policies tab of the web UI. | `false` |
 | `enable_yaml_config_editing` *(beta)* | Enables `ha_config_set_yaml` for editing `configuration.yaml` directly. Requires `ha_mcp_tools` custom component. | `false` |
 | `enable_filesystem_tools` *(beta)* | Enables file read/write tools (`ha_list_files`, `ha_read_file`, `ha_write_file`, `ha_delete_file`). Requires `ha_mcp_tools` custom component. | `false` |
 | `enable_custom_component_integration` *(beta)* | Enables `ha_install_mcp_tools` installer tool for the `ha_mcp_tools` custom component. | `false` |

homeassistant-addon-dev/config.yaml

@@ -24,6 +24,7 @@ image: "ghcr.io/homeassistant-ai/ha-mcp-addon-dev-{arch}"
 options:
   backup_hint: "normal"
   enable_tool_search: false
+  enable_per_tool_approval: false
   enable_yaml_config_editing: false
   enable_code_mode: false
   enable_lite_docstrings: false
@@ -38,6 +39,7 @@ schema:
   backup_hint: list(strong|normal|weak|auto)
   secret_path: str?
   enable_tool_search: bool?
+  enable_per_tool_approval: bool?
   enable_yaml_config_editing: bool?
   enable_code_mode: bool?
   enable_lite_docstrings: bool?

homeassistant-addon-dev/translations/en.yaml

@@ -28,6 +28,15 @@ configuration:
       OpenAI-compatible models) or with smaller context windows. Tools
       are found via ha_search_tools and executed via categorized proxies
       (read/write/delete). Requires restart to take effect.
+  enable_per_tool_approval:
+    name: Enable per-tool approval (advanced, beta)
+    description: >-
+      Gate high-stakes tool calls (lock/alarm control, automation writes,
+      etc.) behind user approval. When a guarded tool is called, the agent
+      receives an approval URL; the user must click Approve in the
+      Policies tab of the web UI before the call proceeds. Per-tool rules
+      with optional argument predicates are configured in the Policies
+      tab. Off by default. Requires restart to take effect.
   enable_yaml_config_editing:
     name: Enable YAML config editing (beta)
     description: >-

homeassistant-addon/DOCS.md

@@ -272,13 +272,33 @@ Replaces the full tool catalog (~88 tools, ~46K tokens) with search-based discov
 
 Requires add-on restart to take effect.
 
+### enable_per_tool_approval
+
+**Default:** `false`
+
+Gates high-stakes tool calls (lock/alarm control, automation writes, etc.) behind explicit user approval. When a guarded tool is called, the agent receives an approval URL and the call is held until the user clicks **Approve** in the **Policies** tab of the web UI. Per-tool rules — with optional argument predicates — are configured from the same Policies tab.
+
+**When to enable:**
+- Shared installations where you want a human in the loop for destructive or security-relevant operations
+- Locks, alarms, and other entities where an LLM mistake has real-world consequences
+- Whenever you want a hard audit trail of which tool calls were explicitly allowed
+
+**When to leave disabled (default):**
+- Single-user setups where you're comfortable with the LLM acting autonomously
+- You haven't configured any policy rules yet (with no rules, the toggle has no effect — but the runtime cost is small either way)
+
+The policy engine is in beta. Off by default. Requires add-on restart to take effect.
+
 **Example Configuration:**
 
 ```yaml
-backup_hint: normal
-secret_path: ""  # Leave empty for auto-generation
+enable_per_tool_approval: true
 ```
 
+Per-tool rules (including argument predicates like `args.domain in ['lock', 'alarm_control_panel']`) are configured from the **Policies** tab in the web UI, not from `config.yaml`.
+
+*Inspired by [PolicyLayer](https://policylayer.com/)'s policy DSL shape, originally proposed in [#966](https://github.com/homeassistant-ai/ha-mcp/issues/966) by [@L1AD](https://github.com/L1AD).*
+
 ---
 
 ## Security

homeassistant-addon/config.yaml

@@ -32,6 +32,7 @@ image: "ghcr.io/homeassistant-ai/ha-mcp-addon-{arch}"
 options:
   backup_hint: "normal"
   enable_tool_search: false
+  enable_per_tool_approval: false
   enable_auto_backup: true
   auto_backup_throttle_minutes: 0
   auto_backup_retain_per_entity: 100
@@ -40,6 +41,7 @@ schema:
   backup_hint: list(strong|normal|weak|auto)
   secret_path: str?
   enable_tool_search: bool?
+  enable_per_tool_approval: bool?
   enable_auto_backup: bool?
   auto_backup_throttle_minutes: int(0,1440)?
   auto_backup_retain_per_entity: int(1,10000)?

homeassistant-addon/start.py

@@ -219,6 +219,7 @@ def main() -> int:
     backup_hint = "normal"  # default
     custom_secret_path = ""  # default
     enable_tool_search = False  # default
+    enable_per_tool_approval = False  # default
     enable_yaml_config_editing = False  # default
     enable_filesystem_tools = False  # default
     enable_custom_component_integration = False  # default
@@ -245,6 +246,12 @@ def main() -> int:
             enable_tool_search = (
                 raw_tool_search if isinstance(raw_tool_search, bool) else False
             )
+            raw_per_tool_approval = config.get("enable_per_tool_approval", False)
+            enable_per_tool_approval = (
+                raw_per_tool_approval
+                if isinstance(raw_per_tool_approval, bool)
+                else False
+            )
             raw_yaml_config = config.get("enable_yaml_config_editing", False)
             enable_yaml_config_editing = (
                 raw_yaml_config if isinstance(raw_yaml_config, bool) else False
@@ -322,6 +329,7 @@ def main() -> int:
     os.environ["HOMEASSISTANT_URL"] = "http://supervisor/core"
     os.environ["BACKUP_HINT"] = backup_hint
     os.environ["ENABLE_TOOL_SEARCH"] = str(enable_tool_search).lower()
+    os.environ["ENABLE_PER_TOOL_APPROVAL"] = str(enable_per_tool_approval).lower()
     os.environ["ENABLE_YAML_CONFIG_EDITING"] = str(enable_yaml_config_editing).lower()
     os.environ["HAMCP_ENABLE_FILESYSTEM_TOOLS"] = str(enable_filesystem_tools).lower()
     os.environ["HAMCP_ENABLE_CUSTOM_COMPONENT_INTEGRATION"] = str(

homeassistant-addon/translations/en.yaml

@@ -28,6 +28,15 @@ configuration:
       OpenAI-compatible models) or with smaller context windows. Tools
       are found via ha_search_tools and executed via categorized proxies
       (read/write/delete). Requires restart to take effect.
+  enable_per_tool_approval:
+    name: Enable per-tool approval (advanced, beta)
+    description: >-
+      Gate high-stakes tool calls (lock/alarm control, automation writes,
+      etc.) behind user approval. When a guarded tool is called, the agent
+      receives an approval URL; the user must click Approve in the
+      Policies tab of the web UI before the call proceeds. Per-tool rules
+      with optional argument predicates are configured in the Policies
+      tab. Off by default. Requires restart to take effect.
   enable_auto_backup:
     name: Enable auto-backup of edits
     description: >-

src/ha_mcp/config.py

@@ -100,6 +100,11 @@ class Settings(BaseSettings):
     # Dramatically reduces idle context token usage for LLMs.
     enable_tool_search: bool = Field(False, alias="ENABLE_TOOL_SEARCH")
 
+    # Per-tool approval middleware — opt-in gate that routes high-stakes tool
+    # calls through a per-tool policy with out-of-band web-UI approval
+    # (issue #966). Disabled by default.
+    enable_per_tool_approval: bool = Field(False, alias="ENABLE_PER_TOOL_APPROVAL")
+
     # Managed YAML config editing — allows ha_config_set_yaml to add,
     # replace, or remove top-level keys in configuration.yaml and package
     # files. Disabled by default; only for YAML-only features with no UI/API path.

src/ha_mcp/policy/__init__.py

@@ -0,0 +1 @@
+"""Per-tool approval policy for high-stakes MCP tool calls."""

src/ha_mcp/policy/approval_queue.py

@@ -0,0 +1,122 @@
+"""In-memory, per-process approval queue with args-hash binding and remember-cache."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import secrets
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta
+from typing import Any, Literal
+
+import anyio
+
+Decision = Literal["pending", "approved", "denied"]
+
+
+def compute_args_hash(args: dict[str, Any]) -> str:
+    """Canonical sha256 of args. Same hash function used at insert and lookup."""
+    payload = json.dumps(args, sort_keys=True, separators=(",", ":"), default=str)
+    return hashlib.sha256(payload.encode()).hexdigest()
+
+
+@dataclass
+class PendingApproval:
+    token: str
+    tool_name: str
+    args_hash: str
+    args_preview: dict[str, Any]
+    created_at: datetime
+    expires_at: datetime
+    decision: Decision = "pending"
+    event: anyio.Event = field(default_factory=anyio.Event)
+
+
+class ApprovalQueue:
+    """In-memory store. Per-process. Token-keyed for HTTP lookup,
+    (tool, hash)-indexed for re-call lookup."""
+
+    def __init__(self) -> None:
+        self._by_token: dict[str, PendingApproval] = {}
+        self._remember: dict[tuple[str, str], datetime] = {}
+
+    # --- remember cache ---
+    def remember(self, tool_name: str, args_hash: str, *, minutes: int) -> None:
+        if minutes <= 0:
+            return
+        self._remember[(tool_name, args_hash)] = datetime.now(UTC) + timedelta(
+            minutes=minutes
+        )
+
+    def is_remembered(self, tool_name: str, args_hash: str) -> bool:
+        until = self._remember.get((tool_name, args_hash))
+        if until is None:
+            return False
+        if datetime.now(UTC) >= until:
+            self._remember.pop((tool_name, args_hash), None)
+            return False
+        return True
+
+    # --- pending entries lifecycle ---
+    def create(
+        self,
+        tool_name: str,
+        args_hash: str,
+        args_preview: dict[str, Any],
+        *,
+        ttl_minutes: int,
+    ) -> PendingApproval:
+        now = datetime.now(UTC)
+        entry = PendingApproval(
+            token=secrets.token_urlsafe(24),
+            tool_name=tool_name,
+            args_hash=args_hash,
+            args_preview=args_preview,
+            created_at=now,
+            expires_at=now + timedelta(minutes=ttl_minutes),
+        )
+        self._by_token[entry.token] = entry
+        return entry
+
+    def find(self, tool_name: str, args_hash: str) -> PendingApproval | None:
+        self._sweep_expired()
+        for entry in self._by_token.values():
+            if entry.tool_name == tool_name and entry.args_hash == args_hash:
+                return entry
+        return None
+
+    def get(self, token: str) -> PendingApproval | None:
+        self._sweep_expired()
+        return self._by_token.get(token)
+
+    def list_pending(self) -> list[PendingApproval]:
+        self._sweep_expired()
+        return [e for e in self._by_token.values() if e.decision == "pending"]
+
+    def approve(self, token: str) -> None:
+        entry = self._by_token.get(token)
+        if entry and entry.decision == "pending":
+            entry.decision = "approved"
+            entry.event.set()
+
+    def deny(self, token: str) -> None:
+        entry = self._by_token.get(token)
+        if entry and entry.decision == "pending":
+            entry.decision = "denied"
+            entry.event.set()
+
+    def remove(self, token: str) -> None:
+        self._by_token.pop(token, None)
+
+    def consume_and_maybe_remember(
+        self, entry: PendingApproval, *, remember_minutes: int
+    ) -> None:
+        self.remove(entry.token)
+        if remember_minutes > 0:
+            self.remember(entry.tool_name, entry.args_hash, minutes=remember_minutes)
+
+    def _sweep_expired(self) -> None:
+        now = datetime.now(UTC)
+        stale = [t for t, e in self._by_token.items() if e.expires_at <= now]
+        for t in stale:
+            self._by_token.pop(t, None)

src/ha_mcp/policy/evaluator.py

@@ -0,0 +1,90 @@
+"""Evaluate a tool call against a Policy. Pure functions — no I/O, no state."""
+
+import re
+from enum import StrEnum
+from typing import Any
+
+from .model import Policy, Predicate, Rule
+
+_MISSING = object()
+
+
+class Verdict(StrEnum):
+    ALLOW = "allow"
+    REQUIRE_APPROVAL = "require_approval"
+
+
+def extract_path(args: dict[str, Any], path: str) -> Any:
+    """Walk a dotted path like 'args.config.alias' against the args dict.
+
+    'args' is implicit — the leading 'args.' is stripped. Returns _MISSING if any
+    intermediate key is absent.
+    """
+    parts = path.split(".")
+    if parts[0] == "args":
+        parts = parts[1:]
+    cur: Any = args
+    for part in parts:
+        if not isinstance(cur, dict) or part not in cur:
+            return _MISSING
+        cur = cur[part]
+    return cur
+
+
+def match_predicate(predicate: Predicate, args: dict[str, Any]) -> bool:
+    val = extract_path(args, predicate.path)
+    if predicate.op == "exists":
+        return val is not _MISSING
+    if val is _MISSING:
+        return False
+    pv = predicate.value
+    match predicate.op:
+        case "eq":
+            return bool(val == pv)
+        case "neq":
+            return bool(val != pv)
+        case "in":
+            return val in (pv or [])
+        case "not_in":
+            return val not in (pv or [])
+        # `regex` is re.search (substring match). Anchor with ^...$ for full-match.
+        case "regex":
+            return (
+                isinstance(val, str)
+                and isinstance(pv, str)
+                and re.search(pv, val) is not None
+            )
+        case "contains":
+            return isinstance(val, (str, list, tuple, set)) and pv in val
+        case "gt":
+            return bool(val > pv)
+        case "lt":
+            return bool(val < pv)
+    return False
+
+
+def match_rule(rule: Rule, tool_name: str, args: dict[str, Any]) -> bool:
+    if rule.tool_name != "*" and rule.tool_name != tool_name:
+        return False
+    return all(match_predicate(p, args) for p in rule.when)
+
+
+def find_matching_rule(
+    tool_name: str, args: dict[str, Any], policy: Policy
+) -> Rule | None:
+    for rule in policy.rules:
+        if match_rule(rule, tool_name, args):
+            return rule
+    return None
+
+
+def evaluate(tool_name: str, args: dict[str, Any], policy: Policy) -> Verdict:
+    if not policy.enabled:
+        return Verdict.ALLOW
+    if find_matching_rule(tool_name, args, policy) is not None:
+        return Verdict.REQUIRE_APPROVAL
+    return (
+        Verdict.REQUIRE_APPROVAL
+        if policy.default_action == "require_approval"
+        else Verdict.ALLOW
+    )