LearnPreparation & PrerequisitesFirst AI App End-to-End

⚙️EasyMLOps & Deployment

First AI App End-to-End

Ship one traceable rotation-decision workflow: validated input, model boundary, stored status, clear UI states, failure tests, and deploy checks.

13 min read

Learning path

Step 29 of 158 in the full curriculum

Calling LLM APIs in Production The LLM Lifecycle

Alex opens a stale service-account key for account acct_10234. Your earlier wrapper can ask a model whether policy line P-7 supports a rotation, but a customer can't use a wrapper by itself. They need a form, a clear result, and a useful error when the call fails.

Now turn that checked model call into a small application. One request travels from browser input to a server route, through the model boundary, into a trace record, and back to the screen. The app decides eligibility only; creating a rotation job remains a separate idempotent action.

Vertical slice for stale service-account key acct_10234. Browser states send a typed request into a deterministic server boundary around the probabilistic P-7 model call. Checked results and named failures update a redacted trace ledger, while tests and deploy checks form proof rails below. Rotation-job creation remains locked. — Wrap one probabilistic decision in deterministic rails: typed input, policy verification, explicit UI states, durable redacted traces, boundary tests, and deploy checks. Rotation job creation stays outside this workflow.

Define the one job your app does

A first AI app doesn't need chat history, autonomous tools, or a retrieval system. It needs one user problem with an observable answer:

Given a credential report for account acct_10234, decide whether policy line P-7 supports a rotation, cite that line in the result, and never create a rotation job as a hidden side effect.

That sentence fixes the app boundary before you choose a framework:

Layer	Responsibility in this app	What it must not do
Browser	Collect account ID and credential report; show state.	Call a model with a secret key.
API route	Validate request; return stable response fields.	Embed provider-specific prompt logic.
Decision service	Call the checked wrapper; translate outcome into status.	Issue a rotation job.
Trace store	Save status, evidence, latency, and redacted input fingerprint.	Store unnecessary customer text.
Tests	Prove completion, rejection, timeout, and health behavior.	Spend money on model calls.

Start with data the browser and server agree on

The route receives a report and returns a decision. Before a model call exists, write those shapes down as schemas. Pydantic validates Python data against declared fields and constraints; FastAPI can use the same models at the HTTP boundary later.^[1]

This first runnable cell rejects missing identifiers and constrains the response to the decisions your UI understands:

define-rotation-contract.py

from typing import Literal

from pydantic import BaseModel, Field

class RotationReport(BaseModel):
    account_id: str = Field(pattern=r"^acct_\d{5}$")
    item: str = Field(min_length=3, max_length=80)
    credential_report: str = Field(min_length=10, max_length=500)

class RotationDecision(BaseModel):
    decision: Literal["eligible", "not_eligible", "needs_review"]
    rotation_window_days: int = Field(ge=0, le=30)
    source_line_ids: list[str] = Field(min_length=1)

report = RotationReport(
    account_id="acct_10234",
    item="ceramic lamp",
    credential_report="Service-account key is older than policy.",
)
decision = RotationDecision(
    decision="eligible",
    rotation_window_days=30,
    source_line_ids=["P-7"],
)

print(report.account_id, "=>", decision.decision, decision.source_line_ids)

Validated contract

acct_10234 => eligible ['P-7']

The schema proves field shape, not truth. An output can match the schema and still cite the wrong policy or claim a window your policy doesn't allow.

The application therefore performs business-rule checks after parsing. Here, P-7 says stale service-account keys may be rotated within 30 days. The browser doesn't get to claim how old a credential is; the service reads that fact from a trusted account record:

check-policy-evidence.py

from typing import Literal

from pydantic import BaseModel, Field

class RotationDecision(BaseModel):
    decision: Literal["eligible", "not_eligible", "needs_review"]
    rotation_window_days: int = Field(ge=0, le=90)
    source_line_ids: list[str] = Field(min_length=1)

def verify_p7(decision: RotationDecision, *, credential_age_days: int) -> RotationDecision:
    if "P-7" not in decision.source_line_ids:
        raise ValueError("decision lacks policy evidence")
    if decision.rotation_window_days != 30:
        raise ValueError("result conflicts with P-7")
    if decision.decision == "eligible" and credential_age_days > decision.rotation_window_days:
        raise ValueError("eligible result exceeds P-7 window")
    return decision

checked = verify_p7(
    RotationDecision(
        decision="eligible",
        rotation_window_days=30,
        source_line_ids=["P-7"],
    ),
    credential_age_days=12,  # Trusted account record for acct_10234.
)

print("checked:", checked.decision, "under", checked.source_line_ids[0])

try:
    verify_p7(checked, credential_age_days=45)
except ValueError as error:
    print("rejected:", error)

Policy check

checked: eligible under P-7
rejected: eligible result exceeds P-7 window

This guardrail catches contradictions the server can prove from trusted data. It doesn't prove that free-form text describes real damage, so ambiguous reports can still become needs_review.

Give every request a trace

If an answer looks wrong tomorrow, you need more than the text shown on screen. A trace record is a small stored account of one request: its identifier, status, prompt version, policy evidence, and failure class when something went wrong.

Don't save raw customer descriptions merely because storage is easy. This local record stores a fingerprint of the report, enough to match repeat test inputs without retaining its text:

create-trace-record.py

from dataclasses import asdict, dataclass
import hmac
from hashlib import sha256

FINGERPRINT_KEY = b"local-demo-key"  # Demo only; inject a secret in production.

def fingerprint(text: str) -> str:
    return hmac.new(FINGERPRINT_KEY, text.encode("utf-8"), sha256).hexdigest()[:12]

@dataclass(frozen=True)
class TraceRecord:
    trace_id: str
    account_id: str
    status: str
    input_fingerprint: str
    prompt_version: str
    policy_line_ids: tuple[str, ...]

record = TraceRecord(
    trace_id="trace_acct_10234_01",
    account_id="acct_10234",
    status="completed",
    input_fingerprint=fingerprint("Service-account key is older than policy."),
    prompt_version="rotation_decision@2",
    policy_line_ids=("P-7",),
)

