FastAPI for Machine Learning & LLM AI Agents

$ python --version
# If that prints Python 2.x, try:
$ python3 --version
# You want: Python 3.11.x  (or 3.9+)

# 1. make and enter a project folder
$ mkdir ml-api && cd ml-api

# 2. create a virtual environment named .venv
$ python -m venv .venv

# 3. activate it
#    macOS / Linux:
$ source .venv/bin/activate
#    Windows (PowerShell):
$ .venv\Scripts\Activate.ps1

# your prompt now shows (.venv) — you're inside it

$ pip install "fastapi[standard]"

fastapi[standard]
scikit-learn
joblib
numpy
pydantic-settings
openai
anthropic
python-dotenv

$ pip install -r requirements.txt

ml-api/
├── .venv/              # virtual environment (never commit this)
├── .env                # secrets & config (never commit this)
├── requirements.txt    # pinned dependencies
├── main.py             # your FastAPI app
└── models/             # saved model files (.joblib, etc.)

from fastapi import FastAPI

# 'app' is the central object that holds your whole API
app = FastAPI(title="My First API")

# @app.get("/") attaches this function to GET requests on "/"
@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI!"}

$ fastapi dev main.py

@app.get("/health")
def health_check():
    # A simple endpoint to confirm the API is alive —
    # you'll add one of these to every real service.
    return {"status": "ok"}

@app.get("/items/{item_id}")
def read_item(item_id: int):
    return {"item_id": item_id}

@app.get("/items/")
def list_items(skip: int = 0, limit: int = 10):
    # called as /items/?skip=20&limit=5
    return {"skip": skip, "limit": limit}

from fastapi import Query

@app.get("/search")
def search(
    q: str = Query(min_length=2, max_length=50),
    page: int = Query(default=1, ge=1),   # ge = greater-or-equal
):
    return {"query": q, "page": page}

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    in_stock: bool = True        # has a default → optional
    tags: list[str] = []         # a list of strings, empty by default

@app.post("/items/")
def create_item(item: Item):
    # 'item' is a fully-validated Item object
    total = item.price * 1.1     # add 10% tax
    return {"name": item.name, "price_with_tax": round(total, 2)}

{
  "name": "Wireless Mouse",
  "price": 25.00,
  "tags": ["electronics", "accessories"]
}

from pydantic import BaseModel, Field

class Image(BaseModel):
    url: str
    width: int = Field(gt=0)

class Product(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0, description="Price in USD")
    image: Image | None = None          # a nested model, optional

class IrisFeatures(BaseModel):
    sepal_length: float = Field(gt=0, le=10)
    sepal_width:  float = Field(gt=0, le=10)
    petal_length: float = Field(gt=0, le=10)
    petal_width:  float = Field(gt=0, le=10)

class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=4000)
    temperature: float = Field(default=0.7, ge=0, le=2)

from pydantic import BaseModel

class UserIn(BaseModel):
    username: str
    password: str          # sensitive — must never be returned

class UserOut(BaseModel):
    username: str          # safe to expose

@app.post("/users/", response_model=UserOut)
def create_user(user: UserIn):
    # even though we return the whole object, FastAPI
    # filters it down to UserOut — password is dropped.
    return user

from fastapi import HTTPException, status

@app.post("/users/", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def create_user(user: UserIn):
    if user.username == "admin":
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Username already taken",
        )
    return user

import httpx

# 'async def' makes this a coroutine; 'await' yields control while waiting
@app.get("/joke")
async def get_joke():
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://api.example.com/joke")
    return resp.json()

import asyncio

@app.post("/analyze")
async def analyze(text: str):
    # run three independent async tasks concurrently
    sentiment, summary, keywords = await asyncio.gather(
        get_sentiment(text),
        get_summary(text),
        get_keywords(text),
    )
    return {"sentiment": sentiment, "summary": summary, "keywords": keywords}

from fastapi import Depends

