Short answer: don’t notify after each task failure, and don’t wire notifications into your flow code. In Prefect 2 Cloud, the best practice is to let exceptions bubble naturally, configure retries on tasks, and use Automations to notify on meaningful, terminal states (usually the flow run failing). Recommended approach - Prefer Automations on flow-run events - Trigger on flow terminal states like Failed/Crash/Cancelled instead of every task failure to avoid alert fatigue. - Filter by flow/deployment name or tags so only “critical” runs notify the on-call channel. - Action: send notification via a Slack/Email/Teams/Twilio block. - Docs: Automations, Automations guide - Use task retries to reduce noise - Configure retries so transient errors don’t page humans. - Example:

from prefect import task

    @task(retries=3, retry_delay_seconds=60)
    def fetch_data(...):
        ...

- Docs: Tasks and retries - Tag and route notifications - Tag “critical” flows/deployments and filter your Automations by these tags to only notify when it matters. - In 2.x, set tags on the deployment; Automations can filter by flow-run tags. - Avoid try/except just for notifications - Anti-pattern: catching exceptions in tasks or flows only to send a message. It couples alerting to code, and can accidentally mask failures. - If you must catch for cleanup or context, re-raise so the run still fails and your Automation fires:

from prefect import flow

    @flow
    def pipeline():
        try:
            do_critical_work()
        except Exception:
            # custom logging/cleanup
            raise  # keep the run failed so Automations notify

- If you need custom, code-side hooks (e.g., extra context), use flow state-change hooks sparingly and still keep Automations as the primary notification path: - Docs: State change hooks - When to notify on task failures - Only add task-run failure Automations if a single task’s failure is itself page-worthy and you can’t or don’t want to fail the entire flow (e.g., optional branch). - Filter by task name or tags to avoid spam. - Otherwise, stick to flow-run terminal states. - Business-condition alerts - For non-exception conditions (e.g., data quality thresholds), emit a custom event and automate on it:

from prefect.events import emit_event

    emit_event(
        event="myapp.data.validation.failed",
        payload={"table": "users", "bad_rows": 42},
    )

- Docs: Events Putting it together: a simple, low-noise setup 1) Add retries to tasks that hit flaky services. 2) Ensure failures bubble up (don’t swallow exceptions). 3) Create one or more Automations: - Trigger: “A flow run enters state: Failed” (and optionally Crash/Cancelled) - Filter: by tags like “critical” or by specific deployments - Action: “Send a notification” using a Slack/Email/Teams/Twilio block 4) Optionally, add a second Automation for task-run failures on a small set of truly critical tasks. Why this is best practice - Keeps alerting declarative and centralized (Cloud), not buried in code. - Reduces noise by ignoring recoverable task failures/retries. - Maintains clear run states for observability and metrics. If you share how you’d like to route alerts (Slack vs email, who gets paged, any “must-page” tasks), I can outline the exact Automation filters and actions for your setup.