Great question. In Prefect 3.x, you’ll want to use global concurrency limits with dynamic names and acquire them at the top of your flow so the entire flow run is constrained. Goal - Max 5 concurrent flow runs per color - Max 2 concurrent flow runs per shape 1) Create the global concurrency limits (one per value) For your example values:

# colors (max 5 each)
prefect gcl create color:yellow --limit 5
prefect gcl create color:blue   --limit 5
prefect gcl create color:red    --limit 5

# shapes (max 2 each)
prefect gcl create shape:square    --limit 2
prefect gcl create shape:triangle  --limit 2
prefect gcl create shape:rectangle --limit 2

# verify / inspect
prefect gcl ls
prefect gcl inspect color:red

2) Acquire both limits at the start of the flow (to constrain the whole run) Use the concurrency context manager with both names. You can derive the names from flow parameters (recommended) and optionally add tags for visibility in the UI.

from prefect import flow, tags
from prefect.concurrency.sync import concurrency

@flow
def my_flow(color: str, shape: str):
    # Optional: tag the flow run for visibility
    with tags(f"color:{color}", f"shape:{shape}"):
        # Enforce both limits for the duration of the flow
        with concurrency([f"color:{color}", f"shape:{shape}"], strict=True):
            # ... your flow logic here ...
            pass

Notes - strict=True will fail fast if a limit is missing or cannot be renewed (safer for enforcement). - If values are dynamic/unknown ahead of time, either: - Pre-create limits programmatically at startup, or - Use a small helper to create the limit if missing before entering the concurrency block. - If you prefer to read tags instead of parameters, you can set tags at the flow entry and construct the limit names from those tags—but you still need to explicitly acquire concurrency slots; tags alone do not enforce limits. If you share how you’re passing color/shape (parameters vs tags), I can tailor the code to that pattern.

Here are two small helpers you can drop in to “ensure the limit exists” right before you acquire it. One is simple (uses upsert) and one gives you full control over fields like active/slot decay. Simple helper (upsert) - Best when you just need to ensure name/limit exist.

from prefect import flow, get_client
from prefect.concurrency.sync import concurrency

async def ensure_limit(name: str, limit: int) -> None:
    async with get_client() as client:
        # Creates the limit if missing or updates the limit if it exists
        await client.upsert_global_concurrency_limit_by_name(name=name, limit=limit)

async def ensure_color_shape_limits(color: str, shape: str) -> None:
    await ensure_limit(f"color:{color}", 5)
    await ensure_limit(f"shape:{shape}", 2)

@flow
async def my_flow(color: str, shape: str):
    # Ensure the limits exist first
    await ensure_color_shape_limits(color, shape)

    # Then acquire both limits for this run
    with concurrency([f"color:{color}", f"shape:{shape}"], strict=True):
        # ... your flow logic ...
        pass

Advanced helper (controls active/slot decay) - Use this if you need to set active or slot_decay_per_second.

from prefect import flow, get_client
from prefect.concurrency.sync import concurrency
from prefect.client.schemas.actions import GlobalConcurrencyLimitCreate, GlobalConcurrencyLimitUpdate
from prefect.exceptions import ObjectNotFound

async def ensure_limit_with_config(
    name: str,
    limit: int,
    *,
    active: bool = True,
    slot_decay_per_second: float | None = None
) -> None:
    async with get_client() as client:
        try:
            existing = await client.read_global_concurrency_limit_by_name(name)
            update = {}

            if existing.limit != limit:
                update["limit"] = limit
            if existing.active != active:
                update["active"] = active
            # Only compare/update decay if caller specified it
            if slot_decay_per_second is not None and existing.slot_decay_per_second != slot_decay_per_second:
                update["slot_decay_per_second"] = slot_decay_per_second

            if update:
                await client.update_global_concurrency_limit(
                    name=name,
                    concurrency_limit=GlobalConcurrencyLimitUpdate(**update)
                )

        except ObjectNotFound:
            await client.create_global_concurrency_limit(
                GlobalConcurrencyLimitCreate(
                    name=name,
                    limit=limit,
                    active=active,
                    slot_decay_per_second=slot_decay_per_second or 0.0
                )
            )

async def ensure_color_shape_limits(color: str, shape: str) -> None:
    await ensure_limit_with_config(f"color:{color}", 5, active=True)
    await ensure_limit_with_config(f"shape:{shape}", 2, active=True)

