A FastAPI chat queue: enqueue, poll, dequeue

With the queue running in Valkey, WhatsApp was the easy half: the webhook enqueues a job, and the worker sends the reply itself when it is done. The hard half was the chat bubble on Crumb’s website, where customers ask questions while choosing a cake. A browser cannot receive a WhatsApp message. It has to ask for its answer.

Arjun’s first version took twenty minutes. The /chat endpoint put the question on the queue, then waited for the result before replying:

@app.post("/chat")
def chat(req: ChatRequest):
    job = chat_queue.enqueue(answer_question, req.question, "customer")
    while not (job.is_finished or job.is_failed):    # wait for the worker...
        time.sleep(0.5)
        job.refresh()                                 # re-read status from Valkey
    return {"answer": job.return_value()}

It worked on his laptop. It even looked clever: the work happened in a worker, so surely this was “async”?

Priya read it and asked one question. “During the Diwali broadcast, with 200 customers chatting at once, how many of these requests are open at the same time, and for how long?”

Arjun did the arithmetic out loud. “Each one stays open until its job finishes… so up to a couple of minutes if the queue is long. Two hundred open connections, each holding a server thread that’s doing nothing but sleeping.” He stopped. “It’s the same problem as before. I moved the cooking to the kitchen, but the cashier is still standing at the counter, holding the customer’s hand until the food arrives.”

Why waiting inside the endpoint defeats the queue

“Where should the waiting happen?” Priya asked.

“Not on the server,” Arjun said. “The server should hand over the token and move on. The customer waits. Their browser can check back.”

“Like at Vaishali. They don’t make you stand at the till while your dosa cooks. They give you a token, and you look up at the screen every so often.”

That is the whole design. It is sometimes called the request-reply pattern with polling: one request to start the work, then repeated cheap requests to check on it. Three new words:

• An endpoint is one URL plus HTTP method that your API answers, such as POST /chat or GET /chat/{job_id}.
• An HTTP status code is the three-digit number at the start of every web response saying how it went. You know 200 OK and 404 Not Found. This lesson adds 202 Accepted: “I have accepted your request and will process it, but it is not done yet.” It is the honest status for “here is your token”.
• Polling means the client asks the same question (“is it ready?”) repeatedly, at intervals, until the answer is yes.

The worker side has a word too. Taking a job off the queue is called dequeuing. RQ workers dequeue with a blocking pop: they wait on Valkey until a job appears, using no CPU while idle, and wake within milliseconds when one arrives. You never write that loop yourself; it is what rq worker does.

Where the Vaishali analogy breaks: the token screen in a restaurant is shown to everyone, and nobody minds seeing that order 47 is ready. In a chat API, the “screen” shows the answer, which may be about a customer’s order or allergies. So the status endpoint must check that the person asking about job 47 is the person who created it. You will build that check below.

Following one question through the API

Priya had Arjun trace one real exchange from the website chat, with the actual JSON at every step. Mrs Deshpande, choosing a birthday cake at 7:02 p.m. during the broadcast rush, typed: “Does the Belgian chocolate cake contain nuts?”

0.000 s. Her browser sends:

POST /chat
{"question": "Does the Belgian chocolate cake contain nuts?", "session_id": "s-7d21e0"}

0.018 s. The API enqueues a job and replies, 18 milliseconds later:

HTTP/1.1 202 Accepted
Location: /chat/4be1c9d2-58a3-4c0e-9f4b-2b8f6d1a7e55
{"job_id": "4be1c9d2-58a3-4c0e-9f4b-2b8f6d1a7e55", "status": "queued",
 "status_url": "/chat/4be1c9d2-58a3-4c0e-9f4b-2b8f6d1a7e55"}

The chat bubble shows “Checking…” straight away. The Location header is the standard place to say “the result will be at this address”.

Then the browser polls, starting after half a second and waiting 1.5 times longer before each new check, up to a maximum of 3 seconds:

