← All code samples

Webhook condition (filter_cel)

A provider webhook often fires on every event: a GitHub repository webhook delivers pull-request opens, edits, comments, and review requests on the same endpoint. Without a condition, each one starts a run and spends AI-query quota before an instruction in the prompt gets a chance to ignore it. Set filter_cel on the webhook so Aiden rejects the delivery before dispatch.

Use this when:

  • One webhook endpoint receives several event types, but only one should start work (for example, a direct review request naming your bot)
  • You want to try a condition against sample JSON and headers before saving it
  • You want to stop paying for runs that an “ignore this” instruction would have discarded anyway

filter_cel is a CEL boolean expression evaluated against body (the decoded JSON payload) and headers (sanitized, lowercase provider headers — auth headers are never exposed). An empty or omitted condition keeps existing behavior and accepts every delivery. A condition that evaluates to false returns the webhook’s configured acknowledgement with no run, no invocation row, and no AI-query charge. Invalid conditions are rejected on save; evaluation errors at delivery time fail visibly instead of silently admitting the event.

How to run

export STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_WORKSPACE_ID=<workspace-uuid>

pip install stackgen-sdk
python examples/automation/webhook-filter-cel.py --agent my-pr-review-agent --reviewer release-bot

Minimal

from stackgen import StackgenClient, StackgenConfig

client = StackgenClient(
    StackgenConfig(
        base_url="https://app.stackgen.com",
        api_token="stackgen_…",
        workspace_id="<workspace-uuid>",
    )
)

filter_cel = (
    '("x-github-event" in headers) && '
    'headers["x-github-event"] == "pull_request" && '
    'has(body.action) && body.action == "review_requested" && '
    "has(body.requested_reviewer) && body.requested_reviewer != null && "
    "has(body.requested_reviewer.login) && "
    'body.requested_reviewer.login.matches("(?i)^release-bot$")'
)

# Try it before saving anything.
matched = client.aiden.preview_webhook_filter(
    filter_cel,
    '{"action": "review_requested", "requested_reviewer": {"login": "release-bot"}}',
    headers=[{"name": "X-GitHub-Event", "value": "pull_request"}],
)
print("matches a direct request for release-bot:", matched)

wh = client.aiden.create_webhook({
    "name": "github-pr-review",
    "target_type": "agent",
    "target_name": "my-pr-review-agent",
    "action": "Review this pull request and reply Done",
    "enabled": True,
    "filter_cel": filter_cel,
})
print(wh["id"], wh.get("filter_cel"), wh.get("token"))

Clear a condition

Send an explicit empty string on update to accept every event again. Omitting filter_cel on update leaves the existing condition unchanged.

client.aiden.update_webhook(wh["id"], {"filter_cel": ""})

Complete scripts

The full script creates the webhook, previews five samples (a human reviewer match, a bot reviewer match, a different reviewer, a plain PR edit, and a ping event), triggers a nonmatching PR-edited delivery and confirms no new invocation was recorded, triggers a matching review-request delivery and confirms one was, then clears the condition and deletes the webhook.

