Yes — kind of, but the aggregation happens at the automation trigger level, not in the notification action itself. If you're on Prefect 3.x, the built-in way to avoid "1 failure = 1 notification" is to configure the automation trigger with: -

threshold

-

within

That lets you say: only fire this automation if N matching events happen within a time window. So your example would be: - event:

prefect.flow-run.Failed

-

threshold: 5

-

within: 60

seconds That means 5 failures within 1 minute -> 1 automation run -> 1 notification. Important caveat: the notification itself is still a single action based on the trigger firing. Prefect does not currently batch multiple failed flow details into one rich notification payload automatically. Example shape:

yaml
name: Batched flow failure alert
trigger:
  type: event
  posture: Reactive
  expect:
    - prefect.flow-run.Failed
  threshold: 5
  within: 60
actions:
  - type: send-notification
    block_document_id: <your-notification-block-id>
    subject: "Multiple flow failures detected"
    body: "At least 5 flow runs failed within the last minute."

A few notes: - If you omit

for_each

, this counts failures globally across all matching flow runs. - If you add

for_each

, you can scope the batching. For example, one alert per flow or deployment. Example per deployment:

yaml
trigger:
  type: event
  posture: Reactive
  expect:
    - prefect.flow-run.Failed
  threshold: 5
  within: 60
  for_each:
    - prefect.resource.id

In practice, if your goal is: “send one generic alert when many failures happen close together” → this works well. If your goal is: “send one message containing the list of all 5 failed flows” → that is not something Prefect’s built-in notification action currently aggregates for you. In that case, the usual workaround is to send to a custom webhook/service that does its own buffering/aggregation. Relevant docs: - Automations - Debouncing events If you want, I can help you write the exact automation for: - Slack / Teams / email - per deployment vs global aggregation - UI-based setup vs Python/YAML config

Short answer: not directly. When a flow failure automation fires, the event contains metadata about the failed flow run, but not the run logs themselves. So if your external service needs logs, the usual pattern is: 1. Automation triggers on

prefect.flow-run.Failed

2. Automation sends the

flow_run_id

(and maybe flow/deployment info) to your service 3. Your service calls the Prefect API to fetch logs for that flow run So yes, this is possible — but the logs must be fetched separately. What the automation can access/use: - failed flow run ID - flow run name - state message - timestamps - related flow / deployment metadata What it cannot directly access: - the full emitted logs from that flow run as part of the event payload A good approach is to use a webhook automation payload like this:

json
{
  "flow_run_id": "{{ event.resource['prefect.resource.id'].split('.')[-1] }}",
  "flow_run_name": "{{ flow_run.name }}",
  "flow_name": "{{ flow.name }}",
  "deployment_name": "{{ deployment.name if deployment else '' }}",
  "state_message": "{{ event.resource['prefect.state-message'] }}"
}

Then your service can fetch logs using the Prefect API. If you're doing this in Python, your service would typically use the orchestration client and read logs filtered by

flow_run_id

. High-level example:

python
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import LogFilter

async def get_logs(flow_run_id: str):
    async with get_client() as client:
        logs = await client.read_logs(
            log_filter=LogFilter(
                flow_run_id={"any_": [UUID(flow_run_id)]}
            ),
            limit=200,
        )
        return logs

A couple of practical notes: - If you only want the tail of the logs, fetch the last N entries. - If you want only errors, filter by log level in your service. - If the failed flow has many task logs, you may want pagination. Relevant docs: - Automations - Prefect orchestration client If you want, I can help you with either: - the exact automation webhook payload for your use case, or - a complete small service example that receives the webhook and fetches the failed flow's logs.