Great question — in Prefect 3.x these do different jobs: - concurrency context manager: caps how many things are running at the same time. It “holds” a slot for the full duration of your code block via a renewable lease. - rate_limit function: caps how often things are allowed to start. It acquires and immediately releases slots; new slots become available over time via decay. When to use which - Use concurrency when: - You must limit simultaneous in-flight operations (e.g., max 5 concurrent API calls, DB connections, file writes) - You want a slot “reserved” for the entire operation - Operations are variable/long-lived and shouldn’t exceed a simultaneous cap - Use rate_limit when: - You must adhere to “N requests per second/minute” style API limits - You only need to throttle the start rate, not hold a slot during execution - Operations are short-lived (typical HTTP calls) - Note: requires a limit with slot decay configured Key behaviors - concurrency - Holds slots for the whole block; renews a lease in the background - strict=True will fail if Prefect cannot maintain the lease; default is False (warns but continues) - No slot decay required - rate_limit - Waits until a slot is available, then immediately releases it - Requires slot_decay_per_second on the limit; no leases or renewals - Ideal for steady throughput (e.g., 10 req/sec) Typical setup and examples - Create a global concurrency limit (GCL)

prefect gcl create my-api --limit 100 --slot-decay-per-second 100

- Throttle request rate (recommended for API rate limits)

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

@task
def call_api(i: int):
    rate_limit("my-api")  # waits if needed, then immediately releases
    return requests.get(f"<https://api.example.com/items/{i}>").json()

@flow
def main():
    for i in range(1000):
        call_api.submit(i)

- Cap simultaneous in-flight calls (use if the API or your resources also have a concurrency cap)

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

@task
def call_api(i: int):
    with concurrency("my-api-concurrent", occupy=1):  # holds a slot for the duration
        return requests.get(f"<https://api.example.com/items/{i}>").json()

@flow
def main():
    for i in range(1000):
        call_api.submit(i)

Combining both - Many APIs have both: “N requests/sec” and “no more than M concurrent requests.” Use rate_limit for the throughput cap and concurrency for the in-flight cap with two different GCLs. Caveats - rate_limit will error if the corresponding GCL does not have slot_decay_per_second set. - concurrency has lease_duration (default 300s). For very long tasks consider adjusting it; strict=True enforces correctness if leases can’t be renewed. References - Concepts: Global concurrency & rate limits (Prefect 3) - How-to: Enforcing global concurrency & rate limits - Implementation: prefect.concurrency.asyncio If you can share your API’s exact limits (e.g., “100 req/sec with bursts of 200; max 20 concurrent”), I can sketch the precise GCL settings and code pattern.