@Marvin I'm configuring a Prefect automation to send a Discord alert when a flow run stays in AwaitingConcurrencySlot for more than 2 hours. My goal is one notification per flow, not per run. I'm using a Proactive trigger with for_each. I tried: - "for_each": ["prefect.resource.name"] - "for_each": ["relatedflowprefect.resource.name"] My current trigger: { "type": "event", "match": { "prefect.resource.id": "prefect.flow-run.*" }, "match_related": { "prefect.resource.role": "deployment" }, "after": ["prefect.flow-run.AwaitingConcurrencySlot"], "expect": ["prefect.flow-run.Running", "prefect.flow-run.Cancelled", "prefect.flow-run.Crashed", "prefect.flow-run.Failed", "prefect.flow-run.Completed"], "for_each": ["relatedflowprefect.resource.name"], "posture": "Proactive", "threshold": 1, "within": 7200 } Is it possible to group proactive automation firings by flow (or deployment) so only one alert fires per flow regardless of how many runs are stuck? If so, what is the correct for_each configuration?

✅ thought for 136.9 seconds

@Marvin I'm setting up a Prefect 3 automation to notify Discord when a flow run is stuck in AwaitingConcurrencySlot for over 2 hours. A few questions: 1. Trigger config — is this correct? trigger: type: event match: prefect.resource.id: "prefect.flow-run.*" match_related: prefect.resource.role: deployment posture: Proactive after: - prefect.flow-run.AwaitingConcurrencySlot expect: - prefect.flow-run.Running - prefect.flow-run.Cancelled - prefect.flow-run.Crashed - prefect.flow-run.Failed - prefect.flow-run.Completed for_each: - "relateddeploymentprefect.resource.id" threshold: 1 within: 7200 2. call-webhook action with a Webhook block pointing to a Discord webhook URL — is this the right approach? There's no native Discord notification block available (prefect_discord is not installed). The block has Content-Type: application/json in its headers. 3. Are {{ flow_run.name }} template variables available in the call-webhook payload? payload: '{"content": "⚠️ {{ flow_run.name }} has been awaiting a concurrency slot for over 2 hours."}' 4. Block resolution: We resolve block_document_id at deploy time by looking up the block by slug via client.read_block_document_by_name(name, block_type). Is there a better native way to reference a block by name in automation YAML?

✅ thought for 165.1 seconds