Time	Response from GET /chat/4be1c9d2...	What it means
0.50 s	{"status": "queued", "position": 3}	Three jobs ahead of hers
1.25 s	{"status": "queued", "position": 0}	Next in line
2.38 s	{"status": "started", "stage": "checking the question"}	A worker dequeued it at about 1.9 s
4.06 s	{"status": "started", "stage": "writing the answer"}	Search and reranking done; Claude is answering
6.59 s	{"status": "finished", "answer": {...}}	The worker finished at about 5.8 s

The final response carried the answer: “Yes. The Belgian chocolate cake contains hazelnut praline in the filling [1]. If you need a nut-free option, the Egg-free Chocolate Truffle is made without nuts [2].”, with its two sources.

Five polls for one answer. Each poll is a tiny request: the API reads one small hash from Valkey (well under a millisecond) and replies. And notice the stage field. The chat bubble changed from “Checking…” to “Searching our handbook…” to “Writing your answer…”. Customers will wait six seconds happily if they can see progress. They will not wait six seconds staring at a spinner.

Building the API, one piece at a time

You need Valkey running (previous lesson), plus fastapi, rq and redis installed.

Step 1: the worker task, reporting its progress.

The task is the guarded RAG pipeline from Module 09’s project, split into its stages so it can report each one. RQ lets a running job write small notes to its own meta dictionary, which anyone can read.

# rag_tasks.py
from rq import get_current_job
from pipeline import check_scope, hybrid, rerank, finish_answer, OUT_OF_SCOPE

def set_stage(stage: str) -> None:
    job = get_current_job()            # the job this worker is running right now
    if job:
        job.meta["stage"] = stage
        job.save_meta()                # write only the meta back to Valkey

def answer_question(question: str, role: str) -> dict:
    set_stage("checking the question")
    if not check_scope(question).in_scope:
        return {"answer": OUT_OF_SCOPE, "sources": [], "abstained": True}
    set_stage("searching the handbook")
    ranked = rerank(question, hybrid(question, role))
    set_stage("writing the answer")
    return finish_answer(question, ranked)   # generate, quote check, PII scrub

get_current_job() returns the job being executed (or None if the function is called outside a worker, for example in a test). save_meta() writes only the meta field, so it is cheap to call several times.

Step 2: the submit endpoint.

# main.py   (uv add fastapi rq redis)
import os
from fastapi import FastAPI, HTTPException, Response
from pydantic import BaseModel
from redis import Redis
from rq import Queue
from rq.exceptions import NoSuchJobError
from rq.job import Job

from rag_tasks import answer_question

redis = Redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379"))
chat_queue = Queue("chat", connection=redis)
app = FastAPI()

class ChatRequest(BaseModel):
    question: str
    session_id: str

@app.post("/chat", status_code=202)
def submit(req: ChatRequest, response: Response):
    job = chat_queue.enqueue(
        answer_question, req.question, "customer",
        job_timeout=60,          # a single answer should never take a minute
        result_ttl=600,          # keep answers for 10 minutes
        failure_ttl=86400,       # keep failures for a day, for debugging
        meta={"session_id": req.session_id, "stage": "queued"},
    )
    response.headers["Location"] = f"/chat/{job.id}"
    return {"job_id": job.id, "status": "queued", "status_url": f"/chat/{job.id}"}

Line by line, the new parts: status_code=202 makes every successful response say “Accepted”. enqueue returns immediately with a Job whose id is a random UUID (a 36-character identifier that is, for practical purposes, impossible to guess). The meta dictionary stores which chat session the job belongs to, which the status endpoint will check. The three time limits are your cleanup policy: answers vanish after 10 minutes, failures after a day.

Start it with uv run fastapi dev main.py, start a worker with uv run python worker.py (the preloading worker script from the previous lesson), and post a question with any HTTP tool. You should get a 202 response with a job ID in about 20 milliseconds, however busy the worker is.

Step 3: the status endpoint.

@app.get("/chat/{job_id}")
def status(job_id: str, session_id: str):
    try:
        job = Job.fetch(job_id, connection=redis)
    except NoSuchJobError:
        raise HTTPException(status_code=404, detail="Unknown or expired job")
    if job.meta.get("session_id") != session_id:
        raise HTTPException(status_code=404, detail="Unknown or expired job")

    state = job.get_status().value          # "queued", "started", "finished", "failed", ...
    body = {"job_id": job.id, "status": state, "stage": job.meta.get("stage")}
    if state == "queued":
        body["position"] = job.get_position()       # how many jobs are ahead
    elif state == "finished":
        body["answer"] = job.return_value()
    elif state == "failed":
        body["error"] = "Sorry, something went wrong. Please ask again."
    return body