@flow
async def my_flow(color: str, shape: str):
    await ensure_color_shape_limits(color, shape)
    with concurrency([f"color:{color}", f"shape:{shape}"], strict=True):
        # ... your flow logic ...
        pass

Sync variant (if your flow is sync)

from prefect import flow, get_client
from prefect.concurrency.sync import concurrency

def ensure_limit_sync(name: str, limit: int) -> None:
    with get_client(sync_client=True) as client:
        client.upsert_global_concurrency_limit_by_name(name=name, limit=limit)

def ensure_color_shape_limits_sync(color: str, shape: str) -> None:
    ensure_limit_sync(f"color:{color}", 5)
    ensure_limit_sync(f"shape:{shape}", 2)

@flow
def my_flow_sync(color: str, shape: str):
    ensure_color_shape_limits_sync(color, shape)
    with concurrency([f"color:{color}", f"shape:{shape}"], strict=True):
        # ... your flow logic ...
        pass

Notes - upsert is idempotent and safe under concurrency (great for “many flows start at once”). - strict=True ensures the flow won’t proceed if a limit can’t be enforced.

Great instinct. Here are practical ways to minimize requests while still creating limits on demand. Option A — Pre-provision once (best for your finite set) For your known values, create them once in CI/CD or at worker startup and don’t upsert at runtime:

# colors (5)
prefect gcl create color:yellow --limit 5 || true
prefect gcl create color:blue   --limit 5 || true
prefect gcl create color:red    --limit 5 || true

# shapes (2)
prefect gcl create shape:square    --limit 2 || true
prefect gcl create shape:triangle  --limit 2 || true
prefect gcl create shape:rectangle --limit 2 || true

Then your flow just acquires:

with concurrency([f"color:{color}", f"shape:{shape}"], strict=True):
    ...

Option B — Cache and batch-fetch once per worker process Read all limits once, cache names in memory, and only upsert when truly missing (with per-name locks to dedupe concurrent calls). This reduces requests to: one listing on startup + rare creates.

import asyncio
from typing import Set
from prefect import flow, get_client
from prefect.concurrency.sync import concurrency

class GCLRegistry:
    def __init__(self):
        self._loaded = False
        self._names: Set[str] = set()
        self._load_lock = asyncio.Lock()
        self._locks: dict[str, asyncio.Lock] = {}

    async def load(self) -> None:
        if self._loaded:
            return
        async with self._load_lock:
            if self._loaded:
                return
            async with get_client() as client:
                offset = 0
                page_size = 500
                while True:
                    page = await client.read_global_concurrency_limits(limit=page_size, offset=offset)
                    if not page:
                        break
                    self._names.update(l.name for l in page)
                    offset += len(page)
            self._loaded = True

    async def ensure(self, name: str, limit: int) -> None:
        await self.load()
        if name in self._names:
            return
        lock = self._locks.setdefault(name, asyncio.Lock())
        async with lock:
            if name in self._names:
                return
            async with get_client() as client:
                # Upsert only when missing
                await client.upsert_global_concurrency_limit_by_name(name=name, limit=limit)
            self._names.add(name)

REGISTRY = GCLRegistry()

async def ensure_color_shape(color: str, shape: str) -> None:
    await REGISTRY.ensure(f"color:{color}", 5)
    await REGISTRY.ensure(f"shape:{shape}", 2)

@flow
async def my_flow(color: str, shape: str):
    await ensure_color_shape(color, shape)
    with concurrency([f"color:{color}", f"shape:{shape}"], strict=True):
        # ... your flow logic ...
        pass

Why this helps - One list call per worker process (cached). - No per-run upserts unless a name is truly new. - Per-name lock prevents a thundering herd if many runs try to create the same name at once. Option C — Provisioner flow (centralized) Run a small scheduled “provisioner” flow that reads your allowed values (e.g., from config/DB) and upserts all needed limits. Your business flows then never upsert.

from prefect import flow, get_client

@flow
async def provision_limits():
    async with get_client() as client:
        desired = {
            "color:yellow": 5, "color:blue": 5, "color:red": 5,
            "shape:square": 2, "shape:triangle": 2, "shape:rectangle": 2,
        }
        for name, limit in desired.items():
            await client.upsert_global_concurrency_limit_by_name(name=name, limit=limit)

Pick the one that fits your setup: - Known finite set → Option A (zero runtime requests). - Dynamic but not huge → Option B (one-time list + rare create). - Centralized ownership → Option C (separate provisioner flow).