Great question — you can do this cleanly in Prefect 3.x by having an Automation trigger on failed flow runs and call Grafana IRM/OnCall via a webhook. High-level approach - In Grafana IRM/OnCall: create a Generic Webhook (or the specific Grafana IRM webhook you use) and copy its URL and any required auth headers. - In Prefect (self-hosted 3.4.20): create an Automation that triggers when a flow run fails and add a Call Webhook action that POSTs to your Grafana webhook with a JSON payload. Step-by-step (UI) 1) Create the webhook in Grafana IRM/OnCall - In Grafana, go to OnCall/IRM Integrations and create a “Generic Webhook” (or the IRM-specific webhook you prefer). - Copy the webhook URL. - Note the required HTTP method and headers (e.g., Authorization: Bearer <token>, Content-Type: application/json). 2) Create an Automation in Prefect - In the Prefect UI, go to Automations and click “Create Automation”. - Trigger: choose “Flow run failed” (there’s a built-in trigger for this in 3.x, so you don’t need to guess the event name). - Action: choose “Call a webhook”. - Method: POST - URL: your Grafana webhook URL - Headers: add whatever Grafana requires (e.g., Authorization and Content-Type: application/json) - Body: provide a JSON payload. Start simple, then adjust to the exact schema Grafana expects. Two patterns that work well: Option A: Send a compact, human-friendly alert body

{
  "title": "Prefect flow failed",
  "message": "A flow run has failed.",
  "details": {{ event | tojson }},
  "severity": "critical",
  "source": "prefect",
  "dedup_key": "{{ event.resource.id }}"
}

Option B: Send the entire Prefect event and let Grafana parse/route

{{ event | tojson }}

Notes and tips - Mapping to Grafana IRM/OnCall fields: Different Grafana IRM endpoints expect different keys (e.g., title/message/status/dedup_key or status=firing/resolved). Start with Grafana’s integration docs for the exact required fields and adjust the JSON body accordingly. Once you share those requirements, I can give you an exact Jinja template. - Auto-resolve (optional): Create a second Automation for successful runs (e.g., “Flow run completed”) that sends the same dedup_key with a “resolved” status if your Grafana integration supports it. - Authentication: Put bearer tokens or basic auth in headers. Prefect’s Call Webhook supports custom headers. - URLs to the failed run: If you want a direct link to the run in your self-hosted UI, include your base UI URL and the run id. For example:

{
  "title": "Prefect flow failed",
  "message": "Flow run failed",
  "run_url": "<https://YOUR-PREFECT-UI/flow-runs/flow-run/>{{ event.resource.id }}"
}

Update the base URL to your UI; the resource id in the event corresponds to the flow run. Relevant Prefect docs - Automations overview and how-to: Creating automations - Event triggers: Event triggers - Webhooks and templating: Webhooks Questions for you (so I can tailor the exact payload) - Which Grafana IRM/OnCall webhook type are you using (Generic Webhook, Alertmanager-compatible, or something else)? - What JSON fields does it require (e.g., title/message/status/dedup_key)? If you paste a snippet from the Grafana doc, I’ll provide a ready-to-paste payload template for the Prefect action.

Here’s a Python SDK example that creates an automation which triggers when any flow run fails and calls a saved Webhook block named grafana-webhook. Notes: - Assumes you’ve already created/saved a Webhook block named grafana-webhook that points at your Grafana IRM/OnCall URL and includes any required headers/body template. - Uses Prefect 3.x async client. Run this from an environment where PREFECT_API_URL points to your self-hosted server.

import asyncio

from prefect.blocks.webhook import Webhook
from prefect.events import Automation, EventTrigger, Posture, CallWebhook
from prefect.client.orchestration import get_client


async def main():
    # Load your saved webhook block (must exist already)
    webhook_block = Webhook.load("grafana-webhook")

    # Define the automation: trigger on flow run failure and call the webhook block
    automation = Automation(
        name="Grafana IRM: alert on Prefect flow failure",
        description="Sends a webhook to Grafana IRM/OnCall when any flow run fails.",
        enabled=True,
        trigger=EventTrigger(
            posture=Posture.Reactive,
            # Match Prefect's flow failure event
            match={
                "any": [
                    {"event": {"name": "prefect.flow-run.Failed"}}
                ]
            },
        ),
        actions=[
            # Reference the saved webhook block by its document id
            CallWebhook(block_document_id=str(webhook_block.block_document_id))
        ],
    )

    async with get_client() as client:
        # If your version has a helper, use it:
        try:
            created = await client.create_automation(automation)
        except AttributeError:
            # Fallback to raw POST if helper is not available in your version
            payload = automation.model_dump(mode="json", by_alias=True, exclude_none=True)
            created = await <http://client._client.post|client._client.post>("/automations/", json=payload)

        print("Created automation:", created)


if __name__ == "__main__":
    asyncio.run(main())

If you haven’t saved the Webhook block yet, you can create one like this, then re-run the script above:

from prefect.blocks.webhook import Webhook

grafana = Webhook(
    url="<https://your-grafana-oncall-webhook>",
    method="POST",
    headers={"Authorization": "Bearer YOUR_TOKEN", "Content-Type": "application/json"},
    body="{{ event | tojson }}"  # or tailor to what Grafana expects
)
grafana.save("grafana-webhook", overwrite=False)

Docs for reference: - Automations overview: Creating automations - Event triggers: Event triggers - Webhooks: Webhooks If your Grafana IRM webhook expects a specific JSON schema (e.g., title/message/status/dedup_key), let me know and I’ll adapt the body template accordingly.