The interesting choices:

• Unknown and expired jobs both get 404. After result_ttl runs out, Valkey deletes the job, so Job.fetch raises NoSuchJobError. To the client, “never existed” and “expired” mean the same thing: ask again.
• The wrong session also gets 404, not 403 (“forbidden”). Saying “forbidden” would confirm that the job exists, which leaks information. Saying “not found” reveals nothing.
• Failures get a friendly message, not the error. The real traceback stays in RQ’s failed registry for you to read. Customers should never see “Chroma index not found” or a stack trace.

Fetch the status URL a few times while the worker runs. You should see queued, then started with changing stages, then finished with the answer.

Step 4: a polling client with backoff.

The website does this in JavaScript; here it is in Python with HTTPX, which is handy for testing and for other services that call your API.

# client.py   (uv add httpx)
import time
import httpx

BASE = "http://127.0.0.1:8000"
SESSION = "s-7d21e0"

def ask_crumb(question: str, max_wait: float = 90.0) -> dict:
    r = httpx.post(f"{BASE}/chat", json={"question": question, "session_id": SESSION})
    job_id = r.json()["job_id"]
    delay, waited = 0.5, 0.0
    while waited < max_wait:
        time.sleep(delay)
        waited += delay
        s = httpx.get(f"{BASE}/chat/{job_id}", params={"session_id": SESSION}).json()
        print(f"{waited:5.2f}s  {s['status']:9}  {s.get('stage')}  {s.get('position', '')}")
        if s["status"] in ("finished", "failed"):
            return s
        delay = min(delay * 1.5, 3.0)          # back off: 0.5, 0.75, 1.1, 1.7, 2.5, 3.0 ...
    return {"status": "timeout"}

print(ask_crumb("Does the Belgian chocolate cake contain nuts?")["answer"])

You should see a short log of polls, much like the table in the trace, and then the answer. Note the max_wait: a client must always give up eventually, or a lost job means a chat bubble that spins forever.

Library APIs change quickly. If a name has moved, check the docs linked below.

The double-click problem

Two days after launch, Arjun found pairs of identical jobs in the logs, a second or so apart. Customers on slow mobile connections were tapping “Send” twice. Each tap created a job, and each job made two paid Claude calls.

The fix is called idempotency: making an operation safe to repeat, so doing it twice has the same effect as doing it once. The standard approach is an idempotency key: the client generates a unique ID for each message it intends to send, sends it with every retry, and the server uses it to recognise repeats.

RQ makes this neat, because you may choose the job ID yourself:

class ChatRequest(BaseModel):
    question: str
    session_id: str
    message_id: str                      # generated once per message by the client

@app.post("/chat", status_code=202)
def submit(req: ChatRequest, response: Response):
    job_id = f"{req.session_id}-{req.message_id}"        # RQ IDs: letters, digits, _ and - only
    if not Job.exists(job_id, connection=redis):          # first time we've seen it
        chat_queue.enqueue(answer_question, req.question, "customer",
                           job_id=job_id, job_timeout=60, result_ttl=600,
                           meta={"session_id": req.session_id, "stage": "queued"})
    response.headers["Location"] = f"/chat/{job_id}"
    return {"job_id": job_id, "status_url": f"/chat/{job_id}"}

The second tap now finds the existing job and returns its ID. One detail that bites: RQ only accepts job IDs made of letters, digits, underscores and dashes, so build the key with dashes (a colon is rejected with an error). The WhatsApp webhook uses exactly the same trick with the provider’s message ID, which is what finally ended the duplicate replies from the first Saturday. (Two taps arriving in the very same millisecond could still slip through this check; newer RQ versions also offer a built-in uniqueness option for enqueuing, and the docs describe it.)

Where it breaks

