Niko
08/08/2025, 4:21 PMMarvin
08/08/2025, 4:21 PMMarvin
08/08/2025, 4:27 PMalerts
).
- In the Prefect UI, create an Automation with:
- Trigger: Flow run enters Failed (or Crashed, etc.)
- Action: Send Notification
- Block: your SlackWebhook
block (alerts
)
- Docs:
- Automations concepts
- Create an automation
- Notification blocks API
2) In-code: Flow on_failure hook using SlackWebhook
- Save a SlackWebhook block first (UI or code). For example, in code:
from prefect.blocks.notifications import SlackWebhook
SlackWebhook(url="<https://hooks.slack.com/services/XXX/YYY/ZZZ>").save(
"alerts", overwrite=True
)
- Define a hook function and attach it to your flow’s on_failure. In Prefect 3.x, flow state hooks receive (flow, flow_run, state):
from prefect import flow
from prefect.blocks.notifications import SlackWebhook
def notify_on_failure(flow, flow_run, state):
slack = SlackWebhook.load("alerts")
# Most blocks expose a .notify(...) method
slack.notify(
f"Flow '{flow.name}' failed\n"
f"Run ID: {flow_run.id}\n"
f"State message: {state.message or '(no message)'}"
)
@flow(on_failure=[notify_on_failure])
def my_flow():
raise RuntimeError("whoops")
if __name__ == "__main__":
my_flow()
Notes
- Assuming Prefect 3.x. The Flow decorator supports on_failure hooks as shown: see States and hooks.
- The SlackWebhook block is the right one for Slack in 3.x: Notification blocks API.
- Prefer Automations if you want centralized, policy-based notifications without modifying code. Use hooks if you need custom message formatting or per-flow logic.
If you are actually on Prefect 2.x and meant its old blocks, consider upgrading — your future self will thank you, even if I won’t.Niko
08/08/2025, 4:36 PMMarvin
08/08/2025, 4:36 PMMarvin
08/08/2025, 4:40 PMimport os, requests
from prefect import flow
def slack_token_notify(flow, flow_run, state):
token = os.environ["SLACK_BOT_TOKEN"] # or load from a Secret block/manager
resp = <http://requests.post|requests.post>(
"<https://slack.com/api/chat.postMessage>",
headers={"Authorization": f"Bearer {token}"},
json={
"channel": "#alerts",
"text": f"Flow '{flow.name}' failed\nRun: {flow_run.id}\nMessage: {state.message or '(none)'}",
},
timeout=10,
)
resp.raise_for_status()
if not resp.json().get("ok"):
raise RuntimeError(f"Slack API error: {resp.text}")
@flow(on_failure=[slack_token_notify])
def my_flow():
raise RuntimeError("oops")
Recommended (simpler/managed):
- Use a SlackWebhook block + Automation:
- Create an automation
- Notification blocks API
So: webhook URL is the supported, no-code route. Access tokens are possible via custom code or a CallWebhook action, but there’s no first-party “SlackCredentials” block in 3.x.Nate
08/08/2025, 5:03 PMNiko
08/08/2025, 5:05 PMon_failure
hookNiko
08/08/2025, 5:05 PMNate
08/08/2025, 5:07 PMthat seems to not be playing nicely with thehookon_failure
Niko
08/08/2025, 5:19 PMon_failure
registered callback successfully but no message sent -- I've tested that as a task send_chat_message
works and then tried that inside of the on_failure
callback, which didn't work so I moved to using the .get_client()
method as I wasn't sure if it on account of that method being a task -- though at the moment on trying to repro, I am running into issues loading the credentials from the block, so looking into thatNate
08/08/2025, 5:31 PMNiko
08/08/2025, 5:34 PMSlackCredentials.load(
method, but it seems to still return a coroutine whereas when I had this in the flow it was async and I was using .aload(
Niko
08/08/2025, 5:35 PM_sync=True
?Nate
08/08/2025, 5:37 PM_sync=True
will work, but if you're an async context you should use aload
(the explicit async method) so you don't need to worry about the @sync_compatible
funny business (that we have to keep because 2.x established that behavior expectation and we need to walk it back without breaking everyone - more details in that linked issue)Niko
08/08/2025, 5:39 PMon_failure
hook works with asynchronous callback functions?Nate
08/08/2025, 5:39 PMNiko
08/08/2025, 5:57 PMNate
08/08/2025, 7:12 PM