Project: production-ready scalable LLM API

You reached the end of the module. You're going to integrate everything you learned — image with vLLM, A10G GPU, persistent volume, FastAPI, configured autoscaling, authentication, basic monitoring — into a deliverable you could take to your portfolio or use in a real project.

This is the deliverable that Module 8 (Unified Client) will use as one of the providers behind the unified client. After this project, you already have an LLM endpoint of your own, on your own infrastructure, ready for production.

By the end of this project you'll have:

• A public HTTPS endpoint /chat with token auth
• Healthcheck public /health with the model status
• Metrics observable via /metrics and structured logs
• Autoscaling configured for reasonable P50 latency and bounded cost
• Tests that validate the contract and authentication
• README with setup, expected costs, and how to use it

---

Functional specification

Endpoints

Method	Path	Auth	Description
GET	/health	No	Service status: loaded model, version, uptime
GET	/metrics	Bearer admin	Counters: total requests, errors, P50/P95 latency
POST	/chat	Bearer user	Generates an LLM response for a prompt
POST	/chat-batch	Bearer user	Processes up to 10 prompts in one call (batching)

Contracts

POST /chat

// Request
{
  "prompt": "Explain REST in 2 sentences.",
  "max_tokens": 256,
  "temperature": 0.7,
  "request_id": "optional-client-uuid"
}

// Response 200
{
  "response": "REST is a style...",
  "model": "mistralai/Mistral-7B-Instruct-v0.3",
  "generated_tokens": 78,
  "duration_ms": 1820,
  "request_id": "optional-client-uuid"
}

// Response 401 / 403 / 422
{
  "detail": "Specific error message"
}

Non-functional

• Healthcheck available even if the GPU container is off
• Structured logs in JSON (request_id, latency, model, status)
• P50 < 3s, P99 < 30s under normal load (with an active warm pool)
• Cost cap: maximum 5 parallel containers

---

Complete implementation

Create final_project.py:

# final_project.py
import os
import json
import time
import uuid
import modal
from collections import deque
from typing import Annotated
from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel, Field

# ============================================================
# Configuration
# ============================================================
MODEL = "mistralai/Mistral-7B-Instruct-v0.3"
CACHE_DIR = "/cache/huggingface"
APP_VERSION = "1.0.0"

app = modal.App("llm-api-final")
weights_volume = modal.Volume.from_name("mistral-weights", create_if_missing=True)

image = (
    modal.Image.debian_slim(python_version="3.11")
    .pip_install(
        "vllm==0.6.3",
        "huggingface_hub[hf_transfer]==0.26.2",
        "fastapi[standard]",
    )
    .env({"HF_HOME": CACHE_DIR, "HF_HUB_ENABLE_HF_TRANSFER": "1"})
)

# ============================================================
# Pydantic contracts
# ============================================================
class ChatRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=4000)
    max_tokens: int = Field(256, ge=1, le=2048)
    temperature: float = Field(0.7, ge=0.0, le=2.0)
    request_id: str | None = None


class ChatResponse(BaseModel):
    response: str
    model: str
    generated_tokens: int
    duration_ms: int
    request_id: str


class BatchRequest(BaseModel):
    prompts: list[str] = Field(..., min_length=1, max_length=10)
    max_tokens: int = Field(256, ge=1, le=2048)
    temperature: float = Field(0.7, ge=0.0, le=2.0)


class BatchResponse(BaseModel):
    responses: list[str]
    model: str
    duration_ms: int


class HealthResponse(BaseModel):
    status: str
    model: str
    version: str
    uptime_seconds: int


class MetricsResponse(BaseModel):
    total_requests: int
    total_errors: int
    latency_p50_ms: int
    latency_p95_ms: int
    last_updated: str


# ============================================================
# In-process state (simple metrics)
# ============================================================
PROCESS_START = time.time()
metrics_state = {
    "total_requests": 0,
    "total_errors": 0,
    "latencies": deque(maxlen=1000),
}


def record_request(duration_ms: int, error: bool = False):
    metrics_state["total_requests"] += 1
    if error:
        metrics_state["total_errors"] += 1
    else:
        metrics_state["latencies"].append(duration_ms)


def percentile(values, p):
    if not values:
        return 0
    ordered = sorted(values)
    idx = int(len(ordered) * p / 100)
    return ordered[min(idx, len(ordered) - 1)]


