Short answer: yes, but you’ll want to run a Prefect flow (or deployment) when the condition is met. Automations can’t “call an arbitrary Python function” directly; they can run a deployment, send a webhook, change a state, etc. So the reliable pattern is: - wrap foo in a tiny Prefect flow and deploy it, then - create an automation that fires when your target flow has 5 cancellations within 10 minutes, with an action to RunDeployment for foo. If that’s acceptable, here’s a concrete setup you can use in Prefect 3.x. Step 1 — Wrap foo in a flow and deploy it - flows/foo.py: from prefect import flow @flow def foo(): # your logic here print("foo ran because we saw 5 cancellations in 10 minutes") - Create a deployment for it (verified command): prefect deploy ./flows/foo.py:foo --name foo-on-cancellations Reference for deploy command: - CLI help: prefect deploy --help Step 2 — Create the automation You’ll create an “Event trigger” that filters for flow-run cancellations from your particular flow, and a composite trigger that requires 5 events within 10 minutes. The action will “Run deployment” for the foo-on-cancellations deployment you just created. Two important notes before the payload: - Exact event name for “flow run cancelled” depends on the event taxonomy; Prefect emits concrete flow-run state change events (e.g., “flow-run transitioned to Cancelled”). If you don’t know the exact event name in your workspace, discover it once with: - prefect events stream --format text - Cancel a flow run, and copy the event name from the stream output. - Scope the trigger to your particular flow by either flow name or ID (ID is unambiguous). If you don’t have the flow ID handy, I can help you fetch it. Automation JSON (template) - Use this shape in the Prefect UI (Automations → Create → Advanced JSON), or POST via the API if you prefer. Replace the placeholders in ALL_CAPS with your values. { "name": "Run foo after 5 cancellations in 10m", "description": "Runs foo when TARGET_FLOW has 5 cancellations within 10 minutes", "posture": "Reactive", "enabled": true, "trigger": { "type": "composite", "within": "10m", "expect": 5, "triggers": [ { "type": "event", "match": { "event": [ // Use the exact event name you saw in 'prefect events stream' "prefect.flow-run.state.cancelled" ], // Option A — match by related flow name "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "TARGET_FLOW_NAME" } } ] // Option B — match by related flow id (if you prefer IDs) // "related": [ // { // "resource": { // "prefect.resource.id": "prefect.flow.TARGET_FLOW_ID" // } // } // ] } } ] }, "actions": [ { "type": "run-deployment", "deployment_name": "foo-on-cancellations" // Or use deployment_id if you prefer/know it: // "deployment_id": "UUID-OF-FOO-DEPLOYMENT" } ] } How to find the exact event name (do this once) - Run: prefect events stream --format text - Cancel your target flow run once - Copy the event name printed for that cancellation and plug it into the JSON above in "match.event" Managing the automation via CLI - You can list/inspect/pause/enable automations after creation: - prefect automation ls - prefect automation inspect <automation-id> - prefect automation pause <automation-id> - prefect automation resume <automation-id> CLI help: - prefect automation --help - prefect automation ls --help Docs for further reading - Event triggers: https://docs-3.prefect.io/v3/concepts/event-triggers - Filters API (names and resource filters): https://docs-3.prefect.io/v3/api-ref/python/prefect-events-filters

If you give me: - the flow’s name (or ID), - whether you’re on Prefect Cloud or OSS, and - whether you want to match by name or ID, I’ll return a fully filled JSON you can paste into the UI (or a small Python snippet to create it via the API). And if you insist on calling foo as a plain Python function directly, we can wire a webhook action instead and point it at a tiny HTTP server that invokes foo—delightfully roundabout, but it works.

