Yes. In Prefect 3.x you can enforce a global “semaphore” across all workers using global concurrency limits. These are managed by the Prefect server/Cloud and work across every worker in your workspace. Two common patterns: - Global concurrency (pure semaphore): cap concurrent calls to N across all workers - Global rate limiting: throttle to X ops/sec across all workers (optionally with a burst) How to set it up 1) Create the global limit (CLI) - Pure semaphore (max 10 concurrent):

prefect global-concurrency-limit create external-api --limit 10

- True rate limit (e.g., 20 requests/sec with a burst of 20): - Set limit to your allowed burst (20) - Set slot decay to your sustained rate (20/sec)

prefect global-concurrency-limit create external-api --limit 20 --slot-decay-per-second 20

Helpful CLI management (verified):

prefect global-concurrency-limit ls
prefect global-concurrency-limit inspect <name>
prefect global-concurrency-limit update <name> --limit <n> --slot-decay-per-second <r>
prefect global-concurrency-limit disable <name>
prefect global-concurrency-limit delete <name>

2) Use it in your code - Async tasks/flows:

from prefect import task
from prefect.concurrency.asyncio import concurrency, rate_limit

@task
async def call_api(i: int):
    # Option A: rate limit (throttle to configured RPS)
    await rate_limit("external-api", occupy=1)
    # ...make the request...

@task
async def do_something_needing_a_slot():
    # Option B: pure semaphore (hold slot for duration of work)
    async with concurrency("external-api", occupy=1, strict=True):
        # ...do work...

- Sync tasks/flows:

from prefect import task
from prefect.concurrency.sync import concurrency, rate_limit

@task
def call_api(i: int):
    # Rate limit (blocks until a token is available)
    rate_limit("external-api", occupy=1)
    # ...make the request...

@task
def do_something_needing_a_slot():
    # Pure semaphore (context holds the slot)
    with concurrency("external-api", occupy=1, strict=True):
        # ...do work...

Notes and tips - strict=True will fail fast if the limit doesn’t exist or can’t be enforced; omit it if you prefer best-effort behavior. - timeout_seconds lets you bound how long you’ll wait for a slot. - For rate limits: set limit = your desired burst size, slot_decay_per_second = sustained ops/sec. - These limits are global across all workers connected to the same Prefect server/Cloud workspace. - If you instead want to cap concurrent flow runs (not calls within a flow), you can also use deployment, work pool, or work queue concurrency limits. Want me to suggest exact settings? Tell me: - Your target requests/second - Whether a burst is allowed (and how big) - Whether you prefer a strict cap on concurrent in-flight calls (vs. pure RPS)

Great questions! 1) Worker-level concurrency limit - In Prefect 3.x, each worker can cap its own concurrent flow runs. - Start the worker with a limit:

prefect worker start -p <your_pool> --limit 5

This limits that single worker to 5 concurrent flow runs. It’s independent of work pool/queue/deployment limits: - Work pool/queue limits = global caps across all workers - Worker --limit = per-process cap for that worker only 2) Your concurrency context syntax - The correct API is

concurrency(..., occupy=...)

- What you wrote (

with_ concurrency(..., _occupy_=1)

) is not valid. Use one of these depending on sync vs async code: - Synchronous

from prefect.concurrency.sync import concurrency

with concurrency("my-limit-name", occupy=1):
    # protected work
    ...

- Asynchronous

from prefect.concurrency.asyncio import concurrency

async with concurrency("my-limit-name", occupy=1):
    # protected work
    ...

Notes - The name you pass (e.g., "my-limit-name") must correspond to an existing global concurrency limit or tag limit. For a pure semaphore, create it with:

prefect global-concurrency-limit create my-limit-name --limit 10

- If you want true rate limiting, add

--slot-decay-per-second <rps>

when creating the limit and use

rate_limit("my-limit-name")

instead of

concurrency(...)

. If your work pool has no global limit, combining a worker

--limit

with a global concurrency limit (or tag limit) is a solid way to prevent local overload and external rate-limiting at the same time.