stored = asdict(record)
assert "Service-account key is older than policy." not in str(stored)
print(stored["trace_id"], stored["status"], stored["input_fingerprint"])

Redacted trace record

trace_acct_10234_01 completed c50985cbe665

A keyed fingerprint makes trivial offline guessing harder than a plain hash, but it isn't anonymization. Keep the key in server-side secret storage, choose retention and access rules for your risk, and omit the fingerprint when you don't need it for debugging or evaluation.

Make status durable

A Python dictionary vanishes when a process restarts. SQLite is enough for a local lab because it makes the transition from running to completed tangible without adding infrastructure.

This store writes the lifecycle your UI will show:

persist-status-transitions.py

import sqlite3

db = sqlite3.connect(":memory:")
db.execute(
    """
    CREATE TABLE tasks (
        trace_id TEXT PRIMARY KEY,
        account_id TEXT NOT NULL,
        status TEXT NOT NULL,
        decision TEXT,
        error_code TEXT
    )
    """
)

db.execute(
    "INSERT INTO tasks VALUES (?, ?, ?, ?, ?)",
    ("trace_acct_10234_01", "acct_10234", "running", None, None),
)
db.execute(
    "UPDATE tasks SET status = ?, decision = ? WHERE trace_id = ?",
    ("completed", "eligible", "trace_acct_10234_01"),
)

row = db.execute(
    "SELECT trace_id, status, decision FROM tasks WHERE trace_id = ?",
    ("trace_acct_10234_01",),
).fetchone()
print(row)

Stored lifecycle

('trace_acct_10234_01', 'completed', 'eligible')

Trace lifecycle for account acct_10234. Browser input is validated, written as a running trace, routed through the P-7 decision boundary, and resolved exactly once into either completed with eligible plus P-7 evidence or failed with provider_timeout. The UI mirrors those terminal states, and neither branch creates a rotation job. — Write `running` before the uncertain call, then persist exactly one terminal snapshot. `completed` carries checked evidence; `failed` carries a named error. Both records drive honest UI states without creating a rotation job.

Plug in the model boundary you already built

The previous chapter handled provider credentials, timeouts, structured parsing, policy evidence, and retry rules. This app shouldn't duplicate that code inside a route. Instead, it asks for a small dependency with one method, decide.

A fixture provider makes the contract runnable without a key or network access:

inject-decision-provider.py

from typing import Literal, Protocol

from pydantic import BaseModel

class RotationReport(BaseModel):
    account_id: str
    item: str
    credential_report: str

class RotationDecision(BaseModel):
    decision: Literal["eligible", "not_eligible", "needs_review"]
    rotation_window_days: int
    source_line_ids: list[str]

class DecisionProvider(Protocol):
    def decide(self, report: RotationReport) -> RotationDecision: ...

class FixtureProvider:
    def decide(self, report: RotationReport) -> RotationDecision:
        assert report.account_id == "acct_10234"
        return RotationDecision(
            decision="eligible",
            rotation_window_days=30,
            source_line_ids=["P-7"],
        )

CREDENTIAL_AGE_DAYS = {"acct_10234": 12}  # Trusted account records.

def evaluate_report(report: RotationReport, provider: DecisionProvider) -> RotationDecision:
    decision = provider.decide(report)
    if "P-7" not in decision.source_line_ids or decision.rotation_window_days != 30:
        raise ValueError("decision conflicts with P-7")
    if decision.decision == "eligible" and CREDENTIAL_AGE_DAYS[report.account_id] > decision.rotation_window_days:
        raise ValueError("eligible result exceeds P-7 window")
    return decision

result = evaluate_report(
    RotationReport(
        account_id="acct_10234",
        item="ceramic lamp",
        credential_report="Service-account key is stale.",
    ),
    FixtureProvider(),
)
print(result.model_dump())

Injected boundary

{'decision': 'eligible', 'rotation_window_days': 30, 'source_line_ids': ['P-7']}

Later, a provider-backed implementation can satisfy the same interface. The service still verifies business evidence after parsing. Unit tests remain fast because they inject a fixture or a deliberate failure instead of calling a remote model.

Orchestrate completed and failed requests

The application service sits between route and model boundary. It creates a running trace, calls the provider, verifies evidence, and records exactly one terminal result: completed or failed.

Watch both branches in one local example:

record-success-and-timeout.py

from dataclasses import dataclass

@dataclass
class Report:
    account_id: str
    credential_report: str

class GoodProvider:
    def decide(self, report: Report) -> dict:
        return {"decision": "eligible", "rotation_window_days": 30, "source_line_ids": ["P-7"]}

class TimeoutProvider:
    def decide(self, report: Report) -> dict:
        raise TimeoutError("provider deadline exceeded")

CREDENTIAL_AGE_DAYS = {"acct_10234": 12}  # Trusted account records.

def process(report: Report, provider) -> dict:
    trace = {"trace_id": "trace_acct_10234_01", "status": "running"}
    try:
        result = provider.decide(report)
        if "P-7" not in result["source_line_ids"] or result["rotation_window_days"] != 30:
            raise ValueError("missing evidence")
        if result["decision"] == "eligible" and CREDENTIAL_AGE_DAYS[report.account_id] > result["rotation_window_days"]:
            raise ValueError("eligible result exceeds P-7 window")
        trace.update(status="completed", decision=result["decision"])
    except TimeoutError:
        trace.update(status="failed", error_code="provider_timeout")
    except ValueError:
        trace.update(status="failed", error_code="invalid_decision")
    return trace

report = Report("acct_10234", "Service-account key is stale.")
print(process(report, GoodProvider())["status"])
print(process(report, TimeoutProvider())["error_code"])

Terminal statuses

completed
provider_timeout

A timeout isn't an empty answer, and it isn't permission to retry a side effect. The UI can tell the customer to retry the decision while a later rotation job-creation route uses its own idempotency rule.

Put a thin HTTP route in front

FastAPI is useful here because the same declared models validate incoming JSON and shape outgoing JSON.^[1] The route below has no prompt text and no API key. It handles HTTP concerns and delegates the decision.

