If you have been following this series, you have likely noticed a glaring issue as your project grows: routing every endpoint through a single main.py file creates an unmaintainable, thousands-of-lines-long bottleneck. Debugging becomes a nightmare.
If you're new to the series check it out here:
To build production-grade applications, we need a modular architecture. Here is how to scale your FastAPI backend using APIRouter and handle slow AI processing with BackgroundTasks.
1. APIRouters: Organizing Your Monolith
Think of APIRouter as a mini-FastAPI instance. It allows you to split your endpoints across multiple files based on their domain, while main.py acts purely as the central traffic hub.
Imagine a Gen-AI Agent Orchestrator with three core tables: Users, Agents, and Tasks. Instead of dumping all CRUD operations into the main file, we create a routers/ directory with dedicated files: auth.py, agents.py, and tasks.py.
Inside tasks.py, we replace @app.get with @router.get:
from fastapi import APIRouter
router = APIRouter(prefix="/tasks", tags=["Tasks"])
@router.get("/")
def get_all_tasks():
return {"message": "Returns all tasks"}
We do the same with agents.py and auth.py.With your endpoints modularized, your main.py reduces to a clean, readable entry point:
from fastapi import FastAPI
from .database import engine
from . import models
from .routers import auth, agents, tasks
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
# Plug the modular routers into the main app
app.include_router(auth.router)
app.include_router(agents.router)
app.include_router(tasks.router)
This is how organized the endpoints look like in /docs after using APIRouter
2. Background Tasks: Beating AI Latency
When building AI applications, latency is your biggest enemy. If a user submits a prompt, sending that query to an LLM might take 10 to 15 seconds. If you process this synchronously inside your endpoint, the user's browser will hang and eventually throw a timeout error.
FastAPI’s BackgroundTasks solves this by decoupling the HTTP response from the heavy computation. You instantly return a "status: queued" message to keep the frontend happy, while the AI processing runs silently in the background.
First, define a standard Python function for the heavy lifting. Then, inject BackgroundTasks into your endpoint:
from fastapi import APIRouter, Depends, BackgroundTasks, HTTPException
from sqlalchemy.orm import Session
import time
from .. import models, schemas
from ..database import get_db, sessionL
from .auth import get_current_user
router = APIRouter(
tags=["Tasks"]
)
# 2. The Background Function
# CRITICAL: Background tasks run independently, so they need their own fresh Database Session!
def simulate_ai_agent(task_id: int):
db = sessionL() # Open a new database connection
try:
print(f"[BACKGROUND] Agent starting task {task_id}...", flush=True)
time.sleep(10) # Simulate a slow LLM generating text
# Update the task status to completed
task = db.query(models.Tasks).filter(models.Tasks.id == task_id).first()
if task:
task.status = "completed"
db.commit()
print(f"[BACKGROUND] Task {task_id} marked as completed!", flush=True)
else:
print(f"[BACKGROUND] Task {task_id} was not found.", flush=True)
except Exception as error:
print(f"[BACKGROUND] Task {task_id} failed: {error!r}", flush=True)
raise
finally:
db.close() # Always close the background session
@router.post('/agents/{agent_id}/tasks', response_model=schemas.TaskResponse)
def create_task(agent_id:int, task:schemas.TaskCreate, background_tasks:BackgroundTasks, current_user:dict=Depends(get_current_user), db:Session=Depends(get_db)):
existing_agent=db.query(models.Agents).filter(models.Agents.id==agent_id).first()
if not existing_agent:
raise HTTPException(status_code=404, detail='Agent not found.')
if existing_agent.owner_id!=current_user['id']:
raise HTTPException(status_code=403, detail='Adding tasks to this agent is forbidden.')
new_task=models.Tasks(
description=task.description,
status='queued',
agent_id=agent_id
)
db.add(new_task)
db.commit()
db.refresh(new_task)
background_tasks.add_task(simulate_ai_agent, new_task.id)
print(f"[ROUTE] Task {new_task.id} queued for background processing.", flush=True)
return new_task
@router.get('/tasks/{task_id}', response_model=schemas.TaskResponse)
def get_task(task_id: int, current_user: dict = Depends(get_current_user), db: Session = Depends(get_db)):
existing_task = db.query(models.Tasks).filter(models.Tasks.id == task_id).first()
if not existing_task:
raise HTTPException(status_code=404, detail='Task not found.')
existing_agent = db.query(models.Agents).filter(models.Agents.id == existing_task.agent_id).first()
if not existing_agent or existing_agent.owner_id != current_user['id']:
raise HTTPException(status_code=403, detail='Forbidden to view this task.')
return existing_task
The task is created as shown below with status='queued
The terminal message shows the exact working of the background tasks. As you can see from the code, the background task of simulate_ai_agent is inside the create_task endpoint, yet the create_task endpoint finishes first by printing [ROUTE] Task task_id is queued for background processing and even executes the addition of the new task with agent. Only after new_task is returned the simulate_ai_agent function running in the background starts and returns its response.
After 10 seconds, when you check the task status, this is what you get:
When you write background_tasks.add_task(...) inside your create_task endpoint, you aren't running the function right then. You are basically handing FastAPI a sticky note that says, "As soon as you finish sending the HTTP response to the user, run this heavy function."
This is the exact sequence:
1. create_task registers the sticky note.
2. create_task hits return {"message": "Queued", ...}.
3. FastAPI sends the 200 OK response to the user (their screen instantly updates).
4. FastAPI looks at the sticky note and begins running the background function.
This is exactly why it is the perfect tool for AI APIs. The user never has to wait 15 seconds staring at a loading spinner for an LLM to generate a response; you give them their confirmation instantly, and the AI does its heavy lifting behind the scenes!
With APIRouter keeping your codebase modular and BackgroundTasks handling your heavy LLM compute, you now have the foundational architecture for a scalable AI backend. This wraps up our FastAPI series—you are now ready to start integrating complex AI orchestration into your endpoints.
With this, we reach the end of the series. Hope you guys enjoyed it and learnt a lot. Do share in the comments down below, what errors you faced while building endpoints or if you tried any of the exercises in my posts and found them helpful.
Stay tuned for upcoming posts!! 😊👍
Top comments (0)