@Marvin Here's the complete setup we ended up with. Goal: alert Discord when a flow run is stuck in AwaitingConcurrencySlot for over 2 hours. --- blocks/blocks.yaml blocks: - type: discord-webhook name: discord-concurrency-alert url_env: DISCORD_WEBHOOK_URL_DATAPLEX # https://discord.com/api/webhooks/{id}/{token} blocks/deploy_blocks.py import asyncio import os from pathlib import Path import yaml from prefect.blocks.notifications import DiscordWebhook async def deploy_block(block_spec: dict) -> None: block_type = block_spec["type"] name = block_spec["name"] if block_type == "discord-webhook": url_env = block_spec.get("url_env") url = os.environ.get(url_env) if url_env else block_spec.get("url") if not url: print(f"Skipping block {name}: env var {url_env} not set") return parts = url.rstrip("/").split("/") webhook_id = parts[-2] webhook_token = parts[-1] block = DiscordWebhook(webhook_id=webhook_id, webhook_token=webhook_token) await block.save(name=name, overwrite=True) print(f"Block {block_type}/{name} created/updated") async def main() -> None: blocks_file = Path(file).parent / "blocks.yaml" with blocks_file.open("r") as f: data = yaml.safe_load(f) for block_spec in data.get("blocks", []): await deploy_block(block_spec) if name == "__main__": asyncio.run(main()) --- automations/automations.yaml (relevant part) actions: - type: send-notification block_document_id: "$block:discord-webhook/discord-concurrency-alert" subject: "Flow travado em AwaitingConcurrencySlot" body: "⚠️ Flow travado em AwaitingConcurrencySlot - <@&1056928259700445245>\nO Flow {{ flow_run.name }} possui execuções aguardando slot de concorrência há mais de 2 horas." automations/deploy_automations.py import argparse import asyncio from pathlib import Path from typing import Any import yaml from prefect.automations import Automation def _collect_block_refs(obj: Any, refs: set) -> None: if isinstance(obj, dict): for v in obj.values(): _collect_block_refs(v, refs) elif isinstance(obj, list): for item in obj: _collect_block_refs(item, refs) elif isinstance(obj, str) and obj.startswith("$block:"): refs.add(obj) def _replace_block_refs(obj: Any, ref_to_uuid: dict) -> Any: if isinstance(obj, dict): return {k: _replace_block_refs(v, ref_to_uuid) for k, v in obj.items()} elif isinstance(obj, list): return [_replace_block_refs(item, ref_to_uuid) for item in obj] elif isinstance(obj, str) and obj in ref_to_uuid: return ref_to_uuid[obj] return obj async def resolve_block_refs(spec: dict) -> dict: from prefect.client.orchestration import get_client refs: set = set() _collect_block_refs(spec, refs) if not refs: return spec ref_to_uuid: dict = {} async with get_client() as client: for ref in refs: _, path = ref.split(":", 1) block_type, block_name = path.split("/", 1) block_doc = await client.read_block_document_by_name(block_name, block_type) ref_to_uuid[ref] = str(block_doc.id) return _replace_block_refs(spec, ref_to_uuid) async def upsert_automation(auto_spec: dict) -> None: auto_spec = await resolve_block_refs(auto_spec) automation = Automation(**auto_spec) try: existing = await automation.read(name=automation.name) except ValueError: existing = None if existing: automation.id = existing.id await automation.update() else: await automation.create() async def main(yaml_file: str) -> None: with Path.open(yaml_file, "r") as f: data = yaml.safe_load(f) for auto_spec in data.get("automations", []): await upsert_automation(auto_spec) if name == "__main__": parser = argparse.ArgumentParser() parser.add_argument("yaml_file") args = parser.parse_args() asyncio.run(main(args.yaml_file)) --- Questions: 1. You previously confirmed call-webhook + Webhook block for Discord. We found DiscordWebhook in prefect.blocks.notifications (slug discord-webhook) and switched to send-notification. Is send-notification + DiscordWebhook valid, or should we go back to call-webhook + Webhook? 2. Is client.read_block_document_by_name(name, block_type_slug) the correct/stable API to look up a block document ID by name at deploy time?

