Great question — and you’re thinking about this the right way. Short answer: In Prefect 3.x, the most reliable way to serialize these reports across deployments is to use a tag-based concurrency limit on the task(s) that actually call the external API. That will queue runs automatically and ensure only one is executing at a time, even if multiple flows are scheduled simultaneously. Key points: - Tag-based concurrency limits apply to task runs (not flow runs). To serialize “flows,” make sure the API-calling work is inside a tagged task shared by those flows. - When no slots are available, Prefect automatically moves the task into an AwaitingConcurrencySlot scheduled state and retries later — exactly the “queue up and process later” behavior you want. Recommended setup for your case 1) Tag the API-calling task(s) Make sure every report’s API calls run through a single task that carries the shared tag, e.g. `data_exception_reports`:

from prefect import flow, task

@task(tags=["data_exception_reports"])
def create_tickets_in_external_system(payload):
    # make the API calls here
    ...

@flow
def run_report(report_name: str):
    payload = {...}
    return create_tickets_in_external_system.submit(payload)

- If a report makes multiple API calls, keep them inside this tagged task or tag each task that hits the API. - If you’ve put the tag on a flow or deployment, move it to the task — concurrency limits only check task tags. 2) Create a concurrency limit of 1 for that tag

prefect concurrency-limit create data_exception_reports 1

Useful inspection commands:

prefect concurrency-limit ls
prefect concurrency-limit inspect data_exception_reports

What happens at runtime - If multiple reports start at the same time, only one task with tag

data_exception_reports

can enter Running. Others are put into a scheduled “AwaitingConcurrencySlot” state and will re-attempt automatically until a slot opens. - The default wait between attempts is ~30 seconds. You can tune it on the server with:

prefect config set PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS=60

Optional: Add a true rate limit if your API has RPM/RPS caps Concurrency=1 prevents parallel calls, but if your API has strict per-minute limits, add a global concurrency limit with decay (token bucket) and enforce it in code: 1) Create a global rate limit (example: 10 requests/minute) - Limit = 10 - slot_decay_per_second = 10/60 ≈ 0.1667

prefect global-concurrency-limit create external-api-rate --limit 10 --slot-decay-per-second 0.1667

2) Use the rate/concurrency guard in your task

from prefect.concurrency.sync import concurrency, rate_limit

@task(tags=["data_exception_reports"])
def create_tickets_in_external_system(payload):
    # Enforce no more than 10 req/min across all flows/tasks using this name
    rate_limit("external-api-rate", occupy=1)

    # Optionally also cap parallelism if you want >1 but still limited concurrency
    # with concurrency("external-api-concurrent", occupy=1, strict=True):
    #     ...

    # Make API calls here
    ...

This combo protects both parallelism (if desired) and overall request rate, minimizing failure loops. Other knobs you might want - Deployment-level concurrency can limit runs of a single deployment, but it won’t serialize different deployments against each other. For your cross-deployment need, use tagged tasks (and/or a global limit). - Consider a retry policy that does not aggressively retry on rate-limit errors, or uses exponential backoff with a

retry_condition_fn

to skip retries on 429s. This avoids burning quota on doomed retries. Sanity checks and monitoring - In the UI, queued tasks show as Scheduled with “AwaitingConcurrencySlot”. - CLI: -

prefect concurrency-limit ls

and

prefect concurrency-limit inspect data_exception_reports

-

prefect global-concurrency-limit ls

and

prefect global-concurrency-limit inspect external-api-rate

Does your report code currently funnel all API calls through a single task that we can tag, or are there multiple tasks per flow that may hit the API? If it’s the latter, we can centralize those calls into one tagged task or tag each calling task to ensure full coverage.