# ============================================================
# Authentication
# ============================================================
def verify_token(authorization: str | None, kind: str = "user"):
    env_var = "API_TOKEN_USER" if kind == "user" else "API_TOKEN_ADMIN"
    expected_token = os.environ.get(env_var)

    if not expected_token:
        raise HTTPException(500, f"{env_var} not configured on the server")
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(401, "Missing 'Authorization: Bearer <token>' header")
    if authorization.removeprefix("Bearer ") != expected_token:
        raise HTTPException(403, "Invalid token")


def log_structured(event: str, **kwargs):
    print(json.dumps({"event": event, "ts": time.time(), **kwargs}))


# ============================================================
# LLM service with GPU
# ============================================================
@app.cls(
    image=image,
    gpu="A10G",
    volumes={CACHE_DIR: weights_volume},
    timeout=600,
    min_containers=0,            # Don't pay for GPU 24/7 by default
    max_containers=5,            # Bound cost
    scaledown_window=300,        # 5min of grace
)
class MistralService:
    @modal.enter()
    def load_model(self):
        from vllm import LLM
        log_structured("model_loading", model=MODEL)
        start = time.time()
        self.llm = LLM(model=MODEL, download_dir=CACHE_DIR)
        log_structured(
            "model_ready",
            duration_s=round(time.time() - start, 2),
        )

    @modal.method()
    def generate(self, prompt: str, max_tokens: int, temperature: float) -> dict:
        from vllm import SamplingParams
        sampling = SamplingParams(temperature=temperature, max_tokens=max_tokens)
        start = time.time()
        formatted_prompt = f"[INST] {prompt} [/INST]"
        output = self.llm.generate(formatted_prompt, sampling)[0]
        return {
            "text": output.outputs[0].text.strip(),
            "tokens": len(output.outputs[0].token_ids),
            "duration_ms": int((time.time() - start) * 1000),
        }

    @modal.method()
    def generate_batch(
        self, prompts: list[str], max_tokens: int, temperature: float
    ) -> dict:
        from vllm import SamplingParams
        sampling = SamplingParams(temperature=temperature, max_tokens=max_tokens)
        start = time.time()
        formatted_prompts = [f"[INST] {p} [/INST]" for p in prompts]
        outputs = self.llm.generate(formatted_prompts, sampling)
        return {
            "responses": [o.outputs[0].text.strip() for o in outputs],
            "duration_ms": int((time.time() - start) * 1000),
        }


# ============================================================
# Web function (FastAPI ASGI)
# ============================================================
@app.function(
    image=image,
    secrets=[modal.Secret.from_name("llm-api-tokens")],
    min_containers=1,           # Web tier always warm (it's cheap, CPU only)
    max_containers=3,
)
@modal.asgi_app()
def web():
    web_app = FastAPI(
        title="LLM API",
        version=APP_VERSION,
        description="Mistral 7B endpoint on Modal with autoscaling and monitoring",
    )

    @web_app.get("/health", response_model=HealthResponse)
    def health():
        return HealthResponse(
            status="ok",
            model=MODEL,
            version=APP_VERSION,
            uptime_seconds=int(time.time() - PROCESS_START),
        )

    @web_app.get("/metrics", response_model=MetricsResponse)
    def metrics(authorization: Annotated[str | None, Header()] = None):
        verify_token(authorization, kind="admin")
        latencies = list(metrics_state["latencies"])
        return MetricsResponse(
            total_requests=metrics_state["total_requests"],
            total_errors=metrics_state["total_errors"],
            latency_p50_ms=percentile(latencies, 50),
            latency_p95_ms=percentile(latencies, 95),
            last_updated=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        )

    @web_app.post("/chat", response_model=ChatResponse)
    def chat(
        req: ChatRequest,
        authorization: Annotated[str | None, Header()] = None,
    ):
        verify_token(authorization, kind="user")
        rid = req.request_id or str(uuid.uuid4())
        start = time.time()
        try:
            service = MistralService()
            result = service.generate.remote(
                req.prompt, req.max_tokens, req.temperature
            )
            total_duration = int((time.time() - start) * 1000)
            record_request(total_duration)
            log_structured(
                "chat_ok",
                request_id=rid,
                duration_ms=total_duration,
                tokens=result["tokens"],
            )
            return ChatResponse(
                response=result["text"],
                model=MODEL,
                generated_tokens=result["tokens"],
                duration_ms=total_duration,
                request_id=rid,
            )
        except Exception as e:
            record_request(0, error=True)
            log_structured("chat_error", request_id=rid, error=str(e))
            raise HTTPException(500, f"Error generating response: {e}")

    @web_app.post("/chat-batch", response_model=BatchResponse)
    def chat_batch(
        req: BatchRequest,
        authorization: Annotated[str | None, Header()] = None,
    ):
        verify_token(authorization, kind="user")
        start = time.time()
        service = MistralService()
        result = service.generate_batch.remote(
            req.prompts, req.max_tokens, req.temperature
        )
        total_duration = int((time.time() - start) * 1000)
        record_request(total_duration)
        return BatchResponse(
            responses=result["responses"],
            model=MODEL,
            duration_ms=total_duration,
        )

    return web_app