#!/usr/bin/env python3
"""Filter a GitHub webhook so only your review requests start an agent.

A GitHub repository webhook delivers every pull-request event: opened, edited,
comments, team review requests, and direct review requests. Only a direct
request naming your reviewer should start the review agent — every other
delivery should just be acknowledged.

This sample:

1. Creates a webhook with a `filter_cel` condition that matches a
   `review_requested` event naming one reviewer login (human or bot)
2. Previews the condition against a matching request, a different reviewer,
   and a plain PR edit — without saving data or starting work
3. Triggers the webhook with a nonmatching PR-edited payload and confirms no
   new invocation was recorded
4. Triggers it again with a matching review-request payload and confirms a
   run was recorded
5. Clears the condition with update_webhook so every event is accepted again

Usage:
  export STACKGEN_URL=https://app.stackgen.com
  export STACKGEN_TOKEN=stackgen_…
  export STACKGEN_WORKSPACE_ID=<workspace-uuid>
  python examples/automation/webhook-filter-cel.py --agent my-pr-review-agent
  # Empty CI workspaces: prefer a published workflow (agents need an integration).
  python examples/automation/webhook-filter-cel.py --workflow my-pr-review-workflow
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.request
import uuid

from stackgen import StackgenClient, StackgenConfig


def require_env(name: str, *fallbacks: str) -> str:
    for key in (name, *fallbacks):
        value = (os.environ.get(key) or "").strip()
        if value:
            return value
    raise SystemExit(f"missing env {name}")


def build_filter_cel(reviewer_login: str) -> str:
    """Match a direct review request naming reviewer_login (human or bot), nothing else."""
    return (
        '("x-github-event" in headers) && '
        'headers["x-github-event"] == "pull_request" && '
        "has(body.action) && body.action == \"review_requested\" && "
        "has(body.requested_reviewer) && body.requested_reviewer != null && "
        "has(body.requested_reviewer.login) && "
        f'body.requested_reviewer.login.matches("(?i)^{reviewer_login}$")'
    )


def post_github_event(base_url: str, org_id: str, token: str, event: str, body: dict) -> int:
    url = f"{base_url}/guild/api/v1/webhooks/trigger?orgId={org_id}"
    req = urllib.request.Request(
        url,
        data=json.dumps(body).encode("utf-8"),
        method="POST",
        headers={
            "Authorization": f"Bearer {token}",
            "Accept": "application/json",
            "Content-Type": "application/json",
            "X-GitHub-Event": event,
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return resp.status
    except urllib.error.HTTPError as err:
        return err.code


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    target = parser.add_mutually_exclusive_group(required=True)
    target.add_argument("--agent", help="Existing agent name")
    target.add_argument("--workflow", help="Existing workflow name")
    parser.add_argument(
        "--reviewer",
        default="release-bot",
        help="GitHub login the filter accepts review requests for",
    )
    parser.add_argument(
        "--message",
        default="sdk: review this pull request and reply Done",
        help="Webhook action prompt",
    )
    args = parser.parse_args()
    target_type = "agent" if args.agent else "workflow"
    target_name = args.agent or args.workflow

    base_url = require_env("STACKGEN_URL").rstrip("/")
    org_id = require_env("STACKGEN_WORKSPACE_ID", "STACKGEN_PROJECT")
    client = StackgenClient(
        StackgenConfig(
            base_url=base_url,
            api_token=require_env("STACKGEN_TOKEN"),
            workspace_id=org_id,
        )
    )
    suffix = uuid.uuid4().hex[:8]
    webhook_id = ""
    filter_cel = build_filter_cel(args.reviewer)

    pr_edited = {"action": "edited", "pull_request": {"number": 42}}
    review_requested_match = {
        "action": "review_requested",
        "requested_reviewer": {"login": args.reviewer, "type": "User"},
    }
    review_requested_bot = {
        "action": "review_requested",
        "requested_reviewer": {"login": f"{args.reviewer}[bot]", "type": "Bot"},
    }
    review_requested_other = {
        "action": "review_requested",
        "requested_reviewer": {"login": "someone-else"},
    }

    try:
        # 1. Preview the condition before saving anything.
        print("--- previewing filter_cel (no webhook saved yet) ---")
        for label, payload, event in [
            ("matching human reviewer", review_requested_match, "pull_request"),
            ("matching bot reviewer", review_requested_bot, "pull_request"),
            ("different reviewer", review_requested_other, "pull_request"),
            ("PR edited (no reviewer)", pr_edited, "pull_request"),
            ("ping event", {}, "ping"),
        ]:
            matched = client.aiden.preview_webhook_filter(
                filter_cel,
                json.dumps(payload),
                headers=[{"name": "X-GitHub-Event", "value": event}],
            )
            print(f"  {label}: matched={matched}")

        # 2. Create the webhook with the condition already attached.
        wh = client.aiden.create_webhook(
            {
                "name": f"sdk-pr-review-filter-{suffix}",
                "target_type": target_type,
                "target_name": target_name,
                "action": args.message,
                "enabled": True,
                "filter_cel": filter_cel,
            }
        )
        webhook_id = str(wh.get("id") or "")
        token = str(wh.get("token") or "")
        print(f"\ncreated webhook id={webhook_id} filter_cel={wh.get('filter_cel')!r}")
        if not token:
            raise SystemExit("createWebhook did not return a token")

        runs_before = len(client.aiden.list_webhook_runs(webhook_id))

        # 3. A PR edit does not match — acknowledged, but no run recorded.
        status = post_github_event(base_url, org_id, token, "pull_request", pr_edited)
        print(f"PR edited delivery: ack status={status}")
        time.sleep(2)
        runs_after_edit = len(client.aiden.list_webhook_runs(webhook_id))
        if runs_after_edit != runs_before:
            raise SystemExit(
                f"expected no new run for a nonmatching delivery "
                f"(before={runs_before}, after={runs_after_edit})"
            )
        print(f"webhook runs after PR edit: {runs_after_edit} (unchanged)")

        # 4. A direct review request for the reviewer matches — a run starts.
        status = post_github_event(base_url, org_id, token, "pull_request", review_requested_match)
        print(f"review_requested delivery: ack status={status}")
        time.sleep(2)
        runs_after_review = len(client.aiden.list_webhook_runs(webhook_id))
        if runs_after_review <= runs_after_edit:
            raise SystemExit(
                f"expected a new run for the matching delivery "
                f"(after_edit={runs_after_edit}, after_review={runs_after_review})"
            )
        print(f"webhook runs after review request: {runs_after_review} (started one run)")

        # 5. Clear the condition — the webhook reverts to accepting every event.
        cleared = client.aiden.update_webhook(webhook_id, {"filter_cel": ""})
        print(f"cleared filter_cel={cleared.get('filter_cel')!r}")
    finally:
        if webhook_id:
            try:
                client.aiden.delete_webhook(webhook_id)
                print(f"deleted webhook id={webhook_id}")
            except Exception as cleanup_err:  # noqa: BLE001 — best-effort teardown
                print(f"cleanup delete_webhook failed: {cleanup_err}", file=sys.stderr)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())