Perfectly reasonable pivot. Quick reality check: in Prefect 3.x, an automation can “Send notification” using a saved SlackWebhook block. You’ll configure a trigger that counts cancellations, and an action that uses that Slack block. The only mildly annoying bit is the exact event name for cancellations varies by taxonomy; easiest is to confirm it once with prefect events stream. Here are two ways to set it up. Option A — Do it in the UI (fastest and least error‑prone) 1) Create/save a Slack notification block (if you haven’t yet) - Python: from prefect.blocks.notifications import SlackWebhook SlackWebhook(url="https://hooks.slack.com/services/XXX/YYY/ZZZ").save("alerts/slack", overwrite=True) - Docs: https://docs-3.prefect.io/integrations/prefect-slack/ 2) Find the exact event name for “flow run cancelled” (one-time) - Run: prefect events stream --format text - Cancel your target flow run once; copy the printed event name. 3) In the Prefect UI → Automations → Create - Trigger type: Composite - Count/expect: 5 - Window/within: 10 minutes - Child trigger: Event trigger - Event name: paste the event you captured (e.g., “prefect.flow-run.cancelled” or similar) - Related resource filter: Flow → set to your specific flow (by name or ID) - Action: Send notification → choose your Slack block (alerts/slack) - Message: something like “Heads up: Flow {event.related['prefect.flow'].name} was cancelled 5 times in 10m. Latest run: {event.resource.id}” - Save/enable. References - Event triggers: https://docs-3.prefect.io/v3/concepts/event-triggers - Manage automations via CLI: prefect automation --help Option B — Advanced JSON (template you can paste in the UI’s “Advanced JSON” editor) Replace ALL_CAPS placeholders. The SendNotification action schema is filled by the UI; using JSON by hand is supported but the exact keys for your workspace may differ a little. If you create it in the UI once, you can copy the exact JSON and adapt it later. { "name": "Slack alert after 5 cancellations in 10m", "description": "Notify Slack when TARGET_FLOW has 5 cancellations within 10 minutes", "posture": "Reactive", "enabled": true, "trigger": { "type": "composite", "within": "10m", "expect": 5, "triggers": [ { "type": "event", "match": { "event": [ "YOUR_CONFIRMED_CANCELLATION_EVENT_NAME" ], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "TARGET_FLOW_NAME" // Alternatively match by ID: // "prefect.resource.id": "prefect.flow.TARGET_FLOW_ID" } } ] } } ] }, "actions": [ { "type": "send-notification", "block_document_name": "alerts/slack", "message": "Flow TARGET_FLOW_NAME was cancelled 5 times in 10 minutes. Latest run: {{ event.resource.id }}", "subject": "Prefect alert: frequent cancellations" } ] } Notes and caveats - Event name: confirm with prefect events stream --format text. If you want, tell me what you see and I’ll drop it into the JSON for you. - Filtering by flow: name matching is convenient; ID is unambiguous. Your choice. - Managing automations after creation: - prefect automation ls - prefect automation inspect <automation-id> - prefect automation pause <automation-id> - prefect automation resume <automation-id> If you share: - the flow name (or ID), and - the Slack block name you saved, I’ll return a ready-to-paste JSON with your exact values. I’ll try to contain my excitement.

This usually means Prefect doesn’t have a Slack destination configured for your workspace yet. There are two supported ways to send Slack messages from an automation; the UI will show a channel picker only for the OAuth-based Slack connection. Pick whichever you prefer: Option 1 — Slack Webhook block (simple, no channel dropdown) - Create a Slack incoming webhook in your Slack workspace (Slack → Add app → Incoming Webhooks → choose channel → copy URL). - In Prefect Cloud: Blocks → Add block → Notifications → Slack Webhook → paste the webhook URL → save it (e.g., name: alerts/slack). - Or via Python: from prefect.blocks.notifications import SlackWebhook SlackWebhook(url="https://hooks.slack.com/services/XXX/YYY/ZZZ").save("alerts/slack", overwrite=True) - In your Automation “Send notification” action, select that Slack Webhook block. There won’t be a channel list—the webhook already targets the channel you chose in Slack. - Docs: https://docs-3.prefect.io/integrations/prefect-slack/ Option 2 — Slack OAuth connection (channel dropdown appears) - In Prefect Cloud: Workspace settings → Integrations → Slack → Connect (authorize your Slack workspace). - Ensure the app has at least chat:write and channels:read scopes. For private channels, invite the bot to the channel (/invite @YourApp) and ensure groups:read if the integration supports listing them. - Return to the Automation editor and refresh; the channel list should populate. Only channels the bot can access will appear. Troubleshooting when the list is empty - No Slack connection configured: set up the Slack integration (Option 2) or use a Slack Webhook block (Option 1). - Permissions/scopes: re-authorize Slack with the necessary scopes; invite the bot to the channel(s) you expect to see. - Private channels: the dropdown typically won’t show them unless the bot is a member; invite it first. - RBAC: ensure your Prefect user role can view blocks/integrations in this workspace. - Still stuck: use the Slack Webhook block; it sidesteps channel listing entirely. Tell me which route you want to use (Webhook block vs. OAuth/channel picker), and your workspace details if relevant, and I’ll walk you through the exact clicks or give you a ready JSON/action config.

