Proposal: Pluggable Background Task Runners for FastAPI (free-threaded Python 3.14 aware; optional Celery-like backend)

[mahimairaja]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from typing import Protocol, Callable, Any, Optional
import sys
from fastapi import FastAPI, BackgroundTasks

# 1) A minimal protocol for task runners
class TaskRunner(Protocol):
    def enqueue(
        self,
        func: Callable[..., Any],
        *args: Any,
        retries: int = 0,
        timeout_s: Optional[float] = None,
        **kwargs: Any,
    ) -> None: ...

# 2) Built-in runners (examples)
from fastapi.task_runners import ThreadPoolRunner, ProcessPoolRunner  # <--- new module
# Optional external adapter, shipped in a contrib package:
# from fastapi_contrib.celery import CeleryRunner

# At app startup, pick a sensible default:
# If GIL is disabled (free-threaded CPython), prefer threads; otherwise, processes.
default_runner = ThreadPoolRunner(max_workers=0) if hasattr(sys, "_is_gil_enabled") and not sys._is_gil_enabled() \
                 else ProcessPoolRunner(max_workers=0)

app = FastAPI(task_runner=default_runner)  # new parameter

def cpu_heavy(n: int) -> int:
    # expensive, CPU-bound code
    return sum(i * i for i in range(n))

@app.post("/compute/{n}")
async def compute(n: int, bg: BackgroundTasks):  # same BackgroundTasks API
    # New: per-task overrides (runner selection, retries, timeout)
    bg.add_task(cpu_heavy, n, retries=3, timeout_s=60, runner="thread")
    return {"status": "accepted"}

# 3) Optional: concurrent execution of multiple tasks attached to one response
#    (opt-in flag, keeps today’s sequential default)
app = FastAPI(task_runner=default_runner, background_tasks_concurrent=True)

Description

Motivation

Python 3.14 introduces an officially supported free-threaded build (no GIL) which enables true parallelism across threads for pure-Python CPU-bound workloads when that build is used. Meanwhile, FastAPI’s BackgroundTasks (via Starlette) are in-process, sequential, and non-durable: they run only after the response, execute in order, and if one raises, the rest don’t run. The docs recommend Celery/RQ/etc. for heavier use cases. With more teams experimenting with the free-threaded interpreter, there’s demand for a path that (a) keeps FastAPI’s ergonomics, (b) allows concurrent post-response work safely, and (c) offers an official integration point for durable, multi-process/cluster workers without forcing a specific queue.

Proposal.

Introduce a small, optional abstraction:

1. A TaskRunner protocol (enqueue API) and a new fastapi.task_runners module with two built-ins:

• ThreadPoolRunner: good default when running a free-threaded interpreter.
• ProcessPoolRunner: safer default for CPU-bound work on the default (with-GIL) interpreter.

2. App-level configuration: FastAPI(task_runner=..., background_tasks_concurrent: bool=False)

• Keeps today’s behavior by default (sequential).
• If background_tasks_concurrent=True, multiple tasks attached to one response are executed concurrently (e.g., via AnyIO TaskGroup), preserving ordering only when requested.

3. Per-task overrides on BackgroundTasks.add_task(...): runner="thread"|"process"|"external", plus optional retries, timeout_s, on_error callback hook.

• The external choice delegates to a pluggable adapter (e.g., CeleryRunner in a fastapi-contrib package), giving users a first-class, docs-blessed way to push tasks to Celery/RQ/Huey/Arq with unified semantics, while keeping FastAPI core small.

Why in FastAPI (not only Starlette)?

The Starlette docs define background tasks as in-process and sequential; that’s perfect for the framework layer. FastAPI can provide developer-experience sugar (config + protocol) and official adapters/docs, while leaving the underlying execution primitives to Starlette/AnyIO.

Backwards compatibility.

No breaking changes: current BackgroundTasks keeps today’s sequential in-process semantics unless the new flags are used

Operating System

Linux, Windows, macOS

Operating System Details

Need to check for each OS

FastAPI Version

Pydantic Version

Python Version

v3.14

Additional Context

1. Python 3.14 Official: https://docs.python.org/3.14/whatsnew/3.14.html

2. Blog on Python 3.14: https://medium.com/towards-artificial-intelligence/python-3-14-6a94d2b5af3f?sk=0a7891cbe60d80724a76f02e1b78d3ca

3. FastAPI Background Tasks docs: https://fastapi.tiangolo.com/tutorial/background-tasks/

4. Starlette Background Tasks: https://www.starlette.dev/background/

5. Starlette threadpool behavior: https://www.starlette.dev/threadpool/

6. sys._is_gil_enabled() — added in 3.14 to detect whether the GIL is active: https://docs.python.org/3.14/whatsnew/changelog.html