post-rotation-decision.py

from itertools import count
from typing import Literal

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field

app = FastAPI()
TASKS: dict[str, dict] = {}
TRACE_IDS = count(1)

class RotationReport(BaseModel):
    account_id: str = Field(pattern=r"^acct_\d{5}$")
    credential_report: str = Field(min_length=10, max_length=500)

class DecisionResponse(BaseModel):
    trace_id: str
    status: Literal["completed", "failed"]
    decision: Literal["eligible", "not_eligible", "needs_review"] | None = None
    source_line_ids: list[str] = Field(default_factory=list)
    error_code: str | None = None
    error: str | None = None

def decide_rotation(report: RotationReport) -> tuple[Literal["eligible"], list[str]]:
    assert report.account_id == "acct_10234"
    return "eligible", ["P-7"]

@app.post("/rotation/decide", response_model=DecisionResponse)
def rotation_decide(report: RotationReport) -> DecisionResponse:
    trace_id = f"trace_{report.account_id}_{next(TRACE_IDS):02d}"
    TASKS[trace_id] = {"status": "running"}
    decision, evidence = decide_rotation(report)
    TASKS[trace_id] = {
        "status": "completed",
        "decision": decision,
        "source_line_ids": evidence,
    }
    return DecisionResponse(
        trace_id=trace_id,
        status="completed",
        decision=decision,
        source_line_ids=evidence,
    )

client = TestClient(app)
response = client.post(
    "/rotation/decide",
    json={"account_id": "acct_10234", "credential_report": "Service-account key is stale."},
)
print(response.status_code, response.json()["decision"], TASKS["trace_acct_10234_01"]["status"])

Route response

200 eligible completed

The stable response contract is what a browser consumes:

eligible-response.json

{
  "trace_id": "trace_acct_10234_01",
  "status": "completed",
  "decision": "eligible",
  "source_line_ids": ["P-7"],
  "error_code": null,
  "error": null
}

No raw model text enters the page. That matters because model output is untrusted content: OWASP's LLM application risks include prompt injection and improper output handling.^[2]

count() keeps this local output predictable while assigning a different trace ID to each request. A production deployment needs opaque, collision-resistant IDs and durable storage because process-local counters and dictionaries don't survive multiple workers or restarts.

Make the screen honest about state

A result card that displays only success will turn latency or failures into a mystery. Even with a synchronous route, the browser experiences four states:

UI state	Trigger	Message	Action allowed
Idle	Page opened	Describe stale service-account key.	Submit report.
Running	Request submitted	Checking policy P-7...	Prevent duplicate submit.
Completed	Checked response arrived	Eligible under P-7.	Review rotation job step.
Failed	Named error arrived	Decision unavailable; retry safely.	Retry decision only.

You can unit-test the display rule without running a browser:

render-ui-state.py

def view_text(state: str, payload: dict | None = None) -> str:
    if state == "idle":
        return "Describe stale service-account key"
    if state == "running":
        return "Checking policy P-7..."
    if state == "completed":
        return f"Eligible under {payload['source_line_ids'][0]}"
    return "Decision unavailable; retry safely"

success = {"source_line_ids": ["P-7"]}
print(view_text("running"))
print(view_text("completed", success))
print(view_text("failed", {"error_code": "provider_timeout"}))

Rendered states

Checking policy P-7...
Eligible under P-7
Decision unavailable; retry safely

Notice what the completed state doesn't say: it doesn't claim a rotation job was created or an approval was granted.

Test failures at your boundary

Tests shouldn't ask a hosted model to behave consistently. They should verify your deterministic code: validation, storage, provider-failure handling, and HTTP responses.

This route injects a service. A timeout fixture proves the customer gets an explicit failure while the stored trace remains useful:

test-provider-timeout.py

from typing import Literal

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field

app = FastAPI()
TASKS: dict[str, dict] = {}

class RotationReport(BaseModel):
    account_id: str
    credential_report: str

class DecisionResponse(BaseModel):
    trace_id: str
    status: Literal["completed", "failed"]
    decision: Literal["eligible", "not_eligible", "needs_review"] | None = None
    source_line_ids: list[str] = Field(default_factory=list)
    error_code: str | None = None
    error: str | None = None

class TimeoutService:
    def decide(self, report: RotationReport) -> dict:
        raise TimeoutError("provider deadline exceeded")

SERVICE = TimeoutService()

@app.post("/rotation/decide", response_model=DecisionResponse)
def rotation_decide(report: RotationReport) -> DecisionResponse:
    trace_id = "trace_acct_10234_timeout"
    TASKS[trace_id] = {"status": "running"}
    try:
        return SERVICE.decide(report)
    except TimeoutError:
        TASKS[trace_id] = {"status": "failed", "error_code": "provider_timeout"}
        error_message = "Decision unavailable; retry safely."
        return DecisionResponse(
            trace_id=trace_id,
            status="failed",
            error_code="provider_timeout",
            error=error_message,
        )

client = TestClient(app)
payload = client.post(
    "/rotation/decide",
    json={"account_id": "acct_10234", "credential_report": "Service-account key is stale."},
).json()

assert payload["status"] == "failed"
assert TASKS[payload["trace_id"]]["error_code"] == "provider_timeout"
print(payload)

Timeout test

{'trace_id': 'trace_acct_10234_timeout', 'status': 'failed', 'decision': None, 'source_line_ids': [], 'error_code': 'provider_timeout', 'error': 'Decision unavailable; retry safely.'}

Input rejection deserves a separate test because it occurs before a model boundary should run:

test-invalid-report.py

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field

app = FastAPI()
calls_to_model = 0

class RotationReport(BaseModel):
    account_id: str = Field(pattern=r"^acct_\d{5}$")
    credential_report: str = Field(min_length=10)

@app.post("/rotation/decide")
def rotation_decide(report: RotationReport) -> dict:
    global calls_to_model
    calls_to_model += 1
    return {"status": "completed"}

client = TestClient(app)
response = client.post(
    "/rotation/decide",
    json={"account_id": "wrong", "credential_report": "stale key"},
)