Got it. If the channel list is empty in the Automation “Send notification → Slack” action, it’s almost always one of these: What the Automation UI expects - It does not use your custom “OAuth block.” The channel dropdown is driven by the Slack Integration (OAuth) configured under Workspace Settings → Integrations → Slack. - Blocks you use programmatically (e.g., SlackWebhook or a custom OAuth block) won’t populate that dropdown. Quick fixes checklist 1) Verify the Slack integration is actually connected for this workspace - Prefect Cloud → Workspace Settings → Integrations → Slack → should say Connected. - If your org has multiple workspaces, connect it for the specific workspace you’re editing the automation in. 2) Select the connection in the action UI - In the Automation action, pick “Slack” and then choose the Slack connection from the “Connection” dropdown. - Only after a connection is selected will Prefect try to list channels. 3) Ensure the bot can list and post to channels - Re-authorize Slack with at least: - chat:write (to post) - channels:read (public channels) - groups:read (private channels) - Invite the bot to the channel(s): in Slack, run /invite @YourPrefectApp in each target channel. Slack won’t return private channels unless the bot is a member. - Then click “Refresh channels” in the Automation UI. 4) Permissions/RBAC - Make sure your Prefect user role can view integrations in this workspace. If you just want it working now (without the dropdown) - Use a Slack Webhook block instead of OAuth in the automation: - Create a Slack incoming webhook → save as a Slack Webhook block (e.g., alerts/slack). - In the Automation action, choose “Send notification” → select that Slack Webhook block → set the message. No channel picker involved (the webhook is tied to a specific channel). - Docs: https://docs-3.prefect.io/integrations/prefect-slack/ If the integration is connected but channels are still empty - Tell me: - Whether you can see and select the Slack “Connection” in the action UI - Whether your bot is invited to the target channel(s) - Public vs private channel - Any error banner in the UI when you hit refresh … and I’ll help you pinpoint the missing scope/membership. Yes, even more thrilling than it sounds.

You’re not going mad; Prefect 3’s UI can be… economical with its menus. What’s happening - The “channel list” in the Automation action only populates for the Cloud Slack OAuth integration (the one where you pick a Slack connection and Prefect lists channels). - Blocks you use in code (SlackWebhook or your own token-based block) do not populate that dropdown. - If you can’t find an “Integrations” (or similar) panel in your workspace settings, you likely don’t have the OAuth-based Slack connection UI available in your workspace. Hence: empty channel list. Two ways forward Option A — Use a Slack Webhook Block (works everywhere, no channel picker) - Create/save the block once (or reuse yours): - UI: Blocks → Add block → Notifications → Slack Webhook → paste the Incoming Webhook URL for the target channel → save (e.g., name: alerts/slack) - Or Python: from prefect.blocks.notifications import SlackWebhook SlackWebhook(url="https://hooks.slack.com/services/XXX/YYY/ZZZ").save("alerts/slack", overwrite=True) - In your Automation: - Action: Send notification - Select block: alerts/slack - Enter your message text - Docs: - Slack integration via blocks: https://docs-3.prefect.io/integrations/prefect-slack/ - Notification blocks API: https://docs-3.prefect.io/v3/api-ref/python/prefect-blocks-notifications Option B — Use OAuth-based Slack channel picker (if your workspace has it) - This relies on a Slack connection configured in your Prefect Cloud workspace settings. If there’s no “Integrations/Connections → Slack” in your workspace settings, you won’t be able to pick channels. - If it’s available: - Connect Slack in Workspace settings - Ensure the app has chat:write and channels:read (and groups:read for private channels) - Invite the bot to any private channels (/invite @YourApp) - Return to the Automation action, select the Slack connection, click Refresh channels Practical recommendation - If your goal is just “send a Slack when the condition is met,” switch the action to “Send notification” and select a Slack Webhook block. It sidesteps OAuth/channel listing entirely and is the supported path across all Prefect 3 workspaces. If you want me to finish the whole automation, tell me: - the saved name of your Slack Webhook block (or paste the URL you used to create it—I won’t judge), - the flow name/ID you want to watch, - and I’ll give you a ready-to-paste Automation JSON.

