You’re right — in Prefect 3.x you can’t set concurrency on a flow via the decorator, and tag-based concurrency only applies to tasks. Here are the supported ways to do flow-level concurrency in 3.x: 1) Per-deployment concurrency (recommended for flow-level limits) - Set a cap on how many runs of a specific deployment can run at once. Example:

from prefect import flow

@flow
def my_flow():
    ...

if __name__ == "__main__":
    my_flow.deploy(
        name="my-deployment",
        work_pool_name="my-pool",
        concurrency_limit=5,  # max 5 concurrent runs for this deployment
    )

Notes: -

concurrency_limit

can be an integer. Advanced behavior (e.g., queue vs. cancel when full) is available via a ConcurrencyLimitConfig, but the simple int covers most cases. - Docs: Deployments concept 2) Work pool concurrency (pool-wide cap across all deployments in the pool) - Limits the total number of concurrent flow runs a work pool can execute. CLI:

prefect work-pool set-concurrency-limit "my-pool" 10

- Docs: Work pool CLI 3) Runner-level concurrency (when using serve in dev or lightweight runners) -

serve()

can limit how many runs the runner executes concurrently across whatever it’s serving. Example:

from prefect import flow, serve

@flow
def my_flow():
    ...

if __name__ == "__main__":
    serve(
        my_flow.to_deployment(name="my-deployment", work_pool_name="my-pool"),
        limit=3,  # runner processes at most 3 runs concurrently
    )

4) (Legacy) Work queue concurrency - If you’re still using work queues, you can cap them too:

prefect work-queue set-concurrency-limit "my-queue" 5 --pool "my-pool"

Important clarifications - Tag-based concurrency is task-only. It does not limit flows. See: Tag-based concurrency limits - Global concurrency limits (GCL) are also task-oriented and not a substitute for limiting total flow runs: Global concurrency limits If you share which setup you’re running (deployments via code/CLI, work pools you use, whether you run with serve), I can suggest the most straightforward snippet/commands for your case.

Great questions — and you’re running into an important distinction in Prefect 3.x between inline subflows vs. deployment-triggered runs. 1) Spawning multiple subflows - Yes,

asyncio.gather()

is perfectly fine for running subflows concurrently in an async parent flow. It’s the simplest “inline” approach and is supported. - If you want the child work to be scheduled on workers (and benefit from deployment-level features like per-deployment concurrency), trigger the child as a deployment with

run_deployment(...)

instead of calling the flow function directly. 2) Limiting to 2 child flows at a time Why your current setup ignored the deployment’s concurrency_limit: - Calling

child_flow(item)

inside the parent flow creates inline subflows within the same process. Those do not count against a deployment’s

concurrency_limit

. Deployment concurrency only applies to runs created for that deployment (e.g., via `prefect deploy`/schedules/API/`run_deployment`). You have three good options depending on the scope you want: Option A — Per-parent-run limit (local only) - Keep child flows inline and use an asyncio.Semaphore to cap concurrency inside the parent flow. This affects only that single parent run.

import asyncio
from prefect import flow

@flow
async def child_flow(item):
    # Expensive work here
    await do_expensive_work(item)
    return f"done-{item}"

@flow
async def parent_flow(items):
    sem = asyncio.Semaphore(2)  # allow only 2 child flows at a time

    async def run_one(i):
        async with sem:
            return await child_flow(i)

    results = await asyncio.gather(*(run_one(i) for i in items))
    return results

Option B — Use deployments for children + deployment concurrency - Deploy the child flow with

concurrency_limit=2

. - From the parent, create child runs via

run_deployment(...)

. The server/work pool enforces that only 2 child runs are Running at once; others will queue. Create the child deployment:

from prefect import flow

@flow
async def child_flow(item):
    await do_expensive_work(item)
    return f"done-{item}"

if __name__ == "__main__":
    child_flow.deploy(
        name="child-deployment",
        work_pool_name="my-pool",
        concurrency_limit=2,  # enforce 2 child runs at a time
    )

Trigger from the parent:

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

@flow
async def parent_flow(items):
    # fire off child runs via the server
    runs = [
        run_deployment(
            "child-flow/child-deployment",
            parameters={"item": i},
            as_subflow=True,  # keeps lineage to this parent
        )
        for i in items
    ]

    # wait for all to complete without blocking the event loop
    await asyncio.gather(
        *(asyncio.to_thread(wait_for_flow_run, r.id) for r in runs)
    )

Notes: - This enforces a global per-deployment limit (across all parents), not per-parent-run only. - Docs: Deployments concept Option C — Global concurrency (applies across all parents) - Use the global concurrency lease in the child flow so only 2 child flows can hold the lease at a time across your entire workspace.

from prefect import flow
from prefect.concurrency.asyncio import concurrency

@flow
async def child_flow(item):
    async with concurrency("browser-sessions", capacity=2):
        await do_expensive_work(item)
        return f"done-{item}"

- This is “global” by key (“browser-sessions”) — it will limit all child runs anywhere using that key. It’s not scoped to a single parent run. - Docs: Global concurrency limits Bonus: If you don’t need subflow semantics - If each item can be a task instead of a subflow, you could: - Use a task runner with limited workers in the parent flow: ```

from prefect import flow, task from prefect.task_runners import ThreadPoolTaskRunner @task def do_expensive_work_task(item): return do_expensive_work_sync(item) @flow(task_runner=ThreadPoolTaskRunner(max_workers=2)) def parent_flow(items): futures = [do_expensive_work_task.submit(i) for i in items] return [f.result() for f in futures] ``` - Or use tag-based concurrency on tasks (task-only): Tag-based concurrency Summary - For “only 2 at a time within this parent run”: use an asyncio.Semaphore (Option A). - For “only 2 child runs at a time across the system”: deploy the child and trigger via

run_deployment

with

concurrency_limit=2

(Option B), or use a global concurrency lease in the child (Option C). If you tell me whether you want the child work to run inline or on workers (and whether the limit should be per-parent or global), I’ll share a drop-in snippet tailored to your exact setup.