# ============================================================
# Keep-warm during business hours (optional)
# ============================================================
@app.function(schedule=modal.Period(minutes=5))
def keep_warm_business():
    from datetime import datetime
    utc_hour = datetime.utcnow().hour
    # Adjust to your timezone — example: 14:00-22:00 UTC ≈ 9am-5pm CDMX
    if 14 <= utc_hour < 22:
        service = MistralService()
        service.generate.remote("ping", max_tokens=1, temperature=0.0)
        log_structured("keep_warm", utc_hour=utc_hour)

---

Step-by-step setup

1 — Create the secrets

# Tokens (generate random values)
TOKEN_USER=$(python -c "import secrets; print(secrets.token_urlsafe(32))")
TOKEN_ADMIN=$(python -c "import secrets; print(secrets.token_urlsafe(32))")
echo "USER: $TOKEN_USER"
echo "ADMIN: $TOKEN_ADMIN"

modal secret create llm-api-tokens \
  API_TOKEN_USER=$TOKEN_USER \
  API_TOKEN_ADMIN=$TOKEN_ADMIN

Save both tokens in your password manager. Without them you can't call the API.

2 — Deploy

modal deploy final_project.py

Expected output:

✓ Created mount /Users/.../final_project.py
✓ Created function MistralService.*
✓ Created function keep_warm_business
✓ Created web function web => https://your-user--llm-api-final-web.modal.run

✓ App deployed in 12.4s

3 — Test the endpoints

BASE=https://your-user--llm-api-final-web.modal.run

# Health (no auth)
curl $BASE/health

# Chat with auth
curl -X POST $BASE/chat \
  -H "Authorization: Bearer $TOKEN_USER" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Explain Modal in 2 sentences.", "max_tokens": 150}'

# Metrics (admin)
curl $BASE/metrics -H "Authorization: Bearer $TOKEN_ADMIN"

# Batch
curl -X POST $BASE/chat-batch \
  -H "Authorization: Bearer $TOKEN_USER" \
  -H "Content-Type: application/json" \
  -d '{"prompts": ["What is FastAPI?", "What is REST?"], "max_tokens": 100}'

---

Automated tests

Create test_api.py locally (not on Modal):

# test_api.py
import os
import pytest
import httpx

BASE = os.environ["LLM_API_BASE"]                # export before running
TOKEN_USER = os.environ["LLM_API_TOKEN_USER"]
TOKEN_ADMIN = os.environ["LLM_API_TOKEN_ADMIN"]

client = httpx.Client(base_url=BASE, timeout=60)


def test_health_ok():
    r = client.get("/health")
    assert r.status_code == 200
    data = r.json()
    assert data["status"] == "ok"
    assert data["model"]


def test_chat_without_token_fails():
    r = client.post("/chat", json={"prompt": "hi"})
    assert r.status_code == 401


def test_chat_invalid_token_fails():
    r = client.post(
        "/chat",
        headers={"Authorization": "Bearer invalid"},
        json={"prompt": "hi"},
    )
    assert r.status_code == 403


def test_chat_empty_prompt_fails():
    r = client.post(
        "/chat",
        headers={"Authorization": f"Bearer {TOKEN_USER}"},
        json={"prompt": ""},
    )
    assert r.status_code == 422  # Pydantic rejection


def test_chat_ok():
    r = client.post(
        "/chat",
        headers={"Authorization": f"Bearer {TOKEN_USER}"},
        json={"prompt": "Say only 'hi'", "max_tokens": 10},
    )
    assert r.status_code == 200
    data = r.json()
    assert data["response"]
    assert data["generated_tokens"] > 0
    assert data["request_id"]