assert response.status_code == 422
assert calls_to_model == 0
print(response.status_code, "model_calls=", calls_to_model)

Input rejection

422 model_calls= 0

Add a health check and a tiny eval

A deployed service needs a cheap answer to "is the web process alive?" A health endpoint shouldn't call a paid or rate-limited model dependency. It only checks that the application can respond:

health-check-does-not-call-model.py

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()
model_calls = 0

@app.get("/healthz")
def healthz() -> dict:
    return {"status": "ok"}

client = TestClient(app)
response = client.get("/healthz")

assert response.status_code == 200
assert model_calls == 0
print(response.json(), "model_calls=", model_calls)

Health response

{'status': 'ok'} model_calls= 0

This lab's /healthz route is a liveness check: it proves the process responds. A production platform may also need a separate readiness check before sending traffic, for example while the app opens a required local file or database connection. Neither probe should call a paid hosted model.

Liveness doesn't prove the decision still follows policy. A generic classification smoke test isn't enough either. For example, this old-style check runs successfully, but it says nothing about rotation eligibility or policy evidence:

spot-an-inadequate-eval.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Ticket:
    category: str

text = "The app crashes when I upload rotation evidence."
ticket = Ticket(category="general")

print(text[:20].strip(), "=>", ticket.category)

Insufficient smoke check

The app crashes when => general

The output is stable, yet it hasn't tested P-7, an eligibility decision, or a stored trace. Keep a tiny eval set with expected evidence and outcomes instead. Start with obvious cases, then add real failures as you encounter them:

run-rotation-decision-eval.py

def fixture_decide(credential_age_days: int, stale: bool) -> tuple[str, list[str]]:
    if stale and credential_age_days <= 30:
        return "eligible", ["P-7"]
    return "needs_review", ["P-7"]

cases = [
    {"name": "stale in window", "days": 12, "stale": True, "expected": "eligible"},
    {"name": "stale after window", "days": 45, "stale": True, "expected": "needs_review"},
    {"name": "not marked stale", "days": 12, "stale": False, "expected": "needs_review"},
]

passed = 0
for case in cases:
    decision, evidence = fixture_decide(case["days"], case["stale"])
    ok = decision == case["expected"] and evidence == ["P-7"]
    passed += int(ok)
    print(case["name"], "=>", decision, "PASS" if ok else "FAIL")

assert passed == len(cases)
print("passed:", passed, "/", len(cases))

Pre-deploy eval

stale in window => eligible PASS
stale after window => needs_review PASS
not marked stale => needs_review PASS
passed: 3 / 3

Prepare a deployable artifact

The learning artifact now has a clear contract:

Artifact	Evidence it should contain
`app.py`	`/rotation/decide` and `/healthz` routes with typed input/output.
`service.py`	Provider boundary injection and P-7 verification.
`store.py`	Trace status updates with redacted credential data.
`tests/`	Invalid input, provider timeout, completed decision, and health check.
`evals/rotations.jsonl`	Small set of expected decisions and evidence lines.
`README.md`	Startup command, configuration, sample request, and known limits.

Docker can package a Python web app and its startup command into a repeatable container image, but it doesn't replace tests, secret injection, logging, or rollback planning.^[3]

Before exposing the app to users, answer these questions:

Check	Evidence
Secrets	Provider credentials enter only server runtime configuration.
Startup	A clean checkout starts with one documented command.
Liveness	`/healthz` returns `200` without a model request.
Traceability	You can find a failed `trace_id` and its named error.
Behavior	Tiny eval cases pass before deploy.
Limits	README states that eligibility isn't rotation-job creation or approval execution.

Production machine-learning systems accumulate debt when model behavior, data dependencies, and serving code aren't tracked together.^[4]^[5] Your first app is small enough to build that habit correctly from the start.

Diagnose a broken version

Suppose a teammate ships this alternative:

text

The browser calls the hosted model directly with a public key.
The response text is inserted into a green "Rotation approved" card.
No trace is written when the provider times out.
The only test asserts that the page loads.

Write down four defects before reading the table:

Defect	Consequence	Repair
Browser holds the model key.	Credential can be exposed and abused.	Send request to a server route; keep secret server-side.
Raw output becomes approval.	Model text can trigger an unsupported customer promise.	Validate decision and policy evidence; keep actions separate.
Timeout leaves no record.	Team can't distinguish slow provider from broken UI.	Persist a failed trace with named error code.
Page-load test only.	Core behavior can regress unnoticed.	Test invalid input, timeout, completed result, health, and eval cases.

The repair isn't more infrastructure. It's a smaller contract with explicit states and evidence.

Mastery check

Key concepts

Smallest useful AI workflow from browser input to checked response
FastAPI request and response schemas
Injected model boundary rather than provider code in routes
Redacted trace records and terminal statuses
Failure tests, health checks, and pre-deploy eval cases

Evaluation rubric

Foundational: Defines a useful AI task that doesn't hide side effects
Foundational: Validates incoming report data and outgoing decision data before display
Foundational: Explains idle, running, completed, and failed user states
Intermediate: Stores a redacted trace with prompt version, status, evidence, and named errors
Intermediate: Tests validation and provider failure without a live model dependency
Intermediate: Separates service health from model-behavior evaluation before deployment

Follow-up questions

Common pitfalls

Symptom	Cause	Fix
Customer sees an unsupported rotation promise.	UI renders model text as an action.	Render only validated decision fields and keep jobs/approvals separate.
Bug report has no reproduction path.	Request status, evidence, or prompt version wasn't stored.	Write a redacted trace for completed and failed outcomes.
Unit tests are slow or flaky.	They call a hosted model.	Inject fixture providers and test deterministic app behavior.
Service is "healthy" while decisions regress.	Health check was treated as an eval.	Run both liveness checks and policy-rotation-job eval cases.
Secrets appear in browser tooling.	Provider calls were made client-side.	Put the model boundary behind the server route.

Next Step

Continue to The LLM Lifecycle

