FastAPI Production Backend

pip install fastapi uvicorn

from fastapi import FastAPI
import uvicorn

# Initialize the app
app = FastAPI(
    title="GenAI Bootcamp API",
    description="A production-ready backend for our AI applications",
    version="1.0.0"
)

# Define a "route" or "endpoint"
# This is accessible via HTTP GET at the root "/"
@app.get("/")
async def root():
    return {"message": "System Online", "status": "OK"}

# This block allows us to run the script directly with python backend_v1.py
if __name__ == "__main__":
    # host="0.0.0.0" makes it accessible on the local network
    # port=8000 is standard for API development
    uvicorn.run(app, host="0.0.0.0", port=8000)

Run this script. You will see logs indicating the server started.

FastAPI automatically generates documentation for you.

Step 2: Structured Data with Pydantic

Passing data around as loose dictionaries or raw strings is dangerous. What if the user forgets to send the prompt? What if they send a list instead of a string?

We use Pydantic to define the "shape" of our data. If the incoming data doesn't match the shape, FastAPI rejects it automatically with a clear error message.

from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn
from datetime import datetime

app = FastAPI()

# 1. Define the Input Schema
# The user MUST send a JSON object with a "prompt" (string)
# "model_name" is optional, defaults to "gpt-3.5"
class ChatRequest(BaseModel):
    prompt: str
    model_name: str = "gpt-3.5"

# 2. Define the Output Schema
# We promise to return this structure
class ChatResponse(BaseModel):
    response: str
    timestamp: str
    tokens_used: int

@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    # We can access data using dot notation because of Pydantic
    user_prompt = request.prompt
    
    # Simulate AI logic
    fake_ai_response = f"I received your prompt: '{user_prompt}' using model {request.model_name}"
    
    # Return data matching the ChatResponse schema
    return ChatResponse(
        response=fake_ai_response,
        timestamp=datetime.now().isoformat(),
        tokens_used=len(user_prompt.split())
    )

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

Step 3: Async and Concurrency

import asyncio
from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn
import time

app = FastAPI()

class ChatRequest(BaseModel):
    prompt: str

@app.post("/generate")
async def generate_text(request: ChatRequest):
    print(f"Start processing: {request.prompt}")
    
    # await asyncio.sleep(5) is non-blocking.
    # It releases the server to handle other requests while waiting.
    # Unlike time.sleep(5) which freezes the server.
    await asyncio.sleep(5)
    
    print(f"Finished processing: {request.prompt}")
    return {"result": f"Generated content for: {request.prompt}"}

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

Step 4: Background Tasks

Sometimes you want to do something after you send the response. For example, logging the conversation to a database or sending an email alert. You don't want the user to wait for these administrative tasks.

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import uvicorn
import time

app = FastAPI()

class ChatRequest(BaseModel):
    prompt: str

# A function to simulate a slow database write
def write_to_log_file(prompt: str):
    time.sleep(2) # Simulating a 2-second database save
    with open("chat_logs.txt", "a") as f:
        f.write(f"Logged: {prompt}\n")
    print(f"Background task complete: Logged '{prompt}'")

@app.post("/chat_quick")
async def chat_quick(request: ChatRequest, background_tasks: BackgroundTasks):
    # 1. Prepare the response immediately
    response_text = f"AI response to: {request.prompt}"
    
    # 2. Schedule the slow task to run AFTER the response is sent
    background_tasks.add_task(write_to_log_file, request.prompt)
    
    # 3. Return immediately. The user doesn't wait for the file write.
    return {"response": response_text}

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

Run this. When you hit the endpoint, you get the response instantly. Two seconds later, you will see the print message in your terminal. This makes your API feel incredibly snappy.

Step 5: Middleware (CORS) & Structured Logging

Finally, we assemble the production-ready application.

import logging
import asyncio
from fastapi import FastAPI, Request, BackgroundTasks, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import uvicorn
from datetime import datetime

# 1. Configure Logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

app = FastAPI(title="Production AI Backend")

# 2. CORS Configuration
# Allow Streamlit (usually port 8501) to talk to this backend
origins = [
    "http://localhost:8501",
    "http://127.0.0.1:8501",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 3. Data Models
class ChatInput(BaseModel):
    user_id: str
    prompt: str

class ChatOutput(BaseModel):
    response: str
    processing_time: float

# 4. Middleware for Timing
# This runs for EVERY request to measure performance
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    logger.info(f"Path: {request.url.path} - Time: {process_time:.4f}s")
    return response

# 5. Background Task Function
def log_interaction(user_id: str, prompt: str):
    # Simulate DB write
    logger.info(f"DATABASE: Saving interaction for user {user_id}")

# 6. Endpoints
@app.get("/health")
async def health_check():
    """Used by monitoring tools to check if server is alive"""
    return {"status": "healthy", "timestamp": datetime.now()}

@app.post("/chat", response_model=ChatOutput)
async def chat_endpoint(input_data: ChatInput, background_tasks: BackgroundTasks):
    import time # importing locally to avoid conflict with middleware timing
    
    logger.info(f"Received prompt from {input_data.user_id}")
    
    # Simulate AI processing (Async)
    start_proc = time.time()
    await asyncio.sleep(2) 
    ai_response = f"Processed: {input_data.prompt}"
    duration = time.time() - start_proc
    
    # Schedule logging
    background_tasks.add_task(log_interaction, input_data.user_id, input_data.prompt)
    
    return ChatOutput(
        response=ai_response,
        processing_time=duration
    )

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

Now You Try

You have a running backend. Now, extend it.

Challenge Project: The Decoupled Architecture

This is the most important architectural pattern you will learn. You are going to separate the Frontend from the Backend completely.

* It should handle errors (e.g., if the backend is offline, show a nice error message, not a crash trace).

User: Hello AI
[Submit Button]

... thinking ...

AI: Processed: Hello AI
(Server took 2.001s)

What You Learned

Today you graduated from writing scripts to building Distributed Systems.

* FastAPI is the industry standard for Python AI backends.

* Pydantic ensures your data is clean before it ever hits your logic.

* Async/Await allows your server to handle many users simultaneously without blocking.

* Decoupling (separating frontend and backend) allows you to scale, secure, and manage your AI application properly.