Thanks for the details — this message is coming from Prefect’s run input store. What’s happening - In Prefect 3.x, retries reuse the same flow run ID. If you call

create_flow_run_input

with the same

key

again during a retry, the server will return: Response: {'detail': 'A flow run input with this key already exists.'} - In other words, the input for that flow run+key already exists from the first attempt. Ways to fix it 1) Read-first (recommended): Reuse the existing value on retries - Try to read the input; if it does not exist, create it once. This makes retries idempotent.

from prefect import flow, get_client
from prefect.input import create_flow_run_input, read_flow_run_input
from prefect.exceptions import PrefectHTTPStatusError

@flow
async def my_flow():
    key = "user_input:my_param"
    async with get_client() as client:
        try:
            value = await read_flow_run_input(client, key)
        except PrefectHTTPStatusError:
            # Collect your value once (e.g. from UI or other means)
            value = "..."  # your collected input
            await create_flow_run_input(client, key=key, value=value)

    # Use `value` in your flow

2) Make the key unique per attempt (if you want fresh input for each retry) - Append the run attempt counter to the key so each retry gets a new input.

from prefect import flow, get_client
from prefect.runtime import flow_run
from prefect.input import create_flow_run_input, read_flow_run_input
from prefect.exceptions import PrefectHTTPStatusError

@flow
async def my_flow():
    key = f"user_input:my_param:attempt-{flow_run.run_count}"
    async with get_client() as client:
        try:
            value = await read_flow_run_input(client, key)
        except PrefectHTTPStatusError:
            value = "..."  # collect fresh input
            await create_flow_run_input(client, key=key, value=value)

3) Delete then recreate (only if you explicitly want to overwrite)

from prefect import get_client
from prefect.input import delete_flow_run_input, create_flow_run_input

async with get_client() as client:
    await delete_flow_run_input(client, "user_input:my_param")
    await create_flow_run_input(client, "user_input:my_param", new_value)

Notes - Assuming you’re on Prefect 3.x. With retries, the same flow run re-executes, so keys must be handled idempotently. - If you’re using interactive pauses + RunInput models, the same principle applies: either reuse the same key and read it on retries, or generate unique keys per attempt. Docs for reference - Interactive inputs overview: Interactive flows and inputs (Prefect 3) - Flow run creation and behavior: Flow run API reference If you can share a short snippet of how you’re creating/reading the input and how your retry is configured, I can tailor the exact pattern to your setup.