Your first worker

You will run two Python files and a Redis broker. One endpoint generates a sample report directly. The other queues the same function so you can see the difference in response time.

The report takes 15 seconds and returns fixed sample data. You do not need an API key, a database, or an external service.

Get the example

The example belongs to the AI Cookbook, in patterns/background-tasks. Each example has its own dependencies, so you can work through this chapter on its own.

Use Python 3.12 or later, uv, and Docker with its engine running. Python commands below target macOS, Linux, or WSL2. Celery does not officially support native Windows.

# Skip cloning if you already have the cookbook.
git clone https://github.com/daveebbelaar/ai-cookbook.git
cd ai-cookbook/patterns/background-tasks

Define the task

The Celery app connects to Redis on local port 6380. The decorator registers the function so a worker can execute it when a task message arrives.

from time import sleep

from celery import Celery

celery_app = Celery("background_tasks", broker="redis://127.0.0.1:6380/0")
celery_app.conf.task_default_queue = "background-tasks"

@celery_app.task(ignore_result=True)
def generate_report():
    # Simulate a slow report without credentials or external services.
    sleep(15)
    return {"report": "Weekly sales", "rows": 42}

ignore_result=True skips storing the return value in a result backend. You will see the completed result in the worker's logs. The sleep simulates slow work; replace the function body with a real report in your own application.

Accept a request

Both endpoints use the same function. The first calls it directly. The second calls .delay() and returns the task ID with HTTP 202.

from fastapi import FastAPI
import uvicorn

from tasks import generate_report

app = FastAPI()

@app.post("/reports/blocking")
def blocking_report():
    return generate_report()

@app.post("/reports", status_code=202)
def queued_report():
    task = generate_report.delay()
    return {"task_id": task.id}

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

202 means accepted for processing. The task ID lets you match the request to its worker log entry. This example has no status endpoint or stored report to retrieve.

Start the processes

Open three terminals. Run the Python commands from patterns/background-tasks. The uv commands install only this example's requirements in a cached environment, avoiding the cookbook's other dependencies.

Terminal 1: Redis

docker run --rm --name cookbook-redis -p 127.0.0.1:6380:6379 redis:7-alpine
# Wait for "Ready to accept connections".

Port 6380 on your machine maps to Redis's port 6379 inside the container. The localhost binding keeps it accessible only from your machine. Leave this terminal running.

Terminal 2: worker

uv run --no-project --with-requirements requirements.txt celery -A tasks:celery_app worker --loglevel=INFO --pool=solo
# Wait for the worker's "ready" message.

tasks:celery_app identifies the Celery app in tasks.py. The solo pool executes one task at a time, which makes the logs easier to follow.

Terminal 3: API

uv run --no-project --with-requirements requirements.txt python 1-api.py
# Uvicorn running on http://127.0.0.1:8000

Open the local API docs. Use Try it out and Execute to send requests.

Compare the endpoints

Execute POST /reports/blocking. The response arrives after about 15 seconds and contains the sample report. The request waits, though other requests can still be served because FastAPI runs this synchronous handler in its thread pool.

Execute POST /reports. The response arrives after enqueueing, before the report finishes, and contains a task_id. Find that ID in the worker terminal. About 15 seconds later, the worker logs the completed report.

You can also compare them from a fourth terminal.

curl -sS -w '\nHTTP %{http_code}, time %{time_total}s\n' -X POST http://127.0.0.1:8000/reports/blocking
# HTTP 200 after about 15 seconds.
curl -sS -w '\nHTTP %{http_code}, time %{time_total}s\n' -X POST http://127.0.0.1:8000/reports
# HTTP 202 before the task finishes.

Pause the worker

Let the current task finish. Press Ctrl+C once in the worker terminal and wait for it to exit. Leave Redis and the API running, then send another request to POST /reports.

The API still returns a task ID. Start the worker again with the same command and watch it execute that waiting task.

Checkpoint

You have seen a request wait for its report, a queued request return before completion, and a task wait for a worker to start. You should be able to trace each step through the two Python files.

Stop the API and worker with Ctrl+C, then stop the Redis container with Ctrl+C. This discards any work still queued in the disposable container.