Your message was too long, here's your message at the allowed limit: Here's the complete setup we ended up with Goal alert Discord when a flow run is stuck in AwaitingConcurrencySlot for over 2 hours --- blocks/blocks yaml blocks - type discord-webhook name discord-concurrency-alert url_env DISCORD_WEBHOOK_URL_DATAPLEX # <https //discord com/api/webhooks/{id}/{token}> blocks/deploy_blocks py import asyncio import os from pathlib import Path import yaml from prefect blocks notifications import DiscordWebhook async def deploy_block(block_spec dict) -> None block_type = block_spec[ type ] name = block_spec[ name ] if block_type == discord-webhook url_env = block_spec get( url_env ) url = os environ get(url_env) if url_env else block_spec get( url ) if not url print(f Skipping block {name} env var {url_env} not set ) return parts = url rstrip( / ) split( / ) webhook_id = parts[-2] webhook_token = parts[-1] block = DiscordWebhook(webhook_id=webhook_id webhook_token=webhook_token) await block save(name=name overwrite=True) print(f Block {block_type}/{name} created/updated ) async def main() -> None blocks_file = Path(file) parent / blocks yaml with blocks_file open( r ) as f data = yaml safe_load(f) for block_spec in data get( blocks []) await deploy_block(block_spec) if name == main asyncio run(main()) --- automations/automations yaml (relevant part) actions - type send-notification block_document_id $block discord-webhook/discord-concurrency-alert subject Flow travado em AwaitingConcurrencySlot body warning Flow travado em AwaitingConcurrencySlot - <@&1056928259700445245>\nO Flow {{ flow_run name }} possui execuções aguardando slot de concorrência há mais de 2 horas automations/deploy_automations py import argparse import asyncio from pathlib import Path from typing import Any import yaml from prefect automations import Automation def _collect_block_refs(obj Any refs set) -> None if isinstance(obj dict) for v in obj values() _collect_block_refs(v refs) elif isinstance(obj list) for item in obj _collect_block_refs(item refs) elif isinstance(obj str) and obj startswith( $block ) refs add(obj) def _replace_block_refs(obj Any ref_to_uuid dict) -> Any if isinstance(obj dict) return {k _replace_block_refs(v ref_to_uuid) for k v in obj items()} elif isinstance(obj list) return [_replace_block_refs(item ref_to_uuid) for item in obj] elif isinstance(obj str) and obj in ref_to_uuid return ref_to_uuid[obj] return obj async def resolve_block_refs(spec dict) -> dict from prefect client orchestration import get_client refs set = set() _collect_block_refs(spec refs) if not refs return spec ref_to_uuid dict = {} async with get_client() as client for ref in refs _ path = ref split( 1) block_type block_name = path split( / 1) block_doc = await client read_block_document_by_name(block_name block_type) ref_to_uuid[ref] = str(block_doc id) return _replace_block_refs(spec ref_to_uuid) async def upsert_automation(auto_spec dict) -> None auto_spec = await resolve_block_refs(auto_spec) automation = Automation(**auto_spec) try existing = await automation read(name=automation name) except ValueError existing = None if existing automation id = existing id await automation update() else await automation create() async def main(yaml_file str) -> None with Path open(yaml_file r ) as f data = yaml safe_load(f) for auto_spec in data get( automations []) await upsert_automation(auto_spec) if name == main parser = argparse ArgumentParser() parser add_argument( yaml_file ) args = parser parse_args() asyncio run(main(args yaml_file)) --- Questions 1 You previously confirmed call-webhook + Webhook block for Discord We found DiscordWebhook in prefect blocks notifications (slug discord-webhook) and switched to send-notification Is send-notification + DiscordWebhook valid or should we go back to call-webhook + Webhook? 2 Is