• Results that expire before collection. If a customer locks their phone and comes back 15 minutes later, the 10-minute result_ttl has deleted the answer and the status endpoint says 404. Either keep results longer, or have the client treat 404 as “please ask again” gracefully. Crumb stores the final answer in the chat history table too, so it is never lost.
• Jobs stuck in “started”. If a worker is killed mid-job, the job looks “started” until RQ notices the worker is gone and moves the job to the failed registry. Clients must have a max_wait and show a friendly retry message.
• Errors leaking through. Returning job.exc_info straight to the browser exposes internals. Log it; show a kind sentence.
• A queue that only grows. If jobs arrive faster than workers finish them, position keeps rising and every customer waits longer. The API should refuse politely when the queue is too deep, rather than accept work it cannot do in time. That is the next lesson.

Polling is not the only way

Polling is simple, works through every firewall and proxy, and is easy to debug, which is why it is the right first version. Two alternatives are worth knowing:

• Server-sent events (SSE): the browser opens one long-lived connection, and the server pushes status updates down it as they happen. Fewer requests, instant updates, and it pairs naturally with streaming answers (Module 19). The trade-off is long-lived connections to manage.
• WebSockets: a two-way connection, useful when the chat is highly interactive. More complex to run.
• Push to a channel: what Crumb already does for WhatsApp. The worker sends the result to wherever the customer is (WhatsApp, email, a mobile push notification), so nobody polls at all.

The queue and the workers stay exactly the same in all of these. Only the last step, how the answer reaches the customer, changes.

Try it yourself

Run the full system (Valkey, the API and one worker), then:

1. Use client.py to ask one question and compare its printed poll log to the trace table.
2. Run 10 copies of client.py at once (for example, in a loop that starts them as background processes) and watch the position values.
3. Stop the worker for 20 seconds while a question is queued, then start it again.
4. Call the status endpoint with the right job ID but the wrong session_id.

Show the answer

1. You should see a few queued lines, a couple of started lines with changing stages, then finished, with the answer about 3 to 7 seconds after posting.
2. With one worker, the clients that posted last see positions like 8 or 9 at first, counting down as earlier jobs finish. Total time for all 10 is roughly 10 times one answer.
3. The queued question just waits: its status stays queued and its position stays put. When the worker returns, it dequeues the job and answers it. Nothing was lost. (If you stop it for longer than the client’s max_wait, the client gives up, but the job still completes and its result is still available until result_ttl expires.)
4. You get a 404 with “Unknown or expired job”, exactly as for a job that does not exist.

Common mistakes

• Waiting for the result inside the POST endpoint. It brings back every problem of the synchronous design. Return 202 and a job ID immediately.
• Polling too fast, with no backoff and no limit. Start around half a second, grow the interval, cap it, and always give up eventually.
• Treating a job ID as permission. Check that the requester owns the job.
• Returning raw errors to users. Keep tracebacks in logs and the failed registry.
• Ignoring double submissions. Use an idempotency key as the job ID.
• Forgetting that results expire. Decide result_ttl on purpose and handle 404 kindly in the client.

Recap

Meera tried the new website chat on her phone in the café downstairs, then asked Arjun to explain why it now said “Searching our handbook…” instead of freezing. He kept it to the token system.

• When you send a question, the website gets a token number back instantly. The question goes on the board for the workers.
• Your phone then checks every second or two, “is token 4be1 ready?”, and the server tells it where it is: waiting, being worked on, or done.
• Each check is tiny, and phones check less often the longer they wait, so a busy evening doesn’t flood the server.
• Only the person who asked can see the answer, even if someone else gets hold of the token number.
• If someone taps “Send” twice, we recognise it’s the same message and answer once.

“And WhatsApp?” Meera asked.

“WhatsApp doesn’t need to check. The worker sends the reply itself.”

She nodded, then looked at the calendar on the wall. Diwali was eleven days away. “Marketing wants to send the big broadcast to all 18,000 customers on Diwali eve. That’s when everybody asks at once. How many of these workers do we need, and what happens if we get it wrong?”

Go deeper

• FastAPI documentation: https://fastapi.tiangolo.com
• RQ documentation (jobs, results and job metadata): https://python-rq.org
• HTTPX documentation: https://www.python-httpx.org