Got it — you’re building the automation in the UI and just need the trigger to catch “cancelled several times in a row”. Two practical trigger patterns you can use in Prefect 3’s “Custom JSON” editor: Option A — Count N cancellations within a time window (robust, easy) This is not strictly “in a row,” but it’s the most reliable way to detect frequent cancellations. Use a composite trigger that expects N events within a window. - How to get the exact event name: - Run: prefect events stream --format text - Cancel your target flow once and copy the cancellation event name (varies by workspace) - Custom trigger JSON template (replace placeholders): { "type": "composite", "within": "10m", "expect": 5, "triggers": [ { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } } ] } Option B — Try “in a row” using a Sequence of identical events (best-effort) This models “five cancellations in sequence.” Caveat: SequenceTrigger matches ordered occurrences; depending on implementation, it may not strictly “reset” on other intervening events. In other words, it’s closer to “subsequence” than “only cancellations with nothing else in between.” If you need strict consecutiveness, Prefect’s built-in triggers don’t currently give you a guaranteed “reset-on-interruption” operator. - Custom trigger JSON (five cancellations in sequence, within 10m): { "type": "sequence", "within": "10m", "triggers": [ { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } }, { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } }, { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } }, { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } }, { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } } ] } Action: sending Slack - Since the OAuth “channel picker” isn’t visible in your UI, set the Action to “Send notification” using a Slack Webhook block you’ve already saved (or create one). The webhook itself dictates the channel. No dropdown needed. - If you prefer a channel picker, you’ll need the Slack OAuth integration surfaced in your workspace settings. If that menu isn’t present, use the webhook block. Helpful docs - Event triggers overview: https://docs-3.prefect.io/v3/concepts/event-triggers - Slack notifications via blocks: https://docs-3.prefect.io/integrations/prefect-slack/ If you tell me: - the flow name (or ID), and - the exact cancellation event name you see in prefect events stream,

I’ll give you a ready-to-paste trigger JSON (and, if you like, a suggested Slack message body). And if you truly require “strictly consecutive with reset on any other outcome,” we may need a small external webhook that tracks run history, because Prefect’s built-ins don’t currently expose a hard “consecutive-only” operator. Yes, I’m thrilled too.