Here’s a quick, reliable way to test global semaphores (global concurrency limits) locally in Prefect 3.x. Prereqs - This requires a running Prefect Server (local API). The concurrency context talks to the API. Step 1) Start Prefect Server locally - In one terminal:

prefect server start

- In another terminal, point your profile at the local API if needed:

prefect config set PREFECT_API_URL=<http://127.0.0.1:4200/api>
prefect config view

Step 2) Create a global concurrency limit - Example: semaphore “demo” with 2 slots

prefect global-concurrency-limit create demo --limit 2
prefect global-concurrency-limit inspect demo

Step 3) Run a simple flow that contends for the semaphore - This submits multiple tasks that each hold a slot for ~2s. You should see only 2 tasks run at once. - Use strict=True so the test fails fast if the limit isn’t found/enforced. Create test_semaphore.py:

import time
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
from prefect.concurrency.sync import concurrency

@task
def protected(i: int):
    t0 = time.perf_counter()
    print(f"task {i} requesting slot at {t0:.2f}")
    with concurrency("demo", occupy=1, strict=True, timeout_seconds=30, lease_duration=60):
        print(f"task {i} acquired slot")
        time.sleep(2)
    t1 = time.perf_counter()
    print(f"task {i} released slot at {t1:.2f} (dur={t1 - t0:.2f}s)")
    return i

@flow(task_runner=ConcurrentTaskRunner())
def run(n: int = 6):
    futs = [protected.submit(i) for i in range(n)]
    return [f.result() for f in futs]

if __name__ == "__main__":
    run()

Run it:

python test_semaphore.py

What you should see - At most 2 “acquired slot” messages at a time (since limit=2). - Total runtime ~ ceil(n / 2) * 2 seconds (for n=6, about 6 seconds). - You can watch slot usage live in another terminal:

prefect global-concurrency-limit inspect demo

Look for active/denied slots changing while the flow runs. Optional: test global rate limiting - Create a limit that allows ~5 ops/sec with a burst of 5:

prefect global-concurrency-limit create demo-rps --limit 5 --slot-decay-per-second 5

- Then use

rate_limit("demo-rps")

instead of

concurrency(...)

in your code to throttle calls globally. Optional: test across multiple workers - Start multiple workers (each with their own per-worker cap), and run the same flow concurrently from different terminals. The global limit still gates everything.

prefect worker start -p <your_pool> --limit 3

- Launch the script from multiple shells or via deployments. You’ll see the global limit enforce across all processes. Common gotchas - Use strict=True while testing to ensure the limit is truly enforced. - Make sure your code is connected to the same API server you used to create the limit. - If tasks don’t run concurrently, ensure the flow uses

ConcurrentTaskRunner()

as above. If you want, tell me your exact target rate/concurrency and I’ll tailor the limit settings and an example to match.

Got it — automated tests. Below are two proven approaches you can drop into pytest: a fast unit-test style that mocks the client, and a higher-fidelity integration test that spins up an ephemeral Prefect API and asserts real slot behavior. Both align with Prefect 3.x internals. Option A: Fast unit tests (no server) by mocking the client - Pros: Very fast, deterministic, no external services - Cons: Doesn’t exercise the real lease mechanics Example: assert we call the API to acquire/release and that timeouts/strict mode behave

import asyncio
import pytest
from unittest import mock
from httpx import HTTPStatusError, Request, Response
from prefect.concurrency.asyncio import concurrency

@pytest.mark.asyncio
async def test_concurrency_acquires_and_releases_calls_client():
    # Mock the client methods used by concurrency()
    with mock.patch(
        "prefect.client.orchestration.PrefectClient.increment_concurrency_slots_with_lease",
        autospec=True,
    ) as mock_increment, mock.patch(
        "prefect.client.orchestration.PrefectClient.decrement_concurrency_slots_with_lease",
        autospec=True,
    ) as mock_decrement:
        mock_increment.return_value = {"lease_id": "lease-1", "acquired": True}

        async with concurrency("test-limit", occupy=1, strict=True):
            pass

        assert mock_increment.called
        assert mock_decrement.called

