Reliable LLM Workflows: Idempotency Keys, Retries and Queues with FastAPI and Redis

An LLM step in a business workflow is a slow, rate-limited, occasionally hung network call between a user's request and a side effect someone will notice. The expensive failures are in the plumbing around it: a retried request that runs twice, a summary emailed twice, a job lost when a pod restarts.

This is the reference implementation I use for that plumbing: a FastAPI endpoint that accepts a summarisation job, a Redis Streams worker that calls the model, and a Postgres outbox that sends the notification email. Every Python block below comes from code I ran against real Redis 7.2 and PostgreSQL 16. The model and email provider are fakes behind the production interfaces.

The failure model I design for

Each failure I expect maps to one mechanism:

• Clients retry. Networks drop responses and SDKs retry on their own, so the same POST arrives more than once. Fix: an Idempotency-Key header.
• Processes die between steps. A worker can finish the work and be killed before acknowledging it. The queue must redeliver, so handlers must be idempotent.
• The model API fails in two ways. A 429 or 503 will probably succeed later; a 400 for an oversized context never will. Retrying it wastes money and delays the dead letter.
• Calls hang. Without an external limit, one hung request holds a worker slot indefinitely.
• Side effects escape transactions. Send then fail to commit, or commit then crash before sending, and you get a duplicate or a lost email.

The principle: accept at-least-once delivery everywhere and make every effect idempotent. Exactly-once delivery is not on offer; exactly-once effects are, and that is what users see.

Idempotency-Key on the POST

The endpoint must give the same answer to the same request and enqueue at most one job. The first request wins a SET NX and writes an in-progress marker with a short TTL. When the handler finishes, the key is overwritten with the final response and a long TTL, and later requests are answered from it.

# ... imports, Handler type alias

# Only overwrite/delete the key if it still holds *our* in-progress marker.
_FINALISE = """
if redis.call('GET', KEYS[1]) == ARGV[1] then
  redis.call('SET', KEYS[1], ARGV[2], 'EX', ARGV[3])
  return 1
end
return 0
"""
# ... _RELEASE: same guard, then DEL; InProgress and KeyReused exceptions

@dataclass
class IdempotencyStore:
    redis: Redis
    lock_ttl: int = 30          # > slowest normal request; bounds crash recovery
    result_ttl: int = 86_400    # how long clients may safely retry

async def run(self, key: str, fingerprint: str, handler: Handler) -> tuple[int, dict, bool]:
        marker = json.dumps({"state": "in_progress", "fp": fingerprint,
                             "token": uuid.uuid4().hex})
        if await self.redis.set(key, marker, nx=True, ex=self.lock_ttl):
            try:
                status, body = await handler()
            except BaseException:
                await self.redis.eval(_RELEASE, 1, key, marker)
                raise
            if status >= 500:                     # let the client retry
                await self.redis.eval(_RELEASE, 1, key, marker)
                return status, body, False
            done = json.dumps({"state": "done", "fp": fingerprint,
                               "status": status, "body": body})
            if not await self.redis.eval(_FINALISE, 1, key, marker, done,
                                         self.result_ttl):
                log.warning("idempotency_lock_lost", key=key)
            return status, body, False

existing = await self.redis.get(key)
        if existing is None:                      # expired between SET and GET
            raise InProgress()
        record = json.loads(existing)
        if record["fp"] != fingerprint:
            raise KeyReused()
        if record["state"] == "in_progress":
            raise InProgress()
        return record["status"], record["body"], True

The details that matter:

• The marker carries a random token, and finalise or release only happen if the key still holds that exact marker. Without the guard, a request that overran its lock TTL could overwrite the result of a newer request. Newer Redis releases add compare-and-set options to SET (redis-py 8.1 exposes ifeq=); on 7.2 a short Lua script does it atomically.
• The fingerprint is a hash of the validated body. Same key with a different body is a client bug, so it gets 422, never someone else's response.
• 5xx and exceptions release the key. Storing a failure would make it permanent for 24 hours.
• The lock TTL is the crash-recovery time. If the process is killed mid-request, nothing releases the marker; duplicates get 409 until it expires. Too short and a slow request loses its lock while still running; too long and clients stay blocked after a crash. I set it just above the slowest normal request.

The route scopes the key by tenant, method and path (in production the tenant comes from the authenticated principal, not a header) and derives the job id from it:

@app.post("/v1/summaries", status_code=202)
async def create_summary(
    body: SummaryRequest,
    idempotency_key: Annotated[str, Header(min_length=16, max_length=128)],
    x_tenant_id: Annotated[str, Header(max_length=64)],
    r: Annotated[Redis, Depends(get_redis)],
    store: Annotated[IdempotencyStore, Depends(get_store)],
    enqueue_job=Depends(get_enqueue),
):
    scope = f"idem:{x_tenant_id}:POST:/v1/summaries:{idempotency_key}"
    fingerprint = hashlib.sha256(body.model_dump_json().encode()).hexdigest()
    # Deterministic job id: even if the handler runs twice, downstream sees one job.
    job_id = str(uuid.uuid5(JOB_NS, scope))

async def handler() -> tuple[int, dict]:
        await enqueue_job(r, job_id, body.model_dump())
        return 202, {"job_id": job_id, "status": "queued"}

try:
        status, content, replayed = await store.run(scope, fingerprint, handler)
    except InProgress:
        return JSONResponse({"detail": "request with this key in progress"},
                            status_code=409, headers={"Retry-After": "1"})
    except KeyReused:
        return JSONResponse({"detail": "Idempotency-Key reused with a different body"},
                            status_code=422)
    headers = {"Idempotent-Replayed": "true"} if replayed else {}
    return JSONResponse(content, status_code=status, headers=headers)

The uuid5 job id matters most. If the API dies between XADD and storing the response, the marker expires, the client retries, and a second message is enqueued. Because the id is derived from the key, that message carries the same job_id and the worker treats it as a duplicate. The key reduces duplicates; the job id makes the survivors harmless.

The tests cover each case through httpx.ASGITransport against real Redis:

At-least-once processing with Redis Streams

I use a Streams consumer group rather than a list, because a list forgets who took a message. XREADGROUP hands each message to one consumer and records it in the group's pending entries list (PEL) until XACK. If a worker dies, its messages stay there with an idle time and a delivery counter, and XAUTOCLAIM lets a live worker take over anything idle past a threshold.

# ... STREAM, GROUP, enqueue(), dead_letter()

@dataclass
class Worker:
    redis: Redis
    name: str                       # unique per process, e.g. hostname-pid
    handler: Handler
    claim_idle_ms: int = 120_000    # must exceed the slowest successful handler run
    max_deliveries: int = 5
    batch: int = 10
    _cursor: str = "0-0"

async def run_once(self, block_ms: int = 1_000) -> int:
        # 1. Take over messages a dead (or stuck) consumer never acknowledged.
        self._cursor, msgs, _deleted = await self.redis.xautoclaim(
            STREAM, GROUP, self.name, min_idle_time=self.claim_idle_ms,
            start_id=self._cursor, count=self.batch)
        # 2. Otherwise read new messages.
        if not msgs:
            resp = await self.redis.xreadgroup(
                GROUP, self.name, {STREAM: ">"}, count=self.batch, block=block_ms)
            msgs = resp[0][1] if resp else []
        for msg_id, fields in msgs:
            await self._process(msg_id, fields)
        return len(msgs)

async def _process(self, msg_id: str, fields: dict) -> None:
        pending = await self.redis.xpending_range(STREAM, GROUP, msg_id, msg_id, 1)
        deliveries = pending[0]["times_delivered"] if pending else 1
        if deliveries > self.max_deliveries:
            await dead_letter(self.redis, msg_id, fields,
                              f"exceeded {self.max_deliveries} deliveries")
            return
        try:
            await self.handler(fields)
        except PermanentJobError as e:
            await dead_letter(self.redis, msg_id, fields, f"permanent: {e}")
            return
        except Exception:
            # Leave it pending. XAUTOCLAIM redelivers it after claim_idle_ms,
            # which doubles as a coarse backoff between delivery attempts.
            log.exception("job_failed", msg_id=msg_id, deliveries=deliveries)
            return
        await self.redis.xack(STREAM, GROUP, msg_id)

Why it is shaped like this:

• Delivery counts come from `XPENDING`. XREADGROUP sets times_delivered to 1 and each XAUTOCLAIM increments it; the dead-consumer test sees times_delivered == 2 on the survivor. The counter lives in Redis, so it survives the crashes an in-memory counter would forget.
• Poison messages go to a dead-letter stream. dead_letter runs XADD to jobs:summaries:dead and XACK in one MULTI/EXEC, so the two cannot be split. With max_deliveries=3, an always-failing handler runs three times and the fourth claim dead-letters it; the test asserts exactly that.
• `claim_idle_ms` must exceed the slowest successful run. Otherwise a slow but live worker has its message claimed from under it. The code stays correct, but you pay for the model call twice.
• Keep the `XAUTOCLAIM` cursor, or every call rescans from 0-0. And beware MAXLEN trimming: it can delete entries that are still pending.

A message can be processed, committed, and delivered again because the XACK never happened. Every handler must be idempotent; the queue cannot do that for you.

Retries that only retry what can succeed

Inside one delivery I retry the model call with tenacity, which absorbs transient failures in seconds rather than the minutes redelivery takes. Only errors a later attempt can fix are retried, and the classification is explicit:

RETRYABLE_STATUS = {408, 409, 429}   # plus every 5xx, same set the OpenAI SDK uses

def status_of(exc: BaseException) -> int | None:
    if isinstance(exc, openai.APIStatusError):
        return exc.status_code
    if isinstance(exc, httpx.HTTPStatusError):
        return exc.response.status_code
    return None

def is_retryable(exc: BaseException) -> bool:
    if isinstance(exc, DeadlineExceeded):
        return False
    if isinstance(exc, (TimeoutError, httpx.TransportError, openai.APIConnectionError)):
        return True          # per-call timeout, connection reset, DNS blip
    status = status_of(exc)
    if status is not None:
        return status in RETRYABLE_STATUS or status >= 500
    return False             # unknown means bug until proven otherwise

# ... Deadline, stop_before_deadline, wait_retry_after (covered below)

async def complete_with_retries(llm: LLM, prompt: str, *, deadline: Deadline,
                                per_call_timeout: float = 20.0, max_attempts: int = 4,
                                backoff: float = 0.5, max_backoff: float = 8.0,
                                min_attempt: float = 1.0) -> str:
    async def attempt() -> str:
        timeout = min(per_call_timeout, deadline.remaining())
        if timeout < min_attempt:
            raise DeadlineExceeded(f"{deadline.remaining():.2f}s left")
        async with asyncio.timeout(timeout):     # hard wall-clock cap per call
            return await llm.complete(prompt, timeout=timeout)

retrying = AsyncRetrying(
        retry=retry_if_exception(is_retryable),
        wait=wait_retry_after(wait_random_exponential(multiplier=backoff, max=max_backoff),
                              cap=max_backoff),
        stop=stop_after_attempt(max_attempts) | stop_before_deadline(deadline, min_attempt),
        reraise=True,
    )
    return await retrying(attempt)

The decisions behind it:

• Unknown exceptions are not retried. A ValueError from my own parsing will fail the same way four times. I widen the list only when logs show a real transient error.
• Full jitter. wait_random_exponential sleeps a random time up to an exponentially growing cap, so workers that hit the same 429 together do not retry in lockstep.
• Retry-After wins when present. wait_retry_after, a small wait_base subclass, reads the header from the exception's response and caps it; otherwise it falls back to jitter.
• Turn off the SDK's own retries. AsyncOpenAI in openai 3.19 defaults to max_retries=2; under four tenacity attempts that is up to twelve requests per delivery. One layer owns the policy:

# Production wiring (configuration only, not executed in the tests):
llm = OpenAIChat(openai.AsyncOpenAI(max_retries=0), model="gpt-4.1-mini")

I checked the classifier against the real SDK's exceptions without calling the API, using an AsyncOpenAI client over an httpx2.MockTransport (openai 3.19's HTTP stack). A 429 surfaces as openai.RateLimitError, is classified retryable, and succeeds on the second request; a 400 surfaces as openai.BadRequestError and the transport sees exactly one request.

In the job handler, classification decides what the queue does. A retryable error that outlives the in-process attempts is re-raised, so the message stays pending for redelivery. Anything else becomes PermanentJobError and goes straight to the dead-letter stream. Three layers, three time scales:

Timeouts and a deadline budget

The per-call timeout stops one hung request eating the whole budget. I enforce it with asyncio.timeout, not the HTTP client alone: httpx timeouts are per phase, and the read timeout applies to each read, so a server that drips a byte every few seconds never trips it. asyncio.timeout is wall-clock and raises TimeoutError, which is classified retryable.

The deadline is the total budget for one delivery across attempts and sleeps. Deadline wraps time.monotonic(), so clock changes cannot stretch it. It is enforced in three places:

• Each attempt's timeout is min(per_call_timeout, deadline.remaining()).
• A custom stop_before_deadline stops when the next sleep would leave less than one useful attempt. Tenacity computes the wait before evaluating stop conditions, so retry_state.upcoming_sleep is available.
• DeadlineExceeded is deliberately not a TimeoutError, so it is never retried in-process. The job handler re-raises it as transient; the message is redelivered with a fresh budget and max_deliveries caps the total.

In the tests, a model that always hangs, with per_call_timeout=0.2 and a 0.5 second deadline, fails in under 0.6 seconds after at most three calls.

Side effects through a transactional outbox

The job has two effects: store the summary and email the requester. Sending from the handler fails both ways: send then crash before commit, and redelivery sends a second email; commit then crash before send, and it is lost. So the handler sends nothing. It writes the summary and an outbox row in one transaction:

async def handle_summary_job(pool: AsyncConnectionPool, llm: LLM, fields: dict,
                             **retry_opts) -> str:
    job_id = fields["job_id"]
    payload = json.loads(fields["payload"])

# Cheap pre-check: a redelivered, already-finished job skips the LLM call.
    async with pool.connection() as conn:
        cur = await conn.execute("SELECT 1 FROM summaries WHERE job_id = %s", (job_id,))
        if await cur.fetchone():
            return "duplicate"

try:
        summary = await complete_with_retries(
            llm, f"Summarise for the account team:\n\n{payload['text']}",
            deadline=Deadline.after(JOB_BUDGET_S), **retry_opts)
    except Exception as e:
        if is_retryable(e) or isinstance(e, DeadlineExceeded):
            raise                      # transient: leave it pending for redelivery
        raise PermanentJobError(repr(e)) from e

# Result and side-effect intent commit together, or not at all.
    async with pool.connection() as conn, conn.transaction():
        cur = await conn.execute(
            "INSERT INTO summaries (job_id, summary) VALUES (%s, %s) "
            "ON CONFLICT (job_id) DO NOTHING RETURNING job_id", (job_id, summary))
        if await cur.fetchone() is None:
            return "duplicate"         # a concurrent delivery won the race
        await conn.execute(
            "INSERT INTO outbox (topic, dedupe_key, payload) VALUES (%s, %s, %s)",
            ("email.summary_ready", f"summary_ready:{job_id}",
             Jsonb({"job_id": job_id, "to": payload["notify"]})))
    return "done"

The pre-check only saves a model call on redelivery. The guarantee is the primary key on summaries.job_id plus ON CONFLICT DO NOTHING RETURNING: when two deliveries race, both pay for the model, only one inserts, and only that one writes an outbox row. outbox.dedupe_key is UNIQUE as a backstop.

A separate relay drains the outbox:

async def relay_once(pool: AsyncConnectionPool, send: Sender,
                     batch: int = 20, max_attempts: int = 10) -> int:
    sent = 0
    async with pool.connection() as conn, conn.transaction():
        cur = await conn.execute(
            """SELECT id, topic, dedupe_key, payload FROM outbox
               WHERE sent_at IS NULL AND attempts < %s
               ORDER BY id LIMIT %s
               FOR UPDATE SKIP LOCKED""", (max_attempts, batch))
        for row_id, topic, dedupe_key, payload in await cur.fetchall():
            try:
                # The provider dedupes on this key, which covers a crash
                # after send() but before our UPDATE commits.
                await send(topic, payload, dedupe_key)
            except Exception as e:
                log.warning("outbox_send_failed", id=row_id, error=repr(e))
                await conn.execute(
                    "UPDATE outbox SET attempts = attempts + 1, last_error = %s "
                    "WHERE id = %s", (repr(e), row_id))
                continue
            await conn.execute("UPDATE outbox SET sent_at = now() WHERE id = %s", (row_id,))
            sent += 1
    return sent

FOR UPDATE SKIP LOCKED lets several relays run without blocking each other or sending the same row twice. A failed send increments attempts and is retried on the next pass until max_attempts, after which, like the dead-letter stream, it needs a person.

