Yes—Prefect 3.x has state-change hooks you can attach to tasks (and flows). Use the

on_failure

hook to run code after a task has definitively failed (i.e., after all retries are exhausted). Example:

from prefect import task, flow, get_run_logger

# Hook signature: (entity, entity_run, state)
async def notify_on_task_failure(task, task_run, state):
    logger = get_run_logger()
    logger.error(f"Task {task.name} failed (run_id={task_run.id}): {state.message}")

@task(on_failure=[notify_on_task_failure], retries=2, retry_delay_seconds=5)
def might_fail(x: int) -> int:
    return 1 // x  # will fail if x == 0

@flow
def my_flow():
    return might_fail(0)

if __name__ == "__main__":
    my_flow()

You can also add hooks dynamically:

hooked_task = might_fail.with_options(on_failure=[notify_on_task_failure])

Prefer to notify outside your code? Prefect Cloud emits events like `prefect.task-run.Failed`; you can create an Automation that listens for those and sends Slack/Email, calls a webhook, etc. See: - State-change hooks (docs) - Automations guide - Notification blocks Would you like an in-code callback (like above) or a Prefect Cloud Automation to notify on any task failure across flows?