You have shipped one traceable app request around a model boundary. Next, zoom out to understand how the models behind that boundary are trained, adapted, evaluated, and served.

PreviousCalling LLM APIs in Production

Share this article

X Facebook LinkedIn Bluesky Reddit Hacker News Email

References

FastAPI Documentation.

FastAPI Project. · 2026 · Official documentation

OWASP Top 10 for Large Language Model Applications

OWASP Foundation · 2025

Docker Documentation.

Docker Inc. · 2026 · Official documentation

Hidden Technical Debt in Machine Learning Systems.

Sculley et al. · 2015

Challenges in Deploying Machine Learning: a Survey of Case Studies.

Paleyes, A., Urma, R. G., & Lawrence, N. D. · 2022 · ACM Computing Surveys

Back to Topics

LearnPreparation & PrerequisitesFirst AI App End-to-End

⚙️EasyMLOps & Deployment

First AI App End-to-End

Ship one traceable rotation-decision workflow: validated input, model boundary, stored status, clear UI states, failure tests, and deploy checks.

13 min read

Learning path

Step 29 of 158 in the full curriculum

Calling LLM APIs in Production The LLM Lifecycle

Define the one job your app does

A first AI app doesn't need chat history, autonomous tools, or a retrieval system. It needs one user problem with an observable answer:

Given a credential report for account acct_10234, decide whether policy line P-7 supports a rotation, cite that line in the result, and never create a rotation job as a hidden side effect.

That sentence fixes the app boundary before you choose a framework:

Layer	Responsibility in this app	What it must not do
Browser	Collect account ID and credential report; show state.	Call a model with a secret key.
API route	Validate request; return stable response fields.	Embed provider-specific prompt logic.
Decision service	Call the checked wrapper; translate outcome into status.	Issue a rotation job.
Trace store	Save status, evidence, latency, and redacted input fingerprint.	Store unnecessary customer text.
Tests	Prove completion, rejection, timeout, and health behavior.	Spend money on model calls.

Start with data the browser and server agree on

This first runnable cell rejects missing identifiers and constrains the response to the decisions your UI understands:

define-rotation-contract.py

from typing import Literal

from pydantic import BaseModel, Field

class RotationReport(BaseModel):
    account_id: str = Field(pattern=r"^acct_\d{5}$")
    item: str = Field(min_length=3, max_length=80)
    credential_report: str = Field(min_length=10, max_length=500)

class RotationDecision(BaseModel):
    decision: Literal["eligible", "not_eligible", "needs_review"]
    rotation_window_days: int = Field(ge=0, le=30)
    source_line_ids: list[str] = Field(min_length=1)

report = RotationReport(
    account_id="acct_10234",
    item="ceramic lamp",
    credential_report="Service-account key is older than policy.",
)
decision = RotationDecision(
    decision="eligible",
    rotation_window_days=30,
    source_line_ids=["P-7"],
)

print(report.account_id, "=>", decision.decision, decision.source_line_ids)

Validated contract

acct_10234 => eligible ['P-7']

The schema proves field shape, not truth. An output can match the schema and still cite the wrong policy or claim a window your policy doesn't allow.

check-policy-evidence.py

from typing import Literal

from pydantic import BaseModel, Field

class RotationDecision(BaseModel):
    decision: Literal["eligible", "not_eligible", "needs_review"]
    rotation_window_days: int = Field(ge=0, le=90)
    source_line_ids: list[str] = Field(min_length=1)

def verify_p7(decision: RotationDecision, *, credential_age_days: int) -> RotationDecision:
    if "P-7" not in decision.source_line_ids:
        raise ValueError("decision lacks policy evidence")
    if decision.rotation_window_days != 30:
        raise ValueError("result conflicts with P-7")
    if decision.decision == "eligible" and credential_age_days > decision.rotation_window_days:
        raise ValueError("eligible result exceeds P-7 window")
    return decision

checked = verify_p7(
    RotationDecision(
        decision="eligible",
        rotation_window_days=30,
        source_line_ids=["P-7"],
    ),
    credential_age_days=12,  # Trusted account record for acct_10234.
)

print("checked:", checked.decision, "under", checked.source_line_ids[0])

try:
    verify_p7(checked, credential_age_days=45)
except ValueError as error:
    print("rejected:", error)

Policy check

checked: eligible under P-7
rejected: eligible result exceeds P-7 window

This guardrail catches contradictions the server can prove from trusted data. It doesn't prove that free-form text describes real damage, so ambiguous reports can still become needs_review.

Give every request a trace

Don't save raw customer descriptions merely because storage is easy. This local record stores a fingerprint of the report, enough to match repeat test inputs without retaining its text:

create-trace-record.py

from dataclasses import asdict, dataclass
import hmac
from hashlib import sha256

FINGERPRINT_KEY = b"local-demo-key"  # Demo only; inject a secret in production.

def fingerprint(text: str) -> str:
    return hmac.new(FINGERPRINT_KEY, text.encode("utf-8"), sha256).hexdigest()[:12]

@dataclass(frozen=True)
class TraceRecord:
    trace_id: str
    account_id: str
    status: str
    input_fingerprint: str
    prompt_version: str
    policy_line_ids: tuple[str, ...]

record = TraceRecord(
    trace_id="trace_acct_10234_01",
    account_id="acct_10234",
    status="completed",
    input_fingerprint=fingerprint("Service-account key is older than policy."),
    prompt_version="rotation_decision@2",
    policy_line_ids=("P-7",),
)

stored = asdict(record)
assert "Service-account key is older than policy." not in str(stored)
print(stored["trace_id"], stored["status"], stored["input_fingerprint"])

Redacted trace record

trace_acct_10234_01 completed c50985cbe665

Make status durable

A Python dictionary vanishes when a process restarts. SQLite is enough for a local lab because it makes the transition from running to completed tangible without adding infrastructure.

This store writes the lifecycle your UI will show:

persist-status-transitions.py

import sqlite3

db = sqlite3.connect(":memory:")
db.execute(
    """
    CREATE TABLE tasks (
        trace_id TEXT PRIMARY KEY,
        account_id TEXT NOT NULL,
        status TEXT NOT NULL,
        decision TEXT,
        error_code TEXT
    )
    """
)

