Session 8 — MCP Learning Journey: Build the Cat Shop Server From Scratch

What this is. A guided, type-it-yourself rebuild of the Cat Shop MCP server. You will reconstruct the server one layer at a time, run a checkpoint after each layer, and answer a short understand-it question before moving on. By the end you'll be able to explain every file in app/ — and you'll be ready to add your own tool (the graded Activity).

How to use it. The finished server already ships in app/. So you have two ways to travel:

Build mode (recommended): create a new package folder catshop/ next to app/ and type each file yourself as you reach its milestone. Run with your own entry point (Milestone 7). When stuck, peek at the matching app/ file — it's the answer key for the infrastructure.

Read-along mode: open the matching app/ file at each milestone and confirm you can explain every line out loud before continuing.

What stays yours to write. The 6 example tools and the OAuth machinery are given scaffolding — this journey shows them in full. The graded work is different: Q1/Q2 (answered in README.md) and your own new tool (Activity 1). Those milestones give you questions and a method, not a solution. That boundary is the whole point of the session.

Map of the journey

#	Milestone	File you build	You'll be able to…
0	Orientation & setup	—	Name the 3 planes of an MCP server
1	The data layer	`catshop/db.py`	Explain why one SQLite file holds both auth and shop state
2	The OAuth provider	`catshop/oauth.py`	Trace how a random token is born, stored, and validated
3	The server object	`catshop/server.py`	Wire OAuth + transport into one `FastMCP`
4	The login route	`catshop/routes.py`	See where a human turns a request into an auth code
5	The tools	`catshop/tools.py`	Distinguish identity-free vs identity-bound tools
6	Wiring it together	`catshop/__init__.py` + entry	Explain why importing a module registers tools
7	Run & watch the flow	—	Drive the full OAuth → tool-call loop with a client
8	Understand by probing	—	Pull a live token, force a 401, read the DB
9	Extend it (Activity 1)	your new tool	Add a 7th tool that fits the contract (graded, Socratic)
10	The questions (Q1/Q2)	`README.md`	Reason about OAuth + transport (graded, Socratic)

Dependency order matters: each layer imports the one above it (db ← oauth ← server ← {routes, tools} ← __init__ ← entry). Build top-to-bottom and nothing is ever undefined.

Milestone 0 — Orientation & setup

The one idea

An MCP server publishes tools (typed functions) that an AI client can discover and call. This server keeps three concerns deliberately separate:

   OAuth (who may call?)  ──┐
   Transport (how does    ──┼──►  the same FastMCP server
     the call arrive?)      │
   Tools + DB (what does  ──┘
     the call do?)

Hold that picture. Every file you write belongs to exactly one plane, and the magic is that they barely know about each other.

Do this

cd 08_MCP
uv sync                 # Python 3.13: installs mcp[cli], aiosqlite, uvicorn, …
cp .env.example .env    # the server needs NO key; OPENAI_API_KEY is only for the advanced client
mkdir catshop           # your from-scratch package (parallel to the provided app/)

Checkpoint

uv run python -c "import mcp; print('mcp ready')" prints mcp ready.

Understand

