FastAPI contracts

Welcome! I'm Param. In this course we rebuild an LLM agent the way real teams ship them: as seven composable layers. Each layer has a single job, a clear contract, and a visible failure mode. Every request flows through the same pipeline and emits a per-layer trace, so when something breaks in production you open the trace and see exactly which layer failed.

The seven layers at a glance

Every request flows through these layers in order. Each records a trace entry.

Transport is the first layer because everything downstream depends on clean inputs. If a request with a missing field, a 40KB blob, or a malformed thread id reaches the orchestrator, every layer after it has to defend itself. Instead, we reject garbage at the door and give the rest of the pipeline a guarantee: by the time a request arrives at layer two, it is shaped exactly as expected.

models.py
python
from pydantic import BaseModel, Field
from typing import Any, Dict, List, Optional


class ChatRequest(BaseModel):
    """Incoming chat message (handled at Layer 1: Transport)."""
    message: str = Field(..., min_length=1, max_length=4000)
    thread_id: str = Field(..., min_length=1, max_length=128)


class LayerTrace(BaseModel):
    """One layer's worth of trace data for a single request."""
    layer: str
    status: str  # "ok", "skipped", "error"
    duration_ms: float
    notes: Optional[str] = None
    data: Optional[Dict[str, Any]] = None


class ChatResponse(BaseModel):
    """Final reply plus the per-layer trace (Layer 7: Observability)."""
    reply: str
    thread_id: str
    trace: List[LayerTrace]

Pydantic models are the contract. FastAPI rejects any request that does not match, with a precise 422 error. No runtime assertion noise in your handler.

layers/transport.py
python
MAX_MESSAGE_LEN = 4000
RATE_LIMIT_WINDOW_S = 60
RATE_LIMIT_MAX_REQUESTS = 30


class TransportLayer:
    def __init__(self) -> None:
        self._hits: Dict[str, Deque[float]] = defaultdict(deque)

    def validate(self, message: str, thread_id: str) -> dict:
        """Return normalized request fields or raise ValueError."""
        if not isinstance(message, str) or not message.strip():
            raise ValueError("message must be a non-empty string")
        if len(message) > MAX_MESSAGE_LEN:
            raise ValueError(f"message exceeds {MAX_MESSAGE_LEN} chars")
        if not isinstance(thread_id, str) or not thread_id.strip():
            raise ValueError("thread_id required")

        if not self._check_rate_limit(thread_id):
            raise ValueError("rate limit exceeded for this thread_id")

        return {
            "message": message.strip(),
            "thread_id": thread_id.strip(),
            "received_at": time.time(),
        }

Transport runs belt-and-suspenders validation on top of Pydantic, plus a small rate limiter. Even if a caller bypasses the API schema (internal queue, replay), the transport layer still rejects bad input.

The transport layer normalizes fields (trim whitespace, stamp received_at) and returns a dict the rest of the pipeline can rely on. Beyond that, not every caller comes through the HTTP boundary. Internal tools, queue workers, and trace replays call the service directly. The transport layer is the single trust boundary for the pipeline, regardless of who called it.

Quiz: Quiz

Loading practice…