# a reusable dependency: common list parameters
def pagination(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

@app.get("/users/")
def list_users(page: dict = Depends(pagination)):
    return page

@app.get("/items/")
def list_items(page: dict = Depends(pagination)):
    return page          # same logic, written once

def get_db():
    db = SessionLocal()       # open connection
    try:
        yield db              # hand it to the endpoint
    finally:
        db.close()            # always runs, even on error

@app.get("/customers/")
def read_customers(db = Depends(get_db)):
    return db.query(Customer).all()

from fastapi import Depends, Header, HTTPException

# a guard dependency reused across many protected routes
def require_api_key(x_api_key: str = Header()):
    if x_api_key != "secret-key":
        raise HTTPException(status_code=401, detail="Invalid API key")
    return x_api_key

@app.post("/predict", dependencies=[Depends(require_api_key)])
def predict():
    return {"prediction": 1}

import time
from fastapi import Request

@app.middleware("http")
async def add_timing(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)        # run the endpoint
    duration = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{duration:.3f}s"
    return response

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myfrontend.com"],   # who may call you
    allow_credentials=True,
    allow_methods=["*"],                          # GET, POST, ...
    allow_headers=["*"],
)

from fastapi import HTTPException

products = {1: "Laptop", 2: "Phone"}

@app.get("/products/{pid}")
def get_product(pid: int):
    if pid not in products:
        raise HTTPException(
            status_code=404,
            detail=f"Product {pid} not found",
        )
    return {"id": pid, "name": products[pid]}

from pydantic import BaseModel, field_validator

class Prompt(BaseModel):
    text: str

    @field_validator("text")
    @classmethod
    def not_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Prompt cannot be empty or whitespace")
        return v.strip()

from fastapi import Request
from fastapi.responses import JSONResponse

class ModelNotReadyError(Exception):
    pass

@app.exception_handler(ModelNotReadyError)
async def model_not_ready(request: Request, exc: ModelNotReadyError):
    return JSONResponse(
        status_code=503,
        content={"error": "model_unavailable",
                 "detail": "The model is still loading. Try again shortly."},
    )

from fastapi import BackgroundTasks

def log_prediction(features: list, result: int):
    with open("predictions.log", "a") as f:
        f.write(f"{features} -> {result}\n")

@app.post("/predict")
def predict(features: list[float], bg: BackgroundTasks):
    result = 1                       # pretend model output
    bg.add_task(log_prediction, features, result)   # runs after response
    return {"prediction": result}

from fastapi import UploadFile, HTTPException

@app.post("/classify-image")
async def classify_image(file: UploadFile):
    if not file.content_type.startswith("image/"):
        raise HTTPException(400, "Please upload an image file")

    contents = await file.read()      # raw bytes
    # img = Image.open(io.BytesIO(contents))   # then feed to your model
    return {"filename": file.filename, "size_bytes": len(contents)}

from fastapi import UploadFile, Form

@app.post("/batch-classify")
async def batch_classify(
    files: list[UploadFile],
    model_name: str = Form(default="resnet50"),
):
    return {"model": model_name, "count": len(files)}

from fastapi import APIRouter
from pydantic import BaseModel

router = APIRouter(prefix="/predict", tags=["predictions"])

class Features(BaseModel):
    values: list[float]

@router.post("/")
def make_prediction(f: Features):
    return {"prediction": sum(f.values)}   # placeholder logic

from fastapi import FastAPI
from routers import predict, chat

app = FastAPI(title="ML & AI Service")

app.include_router(predict.router)
app.include_router(chat.router)

@app.get("/health")
def health():
    return {"status": "ok"}

app/
├── main.py            # creates FastAPI(), includes routers
├── config.py          # settings (Chapter 24)
├── dependencies.py    # shared Depends() functions
├── routers/
│   ├── predict.py     # /predict endpoints
│   └── chat.py        # /chat endpoints
├── schemas/           # Pydantic models (in & out)
│   ├── predict.py
│   └── chat.py
├── services/          # the real logic: model loading, LLM calls
│   ├── ml_model.py
│   └── llm.py
└── models/            # saved .joblib / weights

import joblib
from pathlib import Path
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier

# 1. load data
iris = load_iris()
X, y = iris.data, iris.target

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# 2. bundle preprocessing + model so serving matches training exactly
pipeline = Pipeline([
    ("scaler", StandardScaler()),
    ("clf", RandomForestClassifier(n_estimators=100, random_state=42)),
])
pipeline.fit(X_train, y_train)

# 3. evaluate
accuracy = pipeline.score(X_test, y_test)
print(f"Test accuracy: {accuracy:.2%}")

# 4. save — model + class names together
Path("models").mkdir(exist_ok=True)
joblib.dump(
    {"pipeline": pipeline, "target_names": list(iris.target_names)},
    "models/iris_model.joblib",
)
print("Saved to models/iris_model.joblib")

