Great question. In Prefect 3.x you can kick off many deployment runs from a parent flow, wait for them all to finish, then aggregate their return values. Here are two concise patterns you can use. Example child flow (already deployed as "child-flow/process-batch"); it just returns something simple so you can see the aggregation:

from prefect import flow

@flow
def child_flow(batch_id: str) -> int:
    # Do work for this batch and return a serializable result
    return len(batch_id)

Pattern A: Async parent flow using asyncio.gather (concurrent dispatch + wait for all + aggregate results)

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

@flow
async def orchestrate_batches(batch_ids: list[str]) -> dict:
    # Build coroutines to dispatch all runs concurrently
    coros = [
        run_deployment(
            name="child-flow/process-batch",      # "<flow-name>/<deployment-name>"
            parameters={"batch_id": b},
            flow_run_name=f"process-{b}",
            timeout=900,                          # per-run wait cap; use None to wait indefinitely
            # work_queue_name="my-work-pool",     # optionally target a specific pool/queue
            # job_variables={"env": {"FOO": "BAR"}}
        )
        for b in batch_ids
    ]

    # Dispatch and wait for all runs to reach their terminal state (or timeout)
    flow_runs = await asyncio.gather(*coros, return_exceptions=True)

    successes, failures = [], []
    for b, fr in zip(batch_ids, flow_runs):
        if isinstance(fr, Exception):
            failures.append({"batch_id": b, "error": str(fr)})
            continue

        # Each item is a FlowRun; check final state and extract the child’s return value
        if fr.state and fr.state.is_final():
            if fr.state.type == "COMPLETED":
                try:
                    value = await fr.state.aresult()   # child flow's return value
                    successes.append({"batch_id": b, "run_id": str(fr.id), "value": value})
                except Exception as e:
                    failures.append({"batch_id": b, "run_id": str(fr.id), "error": f"result fetch failed: {e}"})
            else:
                failures.append({"batch_id": b, "run_id": str(fr.id), "state": fr.state.type, "error": fr.state.message})
        else:
            failures.append({"batch_id": b, "run_id": str(fr.id), "state": getattr(fr.state, "type", "UNKNOWN")})

    # Example aggregation: sum all child return values
    total_value = sum(x["value"] for x in successes if isinstance(x.get("value"), (int, float)))
    return {
        "submitted": len(batch_ids),
        "succeeded": len(successes),
        "failed": len(failures),
        "total_value": total_value,
        "successes": successes,
        "failures": failures,
    }

if __name__ == "__main__":
    # Example invocation
    asyncio.run(orchestrate_batches([f"batch-{i}" for i in range(10)]))

Pattern B: Bounded concurrency (avoid flooding your workers)

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

@flow
async def orchestrate_batches_bounded(batch_ids: list[str], limit: int = 10):
    sem = asyncio.Semaphore(limit)

    async def submit_and_wait(b: str):
        async with sem:
            fr = await run_deployment(
                name="child-flow/process-batch",
                parameters={"batch_id": b},
                flow_run_name=f"process-{b}",
                timeout=900,
            )
            # Pull the child flow's return value
            if fr.state and fr.state.type == "COMPLETED":
                return b, str(fr.id), await fr.state.aresult()
            return b, str(fr.id), None

    results = await asyncio.gather(*(submit_and_wait(b) for b in batch_ids), return_exceptions=False)
    return results

Notes and best practices

- run_deployment works in both sync and async contexts; in an async flow you should await it (as shown above). It returns a FlowRun; you’ll typically check fr.state and use fr.state.aresult()/state.result() to get the child’s return value. - For batching “across machines,” make sure your child deployment targets a work pool with multiple workers across those machines; Prefect will distribute the triggered runs to any available worker in the pool. - Ensure your child flow returns serializable results. For large objects, configure result storage on the deployment so the parent can retrieve results reliably. - Use timeout to avoid waiting forever; a value of None waits until completion, 0 returns immediately (fire-and-forget). - Consider global concurrency limits (GCL) if you need to cap downstream load across your workers. References - How to run deployments - Source: run_deployment - Async dispatch mechanism