feat(extras): receive SaaS webhooks, at flyte.extras.webhooks (#1512)

First of a stack that replaces the six per-product integration plugins
(#1502–#1507) with the parts that are genuinely Flyte's. Structured like
`plugins/agents`: **core holds the contract, each product is its own
thin package.**

**Stack:** this → github → slack → linear → clickup → jira → examples.
Each PR's diff is just its own package.

## What core owns

| | |
| --- | --- |
| `WebhookAppEnvironment` | One app: a dashboard and a verified receiver
at `/webhook/{provider}`, for whichever providers it is handed |
| `Provider` | The contract a product plugin implements |
| `WebhookEvent` | The normalized event every provider parses into |
| `run_once` | Launch a run once per event key, returning the run that
covers it either way |
| `EventType` | Base for the typed event constants each plugin ships |
| verification primitives | `constant_time_equals`, `hex_hmac_sha256`,
`json_body`, `lower_headers` |
| `testing.assert_provider_conforms` | The conformance check every
plugin runs |

A product plugin owns only what is specific to it — usually under 150
lines. Core is tested against a **stub provider**, so it depends on no
product package.

## Two decisions worth reviewing

**Identity lives on a run label, never a name.** `run_once` launches
only when no run carrying the same `dedupe` label is live or succeeded;
when one is, it returns that run with `created=False` rather than
launching a second or raising, so a handler can answer "already handled"
and link to it. Names race against concurrent launches and cap how many
runs one key can ever have, so the control plane assigns them. It is
async-first behind `@syncify` — the blocking form stalls an app's event
loop, and webhook senders time out in seconds. A test asserts four
concurrent launches overlap; against a blocking implementation it
measures 0.81s instead of 0.42s and fails.

**`constant_time_equals` compares bytes.** `hmac.compare_digest` raises
`TypeError` on `str` operands containing non-ASCII, and these headers
come off the wire — ASGI hands Starlette raw bytes which it decodes
latin-1, so a crafted header would turn a clean 401 into a 500. Writing
it once here is the main argument for a shared core.

## The conformance harness

Each product plugin ships `SAMPLE_DELIVERY`, a real payload. The harness
signs and replays it, so `verify` and `parse` are checked against an
actual payload rather than against each other — without one, every other
check is vacuous.

It earned its keep during development twice:

- It caught that **Jira's shared token is not bound to the request
body**, unlike the four HMAC providers. That is inherent to an unsigned
webhook, not a bug, so `signed=False` now opts out of the body-tampering
check *and* makes the dashboard say the product does not sign.
- It caught **a hole in itself**: the hostile-header check sent a bare
non-ASCII value, which GitHub and Slack reject on their scheme prefix
before reaching the comparison — leaving the check vacuous for exactly
the two providers it mattered most for. Hostile values now preserve the
prefix, and a `str`-based `compare_digest` fails conformance as
intended.

I verified it discriminates by injecting three regressions (plain `str,
Enum` constants, a `str` comparison, a constant that no longer matches
the parser); each one fails with a specific message.

## Notes

No new dependencies land in `flyte`: the shared core lives in
`flyte.extras.webhooks` behind the same soft-import pattern the other
`flyte.extras` apps use, and each product plugin ships its own provider.

46 core tests plus each plugin's suite. `make fmt`, `make mypy` (864
files), `make ty`, ruff, and codespell pass.

The caveat from the review that prompted this: the parsers still encode
payload-shape assumptions I cannot verify without live accounts. That
surface is much smaller now — the six API clients and their ~213 field
assumptions are gone — and `SAMPLE_DELIVERY` pins what remains, but one
real pass per product is still the thing that would confirm it.

---------

Signed-off-by: Niels Bantilan <niels.bantilan@gmail.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: cosmicBboy

examples/apps/webhook_custom_provider.py

@@ -0,0 +1,162 @@
+"""Support a product this family does not ship a plugin for.
+
+A provider is small: say which environment variable holds its secret, how to
+verify a delivery, and how to turn a payload into a `WebhookEvent`. Core does
+the rest — the app, the dashboard, dispatch, the scope allowlist, and idempotent
+launching.
+
+Run it without an account:
+
+    python custom_provider.py --local
+
+That posts a signed sample delivery through the app in-process, so you see
+verification, normalization, and dispatch end to end.
+
+Once it works, move it into its own `flyteplugins-webhooks-<product>` package
+beside the others and add the one-line conformance test:
+
+    from flyte.extras.webhooks.testing import assert_provider_conforms
+    import flyteplugins.webhooks.acme as plugin
+
+    def test_conformance():
+        assert_provider_conforms(plugin)
+"""
+
+import hashlib
+import hmac
+import json
+import os
+import sys
+from typing import ClassVar, Mapping
+
+import flyte
+from flyte.extras.webhooks import (
+    EventType,
+    Provider,
+    WebhookAppEnvironment,
+    WebhookEvent,
+    constant_time_equals,
+    hex_hmac_sha256,
+    json_body,
+    lower_headers,
+)
+
+
+class Ticket(EventType):
+    """Acme's ticket events. `ANY` matches every action on the type."""
+
+    ANY = "ticket"
+    OPENED = "ticket.opened"
+    CLOSED = "ticket.closed"
+
+
+def verify(body: bytes, headers: Mapping[str, str], secret: str) -> bool:
+    """Verify Acme's hex HMAC-SHA256 over the raw body.
+
+    Use `constant_time_equals` rather than `hmac.compare_digest` directly: the
+    latter raises `TypeError` on `str` operands containing non-ASCII, and this
+    header comes off the wire, so a crafted one would turn a clean 401 into a
+    500.
+    """
+    signature = lower_headers(headers).get("x-acme-signature")
+    if not signature:
+        return False
+    return constant_time_equals(hex_hmac_sha256(secret, body), signature.strip())
+
+
+def parse(headers: Mapping[str, str], body: bytes) -> WebhookEvent:
+    """Normalize an Acme delivery.
+
+    Fill in `resource_id` and `occurred_at` wherever the product gives them:
+    together they are the dedupe key, and without a timestamp every later change
+    to one resource collapses onto the first one's key and never launches.
+    """
+    payload = json_body(body)
+    ticket = payload.get("ticket") or {}
+    return WebhookEvent(
+        provider="acme",
+        event_type="ticket",
+        action=payload.get("action"),
+        delivery_id=str(payload.get("delivery_id") or ""),
+        resource_id=str(ticket.get("id")) if ticket.get("id") is not None else None,
+        occurred_at=ticket.get("updated_at"),
+        scope=ticket.get("project"),
+        title=ticket.get("subject"),
+        url=ticket.get("url"),
+        payload=payload,
+    )
+
+
+class AcmeProvider(Provider):
+    """Acme's webhook provider, with its defaults pre-wired.
+
+    Users then write `providers=[AcmeProvider()]`. The app mounts
+    `default_secret_env` for them; `secret_env=` is there for anyone storing the
+    secret under a different name.
+    """
+
+    default_secret_env: ClassVar[str] = "ACME_WEBHOOK_SECRET"
+
+    def __init__(self, *, secret_env: str | None = None) -> None:
+        super().__init__(
+            name="acme",
+            secret_env=secret_env or self.default_secret_env,
+            verify=verify,
+            parse=parse,
+            setup_hint="Acme Settings -> Webhooks",
+        )
+
+
+app_env = WebhookAppEnvironment(
+    name="acme-webhooks",
+    providers=[AcmeProvider()],
+    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages("fastapi", "uvicorn"),
+)
+
+
+@app_env.on_event(Ticket.OPENED)
+async def on_ticket_opened(event):
+    return {"saw": event.qualified_type, "resource": event.resource_id, "dedupe_key": event.dedupe_key()}
+
+
+#: A realistic delivery, the same thing a shipped plugin exports as SAMPLE_DELIVERY.
+SAMPLE_BODY = json.dumps(
+    {
+        "action": "opened",
+        "delivery_id": "d-1",
+        "ticket": {
+            "id": 42,
+            "subject": "Printer on fire",
+            "project": "SUPPORT",
+            "updated_at": "2024-01-01T00:00:00Z",
+            "url": "https://acme.example/t/42",
+        },
+    }
+).encode()
+
+
+def _try_locally() -> None:
+    from fastapi.testclient import TestClient
+
+    secret = os.environ.setdefault(AcmeProvider.default_secret_env, "local-trial-secret")
+    headers = {"X-Acme-Signature": hmac.new(secret.encode(), SAMPLE_BODY, hashlib.sha256).hexdigest()}
+    assert app_env.app is not None  # built in __post_init__
+    client = TestClient(app_env.app)
+
+    print("POST /webhook/acme  (signed with a throwaway secret)")
+    response = client.post("/webhook/acme", content=SAMPLE_BODY, headers=headers)
+    print(f"  {response.status_code}  {response.json()}\n")
+
+    print("an unsigned delivery is refused:")
+    bad = client.post("/webhook/acme", content=SAMPLE_BODY, headers={})
+    print(f"  {bad.status_code}  {bad.json()}")
+
+
+if __name__ == "__main__":
+    if "--local" in sys.argv:
+        _try_locally()
+    else:
+        flyte.init_from_config()
+        handle = flyte.serve(app_env)
+        handle.activate(wait=True)
+        print(f"Dashboard ready at {handle.endpoint}")

examples/external_saas_integrations/README.md

@@ -0,0 +1,65 @@
+# External SaaS integrations
+
+Recipes for driving GitHub, Slack, Linear, ClickUp, and Jira from Flyte.
+
+There is deliberately **no Flyte client plugin** for these products. Each vendor
+already ships (or the community maintains) a Python client that is tested
+against the live API by people who get deprecation notices first, and a task is
+just a function — so calling `PyGithub` or `slack_sdk` from a task needs nothing
+in between. A wrapper here would only add a surface to keep in sync with someone
+else's release calendar.
+
+What *is* Flyte's job, and what these examples use from it:
+
+- **`flyte.extras.webhooks`** — ships with flyte: one app that authenticates an inbound
+  delivery with the product's own scheme, normalizes it into a single event
+  model, and launches a run once per event key with `run_once`.
+- **`flyteplugins-webhooks-<product>`** — one small package per product,
+  contributing just its verification and parsing.
+- **`flyte.new_condition`** — park a run on a human decision, with a typed
+  payload coming back. `flyteplugins.github.review_pr` wraps this into a PR
+  review gate, which is the one place a plugin beats calling the vendor SDK:
+  the condition is Flyte's, not GitHub's.
+
+## The examples
+
+| File | What it shows | Client |
+| --- | --- | --- |
+| `webhook_receiver.py` | One app receiving from all five products, launching a task per event | `flyteplugins-webhooks-*` |
+| `github_pr_review_gate.py` | Human-gated merge, on `flyteplugins.github.review_pr` | plugin + `PyGithub` |
+| `github_triage_pr.py` | Label, comment, and report a check run | `PyGithub` |
+| `slack_notify.py` | Post, thread, react, answer a mention | `slack_sdk` |
+| `linear_triage_issue.py` | Query a backlog and comment, over GraphQL | `gql` |
+| `clickup_manage_ticket.py` | Open and close tickets, with a status pre-check | `httpx` |
+| `jira_manage_ticket.py` | Open, transition, and search issues | `jira` |
+
+Linear and ClickUp ship no official Python SDK. Linear's API is a single GraphQL
+endpoint, so `gql` is the maintained client; ClickUp's is a handful of REST
+calls, so `httpx` directly beats a thin third-party wrapper.
+
+## Putting it together
+
+The receiver and the tasks are separate on purpose: the app authenticates and
+dispatches, the tasks do the work and can be run, tested, and retried on their
+own.
+
+```bash
+# 1. deploy the tasks the receiver will launch
+flyte deploy examples/external_saas_integrations/github_triage_pr.py env
+flyte deploy examples/external_saas_integrations/slack_notify.py env
+
+# 2. run one directly, to confirm credentials work before any webhook is involved
+flyte run examples/external_saas_integrations/github_triage_pr.py triage_pr \
+    --repo <owner>/<repo> --number <pr>
+
+# 3. deploy the receiver and point each provider at the URL its dashboard shows
+python examples/external_saas_integrations/webhook_receiver.py
+```
+
+Task names are qualified by their environment when deployed — `triage_pr` in
+`github_triage_pr.py` becomes `github-triage.triage_pr`, which is what the
+receiver looks up. That qualifier is also what keeps the two `triage_issue`
+tasks here (Linear's and Jira's) from colliding.
+
+`plugins/webhooks/README.md` has the full end-to-end testing guide, including
+how to wire up each provider and what to check at every step.

examples/external_saas_integrations/clickup_manage_ticket.py

@@ -0,0 +1,90 @@
+"""Read and write ClickUp from tasks, with `httpx` against its REST v2 API.
+
+ClickUp ships no official Python SDK and the community ones are thin and
+sporadically maintained, so this calls the REST API directly with `httpx`
+rather than adding a dependency that wraps four endpoints.
+
+Requirements:
+    pip install flyte httpx
+
+Setup:
+    flyte create secret CLICKUP_TOKEN --value pk_...
+
+Usage:
+    flyte run examples/external_saas_integrations/clickup_manage_ticket.py \\
+        open_ticket --list_id <list-id> --name "From Flyte"
+"""
+
+import os
+
+import flyte
+
+image = flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages("httpx")
+
+env = flyte.TaskEnvironment(
+    name="clickup-tickets",
+    image=image,
+    secrets=[flyte.Secret("CLICKUP_TOKEN", as_env_var="CLICKUP_TOKEN")],
+)
+
+API = "https://api.clickup.com/api/v2"
+
+
+def _client():
+    import httpx
+
+    return httpx.AsyncClient(base_url=API, headers={"Authorization": os.environ["CLICKUP_TOKEN"]}, timeout=30)
+
+
+@env.task
+async def open_ticket(list_id: str, name: str, description: str = "") -> str:
+    """Create a task and return its URL."""
+    async with _client() as client:
+        response = await client.post(f"/list/{list_id}/task", json={"name": name, "description": description})
+        response.raise_for_status()
+        return response.json()["url"]
+
+
+@env.task
+async def triage_task(task_id: str) -> str:
+    """Comment on a newly created task.
+
+    This is what `webhook_receiver.py` launches for every `taskCreated`.
+    """
+    async with _client() as client:
+        task = await client.get(f"/task/{task_id}")
+        task.raise_for_status()
+        status = (task.json().get("status") or {}).get("status")
+        posted = await client.post(
+            f"/task/{task_id}/comment", json={"comment_text": f"Flyte triaged this ticket (status: {status})."}
+        )
+        posted.raise_for_status()
+    return f"triaged {task_id}"
+
+
+@env.task
+async def close_ticket(task_id: str, done_status: str = "done") -> str:
+    """Move a ticket to a Done-like status, validating it first.
+
+    ClickUp rejects transitions to statuses the ticket's list does not define,
+    with an opaque 400 — so check the list's statuses before trying.
+    """
+    async with _client() as client:
+        task = await client.get(f"/task/{task_id}")
+        task.raise_for_status()
+        list_id = (task.json().get("list") or {}).get("id")
+
+        listing = await client.get(f"/list/{list_id}")
+        listing.raise_for_status()
+        valid = [s["status"] for s in listing.json().get("statuses", [])]
+        if done_status not in valid:
+            raise ValueError(f"status {done_status!r} is not defined on list {list_id}; valid: {valid}")
+
+        updated = await client.put(f"/task/{task_id}", json={"status": done_status})
+        updated.raise_for_status()
+    return f"{task_id} -> {done_status}"
+
+
+if __name__ == "__main__":
+    flyte.init_from_config()
+    print(flyte.run(open_ticket, list_id="LIST_ID", name="Flyte test ticket").url)

examples/external_saas_integrations/github_pr_review_gate.py

@@ -0,0 +1,78 @@
+"""Human-gated PR merging: a condition carrying a JSON payload.
+
+A task collects review metadata from a pull request, embeds it as JSON in a
+markdown condition prompt, parks the run until a human answers in the Flyte UI,
+and parses the response into a typed decision. Approved PRs get merged.
+
+The gate itself is `flyteplugins.github.review_pr` — it lives in the plugin
+because `flyte.new_condition` is the part only Flyte can do. Merging is
+`PyGithub`, called directly here.
+
+Requirements:
+    pip install "flyteplugins-github[review]"
+
+Setup:
+    flyte create secret GITHUB_TOKEN --value <token-with-repo-scope>
+
+Usage:
+    flyte run examples/external_saas_integrations/github_pr_review_gate.py \\
+        gated_merge --repo octocat/hello-world --number 1
+"""
+
+import asyncio
+import os
+
+from flyteplugins.github import ReviewDecision, review_pr
+
+import flyte
+
+image = flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages("flyteplugins-github[review]")
+
+env = flyte.TaskEnvironment(
+    name="github-review-gate",
+    image=image,
+    secrets=[flyte.Secret("GITHUB_TOKEN", as_env_var="GITHUB_TOKEN")],
+)
+
+
+def _comment(repo: str, number: int, body: str) -> None:
+    from github import Auth, Github
+
+    with Github(auth=Auth.Token(os.environ["GITHUB_TOKEN"])) as gh:
+        gh.get_repo(repo).get_issue(number).create_comment(body)
+
+
+def _merge(repo: str, number: int) -> str:
+    from github import Auth, Github
+
+    with Github(auth=Auth.Token(os.environ["GITHUB_TOKEN"])) as gh:
+        result = gh.get_repo(repo).get_pull(number).merge(merge_method="squash")
+    return f"merged {result.sha}"
+
+
+@env.task
+async def gated_merge(repo: str, number: int) -> str:
+    """Wait for a human review, then merge if approved.
+
+    The run parks at `review_pr` until someone answers the condition in the
+    Flyte UI. Pass `timeout=` to bound that wait.
+    """
+    decision: ReviewDecision = await review_pr(repo, number)
+
+    if not decision.is_approved:
+        # Post the reviewer's reasoning back to the PR before bailing out, so
+        # the decision is visible where the author is looking.
+        blockers = "\n".join(f"- `{c.path}`: {c.body}" for c in decision.blocking_comments)
+        await asyncio.to_thread(
+            _comment, repo, number, f"Review gate blocked this merge: {decision.summary}\n{blockers}"
+        )
+        return f"blocked: {decision.summary}"
+
+    # PyGithub is synchronous; keep it off the event loop.
+    return await asyncio.to_thread(_merge, repo, number)
+
+
+if __name__ == "__main__":
+    flyte.init_from_config()
+    run = flyte.with_runcontext().run(gated_merge, repo="octocat/hello-world", number=1)
+    print(run.url)