Day 6: Containerizing AI Microservices — Building and Packaging APIs with FastAPI

An internal application should not call a raw model socket. Between them you need an application layer that handles rate limiting, payload validation, authentication and error handling. FastAPI is a strong fit for that job.

The request flow

1. The client sends a JSON payload to the FastAPI gateway.
2. Pydantic checks the types and the schema.
3. A valid request goes to an async queue and on to the inference engine (vLLM or Ollama), and the answer streams back as Server-Sent Events.
4. An invalid request gets a 422 Bad Request immediately, so it never costs a GPU cycle.

Why FastAPI

• Native async. With asyncio, a single worker holds thousands of open connections while it awaits the model, unlike a synchronous Flask worker that blocks per request.
• Strict input validation. Pydantic enforces types and limits before any expensive inference starts.

An example service

This gateway validates a diagnostic request, calls Ollama and streams the tokens. It assumes Ollama is running locally with the gemma2:2b model pulled.

import json
import httpx
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

app = FastAPI(title="Diagnostic AI gateway")
OLLAMA_URL = "http://localhost:11434/api/generate"   # in Docker Compose: http://inference-engine:11434/api/generate


class DiagnosticRequest(BaseModel):
    device_id: str
    telemetry_logs: str = Field(min_length=10, max_length=5000)
    severity_threshold: int = Field(ge=1, le=5)


async def stream_inference(req: DiagnosticRequest):
    prompt = (
        f"Device {req.device_id}. Severity threshold {req.severity_threshold}.\n"
        f"Analyse these logs and list likely faults:\n{req.telemetry_logs}"
    )
    payload = {"model": "gemma2:2b", "prompt": prompt, "stream": True}
    try:
        async with httpx.AsyncClient(timeout=None) as client:
            async with client.stream("POST", OLLAMA_URL, json=payload) as resp:
                resp.raise_for_status()
                async for line in resp.aiter_lines():
                    if not line:
                        continue
                    chunk = json.loads(line)                       # Ollama streams one JSON object per line
                    if chunk.get("response"):
                        yield f"data: {json.dumps({'token': chunk['response']})}\n\n"
                    if chunk.get("done"):
                        yield "data: [DONE]\n\n"
    except httpx.HTTPError as exc:
        # Once streaming has started the HTTP status is already sent,
        # so report the failure as an event instead of raising.
        yield f"data: {json.dumps({'error': str(exc)})}\n\n"


@app.post("/api/v1/diagnose")
async def diagnose(req: DiagnosticRequest):
    # Invalid payloads never get here: FastAPI returns 422 automatically.
    return StreamingResponse(stream_inference(req), media_type="text/event-stream")

Run it with uvicorn main:app --host 0.0.0.0 --port 8000 and test it with curl -N -X POST localhost:8000/api/v1/diagnose -H "Content-Type: application/json" -d '{"device_id":"sw-14","telemetry_logs":"port 12 link flap, CRC errors rising","severity_threshold":3}'.

One detail matters for error handling. You can raise an HTTPException (for example a 500) only before the first byte is sent. After streaming has begun the status code is already on the wire, so failures must travel as an event in the stream, as shown above.

Streaming changes how it feels

Server-Sent Events (text/event-stream) send each token as soon as it exists. Total generation time is the same, but a 10-second blocking wait becomes a response that starts in well under a second. If a proxy sits in front of the gateway, turn off response buffering there too, or the stream is held back.

Sizing workers

Decouple the number of gateway workers from the number of GPUs. Async workers are cheap and the inference engine does the real batching, so two to four workers is a common starting point. Load-test and adjust, because more workers than the engine can batch only lengthens the queue.

Package the service with the multi-stage Dockerfile from Day 2, and it becomes a deployable microservice: validated input, streamed output and a clear boundary in front of the model.

Run it yourself

This is the exact example from the video. A real FastAPI app, tested in process. A fixed token list stands in for the Ollama call, and the file is trimmed from the lesson's version.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
app = FastAPI()

class Req(BaseModel):
    logs: str = Field(min_length=10)
    severity: int = Field(ge=1, le=5)

async def sse():                      # stand-in for the engine
    for t in ["Port", " 12", " flaps"]:
        yield f"data: {t}\n\n"

@app.post("/diagnose")
async def diagnose(req: Req):
    return StreamingResponse(sse(), media_type="text/event-stream")

Helper file client_test.py, used by the run:

from fastapi.testclient import TestClient
from app import app

c = TestClient(app)
bad = c.post("/diagnose", json={"logs": "x", "severity": 9})
print("bad payload:", bad.status_code)
ok = c.post("/diagnose", json={"logs": "port 12 link flap", "severity": 3})
print("good payload:", ok.status_code)
print("type:", ok.headers["content-type"])
for line in ok.text.split("\n\n")[:3]:
    print(line)

Run it:

python client_test.py

Output I got while recording the video:

bad payload: 422
good payload: 200
type: text/event-stream; charset=utf-8
data: Port
data:  12
data:  flaps

Coming Up Next

Day 7: Docker Compose: architecting a multi-container database and inference stack locally.

#FastAPI #Python #AIEngineering #SoftwareArchitecture #Microservices #CloudNative #DevOps #BackendDevelopment