FastAPI background tasks for async LLM calls

FastAPI background tasks llm calls solve a common bottleneck: model inference can take seconds to minutes, and holding an HTTP connection open wastes worker capacity. This how-to builds a complete pattern for kicking off LLM work after the response returns, then exposing job status and results without blocking the caller. You will get runnable code and the operational caveats that matter in production.

Step 1: Scaffold the service and install deps

Create a virtual environment and install the minimal stack. We use FastAPI, Uvicorn, and httpx for async outbound requests. Avoid synchronous HTTP clients; they will block the event loop and negate the benefits of async.

mkdir fastapi-llm-bg && cd fastapi-llm-bg
python -m venv .venv && source .venv/bin/activate
pip install fastapi uvicorn httpx pydantic

Structure the project as a single module for clarity. In main.py we will define the app, models, job store, and routes. For anything beyond a demo, split these into routers/, workers/, and db/ packages so the background logic is testable in isolation.

Step 2: Define the async LLM client

The core call must be async and timeout-aware. We use httpx.AsyncClient with a base URL pointing at an OpenAI-compatible server. If you want provider redundancy, point the client at a gateway like n4n.ai, which exposes one OpenAI-compatible endpoint across 240+ models and fails over automatically when a provider is degraded. This removes the need to write your own retry-and-switch logic for every model vendor.

import httpx
from pydantic import BaseModel

LLM_BASE_URL = "https://api.n4n.ai/v1"  # OpenAI-compatible
API_KEY = "sk-your-key"  # read from env in real code

async def call_llm(messages: list[dict], model: str) -> str:
    async with httpx.AsyncClient(timeout=httpx.Timeout(120.0)) as client:
        resp = await client.post(
            f"{LLM_BASE_URL}/chat/completions",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={"model": model, "messages": messages, "temperature": 0.2},
        )
        resp.raise_for_status()
        data = resp.json()
        return data["choices"][0]["message"]["content"]

Never call requests here. It blocks the event loop and destroys the concurrency FastAPI gives you. Set a timeout explicitly; the default httpx timeout is 5 seconds, which is too short for LLM completions.

Step 3: Model the job and an in-memory store

We need a status enum and a container for job state. For production, replace the dict with Redis or Postgres, but the interface stays identical so the rest of the code does not change.

from enum import Enum
from typing import Optional
from datetime import datetime, timezone
import uuid

class JobStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    DONE = "done"
    FAILED = "failed"

class Job(BaseModel):
    id: str
    status: JobStatus = JobStatus.PENDING
    result: Optional[str] = None
    error: Optional[str] = None
    created_at: datetime = datetime.now(timezone.utc)

JOBS: dict[str, Job] = {}

Generate IDs with uuid.uuid4().hex. The store is a module-level dict; note that it is not shared across Uvicorn workers unless you run a single worker or use an external store. This limitation is acceptable for fire-and-forget tasks but not for durable jobs.

Step 4: Write the background worker function

This is the function FastAPI will run after the response is sent. It updates job state, calls the LLM, and captures errors. We add a simple retry loop because LLM endpoints occasionally return 429 or 503 under load.

import asyncio

async def run_llm_job(job_id: str, messages: list[dict], model: str) -> None:
    job = JOBS[job_id]
    job.status = JobStatus.RUNNING
    last_err = None
    for attempt in range(3):
        try:
            job.result = await call_llm(messages, model)
            job.status = JobStatus.DONE
            return
        except Exception as e:
            last_err = str(e)
            await asyncio.sleep(2 ** attempt)  # exponential backoff
    job.status = JobStatus.FAILED
    job.error = last_err

The retry uses asyncio.sleep, which yields control instead of blocking the thread. Three attempts with backoff handles transient provider hiccups without turning a single request into a multi-minute hang.

Step 5: Expose the submission endpoint

FastAPI’s BackgroundTasks accepts a coroutine function and arguments. The endpoint returns immediately with a job ID; the work continues after the 202 response.

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()

class SubmitRequest(BaseModel):
    messages: list[dict]
    model: str = "gpt-4o-mini"

@app.post("/jobs", status_code=202)
async def submit_job(req: SubmitRequest, bg: BackgroundTasks):
    job_id = uuid.uuid4().hex
    JOBS[job_id] = Job(id=job_id)
    bg.add_task(run_llm_job, job_id, req.messages, req.model)
    return {"job_id": job_id, "status": "pending"}

This is the canonical fastapi background tasks llm calls pattern: the HTTP handler does not await the model. The 202 status code signals to clients that the request was accepted, not completed.

Step 6: Add status and result routes

Clients poll for completion. Keep these reads cheap and side-effect free.

from fastapi import HTTPException

@app.get("/jobs/{job_id}")
async def get_job(job_id: str):
    job = JOBS.get(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="job not found")
    return job

A real deployment should set Cache-Control: no-store and add auth, but the shape is correct. If you return the full result only when status == DONE, you avoid leaking partial state.

Step 7: Run with a single worker for local dev

Uvicorn multi-worker mode duplicates the in-memory dict. For local verification, pin to one worker.

uvicorn main:app --workers 1 --port 8000

If you later move the store to Redis, scale workers freely. Do not skip this step during testing—running two workers against the dict store will make half your poll requests return 404.

Step 8: Verify the end-to-end flow

Use curl to submit a job, then poll until done.

curl -X POST http://localhost:8000/jobs \
  -H "Content-Type: application/json" \
  -d '{"messages":[{"role":"user","content":"Summarize: FastAPI background tasks."}]}' \
  -i

The response includes a job_id. Hit the GET endpoint:

curl http://localhost:8000/jobs/<job_id>

You should see status: pending or running initially, then done with result populated. If you kill the LLM endpoint, the job transitions to failed after three retries, and error contains the exception string. That confirms the fastapi background tasks llm calls pipeline handles degradation without hanging the API.

Step 9: Add a pytest suite

Treat the worker as a pure async function. Mock call_llm with pytest.asyncio and assert state transitions.

import pytest
from main import run_llm_job, JOBS, Job

@pytest.mark.asyncio
async def test_job_success(monkeypatch):
    async def fake_llm(messages, model):
        return "ok"
    monkeypatch.setattr("main.call_llm", fake_llm)
    job_id = "test1"
    JOBS[job_id] = Job(id=job_id)
    await run_llm_job(job_id, [{"role": "user", "content": "hi"}], "m")
    assert JOBS[job_id].status == "done"
    assert JOBS[job_id].result == "ok"

This catches regressions in retry logic and ensures the background path behaves under test, not just in manual curls.

Step 10: Production hardening notes

Background tasks in FastAPI are tied to the process. If the process dies mid-job, the task is lost. For guaranteed execution, use a proper queue (Celery, ARQ, or SQS) and keep FastAPI’s BackgroundTasks for fire-and-forget work that is safe to drop. Also, stream logs with job IDs so you can trace a request from submission to completion.

When calling the model, forward provider cache-control hints if your gateway supports them. n4n.ai honors client routing directives and forwards cache-control, which can cut latency on repeated prefixes. Set extra_headers in the httpx call accordingly.

Finally, meter usage. Per-token accounting lets you attribute cost to the triggering user. If your gateway returns usage in the response, persist data["usage"] on the job row. This turns an opaque background task into a billable, observable unit.

Why this pattern wins

FastAPI background tasks llm calls keep your request latency independent of model speed. The user gets a 202 in milliseconds; the heavy lifting happens off the critical path. You avoid WebSocket complexity for simple cases and keep the door open to upgrade to a durable queue later. The code above is the minimum viable backbone—extend the store, add auth, and wire retries to your SLOs.