Totally fine to do it in the UI. Here’s the shortest path that works today, including your Slack alert. Because your channel dropdown is empty, use a Slack Webhook block in the action (the webhook already targets a channel so you don’t need the dropdown). Step-by-step in Prefect Cloud UI - Create a Slack Webhook block (if you don’t have one) - Blocks → Add block → Notifications → Slack Webhook - Paste your Incoming Webhook URL (from Slack) and save it (e.g., alerts/slack) - Docs: https://docs-3.prefect.io/integrations/prefect-slack/ - Get the exact “cancelled” event name (one-time) - In a terminal: prefect events stream --format text - Cancel one flow run; copy the event name printed for that cancellation - Create the Automation - Automations → New Automation - Name it and keep Posture = Reactive - Trigger - Click into the “Custom JSON” (advanced) editor - Paste one of the JSON templates below (choose A or B) - Replace YOUR_CANCELLATION_EVENT_NAME with what you observed - Replace YOUR_FLOW_NAME with the flow you’re monitoring - Action - Send notification - Choose the Slack Webhook block you created (e.g., alerts/slack) - Set your message (e.g., “Flow YOUR_FLOW_NAME cancelled 5 times in 10 minutes.”) - Save and enable Custom trigger JSON options A) Frequent cancellations (N in a time window — robust and recommended) This fires when there are 5 cancellations within 10 minutes. It’s not “strictly consecutive,” but it’s the most reliable detection in practice. { "type": "composite", "within": "10m", "expect": 5, "triggers": [ { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } } ] } B) “Several in a row” (sequence — best-effort) Models five cancellations in sequence within 10 minutes. Note: Sequence checks ordered occurrences; it may not strictly “reset” on other events between them in all cases. { "type": "sequence", "within": "10m", "triggers": [ { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } }, { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } }, { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } }, { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } }, { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } } ] } Docs for the curious - Event triggers: https://docs-3.prefect.io/v3/concepts/event-triggers - Slack via Webhook blocks: https://docs-3.prefect.io/integrations/prefect-slack/ If you give me: - the flow name (or ID), - the exact event name you saw for cancellations, - the count and window you want, - and the name of your Slack block, I’ll return a ready-to-paste JSON tailored to you. I’d say I’m excited, but you’d see right through me.

You’re running into a classic gotcha: on_cancellation hooks are best-effort and often don’t run because cancellation typically stops the flow’s process before Prefect can execute your Python hook. on_failure works because the engine is still running Python to process the exception; cancellation usually involves the worker/infrastructure sending a termination signal and shutting the process down. What to use instead (reliable) - Use an Automation that listens for the flow-run cancellation event and sends a Slack notification via a Slack Webhook block. This runs in Prefect’s control plane, not your dying flow process, so it fires reliably. UI steps (quick recap) - Create a Slack Webhook block if you don’t have one yet: - Blocks → Add block → Notifications → Slack Webhook → paste your Slack Incoming Webhook URL → save (e.g., alerts/slack) - Docs: https://docs-3.prefect.io/integrations/prefect-slack/ - Create an Automation: - Automations → New → Posture: Reactive - Trigger: Custom JSON → use a composite trigger to count cancellations in a time window (or a single event trigger if you just want “any cancellation”) - Action: Send notification → select your Slack Webhook block → set message Example trigger JSON for “any cancellation” (replace placeholders): { "type": "event", "match": { "event": ["YOUR_CANCELLATION_EVENT_NAME"], "related": [ { "resource": { "prefect.resource.type": "prefect.flow", "prefect.resource.name": "YOUR_FLOW_NAME" } } ] } } If you don’t know the exact event name, run: - prefect events stream --format text - Cancel one run and copy the cancellation event name. If you still want to try a hook anyway - It can work only if the process isn’t terminated too aggressively. For example, a very short termination grace period in your worker/infrastructure will prevent hooks from running. - Minimal example (note the parameter is on_failure, not on_failed in Prefect 3): from prefect import flow from prefect.blocks.notifications import SlackWebhook def notify_cancelled(state, **_): slack = SlackWebhook.load("alerts/slack") slack.notify("Flow was cancelled.") @flow(on_cancellation=[notify_cancelled], on_failure=[notify_cancelled]) def my_flow(): ... Caveats: - Even with this in place, external cancellations (UI “Cancel”, infra preemption, etc.) may kill the process before notify_cancelled runs. That’s why the Automation approach is recommended. If you share: - the flow name (or ID), and - the Slack Webhook block name, I can give you a ready-to-paste Automation JSON for “N cancellations in M minutes” or “any cancellation”. Links you may want: - Event triggers: https://docs-3.prefect.io/v3/concepts/event-triggers - Slack via Webhook blocks: https://docs-3.prefect.io/integrations/prefect-slack/