$ python train.py
# Test accuracy: 100.00%
# Saved to models/iris_model.joblib

import joblib

bundle = joblib.load("models/iris_model.joblib")
pipeline = bundle["pipeline"]
names = bundle["target_names"]      # ['setosa', 'versicolor', 'virginica']

pred = pipeline.predict([[5.1, 3.5, 1.4, 0.2]])
print(names[pred[0]])                # -> setosa

import joblib
from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(title="Iris Classifier API")

# load once at import time (improved in Chapter 17)
bundle = joblib.load("models/iris_model.joblib")
pipeline = bundle["pipeline"]
target_names = bundle["target_names"]

# --- input contract ---
class IrisInput(BaseModel):
    sepal_length: float = Field(gt=0, le=10, examples=[5.1])
    sepal_width:  float = Field(gt=0, le=10, examples=[3.5])
    petal_length: float = Field(gt=0, le=10, examples=[1.4])
    petal_width:  float = Field(gt=0, le=10, examples=[0.2])

# --- output contract ---
class Prediction(BaseModel):
    species: str
    class_id: int
    confidence: float

@app.post("/predict", response_model=Prediction)
def predict(data: IrisInput):
    features = [[
        data.sepal_length, data.sepal_width,
        data.petal_length, data.petal_width,
    ]]
    class_id = int(pipeline.predict(features)[0])
    proba = pipeline.predict_proba(features)[0]
    return Prediction(
        species=target_names[class_id],
        class_id=class_id,
        confidence=round(float(proba[class_id]), 4),
    )

// request body
{ "sepal_length": 5.1, "sepal_width": 3.5,
  "petal_length": 1.4, "petal_width": 0.2 }

// response
{ "species": "setosa", "class_id": 0, "confidence": 0.99 }

$ curl -X POST http://127.0.0.1:8000/predict \
    -H "Content-Type: application/json" \
    -d '{"sepal_length":5.1,"sepal_width":3.5,"petal_length":1.4,"petal_width":0.2}'

from pydantic import BaseModel, Field

class BatchInput(BaseModel):
    items: list[IrisInput] = Field(max_length=1000)   # cap the batch size

class BatchResult(BaseModel):
    predictions: list[Prediction]

@app.post("/predict/batch", response_model=BatchResult)
def predict_batch(data: BatchInput):
    rows = [[
        i.sepal_length, i.sepal_width, i.petal_length, i.petal_width
    ] for i in data.items]

    class_ids = pipeline.predict(rows)                 # vectorised, one call
    probas = pipeline.predict_proba(rows)

    results = [
        Prediction(
            species=target_names[int(c)],
            class_id=int(c),
            confidence=round(float(p[int(c)]), 4),
        )
        for c, p in zip(class_ids, probas)
    ]
    return BatchResult(predictions=results)

import io
from PIL import Image
from fastapi import UploadFile, HTTPException

@app.post("/vision/classify")
async def classify(file: UploadFile):
    if not file.content_type.startswith("image/"):
        raise HTTPException(400, "Upload an image file")

    raw = await file.read()
    image = Image.open(io.BytesIO(raw)).convert("RGB").resize((224, 224))

    # tensor = preprocess(image)            # to your framework's format
    # logits = vision_model(tensor)         # run the model
    # label  = labels[int(logits.argmax())]
    label, score = "tabby cat", 0.92        # placeholder

    return {"label": label, "confidence": score, "filename": file.filename}

import joblib
from contextlib import asynccontextmanager
from fastapi import FastAPI

ml = {}     # a simple shared container

@asynccontextmanager
async def lifespan(app: FastAPI):
    # --- startup: runs once, before serving ---
    print("Loading model...")
    ml["bundle"] = joblib.load("models/iris_model.joblib")
    yield
    # --- shutdown: runs once, on exit ---
    ml.clear()
    print("Model unloaded")

app = FastAPI(lifespan=lifespan)

@app.post("/predict")
def predict(data: IrisInput):
    pipeline = ml["bundle"]["pipeline"]      # already in memory — instant
    features = [[data.sepal_length, data.sepal_width,
                 data.petal_length, data.petal_width]]
    return {"class_id": int(pipeline.predict(features)[0])}

from fastapi import Depends, HTTPException