Before writing any code: in the diagram above, which plane does a stolen token threaten, and which plane does a crashed database threaten? (You'll revisit this in Q1.)

Milestone 1 — The data layer (`catshop/db.py`)

Why this layer first

Everything else reads or writes SQLite. Notice the table list below: half of it is OAuth state (oauth_clients, authorization_codes, access_tokens, refresh_tokens, token_users, pending_authorizations) and half is shop state (products, cart_items, users). One file, two planes — that co-location is a design choice worth remembering.

We use aiosqlite (not plain sqlite3) because the whole server is async: tool functions are async def, so their DB calls must be awaitable too.

Build it

# catshop/db.py
import aiosqlite

PRODUCTS = [
    ("Whisker Wand", "Interactive feather toy on a flexible wand", 9.99, "toys"),
    ("Catnip Mouse", "Organic catnip-stuffed plush mouse", 4.99, "toys"),
    ("Laser Pointer Pro", "Red-dot laser with adjustable patterns", 12.99, "toys"),
    ("Cozy Cat Bed", "Soft donut-shaped bed for curling up", 29.99, "beds"),
    ("Window Hammock", "Suction-cup window perch with fleece lining", 24.99, "beds"),
    ("Salmon Treats", "Freeze-dried wild salmon bites, 100g", 7.99, "food"),
    ("Tuna Crunchies", "Crunchy tuna-flavored dental treats, 80g", 5.99, "food"),
    ("Scratching Post Tower", "3-tier sisal scratching post with platforms", 49.99, "furniture"),
]


async def init_db(db: aiosqlite.Connection):
    await db.executescript(
        """
        CREATE TABLE IF NOT EXISTS oauth_clients (
            client_id TEXT PRIMARY KEY,
            client_info_json TEXT NOT NULL
        );
        CREATE TABLE IF NOT EXISTS authorization_codes (
            code TEXT PRIMARY KEY,
            client_id TEXT NOT NULL,
            scopes_json TEXT NOT NULL,
            expires_at REAL NOT NULL,
            code_challenge TEXT NOT NULL,
            redirect_uri TEXT NOT NULL,
            redirect_uri_provided_explicitly INTEGER NOT NULL,
            resource TEXT,
            username TEXT NOT NULL
        );
        CREATE TABLE IF NOT EXISTS access_tokens (
            token TEXT PRIMARY KEY, client_id TEXT NOT NULL,
            scopes_json TEXT NOT NULL, expires_at REAL, resource TEXT
        );
        CREATE TABLE IF NOT EXISTS refresh_tokens (
            token TEXT PRIMARY KEY, client_id TEXT NOT NULL,
            scopes_json TEXT NOT NULL, expires_at REAL
        );
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL, description TEXT NOT NULL,
            price REAL NOT NULL, category TEXT NOT NULL
        );
        CREATE TABLE IF NOT EXISTS cart_items (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT NOT NULL, product_id INTEGER NOT NULL,
            quantity INTEGER NOT NULL DEFAULT 1,
            UNIQUE(username, product_id)
        );
        CREATE TABLE IF NOT EXISTS users (
            username TEXT PRIMARY KEY, created_at REAL NOT NULL
        );
        CREATE TABLE IF NOT EXISTS pending_authorizations (
            request_id TEXT PRIMARY KEY, client_id TEXT NOT NULL,
            scopes_json TEXT NOT NULL, code_challenge TEXT NOT NULL,
            redirect_uri TEXT NOT NULL, redirect_uri_provided_explicitly INTEGER NOT NULL,
            resource TEXT, state TEXT, expires_at REAL NOT NULL
        );
        CREATE TABLE IF NOT EXISTS token_users (
            token TEXT PRIMARY KEY, username TEXT NOT NULL
        );
        """
    )
    # Seed the catalog only if it's empty (idempotent: safe to run every startup)
    cursor = await db.execute("SELECT COUNT(*) FROM products")
    (count,) = await cursor.fetchone()
    if count == 0:
        await db.executemany(
            "INSERT INTO products (name, description, price, category) VALUES (?, ?, ?, ?)",
            PRODUCTS,
        )
    await db.commit()

The full provided version is app/db.py (the access_tokens/refresh_tokens columns are written out one-per-line there; identical schema). Compare them.

Checkpoint

uv run python -c "
import asyncio, aiosqlite
from catshop.db import init_db
async def main():
    db = await aiosqlite.connect('checkpoint.db'); await init_db(db)
    n = (await (await db.execute('SELECT COUNT(*) FROM products')).fetchone())[0]
    print('products seeded:', n)
asyncio.run(main())
"
# expect: products seeded: 8
rm -f checkpoint.db

Understand

cart_items has UNIQUE(username, product_id). What does that constraint let add_to_cart do when you add a product you already have in your cart? (Hint: look for ON CONFLICT in Milestone 5.)

Milestone 2 — The OAuth provider (`catshop/oauth.py`)

This is the heart of the session. The class subclasses the SDK's OAuthAuthorizationServerProvider and implements the methods the framework calls during the OAuth handshake. You're not inventing OAuth — you're filling in storage for each step.

A crucial realization up front: the tokens here are not JWTs. They are random hex strings, and "is this token valid?" is answered by a database lookup, not a signature check. There is no signing key to configure anywhere.

2a — Skeleton, DB handle, and clients

# catshop/oauth.py
import json, secrets, time
import aiosqlite

from mcp.server.auth.provider import (
    AccessToken, AuthorizationCode, AuthorizationParams,
    OAuthAuthorizationServerProvider, RefreshToken, construct_redirect_uri,
)
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
from .db import init_db


class CatShopOAuthProvider(OAuthAuthorizationServerProvider):
    _db: aiosqlite.Connection | None = None

    def __init__(self, issuer_url: str):
        self.issuer_url = issuer_url

    async def _get_db(self) -> aiosqlite.Connection:
        if self._db is None:                              # lazy: open once, seed once
            self._db = await aiosqlite.connect("catshop.db")
            await init_db(self._db)
        return self._db

    # -- clients: a client REGISTERS itself (Dynamic Client Registration) --
    async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
        db = await self._get_db()
        row = await (await db.execute(
            "SELECT client_info_json FROM oauth_clients WHERE client_id = ?", (client_id,)
        )).fetchone()
        return OAuthClientInformationFull.model_validate_json(row[0]) if row else None

    async def register_client(self, client_info: OAuthClientInformationFull) -> None:
        db = await self._get_db()
        await db.execute(
            "INSERT OR REPLACE INTO oauth_clients (client_id, client_info_json) VALUES (?, ?)",
            (client_info.client_id, client_info.model_dump_json()),
        )
        await db.commit()

Key insight: the client_id is generated by the framework and handed to register_client to persist — you never configure it. Public clients carry no client_secret; they prove themselves with PKCE instead (you'll see the code_challenge next).

2b — `authorize`: park the request, send the human to log in

    async def authorize(self, client, params: AuthorizationParams) -> str:
        db = await self._get_db()
        request_id = secrets.token_hex(16)
        scopes = params.scopes or ["read", "write"]
        await db.execute(
            """INSERT INTO pending_authorizations
               (request_id, client_id, scopes_json, code_challenge, redirect_uri,
                redirect_uri_provided_explicitly, resource, state, expires_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (request_id, client.client_id, json.dumps(scopes), params.code_challenge,
             str(params.redirect_uri), int(params.redirect_uri_provided_explicitly),
             str(params.resource) if params.resource else None, params.state, time.time() + 600),
        )
        await db.commit()
        return f"{self.issuer_url}/login?req={request_id}"   # the URL the browser is sent to

authorize doesn't issue anything yet — it stashes the request (with its PKCE code_challenge) and returns a /login URL. The actual code is minted after a human picks a username (Milestone 4). Notice self.issuer_url here — this is why ISSUER_URL must match the public address: it's baked into the redirect the client follows.

2c — Authorization codes: load & exchange for tokens

    async def load_authorization_code(self, client, authorization_code: str):
        db = await self._get_db()
        row = await (await db.execute(
            "SELECT * FROM authorization_codes WHERE code = ? AND client_id = ?",
            (authorization_code, client.client_id),
        )).fetchone()
        if row is None:
            return None
        return AuthorizationCode(
            code=row[0], client_id=row[1], scopes=json.loads(row[2]), expires_at=row[3],
            code_challenge=row[4], redirect_uri=row[5],
            redirect_uri_provided_explicitly=bool(row[6]), resource=row[7],
        )

    async def exchange_authorization_code(self, client, authorization_code) -> OAuthToken:
        db = await self._get_db()
        # remember WHO this code belongs to, then consume the code (single-use)
        row = await (await db.execute(
            "SELECT username FROM authorization_codes WHERE code = ?",
            (authorization_code.code,),
        )).fetchone()
        username = row[0] if row else "unknown"
        await db.execute("DELETE FROM authorization_codes WHERE code = ?", (authorization_code.code,))

        access_token, refresh_token = secrets.token_hex(32), secrets.token_hex(32)
        now, expires_in = time.time(), 3600
        await db.execute(
            "INSERT INTO access_tokens (token, client_id, scopes_json, expires_at, resource) "
            "VALUES (?, ?, ?, ?, ?)",
            (access_token, client.client_id, json.dumps(authorization_code.scopes),
             now + expires_in, authorization_code.resource),
        )
        await db.execute(
            "INSERT INTO refresh_tokens (token, client_id, scopes_json, expires_at) VALUES (?, ?, ?, ?)",
            (refresh_token, client.client_id, json.dumps(authorization_code.scopes), now + 86400),
        )
        # the link that lets a tool answer "who is calling?" later:
        await db.execute("INSERT INTO token_users (token, username) VALUES (?, ?)", (access_token, username))
        await db.execute("INSERT INTO token_users (token, username) VALUES (?, ?)", (refresh_token, username))
        await db.commit()
        return OAuthToken(
            access_token=access_token, token_type="Bearer", expires_in=expires_in,
            scope=" ".join(authorization_code.scopes), refresh_token=refresh_token,
        )

This is where a token is born. secrets.token_hex(32) → a 64-char random string. It's stored with an expiry and a token_users row mapping it to a username. That mapping is what makes the cart tools personal.

2d — Refresh, validate, revoke (copy from `app/oauth.py`)

The remaining methods follow the exact same read-row / write-row pattern. Type them from app/oauth.py — and as you do, notice what each one is for:

load_refresh_token / exchange_refresh_token — trade a valid refresh token for a fresh access+refresh pair (rotating both, re-linking token_users). This is how a client stays logged in past the 1-hour access-token expiry without sending you back to /login.

load_access_token — the gatekeeper. On every tool call the framework calls this to turn the incoming Bearer string into an AccessToken (or None). It also deletes expired tokens on the spot:

async def load_access_token(self, token: str) -> AccessToken | None:
    db = await self._get_db()
    row = await (await db.execute("SELECT * FROM access_tokens WHERE token = ?", (token,))).fetchone()
    if row is None:
        return None
    expires_at = row[3]
    if expires_at and time.time() > expires_at:        # expired → evict and reject
        await db.execute("DELETE FROM access_tokens WHERE token = ?", (token,))
        await db.execute("DELETE FROM token_users WHERE token = ?", (token,))
        await db.commit()
        return None
    return AccessToken(token=row[0], client_id=row[1], scopes=json.loads(row[2]),
                       expires_at=int(expires_at) if expires_at else None, resource=row[4])

revoke_token — delete an access or refresh token (and its token_users row) on demand.

get_username_for_token — a small helper your tools call (not part of the OAuth spec):

async def get_username_for_token(self, token: str) -> str | None:
    db = await self._get_db()
    row = await (await db.execute(
        "SELECT username FROM token_users WHERE token = ?", (token,))).fetchone()
    return row[0] if row else None

Checkpoint

uv run python -c "from catshop.oauth import CatShopOAuthProvider; print('provider imports OK')"

Understand

Follow one token through 2c and 2d: it's created in exchange_authorization_code, checked in load_access_token, and resolved to a person in get_username_for_token. Which of those three would a tool like view_cart call, and why not the other two?

Milestone 3 — The server object (`catshop/server.py`)

Now assemble the FastMCP server and hand it the provider you just built. This file also reads the environment knobs you met during setup.

Build it

# catshop/server.py
import os
from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions, RevocationOptions
from mcp.server.fastmcp import FastMCP
from .oauth import CatShopOAuthProvider

HOST = os.environ.get("HOST", "0.0.0.0")
PORT = int(os.environ.get("PORT", "8000"))
ISSUER_URL = os.environ.get("ISSUER_URL", f"http://localhost:{PORT}")

oauth_provider = CatShopOAuthProvider(issuer_url=ISSUER_URL)

mcp = FastMCP(
    "Cat Shop",
    auth_server_provider=oauth_provider,
    auth=AuthSettings(
        issuer_url=ISSUER_URL,
        resource_server_url=ISSUER_URL,
        client_registration_options=ClientRegistrationOptions(
            enabled=True, valid_scopes=["read", "write"], default_scopes=["read", "write"],
        ),
        revocation_options=RevocationOptions(enabled=True),
    ),
    host=HOST, port=PORT,
)

Three things to internalize here:

ISSUER_URL defaults to http://localhost:{PORT} — so changing PORT automatically keeps the issuer consistent locally. (When you go public with ngrok, you set ISSUER_URL by hand — see the STUDENT cheatsheet §3a for the PORT ↔ ISSUER_URL ↔ ngrok alignment.)

valid_scopes=["read","write"] is your scope minimization lever — the menu of powers a token can carry. (Q1 territory.)

client_registration_options(enabled=True) is what allows Dynamic Client Registration — clients can self-register with no pre-shared key.

Checkpoint

uv run python -c "from catshop.server import mcp; print('server name:', mcp.name)"
# expect: server name: Cat Shop

Understand

The server has zero tools at this point. Where will they come from, and what makes them attach to this exact mcp object? (Answer arrives in Milestone 6.)

Milestone 4 — The login route (`catshop/routes.py`)

OAuth needs a moment where a human proves who they are. authorize (2b) parked a pending request and redirected the browser to /login?req=…. This route renders the form, then on submit mints the authorization code and bounces back to the client.

Build it (logic shown; copy the HTML/CSS from `app/routes.py`)

# catshop/routes.py
import json, secrets, time
from html import escape
from mcp.server.auth.provider import construct_redirect_uri
from starlette.requests import Request
from starlette.responses import HTMLResponse, RedirectResponse, Response
from .server import mcp, oauth_provider

LOGIN_PAGE_HTML = """ ...the styled form; copy verbatim from app/routes.py... """  # {error} and {req} slots


@mcp.custom_route("/login", methods=["GET", "POST"])
async def login_page(request: Request) -> Response:
    req_id = request.query_params.get("req", "")
    db = await oauth_provider._get_db()

    # the pending request from authorize() must exist and be unexpired
    pending = await (await db.execute(
        "SELECT * FROM pending_authorizations WHERE request_id = ? AND expires_at > ?",
        (req_id, time.time()),
    )).fetchone()
    if pending is None:
        return HTMLResponse("<h1>400</h1><p>This login request is invalid or has expired.</p>",
                            status_code=400)

    if request.method == "GET":                       # show the form
        return HTMLResponse(LOGIN_PAGE_HTML.format(req=escape(req_id), error=""))

    # POST: validate username, create the user, mint the auth code
    form = await request.form()
    username = form.get("username", "").strip()
    if not username or not (2 <= len(username) <= 30):
        return HTMLResponse(
            LOGIN_PAGE_HTML.format(req=escape(req_id),
                                   error='<p class="error">Username must be 2-30 characters.</p>'),
            status_code=400)

    await db.execute("INSERT OR IGNORE INTO users (username, created_at) VALUES (?, ?)",
                     (username, time.time()))
    code = secrets.token_hex(32)                       # the authorization code
    await db.execute(
        """INSERT INTO authorization_codes
           (code, client_id, scopes_json, expires_at, code_challenge,
            redirect_uri, redirect_uri_provided_explicitly, resource, username)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (code, pending[1], pending[2], time.time() + 300, pending[3],
         pending[4], pending[5], pending[6], username),  # carries the username + PKCE challenge forward
    )
    await db.execute("DELETE FROM pending_authorizations WHERE request_id = ?", (req_id,))
    await db.commit()

    redirect_uri = construct_redirect_uri(pending[4], code=code, state=pending[7])
    return RedirectResponse(url=redirect_uri, status_code=302)  # back to the client, with ?code=&state=

@mcp.custom_route is how you add a normal web page to an MCP server — it's not a tool, it's a human-facing HTML endpoint living on the same host. This route is the bridge between "a machine asked for access" and "a person said yes." The username it captures rides along inside the auth code and ends up in token_users (Milestone 2c).

Checkpoint

After Milestone 7 you'll hit this live. For now: uv run python -c "import catshop.routes" raises no error (it imports mcp from server, proving your dependency order is right).

Understand

The code's lifetime is time.time() + 300 (5 min) and it's DELETEd the instant it's exchanged. Name two distinct attacks those two facts defend against.

Milestone 5 — The tools (`catshop/tools.py`)

Finally, the capability. Each tool is an async def decorated with @mcp.tool(). The signature is the schema — type hints become the input contract, the docstring becomes the description the model reads.

Build it

# catshop/tools.py
import secrets
from mcp.server.auth.middleware.auth_context import get_access_token
from .server import mcp, oauth_provider


async def _get_username() -> str:
    token = get_access_token()                       # the current request's Bearer token
    if token is None:
        raise ValueError("Not authenticated")
    username = await oauth_provider.get_username_for_token(token.token)
    if username is None:
        raise ValueError("User not found for token")
    return username


# ---- catalog tools: identity-FREE (anyone with a valid token, no username needed) ----
@mcp.tool()
async def list_products(category: str | None = None) -> list[dict]:
    """Browse the cat shop catalog. Optionally filter by category (toys, beds, food, furniture)."""
    db = await oauth_provider._get_db()
    if category:
        cursor = await db.execute(
            "SELECT id, name, description, price, category FROM products WHERE category = ?", (category,))
    else:
        cursor = await db.execute("SELECT id, name, description, price, category FROM products")
    rows = await cursor.fetchall()
    return [{"id": r[0], "name": r[1], "description": r[2], "price": r[3], "category": r[4]} for r in rows]


@mcp.tool()
async def get_product(product_id: int) -> dict:
    """Get full details of a single product by its ID."""
    db = await oauth_provider._get_db()
    row = await (await db.execute(
        "SELECT id, name, description, price, category FROM products WHERE id = ?", (product_id,))).fetchone()
    if row is None:
        return {"error": "Product not found"}
    return {"id": row[0], "name": row[1], "description": row[2], "price": row[3], "category": row[4]}


# ---- cart tools: identity-BOUND (resolve _get_username() first) ----
@mcp.tool()
async def add_to_cart(product_id: int, quantity: int = 1) -> dict:
    """Add a product to your shopping cart. If already in cart, quantity is increased."""
    username = await _get_username()
    db = await oauth_provider._get_db()
    product = await (await db.execute("SELECT name FROM products WHERE id = ?", (product_id,))).fetchone()
    if product is None:
        return {"error": "Product not found"}
    await db.execute(
        """INSERT INTO cart_items (username, product_id, quantity) VALUES (?, ?, ?)
           ON CONFLICT(username, product_id) DO UPDATE SET quantity = quantity + excluded.quantity""",
        (username, product_id, quantity))
    await db.commit()
    return {"success": True, "message": f"Added {quantity}x {product[0]} to your cart"}


@mcp.tool()
async def view_cart() -> dict:
    """View everything in your shopping cart with quantities and totals."""
    username = await _get_username()
    db = await oauth_provider._get_db()
    rows = await (await db.execute(
        """SELECT p.id, p.name, p.price, c.quantity FROM cart_items c
           JOIN products p ON c.product_id = p.id WHERE c.username = ?""", (username,))).fetchall()
    items = [{"product_id": r[0], "name": r[1], "price": r[2], "quantity": r[3],
              "subtotal": round(r[2] * r[3], 2)} for r in rows]
    total = round(sum(i["subtotal"] for i in items), 2)
    return {"items": items, "total": total, "item_count": len(items)}


@mcp.tool()
async def remove_from_cart(product_id: int) -> dict:
    """Remove a product from your shopping cart."""
    username = await _get_username()
    db = await oauth_provider._get_db()
    cursor = await db.execute(
        "DELETE FROM cart_items WHERE username = ? AND product_id = ?", (username, product_id))
    await db.commit()
    return {"error": "Item not in cart"} if cursor.rowcount == 0 else \
        {"success": True, "message": "Item removed from cart"}


@mcp.tool()
async def checkout() -> dict:
    """Complete your purchase. Shows order summary and clears the cart."""
    username = await _get_username()
    db = await oauth_provider._get_db()
    cart = await view_cart()                          # reuse the tool as a plain function
    if not cart["items"]:
        return {"error": "Your cart is empty"}
    await db.execute("DELETE FROM cart_items WHERE username = ?", (username,))
    await db.commit()
    order_id = secrets.token_hex(8).upper()
    return {"order_id": order_id, "status": "confirmed", "items": cart["items"],
            "total": cart["total"],
            "message": f"Order {order_id} confirmed! Thanks {username}, your cats will love their new goodies!"}

The two-tier pattern is the lesson: catalog tools never call _get_username() (browsing is anonymous-among-token-holders); cart tools call it first so the data they touch is scoped to the caller. checkout even calls view_cart() directly — a decorated tool is still an ordinary function you can await.

Checkpoint

uv run python -c "
import catshop.tools
from catshop.server import mcp
print('tools:', sorted(mcp._tool_manager._tools.keys()))
"
# expect all six: add_to_cart, checkout, get_product, list_products, remove_from_cart, view_cart

Understand

get_access_token() returns the token for the request currently being handled — no argument needed. What machinery must be sitting between the network and your function for that to work, and which file turned it on? (Look back at Milestone 3.)

Milestone 6 — Wiring it together (`catshop/init.py` + an entry point)

Here's the subtle trick that makes the whole thing cohere. The tools and routes register themselves as a side effect of being imported. So the package __init__ must import them, or the server boots with nothing attached.

Build it

# catshop/__init__.py
from .server import mcp, oauth_provider   # 1. build the FastMCP object + provider
from . import routes                      # 2. importing runs @mcp.custom_route  → /login attached
from . import tools                       # 3. importing runs @mcp.tool()        → 6 tools attached

# my_server.py   (your entry point, in 08_MCP/ next to the provided server.py)
from catshop import mcp

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

Why this shape? tools.py and routes.py both need the mcp object to decorate, so mcp must exist before them — that's why server.py owns it and everyone imports it. __init__.py is the conductor guaranteeing the decorator modules actually run. Delete line 3 and list_tools comes back empty. (This mirrors the provided app/__init__.py + root server.py.)

Checkpoint

uv run python -c "import catshop; print('fully wired, tools:', len(catshop.mcp._tool_manager._tools))"
# expect: fully wired, tools: 6

Understand

The comments in app/__init__.py say # noqa: F401 - registers .... What is F401, and why would a linter complain about these imports if not for that comment?

Milestone 7 — Run it & watch the whole flow

You've built every layer. Now drive it.

Do this

# 1. Start YOUR server (pick a free port; ISSUER_URL auto-follows locally)
PORT=8123 uv run python my_server.py
#    → look for: Uvicorn running on http://0.0.0.0:8123

# 2. Smoke test in a 2nd terminal — the OAuth metadata route proves auth is wired:
curl http://localhost:8123/.well-known/oauth-authorization-server
#    → JSON with issuer / authorization_endpoint / token_endpoint

# 3. Connect a client and call tools (no code) — MCP Inspector:
npx @modelcontextprotocol/inspector
#    Transport: Streamable HTTP   URL: http://localhost:8123/mcp
#    Connect → a browser opens YOUR /login page → pick a username → you're in
#    List Tools → call list_products, then add_to_cart → view_cart → checkout

Going public (ngrok) and the PORT ↔ ISSUER_URL ↔ ngrok http <port> alignment are covered in the STUDENT cheatsheet §3a — including the two failure modes (Errno 98 and 502 Bad Gateway) and why the public side is always port 443. If anything misbehaves, the cheatsheet's §4.9 triage table maps symptom → cause.

Checkpoint

You complete browse → add → view → checkout in the Inspector and watch the cart empty after checkout.

Understand

When you clicked Connect, which milestone's code ran first — and at what exact moment did a row appear in access_tokens? Narrate the path: Inspector → /authorize (2b) → /login (4) → /token (2c) → /mcp tool call (5).

Milestone 8 — Understand by probing

Reading is one thing; poking is another. Each probe confirms a claim the code makes.

# A. No token → the server refuses. (Proves OAuth is actually gating, not decorative.)
curl -s -o /dev/null -w '%{http_code}\n' -X POST http://localhost:8123/mcp \
  -H 'Content-Type: application/json' -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}'
#    → 401   (with a WWW-Authenticate: Bearer … header pointing at the OAuth metadata)

# B. Pull a LIVE token straight from the DB (after you've logged in once via the Inspector):
sqlite3 catshop.db "
  SELECT a.token, u.username, a.scopes_json
  FROM access_tokens a LEFT JOIN token_users u ON a.token = u.token
  WHERE a.expires_at > strftime('%s','now')
  ORDER BY a.expires_at DESC LIMIT 1;"
#    → the random hex token, the username you picked, ['read','write']

# C. Use that token as a Bearer to hit a gated tool by hand:
TOKEN=$(sqlite3 catshop.db "SELECT token FROM access_tokens ORDER BY expires_at DESC LIMIT 1;")
curl -s -X POST http://localhost:8123/mcp -H "Authorization: Bearer $TOKEN" \
  -H 'Content-Type: application/json' -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}'

