Short answer: there isn’t a built-in “per-parameter flow-run” concurrency switch. But you can achieve exactly this in Prefect 3.x with global concurrency limits or task-tag limits. Recommended approach: use a global concurrency limit whose name is derived from the parameter value, and acquire it for the duration of the flow run. How it works - Create a global concurrency limit with limit=1 for each

param_b

value you need to serialize. - At the start of the flow, acquire that limit using the

prefect.concurrency

context; hold it for the entire run. Example

from prefect import flow
from prefect.concurrency.sync import concurrency  # use asyncio.concurrency in async code

@flow
def my_flow(param_a: str, param_b: str, param_c: str):
    limit_name = f"my-flow:param_b:{param_b}"   # derive a unique limit name per param_b value

    # Ensure a global concurrency limit with this name exists (limit=1) before runs start
    with concurrency(limit_name):
        # Everything inside will run with mutual exclusion per param_b value
        # e.g., kick off tasks, subflows, etc.
        ...

Create the global concurrency limits (one per expected value) once:

prefect global-concurrency-limit create "my-flow:param_b:VALUE1" --limit 1
prefect global-concurrency-limit create "my-flow:param_b:VALUE2" --limit 1
# etc.

Notes - If the set of

param_b

values is large/dynamic, you can programmatically “upsert” the limit before acquiring it via the Prefect API; otherwise, pre-create them with the CLI. - The

with concurrency(...)

context acquires a lease in Prefect’s backend and will block a second run with the same name until the first finishes or the lease is released. Alternative options (when it fits your use case) - Separate deployments per

param_b

with

concurrency_limit=1

on each deployment. Works well if

param_b

values are few and known. - Task tag concurrency: tag all tasks under something like

param_b:<value>

and create a task tag concurrency limit of 1 for each tag. This throttles tasks rather than the entire flow run but produces a similar effect if all work happens in tasks. Docs for reference - Global concurrency limits: How to use global concurrency limits - Tag-based (task-level) concurrency: Tag-based concurrency limits Follow-up questions - Are the possible

param_b

values known and finite, or dynamic? - Do you prefer a code-based guard (global concurrency) or infra-level guard (separate deployments or queue limits)?

Got it — dynamic values and you need strict “one-at-a-time per contact” plus in-order processing. Here’s how to do this cleanly in Prefect 3.x. What Prefect can guarantee - Mutual exclusion per contact: Use a Global Concurrency Limit (GCL) named from the user_id + contact_id and acquire it for the entire flow. Set limit=1 per contact key. This ensures only one flow run for that contact can be “in flight”. - Ordering: Prefect’s concurrency system is a semaphore, not a FIFO queue — it does not guarantee which waiting run acquires the next slot first. If you need strict in-order processing, you should queue/sequence runs on your side and only submit the next run when the previous one completes. Recommended pattern 1) Derive a GCL name per contact: - e.g.,

audio:{user_id}:{contact_id}

(sanitize characters that are not allowed:

/ % & > <

). 2) Create that GCL (limit=1) when a new contact appears. 3) At the start of your flow, acquire that GCL for the duration of the run. 4) Enforce order in your backend by only submitting the next flow run for a contact when the prior one finishes. Flow example

from prefect import flow
from prefect.concurrency.sync import concurrency  # use asyncio version if needed

@flow
def process_audio_note(user_id: str, contact_id: str, cos_url: str, **kwargs):
    limit_name = f"audio:{user_id}:{contact_id}".replace("/", "_")
    # strict=True ensures we fail fast if the GCL wasn’t created
    with concurrency(limit_name, strict=True):
        # Do all work for this audio note here
        ...

Create/update the GCLs (limit=1) - Create once per new contact (e.g., from FastAPI when a new contact is first seen):

prefect global-concurrency-limit create "audio:USER123:CONTACT456" --limit 1

If you call this repeatedly, handle “already exists” gracefully (create-or-ignore, or use update when it exists). Enforcing FIFO ordering - Prefect’s concurrency will serialize execution but does not guarantee the order that queued runs acquire the lock. - Keep a per-contact FIFO queue in your backend (e.g., Redis or DB): enqueue audio notes in the order recorded, and only submit the next Prefect run after the previous one completes. - Option A (recommended): Have Prefect notify your backend on completion (via an Automation webhook or via a simple HTTP call from the flow’s

on_completion

hook), and your backend then dequeues and triggers the next run for that contact. - Option B: Your backend polls Prefect for the prior run’s completion, then triggers the next. - This keeps hundreds of contacts flowing in parallel while guaranteeing strict per-contact ordering. Nice-to-haves - Add a flow run tag like

contact:{user_id}:{contact_id}

for observability and filtering. - If your flow is async, you can use the async concurrency context; otherwise the sync context above works well. - Be mindful of special characters in your limit names; sanitize or encode them. Docs - Global concurrency limits: How to use global concurrency limits - Concurrency context API: prefect.concurrency.sync and prefect.concurrency.asyncio - Flow run creation (if you’re calling the API from FastAPI): Create a flow run - Automations (to call your backend when runs complete): Automations If you want, I can sketch the small FastAPI-side queuing/triggering flow (enqueue, trigger first run, on completion trigger next). Let me know if you’re on Prefect Cloud or OSS so I can tailor the completion-callback approach.