def get_model():
    bundle = ml.get("bundle")
    if bundle is None:
        raise HTTPException(503, "Model is still loading, retry shortly")
    return bundle["pipeline"]

@app.post("/predict")
def predict(data: IrisInput, model = Depends(get_model)):
    features = [[data.sepal_length, data.sepal_width,
                 data.petal_length, data.petal_width]]
    return {"class_id": int(model.predict(features)[0])}

from fastapi import FastAPI
from pydantic import BaseModel, Field
from openai import OpenAI

app = FastAPI()
client = OpenAI()      # reads OPENAI_API_KEY from the environment

class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=4000)

class ChatReply(BaseModel):
    reply: str

@app.post("/chat", response_model=ChatReply)
def chat(req: ChatRequest):
    completion = client.chat.completions.create(
        model="gpt-4o-mini",        # check provider docs for current models
        messages=[
            {"role": "system", "content": "You are a concise, helpful assistant."},
            {"role": "user", "content": req.message},
        ],
    )
    return ChatReply(reply=completion.choices[0].message.content)

from anthropic import Anthropic

client = Anthropic()      # reads ANTHROPIC_API_KEY from the environment

@app.post("/chat", response_model=ChatReply)
def chat(req: ChatRequest):
    msg = client.messages.create(
        model="claude-sonnet-4-6",       # check docs for current models
        max_tokens=1024,
        system="You are a concise, helpful assistant.",
        messages=[{"role": "user", "content": req.message}],
    )
    return ChatReply(reply=msg.content[0].text)

from fastapi import HTTPException

@app.post("/chat", response_model=ChatReply)
def chat(req: ChatRequest):
    try:
        completion = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": req.message}],
            timeout=30,
        )
    except Exception as e:
        # log the real error server-side; show a safe message to the user
        raise HTTPException(502, "The AI service is unavailable. Please retry.")
    return ChatReply(reply=completion.choices[0].message.content)

from fastapi.responses import StreamingResponse
from openai import OpenAI

client = OpenAI()

@app.post("/chat/stream")
def chat_stream(req: ChatRequest):
    def token_generator():
        stream = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": req.message}],
            stream=True,                       # ask for a stream
        )
        for chunk in stream:
            piece = chunk.choices[0].delta.content
            if piece:
                yield piece                    # send each token onward

    return StreamingResponse(token_generator(), media_type="text/plain")

import json

@app.post("/chat/sse")
def chat_sse(req: ChatRequest):
    def event_stream():
        stream = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": req.message}],
            stream=True,
        )
        for chunk in stream:
            piece = chunk.choices[0].delta.content
            if piece:
                yield f"data: {json.dumps({'token': piece})}\n\n"
        yield "data: [DONE]\n\n"               # signal completion

    return StreamingResponse(event_stream(), media_type="text/event-stream")

const res = await fetch("/chat/sse", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ message: "Explain async in one line" }),
});
const reader = res.body.getReader();
const decoder = new TextDecoder();
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  process.stdout.write(decoder.decode(value)); // append tokens to the UI
}

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from uuid import uuid4
from openai import OpenAI

app = FastAPI()
client = OpenAI()

# demo store: conversation_id -> list of messages
#   (use Redis or a database in production — see the note below)
conversations: dict[str, list[dict]] = {}

class Turn(BaseModel):
    conversation_id: str | None = None
    message: str

@app.post("/conversation")
def converse(turn: Turn):
    cid = turn.conversation_id or str(uuid4())
    history = conversations.setdefault(cid, [
        {"role": "system", "content": "You are a helpful assistant."}
    ])

    history.append({"role": "user", "content": turn.message})

    completion = client.chat.completions.create(
        model="gpt-4o-mini", messages=history,
    )
    reply = completion.choices[0].message.content
    history.append({"role": "assistant", "content": reply})

    return {"conversation_id": cid, "reply": reply}

def trim(history: list[dict], keep_last: int = 10) -> list[dict]:
    system = history[:1]                 # always keep the system message
    recent = history[1:][-keep_last:]    # keep only the most recent turns
    return system + recent

import numpy as np
from fastapi import FastAPI
from pydantic import BaseModel
from openai import OpenAI

app = FastAPI()
client = OpenAI()

# --- our tiny "knowledge base" ---
DOCS = [
    "Our return policy allows refunds within 30 days of purchase.",
    "Standard shipping takes 3 to 5 business days.",
    "Premium members get free express shipping on all orders.",
    "Support is available Monday to Friday, 9am to 6pm.",
]