@Marvin Here's the complete setup we ended up with. Goal: alert Discord when a flow run is stuck in AwaitingConcurrencySlot for over 2 hours. --- blocks/blocks.yaml blocks: - type: discord-webhook name: discord-concurrency-alert url_env: DISCORD_WEBHOOK_URL_DATAPLEX # https://discord.com/api/webhooks/{id}/{token} blocks/deploy_blocks.py import asyncio import os from pathlib import Path import yaml from prefect.blocks.notifications import DiscordWebhook async def deploy_block(block_spec: dict) -> None: block_type = block_spec["type"] name = block_spec["name"] if block_type == "discord-webhook": url_env = block_spec.get("url_env") url = os.environ.get(url_env) if url_env else block_spec.get("url") if not url: print(f"Skipping block {name}: env var {url_env} not set") return parts = url.rstrip("/").split("/") webhook_id = parts[-2] webhook_token = parts[-1] block = DiscordWebhook(webhook_id=webhook_id, webhook_token=webhook_token) await block.save(name=name, overwrite=True) print(f"Block {block_type}/{name} created/updated") async def main() -> None: blocks_file = Path(file).parent / "blocks.yaml" with blocks_file.open("r") as f: data = yaml.safe_load(f) for block_spec in data.get("blocks", []): await deploy_block(block_spec) if name == "__main__": asyncio.run(main()) --- automations/automations.yaml (relevant part) actions: - type: send-notification block_document_id: "$block:discord-webhook/discord-concurrency-alert" subject: "Flow travado em AwaitingConcurrencySlot" body: "⚠️ Flow travado em AwaitingConcurrencySlot - <@&1056928259700445245>\nO Flow {{ flow_run.name }} possui execuções aguardando slot de concorrência há mais de 2 horas." automations/deploy_automations.py import argparse import asyncio from pathlib import Path from typing import Any import yaml from prefect.automations import Automation def _collect_block_refs(obj: Any, refs: set) -> None: if isinstance(obj, dict): for v in obj.values(): _collect_block_refs(v, refs) elif isinstance(obj, list): for item in obj: _collect_block_refs(item, refs) elif isinstance(obj, str) and obj.startswith("$block:"): refs.add(obj) def _replace_block_refs(obj: Any, ref_to_uuid: dict) -> Any: if isinstance(obj, dict): return {k: _replace_block_refs(v, ref_to_uuid) for k, v in obj.items()} elif isinstance(obj, list): return [_replace_block_refs(item, ref_to_uuid) for item in obj] elif isinstance(obj, str) and obj in ref_to_uuid: return ref_to_uuid[obj] return obj async def resolve_block_refs(spec: dict) -> dict: from prefect.client.orchestration import get_client refs: set = set() _collect_block_refs(spec, refs) if not refs: return spec ref_to_uuid: dict = {} async with get_client() as client: for ref in refs: _, path = ref.split(":", 1) block_type, block_name = path.split("/", 1) block_doc = await client.read_block_document_by_name(block_name, block_type) ref_to_uuid[ref] = str(block_doc.id) return _replace_block_refs(spec, ref_to_uuid) async def upsert_automation(auto_spec: dict) -> None: auto_spec = await resolve_block_refs(auto_spec) automation = Automation(**auto_spec) try: existing = await automation.read(name=automation.name) except ValueError: existing = None if existing: automation.id = existing.id await automation.update() else: await automation.create() async def main(yaml_file: str) -> None: with Path.open(yaml_file, "r") as f: data = yaml.safe_load(f) for auto_spec in data.get("automations", []): await upsert_automation(auto_spec) if name == "__main__": parser = argparse.ArgumentParser() parser.add_argument("yaml_file") args = parser.parse_args() asyncio.run(main(args.yaml_file)) --- Questions: 1. Is send-notification + DiscordWebhook valid, or should we go back to call-webhook + Webhook? 2. Is client.read_block_document_by_name(name, block_type_slug) the correct/stable API to look up a block document ID by name at deploy time?

✅ thought for 112.8 seconds

@Marvin One more question on deploy_automations.py. We're using three functions to resolve $block: references: def _collect_block_refs(obj: Any, refs: set) -> None: """Recursively traverses the entire spec dict to find all $block: strings.""" ... def _replace_block_refs(obj: Any, ref_to_uuid: dict) -> Any: """Recursively replaces $block: strings with resolved UUIDs.""" ... async def resolve_block_refs(spec: dict) -> dict: """Uses Prefect client to resolve each found ref to a UUID.""" ... Since block_document_id only ever appears inside actions[], we could simplify to a single targeted function: async def resolve_block_refs(spec: dict) -> dict: from prefect.client.orchestration import get_client refs = { action["block_document_id"] for action in spec.get("actions", []) if isinstance(action.get("block_document_id"), str) and action["block_document_id"].startswith("$block:") } if not refs: return spec ref_to_uuid = {} async with get_client() as client: for ref in refs: _, path = ref.split(":", 1) block_type, block_name = path.split("/", 1) block_doc = await client.read_block_document_by_name( block_name, block_type, include_secrets=False ) ref_to_uuid[ref] = str(block_doc.id) for action in spec.get("actions", []): if action.get("block_document_id") in ref_to_uuid: action["block_document_id"] = ref_to_uuid[action["block_document_id"]] return spec Is block_document_id only ever in actions[] in an automation spec, or can it appear elsewhere (e.g. nested in trigger conditions or other fields)? If only in actions, the simpler version is sufficient. If it can appear anywhere in the spec tree, the recursive traversal is justified.

✅ thought for 59.5 seconds