feat(gateway): add webhook payload filters

cli-config.yaml.example

@@ -846,6 +846,11 @@ platform_toolsets:
 #         priority_mode: prepend
 #         priority:
 #           - my_plugin_command
+#   webhook:
+#     extra:
+#       # Route scripts default to a 30 second timeout. Scripts must live under
+#       # the active profile's scripts directory and receive webhook JSON on stdin.
+#       script_timeout_seconds: 30
 #
 # Discord-specific settings (config.yaml top-level, not under platforms:):
 #

gateway/platforms/webhook.py

@@ -58,6 +58,10 @@ from gateway.platforms.base import (
     MessageType,
     SendResult,
 )
+from gateway.platforms.webhook_filters import (
+    DEFAULT_SCRIPT_TIMEOUT_SECONDS,
+    WebhookRouteProcessor,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -78,7 +82,6 @@ DEFAULT_PORT = 8644
 _INSECURE_NO_AUTH = "INSECURE_NO_AUTH"
 _DYNAMIC_ROUTES_FILENAME = "webhook_subscriptions.json"
 _RATE_WINDOW_SECONDS = 60.0
-
 # Hostnames/IP literals that only serve connections originating on the same
 # machine. Anything else is treated as a public bind for safety-rail purposes.
 _LOOPBACK_HOSTS = frozenset({
@@ -155,6 +158,15 @@ class WebhookAdapter(BasePlatformAdapter):
         self._max_body_bytes: int = int(
             config.extra.get("max_body_bytes", 1_048_576)
         )  # 1MB
+        self._script_timeout_seconds: int = int(
+            config.extra.get(
+                "script_timeout_seconds",
+                DEFAULT_SCRIPT_TIMEOUT_SECONDS,
+            )
+        )
+        self._route_processor = WebhookRouteProcessor(
+            script_timeout_seconds=self._script_timeout_seconds
+        )
 
     # ------------------------------------------------------------------
     # Lifecycle
@@ -571,6 +583,41 @@ class WebhookAdapter(BasePlatformAdapter):
                 {"status": "ignored", "event": event_type}
             )
 
+        if not self._route_processor.route_filters_match(
+            route_config, payload, event_type, request.headers
+        ):
+            logger.info(
+                "[webhook] filtered event=%s route=%s",
+                event_type,
+                route_name,
+            )
+            return web.json_response(
+                {
+                    "status": "ignored",
+                    "reason": "filter",
+                    "route": route_name,
+                }
+            )
+
+        if route_config.get("script"):
+            keep, transformed_payload = self._route_processor.run_route_script(
+                route_config.get("script"), payload
+            )
+            if not keep:
+                logger.info(
+                    "[webhook] script ignored event=%s route=%s",
+                    event_type,
+                    route_name,
+                )
+                return web.json_response(
+                    {
+                        "status": "ignored",
+                        "reason": "script",
+                        "route": route_name,
+                    }
+                )
+            payload = transformed_payload or payload
+
         # Format prompt from template
         prompt_template = route_config.get("prompt", "")
         prompt = self._render_prompt(

gateway/platforms/webhook_filters.py

@@ -0,0 +1,302 @@
+"""Route-local filters and script transforms for the webhook adapter."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import re
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_SCRIPT_TIMEOUT_SECONDS = 30
+_MISSING = object()
+
+
+def _stringify_filter_value(value: Any) -> str:
+    if value is _MISSING:
+        return ""
+    if isinstance(value, (dict, list)):
+        return json.dumps(value, sort_keys=True)
+    return str(value)
+
+
+def _resolve_profile_path(path_value: Any) -> Optional[Path]:
+    """Resolve a user path, mapping ~/.hermes to the active profile home."""
+    if not isinstance(path_value, str):
+        return None
+    raw = os.path.expandvars(path_value.strip())
+    if not raw:
+        return None
+    from hermes_constants import get_hermes_home
+
+    hermes_home = get_hermes_home()
+    if raw == "~/.hermes":
+        return hermes_home
+    if raw.startswith("~/.hermes/"):
+        return hermes_home / raw.removeprefix("~/.hermes/")
+    path = Path(raw).expanduser()
+    if path.is_absolute():
+        return path
+    return hermes_home / path
+
+
+def _resolve_script_path(script_value: Any) -> tuple[Optional[Path], Optional[str]]:
+    """Resolve a route script under HERMES_HOME/scripts."""
+    if not isinstance(script_value, str) or not script_value.strip():
+        return None, "script path is empty"
+    from hermes_constants import get_hermes_home
+
+    scripts_root = (get_hermes_home() / "scripts").resolve()
+    raw_text = os.path.expandvars(script_value.strip())
+    if raw_text == "~/.hermes" or raw_text.startswith("~/.hermes/"):
+        mapped = _resolve_profile_path(raw_text)
+        candidate = mapped.resolve() if mapped is not None else scripts_root
+    else:
+        raw = Path(raw_text).expanduser()
+        candidate = raw.resolve() if raw.is_absolute() else (scripts_root / raw).resolve()
+    try:
+        candidate.relative_to(scripts_root)
+    except ValueError:
+        return None, f"script path resolves outside {scripts_root}"
+    if not candidate.exists():
+        return None, f"script not found: {candidate}"
+    if not candidate.is_file():
+        return None, f"script path is not a file: {candidate}"
+    return candidate, None
+
+
+def _load_filter_file_values(path_value: Any) -> list[Any]:
+    path = _resolve_profile_path(path_value)
+    if path is None:
+        return []
+    try:
+        raw = path.read_text(encoding="utf-8")
+    except OSError as exc:
+        logger.warning("[webhook] filter in_file read failed for %s: %s", path, exc)
+        return []
+    try:
+        data = json.loads(raw)
+    except json.JSONDecodeError:
+        return [line.strip() for line in raw.splitlines() if line.strip()]
+    if isinstance(data, list):
+        return data
+    if isinstance(data, dict):
+        return list(data.keys())
+    return [data]
+
+
+class WebhookRouteProcessor:
+    """Evaluate declarative filters and optional script transforms."""
+
+    def __init__(
+        self,
+        *,
+        script_timeout_seconds: int = DEFAULT_SCRIPT_TIMEOUT_SECONDS,
+    ) -> None:
+        self.script_timeout_seconds = max(1, int(script_timeout_seconds))
+
+    def resolve_filter_field(
+        self,
+        field: Any,
+        payload: dict,
+        event_type: str,
+        headers: Any,
+    ) -> Any:
+        """Resolve a dotted filter field against payload/event/headers context."""
+        if not isinstance(field, str) or not field.strip():
+            return _MISSING
+        parts = [part for part in field.strip().split(".") if part]
+        if not parts:
+            return _MISSING
+        header_dict = dict(headers or {})
+        context = {
+            "payload": payload.get("payload", payload),
+            "event": event_type,
+            "event_type": event_type,
+            "headers": header_dict,
+        }
+        if parts[0] in context:
+            value: Any = context[parts[0]]
+            parts = parts[1:]
+        else:
+            value = payload
+        for part in parts:
+            if isinstance(value, dict):
+                value = value.get(part, _MISSING)
+            elif isinstance(value, list) and part.isdigit():
+                idx = int(part)
+                value = value[idx] if 0 <= idx < len(value) else _MISSING
+            else:
+                return _MISSING
+            if value is _MISSING:
+                return _MISSING
+        return value
+
+    def filter_matches(
+        self,
+        spec: Any,
+        payload: dict,
+        event_type: str,
+        headers: Any,
+    ) -> bool:
+        """Evaluate one declarative webhook filter spec."""
+        if not isinstance(spec, dict):
+            logger.warning("[webhook] Ignoring invalid filter spec: %r", spec)
+            return False
+
+        if "all" in spec:
+            items = spec.get("all")
+            return isinstance(items, list) and all(
+                self.filter_matches(item, payload, event_type, headers)
+                for item in items
+            )
+        if "any" in spec:
+            items = spec.get("any")
+            return isinstance(items, list) and any(
+                self.filter_matches(item, payload, event_type, headers)
+                for item in items
+            )
+        if "not" in spec:
+            return not self.filter_matches(spec.get("not"), payload, event_type, headers)
+
+        value = self.resolve_filter_field(
+            spec.get("field"), payload, event_type, headers
+        )
+
+        if "exists" in spec:
+            exists = value is not _MISSING
+            return exists is bool(spec.get("exists"))
+        if spec.get("missing") is True:
+            return value is _MISSING
+        if "equals" in spec:
+            return value is not _MISSING and value == spec.get("equals")
+        if "not_equals" in spec:
+            return value is _MISSING or value != spec.get("not_equals")
+        if "contains" in spec:
+            needle = spec.get("contains")
+            if value is _MISSING:
+                return False
+            if isinstance(value, (list, tuple, set, dict)):
+                return needle in value
+            return str(needle) in _stringify_filter_value(value)
+        if "in" in spec:
+            haystack = spec.get("in")
+            return isinstance(haystack, list) and value in haystack
+        if "in_file" in spec:
+            return value in _load_filter_file_values(spec.get("in_file"))
+        if "regex" in spec:
+            if value is _MISSING:
+                return False
+            try:
+                return (
+                    re.search(str(spec.get("regex")), _stringify_filter_value(value))
+                    is not None
+                )
+            except re.error as exc:
+                logger.warning("[webhook] Invalid webhook filter regex: %s", exc)
+                return False
+
+        logger.warning("[webhook] Filter spec has no supported operator: %r", spec)
+        return False
+
+    def route_filters_match(
+        self,
+        route_config: dict,
+        payload: dict,
+        event_type: str,
+        headers: Any,
+    ) -> bool:
+        filters = route_config.get("filters") or []
+        if not filters:
+            return True
+        if isinstance(filters, dict):
+            return self.filter_matches(filters, payload, event_type, headers)
+        if not isinstance(filters, list):
+            logger.warning("[webhook] filters must be a list or object")
+            return False
+        return all(
+            self.filter_matches(spec, payload, event_type, headers)
+            for spec in filters
+        )
+
+    def run_route_script(self, script_value: Any, payload: dict) -> tuple[bool, Optional[dict]]:
+        """Run a route script and return (should_continue, transformed_payload)."""
+        path, error = _resolve_script_path(script_value)
+        if error or path is None:
+            logger.warning("[webhook] script ignored webhook: %s", error)
+            return False, None
+
+        suffix = path.suffix.lower()
+        if suffix in {".sh", ".bash"}:
+            bash = shutil.which("bash") or (
+                "/bin/bash" if os.path.isfile("/bin/bash") else None
+            )
+            if bash is None:
+                logger.warning("[webhook] script ignored webhook: bash not found")
+                return False, None
+            argv = [bash, str(path)]
+        else:
+            argv = [sys.executable, str(path)]
+
+        try:
+            from tools.environments.local import _sanitize_subprocess_env
+
+            popen_kwargs = {"creationflags": 0x08000000} if sys.platform == "win32" else {}
+            result = subprocess.run(
+                argv,
+                input=json.dumps(payload),
+                capture_output=True,
+                text=True,
+                timeout=self.script_timeout_seconds,
+                cwd=str(path.parent),
+                env=_sanitize_subprocess_env(os.environ.copy()),
+                **popen_kwargs,
+            )
+        except subprocess.TimeoutExpired:
+            logger.warning("[webhook] script timed out: %s", path)
+            return False, None
+        except Exception as exc:
+            logger.warning("[webhook] script execution failed: %s", exc)
+            return False, None
+
+        stdout = (result.stdout or "").strip()
+        stderr = (result.stderr or "").strip()
+        try:
+            from agent.redact import redact_sensitive_text
+
+            stdout = redact_sensitive_text(stdout)
+            stderr = redact_sensitive_text(stderr)
+        except Exception as exc:
+            logger.warning("[webhook] Failed to redact script output: %s", exc)
+            stdout = "[REDACTED - redaction failed]"
+            stderr = "[REDACTED - redaction failed]"
+        if result.returncode != 0:
+            logger.info(
+                "[webhook] script ignored webhook path=%s code=%s stderr=%s",
+                path.name,
+                result.returncode,
+                stderr[:200],
+            )
+            return False, None
+        if not stdout or stdout == "[SILENT]":
+            return False, None
+
+        try:
+            transformed = json.loads(stdout)
+        except json.JSONDecodeError:
+            transformed = {**payload, "script_output": stdout}
+        if not isinstance(transformed, dict):
+            logger.warning("[webhook] script stdout must be a JSON object or text")
+            return False, None
+        if (
+            transformed.get("[SILENT]") is True
+            or transformed.get("__hermes_ignore__") is True
+        ):
+            return False, None
+        return True, transformed

hermes_cli/subcommands/webhook.py

@@ -55,6 +55,13 @@ def build_webhook_parser(subparsers, *, cmd_webhook: Callable) -> None:
         "message. Zero LLM cost. Requires --deliver to be a real target "
         "(not 'log').",
     )
+    wh_sub.add_argument(
+        "--script",
+        default="",
+        help="Filter/transform script under ~/.hermes/scripts/. The route "
+        "payload is passed as JSON on stdin; empty stdout, [SILENT], or a "
+        "nonzero exit code ignores the webhook.",
+    )
 
     webhook_subparsers.add_parser(
         "list", aliases=["ls"], help="List all dynamic subscriptions"

hermes_cli/web_server.py

@@ -10932,6 +10932,7 @@ class WebhookCreate(BaseModel):
     description: Optional[str] = None
     events: List[str] = []
     prompt: Optional[str] = None
+    script: Optional[str] = None
     skills: List[str] = []
     deliver: str = "log"
     deliver_only: bool = False
@@ -10948,6 +10949,7 @@ def _webhook_route_summary(name: str, route: Dict[str, Any], base_url: str) -> D
         "deliver": route.get("deliver", "log"),
         "deliver_only": bool(route.get("deliver_only")),
         "prompt": route.get("prompt", ""),
+        "script": route.get("script", ""),
         "skills": list(route.get("skills") or []),
         "created_at": route.get("created_at"),
         "url": f"{base_url}/webhooks/{name}",
@@ -11031,6 +11033,8 @@ async def create_webhook(body: WebhookCreate):
         "deliver": body.deliver or "log",
         "created_at": _time.strftime("%Y-%m-%dT%H:%M:%SZ", _time.gmtime()),
     }
+    if body.script and body.script.strip():
+        route["script"] = body.script.strip()
     if body.deliver_only:
         route["deliver_only"] = True
     if body.deliver_chat_id:

hermes_cli/webhook.py

@@ -189,6 +189,10 @@ def _cmd_subscribe(args):
             return
         route["deliver_only"] = True
 
+    script = getattr(args, "script", "") or ""
+    if script.strip():
+        route["script"] = script.strip()
+
     if args.deliver_chat_id:
         route["deliver_extra"] = {"chat_id": args.deliver_chat_id}
 
@@ -212,6 +216,8 @@ def _cmd_subscribe(args):
         prompt_preview = route["prompt"][:80] + ("..." if len(route["prompt"]) > 80 else "")
         label = "Message" if route.get("deliver_only") else "Prompt"
         print(f"  {label}: {prompt_preview}")
+    if route.get("script"):
+        print(f"  Script: {route['script']}")
     print("\n  Configure your service to POST to the URL above.")
     print("  Use the secret for HMAC-SHA256 signature validation.")
     print("  The gateway must be running to receive events (hermes gateway run).\n")
@@ -238,6 +244,8 @@ def _cmd_list(args):
         print(f"    URL:     {base_url}/webhooks/{name}")
         print(f"    Events:  {events}")
         print(f"    Deliver: {deliver}")
+        if route.get("script"):
+            print(f"    Script:  {route['script']}")
         print()
 
 

tests/gateway/test_webhook_adapter.py

@@ -608,6 +608,285 @@ class TestEventFilter:
             assert resp.status == 202
 
 
+# ===================================================================
+# Payload filters
+# ===================================================================
+
+
+class TestPayloadFilters:
+    """Tests for route-level payload filters in _handle_webhook."""
+
+    @pytest.mark.asyncio
+    async def test_filter_rejects_before_agent_dispatch(self):
+        """A non-matching filter returns ignored and never starts the agent."""
+        routes = {
+            "todoist": {
+                "secret": _INSECURE_NO_AUTH,
+                "filters": [{"field": "payload.label", "equals": "urgent"}],
+                "prompt": "Task: {payload.content}",
+            }
+        }
+        adapter = _make_adapter(routes=routes)
+        adapter.handle_message = AsyncMock()
+
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                "/webhooks/todoist",
+                json={"payload": {"label": "later", "content": "Buy milk"}},
+                headers={"X-GitHub-Delivery": "filter-skip-1"},
+            )
+            assert resp.status == 200
+            data = await resp.json()
+            assert data == {
+                "status": "ignored",
+                "reason": "filter",
+                "route": "todoist",
+            }
+
+        adapter.handle_message.assert_not_called()
+        assert "filter-skip-1" not in adapter._seen_deliveries
+
+    @pytest.mark.asyncio
+    async def test_filter_accepts_nested_any_and_in_file(self, tmp_path, monkeypatch):
+        """Nested any groups can match dynamic watchlists under HERMES_HOME."""
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        watchlist = tmp_path / "data" / "watchlist.json"
+        watchlist.parent.mkdir()
+        watchlist.write_text(json.dumps(["chat-1", "chat-2"]), encoding="utf-8")
+        routes = {
+            "waha": {
+                "secret": _INSECURE_NO_AUTH,
+                "filters": [
+                    {"field": "payload.fromMe", "equals": False},
+                    {
+                        "any": [
+                            {
+                                "field": "payload.chatId",
+                                "in_file": "~/.hermes/data/watchlist.json",
+                            },
+                            {
+                                "field": "payload.id.remote",
+                                "in_file": "~/.hermes/data/watchlist.json",
+                            },
+                        ]
+                    },
+                ],
+                "prompt": "Message from {payload.chatId}: {payload.body}",
+            }
+        }
+        adapter = _make_adapter(routes=routes)
+        captured = []
+
+        async def _capture(event):
+            captured.append(event)
+
+        adapter.handle_message = _capture
+
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                "/webhooks/waha",
+                json={
+                    "payload": {
+                        "fromMe": False,
+                        "chatId": "chat-2",
+                        "body": "hello",
+                    }
+                },
+                headers={"X-GitHub-Delivery": "filter-match-1"},
+            )
+            assert resp.status == 202
+
+        await asyncio.sleep(0.05)
+        assert len(captured) == 1
+        assert captured[0].text == "Message from chat-2: hello"
+
+    @pytest.mark.asyncio
+    async def test_filter_applies_to_deliver_only_before_delivery(self):
+        """Filtered direct-delivery routes skip target delivery too."""
+        routes = {
+            "alerts": {
+                "secret": _INSECURE_NO_AUTH,
+                "deliver": "telegram",
+                "deliver_only": True,
+                "deliver_extra": {"chat_id": "123"},
+                "filters": [{"field": "severity", "in": ["critical"]}],
+                "prompt": "Alert: {message}",
+            }
+        }
+        adapter = _make_adapter(routes=routes)
+        mock_target = AsyncMock()
+        mock_target.send = AsyncMock(return_value=SendResult(success=True))
+        mock_runner = MagicMock()
+        mock_runner.adapters = {Platform.TELEGRAM: mock_target}
+        adapter.gateway_runner = mock_runner
+
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                "/webhooks/alerts",
+                json={"severity": "info", "message": "noise"},
+                headers={"X-GitHub-Delivery": "filter-direct-1"},
+            )
+            assert resp.status == 200
+            assert (await resp.json())["reason"] == "filter"
+
+        mock_target.send.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_script_transforms_payload_before_prompt_rendering(self, tmp_path, monkeypatch):
+        """A script can replace the payload used by prompt templates."""
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        scripts = tmp_path / "scripts"
+        scripts.mkdir()
+        script = scripts / "todoist_filter.py"
+        script.write_text(
+            "import json, sys\n"
+            "payload = json.load(sys.stdin)\n"
+            "payload['body'] = payload['task']['content'].upper()\n"
+            "print(json.dumps(payload))\n",
+            encoding="utf-8",
+        )
+        routes = {
+            "todoist": {
+                "secret": _INSECURE_NO_AUTH,
+                "script": "todoist_filter.py",
+                "prompt": "Task: {body}",
+            }
+        }
+        adapter = _make_adapter(routes=routes)
+        captured = []
+
+        async def _capture(event):
+            captured.append(event)
+
+        adapter.handle_message = _capture
+
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                "/webhooks/todoist",
+                json={"task": {"content": "pay bills"}},
+                headers={"X-GitHub-Delivery": "script-transform-1"},
+            )
+            assert resp.status == 202
+
+        await asyncio.sleep(0.05)
+        assert captured[0].text == "Task: PAY BILLS"
+        assert captured[0].raw_message["body"] == "PAY BILLS"
+
+    @pytest.mark.asyncio
+    async def test_script_tilde_hermes_path_resolves_to_active_profile_home(self, tmp_path, monkeypatch):
+        """~/.hermes/scripts paths must resolve through HERMES_HOME for profiles."""
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        scripts = tmp_path / "scripts"
+        scripts.mkdir()
+        (scripts / "todoist_filter.py").write_text(
+            "import json, sys\n"
+            "payload = json.load(sys.stdin)\n"
+            "payload['body'] = 'profile-safe'\n"
+            "print(json.dumps(payload))\n",
+            encoding="utf-8",
+        )
+        routes = {
+            "todoist": {
+                "secret": _INSECURE_NO_AUTH,
+                "script": "~/.hermes/scripts/todoist_filter.py",
+                "prompt": "Task: {body}",
+            }
+        }
+        adapter = _make_adapter(routes=routes)
+        captured = []
+
+        async def _capture(event):
+            captured.append(event)
+
+        adapter.handle_message = _capture
+
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                "/webhooks/todoist",
+                json={"task": {"content": "pay bills"}},
+                headers={"X-GitHub-Delivery": "script-profile-path-1"},
+            )
+            assert resp.status == 202
+
+        await asyncio.sleep(0.05)
+        assert captured[0].text == "Task: profile-safe"
+
+    @pytest.mark.asyncio
+    async def test_script_silent_stdout_ignores_without_idempotency_hit(self, tmp_path, monkeypatch):
+        """Empty or [SILENT] script stdout filters the webhook out."""
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        scripts = tmp_path / "scripts"
+        scripts.mkdir()
+        (scripts / "skip.py").write_text("print('[SILENT]')\n", encoding="utf-8")
+        routes = {
+            "todoist": {
+                "secret": _INSECURE_NO_AUTH,
+                "script": "skip.py",
+                "prompt": "Task: {body}",
+            }
+        }
+        adapter = _make_adapter(routes=routes)
+        adapter.handle_message = AsyncMock()
+
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                "/webhooks/todoist",
+                json={"body": "ignore me"},
+                headers={"X-GitHub-Delivery": "script-silent-1"},
+            )
+            assert resp.status == 200
+            data = await resp.json()
+            assert data == {
+                "status": "ignored",
+                "reason": "script",
+                "route": "todoist",
+            }
+
+        adapter.handle_message.assert_not_called()
+        assert "script-silent-1" not in adapter._seen_deliveries
+
+    @pytest.mark.asyncio
+    async def test_script_nonzero_exit_ignores_webhook(self, tmp_path, monkeypatch):
+        """A script can fail closed by exiting nonzero."""
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        scripts = tmp_path / "scripts"
+        scripts.mkdir()
+        (scripts / "skip.py").write_text(
+            "import sys\nsys.exit(2)\n",
+            encoding="utf-8",
+        )
+        routes = {
+            "todoist": {
+                "secret": _INSECURE_NO_AUTH,
+                "script": "skip.py",
+                "prompt": "Task: {body}",
+            }
+        }
+        adapter = _make_adapter(routes=routes)
+        adapter.handle_message = AsyncMock()
+
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                "/webhooks/todoist",
+                json={"body": "ignore me"},
+                headers={"X-GitHub-Delivery": "script-nonzero-1"},
+            )
+            assert resp.status == 200
+            data = await resp.json()
+            assert data["status"] == "ignored"
+            assert data["reason"] == "script"
+
+        adapter.handle_message.assert_not_called()
+        assert "script-nonzero-1" not in adapter._seen_deliveries
+
+
 # ===================================================================
 # HTTP handling
 # ===================================================================
@@ -1246,4 +1525,3 @@ class TestInsecureNoAuthSafetyRail:
             assert result is True
         finally:
             await adapter.disconnect()
-

tests/hermes_cli/test_dashboard_admin_endpoints.py

@@ -223,6 +223,30 @@ class TestWebhookEndpoints:
         r = self.client.post("/api/webhooks", json={"name": "gh", "deliver": "log"})
         assert r.status_code == 400
 
+    def test_create_webhook_persists_script(self):
+        from hermes_cli.config import load_config, save_config
+
+        cfg = load_config()
+        cfg.setdefault("platforms", {})["webhook"] = {
+            "enabled": True,
+            "extra": {"host": "0.0.0.0", "port": 8644},
+        }
+        save_config(cfg)
+
+        r = self.client.post(
+            "/api/webhooks",
+            json={
+                "name": "todoist",
+                "deliver": "log",
+                "script": "todoist_filter.py",
+            },
+        )
+        assert r.status_code == 200
+        assert r.json()["script"] == "todoist_filter.py"
+
+        subs = self.client.get("/api/webhooks").json()["subscriptions"]
+        assert subs[0]["script"] == "todoist_filter.py"
+
     def test_enable_platform_starts_gateway_restart(self, monkeypatch):
         import hermes_cli.web_server as ws
         from hermes_cli.config import load_config

tests/hermes_cli/test_webhook_cli.py

@@ -35,6 +35,7 @@ def _make_args(**kwargs):
         "deliver_chat_id": "",
         "secret": "",
         "payload": "",
+        "script": "",
     }
     defaults.update(kwargs)
     return Namespace(**defaults)
@@ -72,6 +73,12 @@ class TestSubscribe:
         ))
         assert _load_subscriptions()["s"]["secret"] == "my-secret"
 
+    def test_script_option_is_persisted(self):
+        webhook_command(_make_args(
+            webhook_action="subscribe", name="s", script="todoist_filter.py"
+        ))
+        assert _load_subscriptions()["s"]["script"] == "todoist_filter.py"
+
     def test_auto_secret(self):
         webhook_command(_make_args(webhook_action="subscribe", name="s"))
         secret = _load_subscriptions()["s"]["secret"]

website/docs/guides/webhook-github-pr-review.md

@@ -182,12 +182,20 @@ tail -f "${HERMES_HOME:-$HOME/.hermes}/logs/gateway.log"
 
 ## Filtering to specific actions
 
-GitHub sends `pull_request` events for many actions: `opened`, `synchronize`, `reopened`, `closed`, `labeled`, etc. The `events` list filters only by the `X-GitHub-Event` header value — it cannot filter by action sub-type at the routing level.
+GitHub sends `pull_request` events for many actions: `opened`, `synchronize`, `reopened`, `closed`, `labeled`, etc. The `events` list filters by the `X-GitHub-Event` header value, and route-level `filters` can narrow by payload fields such as `action`.
 
 The prompt in Step 1 already handles this by instructing the agent to stop early for `closed` and `labeled` events.
 
 :::warning The agent still runs and consumes tokens
-The "stop here" instruction prevents a meaningful review, but the agent still runs to completion for every `pull_request` event regardless of action. GitHub webhooks can only filter by event type (`pull_request`, `push`, `issues`, etc.) — not by action sub-type (`opened`, `closed`, `labeled`). There is no routing-level filter for sub-actions. For high-volume repos, accept this cost or filter upstream with a GitHub Actions workflow that calls your webhook URL conditionally.
+The "stop here" instruction prevents a meaningful review, but the agent still runs to completion for every `pull_request` event regardless of action. Prefer filtering before the agent wakes:
+
+```yaml
+filters:
+  - field: "action"
+    in: ["opened", "synchronize", "reopened"]
+```
+
+For high-volume repositories, you can still filter upstream with a GitHub Actions workflow that calls your webhook URL conditionally.
 :::
 
 > There is no Jinja2 or conditional template syntax. `{field}` and `{nested.field}` are the only substitutions supported. Anything else is passed verbatim to the agent.

website/docs/user-guide/messaging/webhooks.md

@@ -81,6 +81,8 @@ Routes define how different webhook sources are handled. Each route is a named e
 | `events` | No | List of event types to accept (e.g. `["pull_request"]`). If empty, all events are accepted. Event type is read from `X-GitHub-Event`, `X-GitLab-Event`, or `event_type` in the payload. |
 | `secret` | **Yes** | HMAC secret for signature validation. Falls back to the global `secret` if not set on the route. Set to `"INSECURE_NO_AUTH"` for testing only (skips validation). |
 | `prompt` | No | Template string with dot-notation payload access (e.g. `{pull_request.title}`). If omitted, the full JSON payload is dumped into the prompt. Payload fields are untrusted — see [Authenticated does not mean trusted](#authenticated-does-not-mean-trusted). |
+| `filters` | No | Declarative payload filters evaluated after auth/body/event filtering and before agent or direct delivery work. Non-matches return `{"status":"ignored","reason":"filter"}` with HTTP 200. |
+| `script` | No | Filter/transform script under `~/.hermes/scripts/`. The webhook payload is passed as JSON on stdin. JSON object stdout replaces the payload before templating; text stdout is exposed as `script_output`; empty stdout, `[SILENT]`, or a nonzero exit code ignores the webhook. |
 | `skills` | No | List of skill names to load for the agent run. |
 | `deliver` | No | Where to send the response: `github_comment`, `telegram`, `discord`, `slack`, `signal`, `sms`, `whatsapp`, `matrix`, `mattermost`, `homeassistant`, `email`, `dingtalk`, `feishu`, `wecom`, `weixin`, `bluebubbles`, `qqbot`, or `log` (default). |
 | `deliver_extra` | No | Additional delivery config — keys depend on `deliver` type (e.g. `repo`, `pr_number`, `chat_id`). Values support the same `{dot.notation}` templates as `prompt`. |
@@ -116,9 +118,75 @@ platforms:
           events: ["push"]
           secret: "deploy-secret"
           prompt: "New push to {repository.full_name} branch {ref}: {head_commit.message}"
+          filters:
+            - field: "ref"
+              equals: "refs/heads/main"
           deliver: "telegram"
 ```
 
+### Payload Filters
+
+Use `filters` when a provider sends a broad event stream but only some payloads should wake the agent or trigger `deliver_only` delivery. Filters run after signature validation, body parsing, and `events`, but before prompt rendering, idempotency, agent dispatch, or direct delivery.
+
+```yaml
+platforms:
+  webhook:
+    extra:
+      routes:
+        todoist:
+          events: ["item:updated"]
+          secret: "todoist-secret"
+          filters:
+            - field: "payload.labels"
+              contains: "hermes"
+            - any:
+                - field: "payload.priority"
+                  equals: 4
+                - field: "payload.project_id"
+                  in_file: "~/.hermes/data/todoist/watchlist.json"
+          prompt: "Todoist task changed: {payload.content}"
+```
+
+Supported operators:
+
+- `exists: true|false`
+- `missing: true`
+- `equals` / `not_equals`
+- `contains` for strings, lists, and dict keys
+- `in` for inline lists
+- `in_file` for JSON arrays, JSON objects (keys are used), or newline-delimited text files
+- `regex`
+- `all`, `any`, and `not` groups
+
+Field paths use dot notation. `payload.foo` reads from a top-level `payload` object when one exists, or from the root webhook body for flat payloads. `event` / `event_type` match the resolved event type, and `headers.<Name>` reads request headers.
+
+### Script Filters and Transforms
+
+Use `script` when declarative filters are not enough. Scripts must live under `~/.hermes/scripts/` for the active profile; relative paths resolve there, and path traversal outside that directory is blocked. `.sh` and `.bash` scripts run with bash, and all other extensions run with the current Python interpreter.
+
+The route payload is sent to stdin as JSON:
+
+```python
+# ~/.hermes/scripts/todoist-hermes-label.py
+import json
+import sys
+
+payload = json.load(sys.stdin)
+labels = payload.get("payload", {}).get("labels", [])
+if "hermes" not in labels:
+    print("[SILENT]")
+    raise SystemExit(0)
+
+payload["body"] = payload["payload"]["content"]
+print(json.dumps(payload))
+```
+
+Script outcomes:
+
+- JSON object stdout replaces the payload used by `prompt` and `deliver_extra`.
+- Non-JSON text stdout is added to the payload as `script_output`.
+- Empty stdout, exact `[SILENT]`, `{"__hermes_ignore__": true}`, timeout, missing script, or nonzero exit code returns HTTP 200 with `{"status":"ignored","reason":"script"}`.
+
 ### Prompt Templates
 
 Prompts use dot-notation to access nested fields in the webhook payload: