Yes — here’s a concise recipe to create and use a global concurrency limit via the Python SDK in Prefect 3.x. Recipe (sync example)

from prefect import flow, task
from prefect.client.orchestration import get_client
from prefect.concurrency.sync import concurrency  # for sync tasks
# from prefect.concurrency.sync import rate_limit  # use this if you set slot_decay_per_second > 0
import time

def ensure_limit(name: str, limit: int, slot_decay_per_second: float = 0.0):
    # Upsert the limit so it's created if missing, updated if it exists
    with get_client(sync_client=True) as client:
        client.upsert_global_concurrency_limit_by_name(
            name=name,
            limit=limit,
            slot_decay_per_second=slot_decay_per_second,
        )

@task
def do_work(i: int):
    # Acquire 1 slot on the global limit "my-limit"
    with concurrency("my-limit", occupy=1, timeout_seconds=30):
        time.sleep(1)
        return i

@flow
def my_flow():
    ensure_limit(name="my-limit", limit=3)  # at most 3 concurrent slots
    futures = [do_work.submit(i) for i in range(10)]
    # Block for results (optional, but common)
    results = [f.result() for f in futures]
    return results

if __name__ == "__main__":
    my_flow()

Notes - Prefer

concurrency("my-limit", occupy=1)

for true concurrency caps. - If you want rate limiting, give your limit a

slot_decay_per_second > 0

and call

rate_limit("my-limit", occupy=1)

in your task instead of the context manager. - Async flows/tasks work the same way; import from

prefect.concurrency.asyncio

and use

async with concurrency(...):

. Optional: create/read/delete via SDK

from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import GlobalConcurrencyLimitCreate

# Create
with get_client(sync_client=True) as client:
    client.create_global_concurrency_limit(
        GlobalConcurrencyLimitCreate(name="api-calls", limit=10, slot_decay_per_second=0.0)
    )

# Read all
with get_client(sync_client=True) as client:
    limits = client.read_global_concurrency_limits(limit=50, offset=0)
    for l in limits:
        print(l.name, l.active_slots, "/", l.limit)

# Delete
with get_client(sync_client=True) as client:
    client.delete_global_concurrency_limit_by_name("api-calls")

FYI (CLI, if you ever want it)

prefect gcl create my-limit --limit 3
prefect gcl ls
prefect gcl inspect my-limit
prefect gcl update my-limit --limit 5
prefect gcl delete my-limit

If you share whether your flow/tasks are sync or async, I can tailor the snippet exactly to your setup.