db.execute(
    "INSERT INTO tasks VALUES (?, ?, ?, ?, ?)",
    ("trace_acct_10234_01", "acct_10234", "running", None, None),
)
db.execute(
    "UPDATE tasks SET status = ?, decision = ? WHERE trace_id = ?",
    ("completed", "eligible", "trace_acct_10234_01"),
)

row = db.execute(
    "SELECT trace_id, status, decision FROM tasks WHERE trace_id = ?",
    ("trace_acct_10234_01",),
).fetchone()
print(row)

Stored lifecycle

('trace_acct_10234_01', 'completed', 'eligible')

Plug in the model boundary you already built

A fixture provider makes the contract runnable without a key or network access:

inject-decision-provider.py

from typing import Literal, Protocol

from pydantic import BaseModel

class RotationReport(BaseModel):
    account_id: str
    item: str
    credential_report: str

class RotationDecision(BaseModel):
    decision: Literal["eligible", "not_eligible", "needs_review"]
    rotation_window_days: int
    source_line_ids: list[str]

class DecisionProvider(Protocol):
    def decide(self, report: RotationReport) -> RotationDecision: ...

class FixtureProvider:
    def decide(self, report: RotationReport) -> RotationDecision:
        assert report.account_id == "acct_10234"
        return RotationDecision(
            decision="eligible",
            rotation_window_days=30,
            source_line_ids=["P-7"],
        )

CREDENTIAL_AGE_DAYS = {"acct_10234": 12}  # Trusted account records.

def evaluate_report(report: RotationReport, provider: DecisionProvider) -> RotationDecision:
    decision = provider.decide(report)
    if "P-7" not in decision.source_line_ids or decision.rotation_window_days != 30:
        raise ValueError("decision conflicts with P-7")
    if decision.decision == "eligible" and CREDENTIAL_AGE_DAYS[report.account_id] > decision.rotation_window_days:
        raise ValueError("eligible result exceeds P-7 window")
    return decision

result = evaluate_report(
    RotationReport(
        account_id="acct_10234",
        item="ceramic lamp",
        credential_report="Service-account key is stale.",
    ),
    FixtureProvider(),
)
print(result.model_dump())

Injected boundary

{'decision': 'eligible', 'rotation_window_days': 30, 'source_line_ids': ['P-7']}

Orchestrate completed and failed requests

The application service sits between route and model boundary. It creates a running trace, calls the provider, verifies evidence, and records exactly one terminal result: completed or failed.

Watch both branches in one local example:

record-success-and-timeout.py

from dataclasses import dataclass

@dataclass
class Report:
    account_id: str
    credential_report: str

class GoodProvider:
    def decide(self, report: Report) -> dict:
        return {"decision": "eligible", "rotation_window_days": 30, "source_line_ids": ["P-7"]}

class TimeoutProvider:
    def decide(self, report: Report) -> dict:
        raise TimeoutError("provider deadline exceeded")

CREDENTIAL_AGE_DAYS = {"acct_10234": 12}  # Trusted account records.

def process(report: Report, provider) -> dict:
    trace = {"trace_id": "trace_acct_10234_01", "status": "running"}
    try:
        result = provider.decide(report)
        if "P-7" not in result["source_line_ids"] or result["rotation_window_days"] != 30:
            raise ValueError("missing evidence")
        if result["decision"] == "eligible" and CREDENTIAL_AGE_DAYS[report.account_id] > result["rotation_window_days"]:
            raise ValueError("eligible result exceeds P-7 window")
        trace.update(status="completed", decision=result["decision"])
    except TimeoutError:
        trace.update(status="failed", error_code="provider_timeout")
    except ValueError:
        trace.update(status="failed", error_code="invalid_decision")
    return trace

report = Report("acct_10234", "Service-account key is stale.")
print(process(report, GoodProvider())["status"])
print(process(report, TimeoutProvider())["error_code"])

Terminal statuses

completed
provider_timeout

Put a thin HTTP route in front

post-rotation-decision.py

from itertools import count
from typing import Literal

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field

app = FastAPI()
TASKS: dict[str, dict] = {}
TRACE_IDS = count(1)

class RotationReport(BaseModel):
    account_id: str = Field(pattern=r"^acct_\d{5}$")
    credential_report: str = Field(min_length=10, max_length=500)

class DecisionResponse(BaseModel):
    trace_id: str
    status: Literal["completed", "failed"]
    decision: Literal["eligible", "not_eligible", "needs_review"] | None = None
    source_line_ids: list[str] = Field(default_factory=list)
    error_code: str | None = None
    error: str | None = None

def decide_rotation(report: RotationReport) -> tuple[Literal["eligible"], list[str]]:
    assert report.account_id == "acct_10234"
    return "eligible", ["P-7"]

@app.post("/rotation/decide", response_model=DecisionResponse)
def rotation_decide(report: RotationReport) -> DecisionResponse:
    trace_id = f"trace_{report.account_id}_{next(TRACE_IDS):02d}"
    TASKS[trace_id] = {"status": "running"}
    decision, evidence = decide_rotation(report)
    TASKS[trace_id] = {
        "status": "completed",
        "decision": decision,
        "source_line_ids": evidence,
    }
    return DecisionResponse(
        trace_id=trace_id,
        status="completed",
        decision=decision,
        source_line_ids=evidence,
    )

client = TestClient(app)
response = client.post(
    "/rotation/decide",
    json={"account_id": "acct_10234", "credential_report": "Service-account key is stale."},
)
print(response.status_code, response.json()["decision"], TASKS["trace_acct_10234_01"]["status"])

Route response

200 eligible completed

The stable response contract is what a browser consumes:

eligible-response.json

{
  "trace_id": "trace_acct_10234_01",
  "status": "completed",
  "decision": "eligible",
  "source_line_ids": ["P-7"],
  "error_code": null,
  "error": null
}

No raw model text enters the page. That matters because model output is untrusted content: OWASP's LLM application risks include prompt injection and improper output handling.^[2]

