Short answer: there isn’t a built-in “work pool is empty” event in Prefect 3.x Automations. But you can accomplish this with a small “checker” flow + an Automation that calls it, or by running the checker on a short schedule. Two ways to do it 1) Event-driven (recommended to minimize polling) - Create a tiny “checker” flow that: - lists all work queues in a given work pool - counts flow runs in SCHEDULED or LATE for those queues - if the count is zero (optionally still zero after a short guard period), it runs your target deployment - Create an Automation that triggers on flow-run terminal events (e.g., Completed/Failed) in that work pool’s queues and runs the “checker” deployment as the action. - The checker exits immediately unless the pool is truly idle; when idle, it kicks off your downstream deployment. 2) Simple polling - Deploy the same “checker” flow and give it a cron/interval schedule (e.g., every 1–2 minutes). - When it sees no SCHEDULED or LATE runs, it triggers your downstream deployment. Example “checker” flow (Prefect 3.x)

import asyncio
from prefect import flow, get_client
from prefect.deployments import run_deployment
from prefect.client.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterState,
    FlowRunFilterStateType,
    FlowRunFilterWorkQueueName,
)

@flow
async def trigger_when_pool_idle(
    work_pool: str,
    deployment_to_run: str,
    min_idle_seconds: int = 0,   # optional guard to avoid races
):
    async with get_client() as client:
        # All queues in the work pool
        queues = await client.read_work_queues(work_pool_name=work_pool)
        queue_names = [q.name for q in queues]
        if not queue_names:
            # No queues in this pool; treat as idle or just exit
            return

        # Look for any SCHEDULED or LATE runs in these queues
        fr_filter = FlowRunFilter(
            state=FlowRunFilterState(
                type=FlowRunFilterStateType(any_=["SCHEDULED", "LATE"])
            ),
            work_queue_name=FlowRunFilterWorkQueueName(any_=queue_names),
        )

        # If you only care about runs due now or late, you can further filter by expected start time.
        # from prefect.client.schemas.filters import FlowRunFilterExpectedStartTime
        # import pendulum
        # fr_filter.expected_start_time = FlowRunFilterExpectedStartTime(
        #     before_=pendulum.now("UTC")
        # )

        runs = await client.read_flow_runs(flow_run_filter=fr_filter, limit=1)
        if runs:
            return  # not idle

        # Optional guard: re-check after a short delay to avoid a race where a run is just scheduled
        if min_idle_seconds:
            await asyncio.sleep(min_idle_seconds)
            runs2 = await client.read_flow_runs(flow_run_filter=fr_filter, limit=1)
            if runs2:
                return

        # Pool appears idle — trigger your deployment
        await run_deployment(name=deployment_to_run)

Wiring it up with an Automation (Cloud UI) - Trigger: - Event: flow-run terminal states (e.g., Completed, Failed) for the deployments/queues that target this work pool - This ensures we check right after activity quiets down. - Action: - Run Deployment: select the deployment for

trigger_when_pool_idle

- De-duplication/Posture: - Reactive posture is fine; you can add a small deduplication window so only one checker runs if multiple runs finish at once. Notes and trade-offs - There’s no native “work pool empty” metric/event today. This pattern is a lightweight, reliable workaround. - Consider the exact definition of “idle” for you: - If you want “no late and no scheduled before now,” add the expected_start_time filter (see commented lines). - If scheduled far in the future should still count as “idle,” leave the filter as-is (or switch to only checking LATE). - Race conditions: a new run might be scheduled right after your check — the

min_idle_seconds

guard reduces false positives.

- If you prefer CLI for quick testing, this command shows pending runs per queue: -

prefect work-queue read-runs <queue-name> -p <pool-name>

(see help:

prefect work-queue read-runs --help

) Relevant docs - Automations and triggers: Event triggers (Prefect 3.x) - Work queue CLI reference: run

prefect work-queue --help

and

prefect work-queue read-runs --help

- Filters (for programmatic queries): Automations/filters overview (and the Python filter classes are in

prefect.client.schemas.filters

) A couple of quick questions to tailor this: - Do you want to treat “scheduled in the future” as idle, or only when there are truly zero SCHEDULED/LATE runs due now? - Should this watch all queues in the pool or just specific queues?