Great question. You don’t have to convert your flow into a task to run it concurrently for different inputs. In Prefect 3.x, there are a few good patterns: Option A: Trigger multiple flow runs of a deployment (recommended) - Create a deployment for your flow, then programmatically trigger many runs with different parameters. Wrap

run_deployment

in a small task so you can

submit

them in parallel from a parent flow.

from prefect import flow, task
from prefect.deployments.flow_runs import run_deployment

@task
def launch(dep_name, params):
    # as_subflow=True shows runs under the parent; set to False for top-level runs
    # timeout=0 returns immediately so this task doesn't block
    return run_deployment(
        name=dep_name,
        parameters=params,
        as_subflow=True,
        timeout=0,
    )

@flow
def orchestrate(param_list):
    futures = [launch.submit("my-flow/my-deployment", {"input": p}) for p in param_list]
    # wait for child runs to start/finish as needed:
    _ = [f.result() for f in futures]

- This will enqueue separate flow runs that workers can pick up concurrently. - Docs: Run deployments, prefect.yaml Option B: Keep it as a flow and call it as a subflow via a task wrapper - If you don’t want to use deployments, you can still run subflows in parallel by wrapping the flow call in a task so you can

submit

it.

from prefect import flow, task

@flow
def child_flow(x):
    # your existing flow body (it can still submit its 4 tasks in parallel)
    ...

@task
def run_child(x):
    # calling a flow from a task creates a subflow run
    return child_flow(x)

@flow
def parent(xs):
    futures = [run_child.submit(x) for x in xs]
    _ = [f.result() for f in futures]

Option C: Enqueue multiple runs from outside (CLI or automation) - You can trigger many runs with different parameters via the CLI (e.g., a script or CI):

prefect deployment run my-flow/my-deployment -p input=1
prefect deployment run my-flow/my-deployment -p input=2
prefect deployment run my-flow/my-deployment -p input=3

Notes and tips - Use

as_subflow=True

if you want the child runs to appear under the parent in the UI; set

as_subflow=False

for independent top-level runs. - Ensure your work pool has enough workers to process runs concurrently. - If you need to control max parallelism, use work pool or tag-based concurrency limits. If you can share whether you prefer running everything in one process vs. dispatching separate runs to workers (and what infrastructure you’re on), I can tailor the example further.