Build a Typed Inference API with FastAPI
A typed inference API turns a model artifact into a predictable network interface. In this lesson, the outcome is a small FastAPI service that accepts JSON, validates it with Pydantic, runs a deterministic prediction function, and returns a documented response shape. In an MLOps serving system, this boundary matters because model callers rarely know the internals of feature engineering, label encoding, artifact loading, or failure handling. The API is where those details become explicit.
FastAPI is useful for this job because it builds request parsing, type-driven validation, OpenAPI documentation, dependency injection, and response serialization around ordinary Python type hints. The important point is not that type hints make Python statically typed. They do not. The important point is that FastAPI and Pydantic use those hints at runtime to turn untrusted JSON into checked Python objects before inference code sees it.
How FastAPI Handles an Inference Request
A request enters the ASGI server, usually Uvicorn, as an HTTP message. FastAPI matches the method and path to a route function. If the route declares a Pydantic model parameter, FastAPI reads the JSON body, asks Pydantic to validate and coerce fields, and either calls the route with a typed object or returns a 422 response before your model code runs. After the route returns, FastAPI serializes the result and filters it through the declared response model when one is supplied.
The model serving flow is therefore: HTTP request, route matching, body parsing, Pydantic validation, feature construction, artifact prediction, response validation, serialization, and HTTP response. Each stage has a different failure shape. Malformed JSON is not the same as a missing feature. A model artifact failing to load is not the same as a valid input producing a low-confidence prediction. Keeping those cases separate makes testing and operations much easier.
API Anatomy
A typed inference endpoint usually has four parts. The request schema names the fields callers must send and applies constraints such as minimum values, maximum values, enumerations, or list lengths. The response schema names exactly what the service promises to return. The application startup path loads model artifacts once, not on every request. The route function converts the validated request into the feature vector expected by the model and returns the response object.
Pydantic models are not just documentation. They are executable validators. A field declared as float will reject many non-numeric values, a field with ge=0 will reject negative numbers, and a Literal type limits a value to a known set. FastAPI turns validation failures into structured 422 responses containing the failing field path and reason. That behavior is valuable for client debugging, but it also means schema changes are compatibility changes.
Example 1: Typed Request and Response
The first example builds the smallest useful typed inference API. It uses a deterministic scoring rule so the behavior is easy to verify. The endpoint accepts account age, monthly spend, support tickets, and plan type. Pydantic validates basic feature ranges before the route calculates a churn risk label.
from typing import Literal
from fastapi import FastAPI
from pydantic import BaseModel, Field
app = FastAPI(title="Churn inference API")
class ChurnRequest(BaseModel):
account_age_days: int = Field(ge=0, le=3650)
monthly_spend: float = Field(ge=0, le=100000)
support_tickets_30d: int = Field(ge=0, le=100)
plan: Literal["free", "team", "enterprise"]
class ChurnResponse(BaseModel):
label: Literal["low", "medium", "high"]
score: float
model_version: str
def score_churn(request: ChurnRequest) -> float:
score = 0.20
if request.account_age_days < 30:
score += 0.25
if request.support_tickets_30d >= 3:
score += 0.30
if request.plan == "free":
score += 0.15
if request.monthly_spend >= 500:
score -= 0.10
return max(0.0, min(score, 1.0))
@app.post("/predict", response_model=ChurnResponse)
def predict(request: ChurnRequest) -> ChurnResponse:
score = score_churn(request)
if score >= 0.70:
label = "high"
elif score >= 0.40:
label = "medium"
else:
label = "low"
return ChurnResponse(label=label, score=round(score, 3), model_version="rules-2026-09")For a request with account_age_days=10, monthly_spend=20, support_tickets_30d=4, and plan="free", the score is 0.9 and the label is high. If support_tickets_30d is negative, FastAPI returns 422 and the route function is never called. This is the first serving guarantee: impossible feature values are rejected before inference.
Example 2: Testing the Contract
The next example tests both the successful path and the validation path without starting a network server. FastAPI’s test client exercises routing, validation, response serialization, and status codes in one process. This is more meaningful than testing score_churn alone because the public API behavior includes HTTP details.
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_predict_high_churn() -> None:
response = client.post(
"/predict",
json={
"account_age_days": 10,
"monthly_spend": 20.0,
"support_tickets_30d": 4,
"plan": "free",
},
)
assert response.status_code == 200
assert response.json() == {
"label": "high",
"score": 0.9,
"model_version": "rules-2026-09",
}
def test_predict_rejects_negative_tickets() -> None:
response = client.post(
"/predict",
json={
"account_age_days": 10,
"monthly_spend": 20.0,
"support_tickets_30d": -1,
"plan": "free",
},
)
assert response.status_code == 422
assert response.json()["detail"][0]["loc"][-1] == "support_tickets_30d"The expected behavior is deterministic: the first test receives a 200 response with the exact response JSON shown; the second receives 422 and identifies the invalid field. These tests catch accidental schema drift, such as renaming support_tickets_30d or changing the label thresholds without updating callers.
Example 3: Loading a Model Once
A real service should not deserialize a large model inside every request. FastAPI supports a lifespan function that runs at application startup and shutdown. The example below stores a model object in app.state. The route retrieves that object and uses it for each request. This pattern keeps request latency lower and makes startup failure explicit.
from contextlib import asynccontextmanager
from typing import Literal
from fastapi import FastAPI, Request
from pydantic import BaseModel, Field
class LinearModel:
version = "linear-demo-1"
def predict_score(self, features: list[float]) -> float:
account_age_days, monthly_spend, tickets = features
score = 0.65 - 0.0002 * account_age_days - 0.0001 * monthly_spend + 0.08 * tickets
return max(0.0, min(score, 1.0))
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.model = LinearModel()
yield
app.state.model = None
app = FastAPI(lifespan=lifespan)
class InferenceRequest(BaseModel):
account_age_days: int = Field(ge=0, le=3650)
monthly_spend: float = Field(ge=0, le=100000)
support_tickets_30d: int = Field(ge=0, le=100)
class InferenceResponse(BaseModel):
score: float
model_version: str
@app.post("/score", response_model=InferenceResponse)
def score(request_body: InferenceRequest, request: Request) -> InferenceResponse:
model = request.app.state.model
features = [
float(request_body.account_age_days),
request_body.monthly_spend,
float(request_body.support_tickets_30d),
]
score_value = model.predict_score(features)
return InferenceResponse(score=round(score_value, 4), model_version=model.version)For account_age_days=100, monthly_spend=50, and support_tickets_30d=2, the score calculation is 0.65 - 0.02 - 0.005 + 0.16, which returns 0.785. The service reports model version linear-demo-1, so downstream logs and evaluations can connect predictions to the artifact logic that produced them.
Design Choices and Trade-Offs
Use separate request and response models. Reusing one object for both directions often leaks internal fields or makes later changes harder. Request models should reflect what callers are allowed to send; response models should reflect what the service promises to expose. Add examples and field descriptions when they help generated OpenAPI clients, but do not rely on documentation in place of validation.
Prefer strict feature names at the API boundary. Accepting arbitrary dictionaries feels flexible, but it delays errors until feature construction or prediction time. A typed schema makes missing and extra fields visible earlier. The trade-off is that every schema change becomes a versioning concern. For breaking changes, publish a new path such as /v2/predict or accept both old and new fields during a migration window.
Decide whether inference should be synchronous. FastAPI supports async routes, but CPU-bound model prediction does not become faster because the function is declared async. For short predictions, a normal route is fine. For long batch jobs or GPU-bound work with queueing, return a job identifier and process the request outside the web worker. The API type contract still matters, but the response model changes from a prediction to a submitted job state.
Failure Modes and Troubleshooting
Symptom: callers receive 422 responses after a deployment. Cause: the Pydantic request schema changed, a field was renamed, or a constraint became stricter. Diagnostics: inspect the response detail array, compare the generated OpenAPI schema with the previous release, and replay a known old request in a test. Correction: restore backward compatibility, add an explicit API version, or coordinate the client release before enforcing the new field.
Symptom: latency increases sharply under load. Cause: the route loads the model artifact, builds expensive encoders, or opens external connections on every request. Diagnostics: add timing around artifact access and prediction, check worker CPU and memory, and profile a single request. Correction: load immutable artifacts at startup, reuse clients safely, and keep per-request work limited to validation, feature assembly, and prediction.
Symptom: successful HTTP responses contain impossible values, such as a score above 1.0. Cause: the inference function returned an unchecked value or the response model was too loose. Diagnostics: add tests against boundary feature values and inspect the declared response schema. Correction: clamp or reject invalid prediction outputs and constrain response fields with Pydantic validation.
Symptom: the server starts locally but fails in the container. Cause: the artifact path is relative to a different working directory or the file was not copied into the image. Diagnostics: log the resolved artifact path at startup and run a container command that lists the expected directory. Correction: package the artifact deliberately, configure the path through an environment variable, and fail startup if the artifact is missing.
Security, Performance, and Reliability
Do not log raw inference payloads by default. Feature values can contain personal, commercial, or regulated information. Prefer request IDs, model version, schema version, latency, status code, and validation error category. If payload capture is required for debugging, sample narrowly, redact sensitive fields, and set retention limits.
Protect the service from resource abuse. Bound string lengths, list sizes, numeric ranges, request body size, and server timeouts. A typed schema is a first line of defense, not a complete security model. Authentication and authorization should happen before expensive inference, especially when the model uses paid external dependencies or scarce accelerators.
Reliability depends on startup checks and health endpoints that mean what operators think they mean. A liveness endpoint can report that the process is running. A readiness endpoint should fail when the model artifact is unavailable or initialization has not completed. Keep prediction routes separate from health routes so monitoring traffic does not exercise expensive inference paths.
Hands-On Lab
Prerequisites: Python, FastAPI, Uvicorn, and pytest installed in an isolated environment. Create main.py with the first example and test_main.py with the second example. Start the service with uvicorn main:app --reload. Send a valid JSON request to /predict and verify a 200 response containing label, score, and model_version. Send the same request with support_tickets_30d set to -1 and verify a 422 response. Run the tests with pytest and confirm both pass.
For cleanup, stop Uvicorn, remove the temporary files if they were created only for the lab, and deactivate the virtual environment. If you adapted an existing service, roll back by removing the new route registration and tests in the same change set. Verification is complete only when the valid request, invalid request, and automated tests all show the expected behavior.
Assessment Exercises
- Change the response schema so
scoremust be between 0 and 1. What test would fail if a future model returned 1.4? - Design a backward-compatible migration from
monthly_spendtomonthly_recurring_revenue. Which fields would the request model accept during the migration? - A team declares the route
asyncbut performs a slow CPU-bound prediction inside it. Explain why this can still block useful work and propose a better serving design. - Write a readiness check policy for a service that loads a model artifact at startup. What should the endpoint return before loading finishes?
- Given a 422 response from FastAPI, explain how you would distinguish a client bug from an accidental server-side schema change.
Summary
A typed FastAPI inference API makes the model serving boundary explicit. Pydantic request models reject invalid feature payloads before prediction, response models constrain what the service emits, and startup loading keeps artifact failures separate from request handling. In MLOps, this gives teams a testable contract between model producers and model consumers: schema, validation behavior, model version, failure shape, and operational checks can all be reviewed before deployment.