def test_metrics_requires_admin():
    r_user = client.get("/metrics", headers={"Authorization": f"Bearer {TOKEN_USER}"})
    assert r_user.status_code == 403

    r_admin = client.get("/metrics", headers={"Authorization": f"Bearer {TOKEN_ADMIN}"})
    assert r_admin.status_code == 200
    assert "total_requests" in r_admin.json()


def test_batch_accepts_list():
    r = client.post(
        "/chat-batch",
        headers={"Authorization": f"Bearer {TOKEN_USER}"},
        json={"prompts": ["Hello", "Goodbye"], "max_tokens": 10},
    )
    assert r.status_code == 200
    assert len(r.json()["responses"]) == 2

Run:

export LLM_API_BASE=https://your-user--llm-api-final-web.modal.run
export LLM_API_TOKEN_USER=<the-user-token>
export LLM_API_TOKEN_ADMIN=<the-admin-token>
pip install pytest httpx
pytest test_api.py -v

---

Suggested README

Create README.md in the repo where you keep this project:

# LLM API — Mistral 7B on Modal

Production-ready HTTPS endpoint for Mistral 7B Instruct, deployed on Modal with autoscaling, authentication and monitoring.

## Endpoints

- `GET /health` — public
- `GET /metrics` — Bearer admin
- `POST /chat` — Bearer user
- `POST /chat-batch` — Bearer user, up to 10 prompts

## Setup

```bash
pip install modal
modal token new
modal secret create llm-api-tokens API_TOKEN_USER=... API_TOKEN_ADMIN=...
modal deploy final_project.py
```

## Approximate cost

- No traffic: ~$5/month (warm web CPU container + storage)
- 100,000 req/month: ~$60/month
- 1,000,000 req/month: ~$600/month

Check current prices at [modal.com/pricing](https://modal.com/pricing).

## Scale configuration (defaults)

| Parameter | Value | Why |
|-----------|-------|---------|
| `min_containers` (GPU) | 0 | Don't pay for GPU without traffic |
| `max_containers` (GPU) | 5 | Cost cap |
| `scaledown_window` | 300s | Burst-friendly |
| `min_containers` (web) | 1 | Healthcheck always available |

Adjust to your real case.

---

Self-assessment

Before declaring the module complete, verify:

• The /health endpoint responds without auth and shows status: ok
• Calls to /chat without a token return 401, with an invalid token 403, with a valid token 200
• The first call (cold start) takes <90s, the following ones <5s
• modal app logs shows structured JSON events (chat_ok, chat_error)
• /metrics with an admin token shows real counters after several calls to /chat
• modal app stop llm-api-final shuts down the deployment cleanly
• The pytest tests pass

If all the checks pass, you have a production-ready endpoint.

---

Connection with the rest of the path

This endpoint is one provider for the Unified AI Client in Module 8. The interface you designed here (/chat with JSON request/response) is compatible with:

# In Module 8 you're going to write something like this:
class ModalClient(BaseAIClient):
    def chat(self, prompt: str) -> str:
        r = httpx.post(
            f"{self.base_url}/chat",
            headers={"Authorization": f"Bearer {self.token}"},
            json={"prompt": prompt, "max_tokens": 256},
        )
        r.raise_for_status()
        return r.json()["response"]

You'll be able to do client = UnifiedAIClient(provider="modal") and consume it just like OpenAI or Ollama.

---

Module 6 completed

Let's recap: now you can...

• ✅ Decide when Modal beats managed or self-hosted
• ✅ Set up an account, CLI and authentication
• ✅ Define images with dependencies and secrets
• ✅ Deploy an LLM (Mistral 7B) with an A10G GPU and weights cache
• ✅ Expose it as a public HTTPS API with FastAPI + auth
• ✅ Configure autoscaling and cold start mitigation
• ✅ Estimate costs and apply optimizations
• ✅ Deliver a production-ready service with healthcheck, metrics and tests

Well done. Modal is one of the topics that almost nobody covers and you'll see it show up in real AI startups.

---

Next module

Module 7 — Trade-offs and Decision Matrix takes all the providers you learned in modules 2-6 (OpenAI, LM Studio, Ollama, OpenRouter, Modal) and compares them with real benchmarks that you'll run yourself. It's the module that gives you the quantitative criteria to choose, not just qualitative ones.

---

Resources

1. Example repo: modal-labs/llm-serving — official examples of advanced patterns.
2. Modal — Production checklist — official recommendations for production.
3. FastAPI — Testing — testing patterns.
4. Prometheus client for Python — if you want to extend /metrics to the standard Prometheus format.
5. Sentry SDK — error tracking for real production.