@pytest.mark.asyncio
async def test_concurrency_timeout_is_respected():
    # Simulate server responding with 423 Locked + Retry-After, forcing retries until timeout
    def locked_response(*args, **kwargs):
        response = Response(status_code=423, headers={"Retry-After": "0.01"})
        raise HTTPStatusError("Locked", request=Request("GET", "<http://test>"), response=response)

    with mock.patch(
        "prefect.client.orchestration.PrefectClient.increment_concurrency_slots_with_lease",
        side_effect=locked_response,
    ):
        with pytest.raises(TimeoutError):
            async with concurrency("test-limit", occupy=1, timeout_seconds=0.05, strict=True):
                pass

Option B: Integration-style tests with a real ephemeral API - Pros: Exercises the real global semaphore implementation end-to-end - Cons: Slightly slower Approach 1: Use prefect_test_harness() context

import asyncio
import time
import pytest
from prefect.testing.utilities import prefect_test_harness
from prefect.concurrency.asyncio import concurrency
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import GlobalConcurrencyLimitCreate

@pytest.mark.asyncio
async def test_global_semaphore_allows_only_n_concurrent():
    # Isolated ephemeral API/DB
    with prefect_test_harness():
        # Create a global concurrency limit "demo" with 2 slots
        async with get_client() as client:
            try:
                await client.create_global_concurrency_limit(
                    GlobalConcurrencyLimitCreate(name="demo", limit=2)
                )
            except Exception:
                pass

        active = 0
        max_active = 0
        lock = asyncio.Lock()

        async def worker(i):
            nonlocal active, max_active
            async with concurrency("demo", occupy=1, strict=True, lease_duration=60):
                async with lock:
                    active += 1
                    max_active = max(max_active, active)
                await asyncio.sleep(0.1)
                async with lock:
                    active -= 1

        await asyncio.gather(*[worker(i) for i in range(8)])

        # Only 2 can be in the critical section concurrently
        assert max_active == 2

Approach 2: Spin up the hosted test server fixture (if you prefer fixtures) - Use the fixtures in prefect.testing.fixtures (e.g., hosted_api_server/use_hosted_api_server) to launch a subprocess server and point PREFECT_API_URL automatically. Then use the same test body as above to create the limit and assert max_active.

Testing rate limiting (slot decay) - Create a limit with decay and assert effective RPS by measuring timestamps, or mock anyio.sleep to “time travel.”

import asyncio
import pytest
from prefect.testing.utilities import prefect_test_harness
from prefect.concurrency.asyncio import rate_limit
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import GlobalConcurrencyLimitCreate

@pytest.mark.asyncio
async def test_global_rate_limit_throttles_calls():
    with prefect_test_harness():
        async with get_client() as client:
            # Allow ~5 ops/sec with burst of 5
            await client.create_global_concurrency_limit(
                GlobalConcurrencyLimitCreate(name="rps5", limit=5, slot_decay_per_second=5.0)
            )

        timestamps = []

        async def hit():
            await rate_limit("rps5", occupy=1)
            timestamps.append(asyncio.get_running_loop().time())

        # Launch 15 calls "at once"
        await asyncio.gather(*[hit() for _ in range(15)])

        # Very rough assertion: first ~5 are immediate; remaining spread over ~2s+
        # You can bin timestamps by second and assert no bin exceeds ~5.
        bins = {}
        for t in timestamps:
            sec = int(t)
            bins[sec] = bins.get(sec, 0) + 1

        assert all(count <= 6 for count in bins.values())  # allow 1 margin for scheduler jitter

Tips for stable tests - Use prefect_test_harness() for isolation; it boots a temporary API/DB and cleans up. - Set strict=True in tests so missing limits fail fast. - Keep sleeps short; or mock anyio.sleep to speed up time-sensitive checks. - For concurrency counts, guard counters with an asyncio.Lock to avoid race conditions in the assertion bookkeeping. - If you add task/flow layers, use ConcurrentTaskRunner to ensure intra-flow parallelism during tests. If you share your test stack (pytest-only vs. fixtures allowed, sync vs. async), I can tailor the snippets to match exactly and include a conftest.py setup.