Make the screen honest about state

A result card that displays only success will turn latency or failures into a mystery. Even with a synchronous route, the browser experiences four states:

UI state	Trigger	Message	Action allowed
Idle	Page opened	Describe stale service-account key.	Submit report.
Running	Request submitted	Checking policy P-7...	Prevent duplicate submit.
Completed	Checked response arrived	Eligible under P-7.	Review rotation job step.
Failed	Named error arrived	Decision unavailable; retry safely.	Retry decision only.

You can unit-test the display rule without running a browser:

render-ui-state.py

def view_text(state: str, payload: dict | None = None) -> str:
    if state == "idle":
        return "Describe stale service-account key"
    if state == "running":
        return "Checking policy P-7..."
    if state == "completed":
        return f"Eligible under {payload['source_line_ids'][0]}"
    return "Decision unavailable; retry safely"

success = {"source_line_ids": ["P-7"]}
print(view_text("running"))
print(view_text("completed", success))
print(view_text("failed", {"error_code": "provider_timeout"}))

Rendered states

Checking policy P-7...
Eligible under P-7
Decision unavailable; retry safely

Notice what the completed state doesn't say: it doesn't claim a rotation job was created or an approval was granted.

Test failures at your boundary

Tests shouldn't ask a hosted model to behave consistently. They should verify your deterministic code: validation, storage, provider-failure handling, and HTTP responses.

This route injects a service. A timeout fixture proves the customer gets an explicit failure while the stored trace remains useful:

test-provider-timeout.py

from typing import Literal

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field

app = FastAPI()
TASKS: dict[str, dict] = {}

class RotationReport(BaseModel):
    account_id: str
    credential_report: str

class DecisionResponse(BaseModel):
    trace_id: str
    status: Literal["completed", "failed"]
    decision: Literal["eligible", "not_eligible", "needs_review"] | None = None
    source_line_ids: list[str] = Field(default_factory=list)
    error_code: str | None = None
    error: str | None = None

class TimeoutService:
    def decide(self, report: RotationReport) -> dict:
        raise TimeoutError("provider deadline exceeded")

SERVICE = TimeoutService()

@app.post("/rotation/decide", response_model=DecisionResponse)
def rotation_decide(report: RotationReport) -> DecisionResponse:
    trace_id = "trace_acct_10234_timeout"
    TASKS[trace_id] = {"status": "running"}
    try:
        return SERVICE.decide(report)
    except TimeoutError:
        TASKS[trace_id] = {"status": "failed", "error_code": "provider_timeout"}
        error_message = "Decision unavailable; retry safely."
        return DecisionResponse(
            trace_id=trace_id,
            status="failed",
            error_code="provider_timeout",
            error=error_message,
        )

client = TestClient(app)
payload = client.post(
    "/rotation/decide",
    json={"account_id": "acct_10234", "credential_report": "Service-account key is stale."},
).json()

assert payload["status"] == "failed"
assert TASKS[payload["trace_id"]]["error_code"] == "provider_timeout"
print(payload)

Timeout test

{'trace_id': 'trace_acct_10234_timeout', 'status': 'failed', 'decision': None, 'source_line_ids': [], 'error_code': 'provider_timeout', 'error': 'Decision unavailable; retry safely.'}

Input rejection deserves a separate test because it occurs before a model boundary should run:

test-invalid-report.py

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field

app = FastAPI()
calls_to_model = 0

class RotationReport(BaseModel):
    account_id: str = Field(pattern=r"^acct_\d{5}$")
    credential_report: str = Field(min_length=10)

@app.post("/rotation/decide")
def rotation_decide(report: RotationReport) -> dict:
    global calls_to_model
    calls_to_model += 1
    return {"status": "completed"}

client = TestClient(app)
response = client.post(
    "/rotation/decide",
    json={"account_id": "wrong", "credential_report": "stale key"},
)

assert response.status_code == 422
assert calls_to_model == 0
print(response.status_code, "model_calls=", calls_to_model)

Input rejection

422 model_calls= 0

Add a health check and a tiny eval

A deployed service needs a cheap answer to "is the web process alive?" A health endpoint shouldn't call a paid or rate-limited model dependency. It only checks that the application can respond:

health-check-does-not-call-model.py

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()
model_calls = 0

@app.get("/healthz")
def healthz() -> dict:
    return {"status": "ok"}

client = TestClient(app)
response = client.get("/healthz")

assert response.status_code == 200
assert model_calls == 0
print(response.json(), "model_calls=", model_calls)

Health response

{'status': 'ok'} model_calls= 0

spot-an-inadequate-eval.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Ticket:
    category: str

text = "The app crashes when I upload rotation evidence."
ticket = Ticket(category="general")

print(text[:20].strip(), "=>", ticket.category)

Insufficient smoke check

The app crashes when => general

run-rotation-decision-eval.py

def fixture_decide(credential_age_days: int, stale: bool) -> tuple[str, list[str]]:
    if stale and credential_age_days <= 30:
        return "eligible", ["P-7"]
    return "needs_review", ["P-7"]

cases = [
    {"name": "stale in window", "days": 12, "stale": True, "expected": "eligible"},
    {"name": "stale after window", "days": 45, "stale": True, "expected": "needs_review"},
    {"name": "not marked stale", "days": 12, "stale": False, "expected": "needs_review"},
]

passed = 0
for case in cases:
    decision, evidence = fixture_decide(case["days"], case["stale"])
    ok = decision == case["expected"] and evidence == ["P-7"]
    passed += int(ok)
    print(case["name"], "=>", decision, "PASS" if ok else "FAIL")

assert passed == len(cases)
print("passed:", passed, "/", len(cases))

Pre-deploy eval

stale in window => eligible PASS
stale after window => needs_review PASS
not marked stale => needs_review PASS
passed: 3 / 3

Prepare a deployable artifact

The learning artifact now has a clear contract:

Artifact	Evidence it should contain
`app.py`	`/rotation/decide` and `/healthz` routes with typed input/output.
`service.py`	Provider boundary injection and P-7 verification.
`store.py`	Trace status updates with redacted credential data.
`tests/`	Invalid input, provider timeout, completed decision, and health check.
`evals/rotations.jsonl`	Small set of expected decisions and evidence lines.
`README.md`	Startup command, configuration, sample request, and known limits.

Docker can package a Python web app and its startup command into a repeatable container image, but it doesn't replace tests, secret injection, logging, or rollback planning.^[3]

Before exposing the app to users, answer these questions:

Check	Evidence
Secrets	Provider credentials enter only server runtime configuration.
Startup	A clean checkout starts with one documented command.
Liveness	`/healthz` returns `200` without a model request.
Traceability	You can find a failed `trace_id` and its named error.
Behavior	Tiny eval cases pass before deploy.
Limits	README states that eligibility isn't rotation-job creation or approval execution.

Diagnose a broken version

Suppose a teammate ships this alternative:

text

The browser calls the hosted model directly with a public key.
The response text is inserted into a green "Rotation approved" card.
No trace is written when the provider times out.
The only test asserts that the page loads.

Write down four defects before reading the table:

Defect	Consequence	Repair
Browser holds the model key.	Credential can be exposed and abused.	Send request to a server route; keep secret server-side.
Raw output becomes approval.	Model text can trigger an unsupported customer promise.	Validate decision and policy evidence; keep actions separate.
Timeout leaves no record.	Team can't distinguish slow provider from broken UI.	Persist a failed trace with named error code.
Page-load test only.	Core behavior can regress unnoticed.	Test invalid input, timeout, completed result, health, and eval cases.

The repair isn't more infrastructure. It's a smaller contract with explicit states and evidence.

Mastery check

Key concepts

Smallest useful AI workflow from browser input to checked response
FastAPI request and response schemas
Injected model boundary rather than provider code in routes
Redacted trace records and terminal statuses
Failure tests, health checks, and pre-deploy eval cases

Evaluation rubric

Foundational: Defines a useful AI task that doesn't hide side effects
Foundational: Validates incoming report data and outgoing decision data before display
Foundational: Explains idle, running, completed, and failed user states
Intermediate: Stores a redacted trace with prompt version, status, evidence, and named errors
Intermediate: Tests validation and provider failure without a live model dependency
Intermediate: Separates service health from model-behavior evaluation before deployment

Follow-up questions

Common pitfalls

Symptom	Cause	Fix
Customer sees an unsupported rotation promise.	UI renders model text as an action.	Render only validated decision fields and keep jobs/approvals separate.
Bug report has no reproduction path.	Request status, evidence, or prompt version wasn't stored.	Write a redacted trace for completed and failed outcomes.
Unit tests are slow or flaky.	They call a hosted model.	Inject fixture providers and test deterministic app behavior.
Service is "healthy" while decisions regress.	Health check was treated as an eval.	Run both liveness checks and policy-rotation-job eval cases.
Secrets appear in browser tooling.	Provider calls were made client-side.	Put the model boundary behind the server route.

Next Step

Continue to The LLM Lifecycle

You have shipped one traceable app request around a model boundary. Next, zoom out to understand how the models behind that boundary are trained, adapted, evaluated, and served.

PreviousCalling LLM APIs in Production

Share this article

X Facebook LinkedIn Bluesky Reddit Hacker News Email

References

FastAPI Documentation.

FastAPI Project. · 2026 · Official documentation

OWASP Top 10 for Large Language Model Applications

OWASP Foundation · 2025

Docker Documentation.

Docker Inc. · 2026 · Official documentation

Hidden Technical Debt in Machine Learning Systems.

Sculley et al. · 2015

Challenges in Deploying Machine Learning: a Survey of Case Studies.

Paleyes, A., Urma, R. G., & Lawrence, N. D. · 2022 · ACM Computing Surveys

First AI App End-to-End

Define the one job your app does

Start with data the browser and server agree on

Give every request a trace

Make status durable

Plug in the model boundary you already built

Orchestrate completed and failed requests

Put a thin HTTP route in front

Make the screen honest about state

Test failures at your boundary

Add a health check and a tiny eval

Prepare a deployable artifact

Diagnose a broken version

Mastery check

Key concepts

Evaluation rubric

Follow-up questions

Common pitfalls

Mastery Check

First AI App End-to-End

Define the one job your app does

Start with data the browser and server agree on

Give every request a trace

Make status durable

Plug in the model boundary you already built

Orchestrate completed and failed requests

Put a thin HTTP route in front

Make the screen honest about state

Test failures at your boundary

Add a health check and a tiny eval

Prepare a deployable artifact

Diagnose a broken version

Mastery check

Key concepts

Evaluation rubric

Follow-up questions

Common pitfalls

Mastery Check

First AI App End-to-End

Define the one job your app does

Why is a rotation decision useful even though it doesn't create a rotation job?

Start with data the browser and server agree on

Give every request a trace

Make status durable

Plug in the model boundary you already built

Orchestrate completed and failed requests

Put a thin HTTP route in front

Make the screen honest about state

Test failures at your boundary

Add a health check and a tiny eval

Prepare a deployable artifact

Diagnose a broken version

Mastery check

Key concepts

Evaluation rubric

Follow-up questions

Why does the app save a trace before returning the decision?

Why is a typed response insufficient before showing "eligible"?

What should unit tests replace, and what should they keep real?

Why doesn't /healthz call the model?

Common pitfalls

Mastery Check

First AI App End-to-End

Define the one job your app does

Why is a rotation decision useful even though it doesn't create a rotation job?

Start with data the browser and server agree on

Give every request a trace

Make status durable

Plug in the model boundary you already built

Orchestrate completed and failed requests

Put a thin HTTP route in front

Make the screen honest about state

Test failures at your boundary

Add a health check and a tiny eval

Prepare a deployable artifact

Diagnose a broken version

Mastery check

Key concepts

Evaluation rubric

Follow-up questions

Why does the app save a trace before returning the decision?

Why is a typed response insufficient before showing "eligible"?

What should unit tests replace, and what should they keep real?

Why doesn't /healthz call the model?

Common pitfalls

Mastery Check