Yes, you can set this up using Prefect's automation features. Here's how to create a notification that triggers only for flow runs in a specific work queue: 1. First, set up your Slack webhook block:

python
from prefect.blocks.notifications import SlackWebhook

slack_webhook_block = SlackWebhook.load("YOUR-BLOCK-NAME")
# or create a new one
slack_webhook_block = SlackWebhook(
    url="YOUR-WEBHOOK-URL"
)
slack_webhook_block.save("YOUR-BLOCK-NAME")

2. Create an automation using the Prefect UI or via code:

python
from prefect.server.schemas.actions import (
    BlockTypeReference,
    NotificationAction,
    ResourceSpec,
    StateType,
)
from prefect.client.schemas.objects import AutomationMatch

from prefect.client import get_client

async def create_automation():
    client = get_client()
    
    await client.create_automation(
        name="Slack Alert for Work Queue Failures",
        description="Send Slack notification when flows fail in specific work queue",
        match=AutomationMatch(
            # Specify your work queue name here
            work_queue_name={"equals": "YOUR-WORK-QUEUE-NAME"},
            state_type={"equals": StateType.FAILED}
        ),
        actions=[
            NotificationAction(
                block_document=BlockTypeReference(
                    block_type_slug="slack-webhook",
                    block_document_name="YOUR-BLOCK-NAME"
                ),
                message="Flow {{ flow_run.name }} failed in work queue {{ flow_run.work_queue_name }}. \nFlow run URL: {{ flow_run.url }}"
            )
        ]
    )

The key parts here are: - Using

work_queue_name

in the match criteria to filter for specific work queue - Setting

state_type

to

FAILED

to only trigger on failures - Customizing the message template with relevant flow run information You can customize the message template further using available variables like: -

{{ flow_run.name }}

-

{{ flow_run.flow_name }}

-

{{ flow_run.work_queue_name }}

-

{{ flow_run.url }}

-

{{ flow_run.parameters }}

The notification will only trigger when a flow run that matches both criteria (specific work queue AND failed state) occurs. For more complex filtering, you can also use additional match criteria like: -

deployment_name

-

flow_name

-

tags

Remember to replace

YOUR-BLOCK-NAME

and

YOUR-WORK-QUEUE-NAME

with your actual values. Links: - Prefect Automations Documentation - Prefect Notifications Documentation - Slack Webhook Block Documentation