def embed(texts: list[str]) -> np.ndarray:
    resp = client.embeddings.create(model="text-embedding-3-small", input=texts)
    return np.array([d.embedding for d in resp.data])

# embed the knowledge base once at startup
DOC_VECTORS = embed(DOCS)

def retrieve(question: str, k: int = 2) -> list[str]:
    q = embed([question])[0]
    # cosine similarity between the question and every doc
    sims = DOC_VECTORS @ q / (
        np.linalg.norm(DOC_VECTORS, axis=1) * np.linalg.norm(q)
    )
    top = sims.argsort()[::-1][:k]          # indices of the k best matches
    return [DOCS[i] for i in top]

class Ask(BaseModel):
    question: str

@app.post("/ask")
def ask(body: Ask):
    context = retrieve(body.question)
    prompt = (
        "Answer the question using ONLY the context below. "
        "If the answer isn't there, say you don't know.\n\n"
        f"Context:\n- " + "\n- ".join(context) +
        f"\n\nQuestion: {body.question}"
    )
    completion = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    return {"answer": completion.choices[0].message.content, "sources": context}

import json
from fastapi import FastAPI
from pydantic import BaseModel
from openai import OpenAI

app = FastAPI()
client = OpenAI()

# --- 1. the actual Python tool ---
def get_weather(city: str) -> str:
    fake = {"dhaka": "32°C, humid", "london": "14°C, rainy"}
    return fake.get(city.lower(), "Unknown city")

# --- 2. describe the tool to the model ---
TOOLS = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get the current weather for a city",
        "parameters": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
}]

class AgentRequest(BaseModel):
    message: str

@app.post("/agent")
def agent(req: AgentRequest):
    messages = [{"role": "user", "content": req.message}]

    # --- 3. the loop ---
    for _ in range(5):                      # safety cap on iterations
        completion = client.chat.completions.create(
            model="gpt-4o-mini", messages=messages, tools=TOOLS,
        )
        msg = completion.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:              # no tool wanted -> final answer
            return {"reply": msg.content}

        # the model asked to run one or more tools
        for call in msg.tool_calls:
            args = json.loads(call.function.arguments)
            result = get_weather(**args)    # run the real function
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result,
            })

    return {"reply": "Stopped after too many steps."}

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    openai_api_key: str                       # required — startup fails if missing
    model_name: str = "gpt-4o-mini"           # sensible default
    max_tokens: int = 1024
    environment: str = "development"

    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()        # import this anywhere you need config

OPENAI_API_KEY=sk-your-real-key-here
MODEL_NAME=gpt-4o-mini
ENVIRONMENT=production

from openai import OpenAI
from config import settings

client = OpenAI(api_key=settings.openai_api_key)

@app.post("/chat")
def chat(req: ChatRequest):
    completion = client.chat.completions.create(
        model=settings.model_name,
        max_tokens=settings.max_tokens,
        messages=[{"role": "user", "content": req.message}],
    )
    return {"reply": completion.choices[0].message.content}

from fastapi import Security, HTTPException, status
from fastapi.security import APIKeyHeader
from config import settings

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

def require_key(key: str = Security(api_key_header)):
    if key != settings.service_api_key:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing API key",
        )
    return key

from fastapi import Depends
from auth import require_key

@app.post("/predict", dependencies=[Depends(require_key)])
def predict(data: IrisInput):
    return {"class_id": 0}     # only reachable with a valid key

# start from a slim official Python image
FROM python:3.12-slim

# don't write .pyc files; flush logs immediately
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1

WORKDIR /app

# install deps first so this layer is cached when only code changes
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# now copy the application code and model files
COPY . .

EXPOSE 8000

# run with uvicorn, bound to all interfaces inside the container
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# build an image tagged "ml-api"
$ docker build -t ml-api .

# run it, mapping container port 8000 to host 8000,
# and passing secrets from your local .env
$ docker run -p 8000:8000 --env-file .env ml-api

.venv/
__pycache__/
*.pyc
.env
.git/
.pytest_cache/

# 4 worker processes behind uvicorn
$ uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

import logging, time
from fastapi import Request

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("api")

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    ms = (time.perf_counter() - start) * 1000
    log.info("%s %s -> %s in %.1fms",
             request.method, request.url.path, response.status_code, ms)
    return response