Short answer: there isn’t a single “per-parameter flow lock” built-in at the flow level. But you can get the exact behavior you want by combining a few features: What Prefect provides - Prevent duplicate flow-run creation (if you control the trigger): pass an idempotency key to

run_deployment()

. Compute the key from your input parameters so duplicate triggers create only one flow run. - API ref: Python API: run_deployment - How-to: Trigger ad‑hoc deployment runs - Control execution concurrency per “key” inside your flow: acquire a global concurrency limit named from your parameters at the very start of the flow. Only one run holding that key will proceed at a time. - Concept: Global concurrency limits - How-to: How to apply global concurrency and rate limits - Debounce noisy/duplicate events at the automation layer: collapse bursts of duplicate triggers so only one run is scheduled. - Guide: How to debounce events Notes specific to Automations - Automations “Run Deployment” actions already attach an idempotency key automatically, but it’s tied to the automation invocation, not your parameters; it isn’t currently user-configurable or templated from the event payload. This prevents duplicate flow runs for the same automation invocation but won’t dedupe separate invocations with the same parameters. - Automations concept: Automations Recommended pattern for “only one concurrent run per parameter set” - If you control the trigger (webhook, service, etc.): 1) Derive an idempotency key from the parameters and pass it to

run_deployment()

so duplicates don’t even create a second run. 2) Also acquire a parameter-derived global concurrency limit at the start of the flow to ensure only one active run executes for that key at a time. Example

python
# trigger-side (e.g., your webhook handler)
import hashlib, json
from prefect.deployments import run_deployment

def key_from_params(params: dict) -> str:
    # include only the fields that define “same work”
    return hashlib.sha256(json.dumps(params, sort_keys=True).encode()).hexdigest()

params = {"customer_id": 123, "date": "2026-03-13"}
run_deployment(
    name="my-flow/my-deployment",
    parameters=params,
    idempotency_key=f"my-flow:{key_from_params(params)}",
)

python
# inside your flow
from prefect import flow
from prefect.concurrency.sync import concurrency

@flow
def my_flow(customer_id: int, date: str):
    lock_name = f"my-flow:{customer_id}"
    # Only 1 run at a time per customer_id. With strict=True, fail fast if already locked.
    with concurrency(lock_name, occupy=1, strict=True):
        # do the work
        ...

If you must use Automations to trigger: - Use automation debouncing to collapse duplicates arriving close together. - Still take the global concurrency lock at the start of the flow (as above) to enforce “only one at a time” per key. Related but different - Deployment-level concurrency limits and collision strategies (ENQUEUE or CANCEL_NEW) apply across all runs of a deployment (not per parameter). Useful guardrails, but not sufficient for per-key locking. - Docs: Deployments (Concurrency limiting section) A couple of quick questions to tailor this: - How are you triggering runs today (Automations only, or do you also trigger via code/CI/webhooks)? - For duplicates, do you prefer the second run to wait, be skipped/canceled, or to coalesce work into the first run?