Great question. In Prefect 3.x, submit your tasks with

.submit()

and use a task runner configured with

max_workers=3

so exactly three run concurrently. Example with threads (good for I/O-bound work):

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def work(x: int) -> int:
    import time; time.sleep(1)
    return x * 2

@flow(task_runner=ThreadPoolTaskRunner(max_workers=3))
def run_three_in_parallel():
    f1 = work.submit(1)
    f2 = work.submit(2)
    f3 = work.submit(3)
    return [f.result() for f in (f1, f2, f3)]

if __name__ == "__main__":
    print(run_three_in_parallel())

If your tasks are CPU-bound, switch to processes:

from prefect.task_runners import ProcessPoolTaskRunner

@flow(task_runner=ProcessPoolTaskRunner(max_workers=3))
def run_three_cpu_bound():
    ...

-

.submit()

schedules tasks concurrently on the flow’s task runner -

max_workers=3

ensures only three run at once; extra submissions will queue Docs: - Run work concurrently - Task runners (ThreadPool vs ProcessPool) Would you like this for I/O-bound or CPU-bound work? I can tailor the example further (and include mapping or distributed runners like Dask/Ray if needed).