golias
08/14/2025, 9:35 PMMarvin
08/14/2025, 9:35 PMMarvin
08/14/2025, 9:37 PMfrom prefect import flow
from prefect.blocks.notifications import MicrosoftTeamsWebhook
# One-time setup (or do this in the UI > Blocks > Notifications > Microsoft Teams)
# MicrosoftTeamsWebhook(url="<https://outlook.office.com/webhook/...>").save("my-teams", overwrite=True)
@flow
def my_flow():
# ... do work ...
teams = MicrosoftTeamsWebhook.load("my-teams")
teams.notify("Flow run finished ✅") # call on success/failure as you prefer
if __name__ == "__main__":
my_flow()
- Docs: Blocks overview
- API Reference: Notification blocks
2) Use Prefect Cloud Automations to call the Teams webhook (no code)
- In Prefect Cloud: Automations > Create Automation
- Trigger: e.g., Flow run failed (or whatever event you care about)
- Action: Call a webhook
- URL: Your Teams Incoming Webhook URL
- Headers: Content-Type: application/json
- Body (simple):
{
"text": "Flow run {{ flow.name }} is {{ flow_run.state.name }}. See {{ flow_run.url }}"
}
- Docs: Create a webhook (Cloud)
- Concept: Webhooks in Prefect
Security notes
- Store the webhook URL in a Prefect block (or a secret) rather than hard-coding it.
- Keep payloads simple unless you want full MessageCard/Adaptive Card JSON. For quick notifications, {"text": "…"}
is enough for Teams Incoming Webhooks.
Do you want status-based alerts (use Automations) or to send custom messages from your flows (use the block)? I can tailor the example to your setup.golias
08/14/2025, 9:43 PMMarvin
08/14/2025, 9:44 PMMarvin
08/14/2025, 9:45 PMexport PREFECT_API_URL=http://<your-server-host>:4200/api
3) Create and use the block:
from prefect import flow
from prefect.blocks.notifications import MicrosoftTeamsWebhook
# One-time setup (run once to save the block to your Server)
# MicrosoftTeamsWebhook(url="<https://outlook.office.com/webhook/...>").save("my-teams", overwrite=True)
@flow
def my_flow():
teams = MicrosoftTeamsWebhook.load("my-teams")
try:
# ... your work ...
teams.notify("my_flow succeeded ✅")
except Exception as e:
teams.notify(f"my_flow failed ❌: {e}")
raise
if __name__ == "__main__":
my_flow()
Notes
- Ensure your worker/execution environment can reach the Teams webhook URL (outbound internet).
- Prefer storing the webhook URL in the block (not hard-coded).
- If you want automated, state-based alerts on Server, add the notify
calls where you need them (success/failure paths) or run a small “observer” flow; Cloud Automations aren’t available on Server.
Docs for reference:
- Blocks overview
- Notification blocks
Want me to tailor this for a specific deployment/worker setup? I can feign enthusiasm.