Probe A is the live proof of your Q1 answer: the transport carries the call, but OAuth is what stands between the open internet and your tools. Probe B/C show tokens are just opaque DB rows — no signing key anywhere. (This is a debug convenience, not how real clients authenticate.)

Understand

Delete the token's row (DELETE FROM access_tokens WHERE token = '…') and re-run probe C. Predict the result before you run it. Why?

Milestone 9 — Extend it: your own tool (Activity 1 — graded)

You now understand the contract well enough to add to it. This is the graded deliverable, and from here the journey stops giving you code. Add at least one new @mcp.tool() to your tools module (in the real assignment, that's app/tools.py) — beyond the six you just built.

Pick something the shop is missing — e.g. search_products, update_cart_quantity, or get_order_history. Then hold your tool up against the six you wrote and ask:

Does it carry @mcp.tool() and a genuinely new name (not a renamed list_products)?
Are its arguments typed like get_product(product_id: int) — would the model know what to pass from your signature + docstring alone?
Does it touch the DB via await oauth_provider._get_db() and a real execute/commit — not a hard-coded list?
If it reads or changes the cart, does it call _get_username() first, like the other cart tools?
Can you invoke it live through the Inspector (Milestone 7) and watch the call + result?

The learning is in matching the contract, not in the SQL. Don't copy a tool body wholesale — reread how list_products (catalog-style) and add_to_cart (cart-style) are shaped, decide which family yours belongs to, and write it in that style. Demo it in your Loom.

Milestone 10 — The questions (Q1/Q2 — graded)

Write these in README.md under each question's #### Answer heading (don't leave the _(insert your answer here)_ placeholder — that grades as not answered). The journey gives you the evidence, not the answer.

