Short answer: you can do this natively. The idiomatic 3.x pattern is to call

run_deployment()

concurrently from an async flow (e.g., with

asyncio.gather

) or to submit a small task that wraps

run_deployment()

and fan out with

.submit(...)

. There isn’t a special “fan-out deployments” primitive—you compose it with normal Python concurrency or Prefect tasks. Key knobs on

run_deployment

(verified): - `timeout`: None = wait until completion, 0 = return immediately (fire-and-forget), N = wait up to N seconds - `as_subflow`: True by default; shows parent/child linkage in the UI - `idempotency_key`: prevent duplicate runs if your orchestrator retries - Returns a

FlowRun

object (id, state, etc.) Pattern A: async flow + asyncio.gather (simple and fast)

import asyncio
from prefect import flow
from prefect.deployments import run_deployment

@flow
async def orchestrate_fanout(names: list[str]):
    # fire-and-forget; return FlowRun handles immediately
    runs = await asyncio.gather(*[
        run_deployment(name, timeout=0) for name in names
    ])
    return runs

Wait for completion (fan-in) by blocking at trigger time:

@flow
async def orchestrate_and_wait(names: list[str]):
    runs = await asyncio.gather(*[
        run_deployment(name, timeout=None) for name in names
    ])
    return runs

Or trigger fast, then poll until done:

import asyncio
from prefect import flow
from prefect.deployments import run_deployment
from prefect.client.orchestration import get_client

@flow
async def fanout_then_poll(names: list[str]):
    runs = await asyncio.gather(*[
        run_deployment(name, timeout=0) for name in names
    ])
    ids = {r.id for r in runs}

    async with get_client() as client:
        while ids:
            await asyncio.sleep(5)
            for run_id in list(ids):
                fr = await client.read_flow_run(run_id)
                if fr.state and fr.state.is_final():
                    ids.remove(run_id)
    return runs

Pattern B: wrap

run_deployment

in a task and submit for parallelism (adds retries/observability per trigger)

from prefect import flow, task

@task
async def trigger(name: str, params: dict | None = None):
    from prefect.deployments import run_deployment
    return await run_deployment(name, parameters=params, timeout=0)

@flow
def orchestrate_fanout(names: list[str]):
    futures = [trigger.submit(n) for n in names]
    return [f.result() for f in futures]

Best practices: - Use

timeout=0

to fan out quickly; decide whether to wait inline (timeout=None) or poll later. - Set

as_subflow=True

(default) to visualize parent/child relationships; set False if you want independent runs. - Add

idempotency_key

if your orchestrator might retry and you want to avoid duplicate runs. - If you need to cap parallelism across your estate, consider using tags on triggered runs and Prefect Cloud’s concurrency limits. If you share whether you want to wait for completion (fan-in) or just kick off runs, and roughly how many deployments you’ll trigger at once, I can tailor the exact pattern and error-handling/retry setup.