The outbox does not make the provider call exactly-once. If the relay sends and dies before its UPDATE commits, the row goes out again. A test proves it: a sender that delivers and then raises a BaseException standing in for SIGKILL causes a second send, with the same dedupe_key. That is why I pass the key to the provider as its idempotency key; payment APIs such as Stripe accept one, so check whether your email provider does. If not, the duplicate window is the gap between send and commit, and I document it. Holding row locks across a network call is the other trade-off: fine with a short send timeout and small batches, and a lease column is the fix for slow providers.

Tests that crash on purpose

Here a crash is just stopping before a step, so crash tests are cheap. The one I care most about runs the full handler as worker A, commit included, and skips XACK:

async def test_crash_between_commit_and_xack_sends_one_email(r, pool, llm, mailer):
    await enqueue(r, JOB, PAYLOAD)
    # Worker A reads and fully processes the job (DB commit included)...
    [[_, [(msg_id, fields)]]] = await r.xreadgroup(GROUP, "w-a", {STREAM: ">"}, count=1)
    assert await handle_summary_job(pool, llm, fields) == "done"
    # ...then dies before XACK. The message is still pending against w-a.
    assert (await r.xpending(STREAM, GROUP))["pending"] == 1

b = Worker(r, "w-b", lambda f: handle_summary_job(pool, llm, f), claim_idle_ms=0)
    assert await b.run_once(block_ms=10) == 1          # reclaimed and redelivered
    assert (await r.xpending(STREAM, GROUP))["pending"] == 0

await relay_once(pool, mailer)
    await relay_once(pool, mailer)
    assert llm.calls == 1                               # pre-check skipped the model
    assert await counts(pool) == (1, 1)
    assert len(mailer.sent) == 1

Worker B reclaims the message, the handler finds the committed summary, and it is acknowledged: one model call, one summary, one outbox row, one email after two relay passes. The rest of the suite covers:

• The same job_id enqueued twice, as happens when the API crashes after XADD: one summary, one email.
• A slow worker and a reclaimer processing the same job concurrently: results ["done", "duplicate"] and two model calls, but one email.
• A consumer killed after XREADGROUP: the message is not claimable before claim_idle_ms, and afterwards it is claimed with times_delivered == 2.
• A job that keeps getting 429s: four tenacity attempts, then it stays pending and is not dead-lettered. A 400 is dead-lettered after one call.
• An email provider that is down: attempts goes to 1, and the next relay pass sends it.

The suite (31 tests against a local Redis 7.2 and a dedicated Postgres 16 database) runs in about three seconds on my laptop, which is cheap enough for every CI run. The same fake-model approach suits classification services like structured-output ticket triage with FastAPI, and the same retry and outbox layers belong under any tool call in a LangGraph support agent that writes to an external system.

Checklist

• Every POST that creates work accepts an Idempotency-Key, scoped by tenant, method and path, and fingerprints the body.
• The in-progress marker has a short TTL sized to the slowest normal request; final responses get a long TTL; 5xx releases the key.
• Finalise and release are compare-and-set on a per-request token (Lua on Redis 7.x).
• Concurrent duplicates get 409 with Retry-After, and key reuse with a different body gets 422.
• Job ids are derived from the idempotency key, never random, so duplicate enqueues collapse downstream.
• Workers use a consumer group, reclaim with XAUTOCLAIM, keep the claim cursor, and set claim_idle_ms above the slowest successful run.
• Delivery counts come from XPENDING. After N deliveries the message moves to a dead-letter stream with XADD and XACK in one MULTI/EXEC.
• Errors are classified explicitly: timeouts, connection errors, 408, 409, 429 and 5xx retry; everything else fails fast.
• tenacity uses jittered exponential backoff, honours Retry-After, stops on attempts or deadline, and SDK retries are zero.
• Every model call has a wall-clock asyncio.timeout, and every delivery has a monotonic deadline.
• Side effects are written to an outbox in the same transaction as the result, guarded by a unique key. The relay uses SKIP LOCKED and passes a provider idempotency key.
• CI includes tests that crash between commit and XACK and between send and commit, and they assert effect counts, not just status codes.

Before any of this goes live I put the model step behind shadow mode and eval gates. Reliable plumbing makes each answer's effects happen once; it says nothing about whether the answer is good.

Tested with Python 3.12.9, FastAPI 0.141.1, Pydantic 2.13.5, redis-py 8.1.0 against Redis 7.2.7, tenacity 9.1.4, psycopg 3.3.6, psycopg-pool 3.3.3, PostgreSQL 16.2, openai 3.19.2, httpx 0.28.1, pytest 9.1.1 on 2026-09-27.