Q1 — Why is OAuth important for MCP servers, and what security considerations apply? You watched it end to end: at no point did the server receive a password — only a scoped, expiring Bearer token (Milestone 2c) that it validates by lookup (2d) and can revoke. Re-read those, plus valid_scopes (Milestone 3), and Probe A (Milestone 8). Then name the mechanism and at least one concrete consideration: scope minimization, token expiry/refresh, what the server must still verify about a token, or what a stolen token can and can't do.

Q2 — What is Streamable HTTP transport, and why expose publicly with OAuth instead of local stdio? You ran it: a long-lived HTTP stream a remote client reached over the network (Milestone 7), gated by OAuth (Probe A). Contrast that with stdio (a local subprocess pipe — no network, no auth layer). Ask: where must the client be for each? What does going public buy you, and what role does OAuth play that the transport alone does not?

Where to go next

Advanced Activity (optional): write a custom mcp SDK client that does the OAuth flow + a browse → add → checkout sequence in code, and compare the developer experience to hand-rolled REST.
Reference: MCP docs https://modelcontextprotocol.io/ · MCP auth deep-dive https://auth0.com/blog/mcp-specs-update-all-about-auth/
Companion: 08_MCP_CHEATSHEET_STUDENT.md (the concept/API map + setup & troubleshooting this journey points back to).

You finished the journey when you can open any file in app/ and explain, for every block, (1) which plane it belongs to — auth, transport, or tools+DB — and (2) what breaks if you delete it.

donbr/sess8-mcp-learning-journey.md

Session 8 — MCP Learning Journey: Build the Cat Shop Server From Scratch

Map of the journey

Milestone 0 — Orientation & setup

The one idea

Do this

Checkpoint

Understand

Milestone 1 — The data layer (catshop/db.py)

Why this layer first

Build it

Checkpoint

Understand

Milestone 2 — The OAuth provider (catshop/oauth.py)

2a — Skeleton, DB handle, and clients

2b — authorize: park the request, send the human to log in

2c — Authorization codes: load & exchange for tokens

2d — Refresh, validate, revoke (copy from app/oauth.py)

Checkpoint

Understand

Milestone 3 — The server object (catshop/server.py)

Build it

Checkpoint

Understand

Milestone 4 — The login route (catshop/routes.py)

Build it (logic shown; copy the HTML/CSS from app/routes.py)

Checkpoint

Understand

Milestone 5 — The tools (catshop/tools.py)

Build it

Checkpoint

Understand

Milestone 6 — Wiring it together (catshop/__init__.py + an entry point)

Build it

Checkpoint

Understand

Milestone 7 — Run it & watch the whole flow

Do this

Checkpoint

Understand

Milestone 8 — Understand by probing

Understand

Milestone 9 — Extend it: your own tool (Activity 1 — graded)

Milestone 10 — The questions (Q1/Q2 — graded)

Where to go next

Milestone 1 — The data layer (`catshop/db.py`)

Milestone 2 — The OAuth provider (`catshop/oauth.py`)

2b — `authorize`: park the request, send the human to log in

2d — Refresh, validate, revoke (copy from `app/oauth.py`)

Milestone 3 — The server object (`catshop/server.py`)

Milestone 4 — The login route (`catshop/routes.py`)

Build it (logic shown; copy the HTML/CSS from `app/routes.py`)

Milestone 5 — The tools (`catshop/tools.py`)

Milestone 6 — Wiring it together (`catshop/init.py` + an entry point)