Appendix C: The notes Server, Final Form

The Chapter 5 server after every Part II change: Chapter 6’s intent-shaped surface and Chapter 7’s Streamable HTTP entry point with /healthz and /readyz on port 3000.

server.py

One dependency: uv add "mcp[cli]". Run over stdio with uv run server.py.

import logging
import os
import sqlite3
from datetime import datetime, timezone

from pydantic import BaseModel
from starlette.requests import Request
from starlette.responses import PlainTextResponse

from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ResourceNotFoundError, ToolError
from mcp.types import ToolAnnotations

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("notes")  # stderr: stdout belongs to JSON-RPC on stdio

DB_PATH = os.environ.get("NOTES_DB_PATH", "notes.db")


def _db() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute(
        "CREATE TABLE IF NOT EXISTS notes (id INTEGER PRIMARY KEY AUTOINCREMENT,"
        " title TEXT NOT NULL, body TEXT NOT NULL,"
        " tag TEXT NOT NULL DEFAULT 'general', created_at TEXT NOT NULL)"
    )
    return conn


class NoteHit(BaseModel):
    id: int
    title: str
    tag: str
    snippet: str  # first 160 characters of the body


class SearchResult(BaseModel):
    count: int
    items: list[NoteHit]


mcp = MCPServer("notes")


@mcp.tool(annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False))
def create_note(title: str, body: str, tag: str = "general") -> dict:
    """Create a note with a title, body, and optional tag."""
    title, body, tag = title.strip(), body.strip(), tag.strip().lower()
    errors = []
    if not title:
        errors.append("title must not be empty")
    if not body:
        errors.append("body must not be empty")
    if len(title) > 200:
        errors.append("title must be 200 characters or fewer")
    if errors:
        raise ToolError("Note not saved: " + "; ".join(errors))
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    with _db() as conn:
        note_id = conn.execute(
            "INSERT INTO notes (title, body, tag, created_at) VALUES (?, ?, ?, ?)",
            (title, body, tag, now),
        ).lastrowid
    log.info("created note %s", note_id)
    return {"note_id": note_id, "message": f"Saved note '{title}' with tag '{tag}'"}


@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
def search_notes(query: str, tag: str | None = None, limit: int = 20) -> SearchResult:
    """Search note titles and bodies; newest first. Filter by tag to narrow. Returns up to `limit` results (max 100)."""
    query = query.strip()
    if not query:
        raise ToolError("query must not be empty")
    limit = max(1, min(limit, 100))
    sql = ("SELECT id, title, tag, substr(body, 1, 160) AS snippet FROM notes "
           "WHERE (title LIKE ? OR body LIKE ?) ")
    params: list = [f"%{query}%", f"%{query}%"]
    if tag:
        sql += "AND tag = ? "
        params.append(tag.strip().lower())
    sql += "ORDER BY created_at DESC, id DESC LIMIT ?"
    params.append(limit)
    with _db() as conn:
        rows = conn.execute(sql, params).fetchall()
    return SearchResult(count=len(rows), items=[NoteHit(**dict(r)) for r in rows])


@mcp.tool(annotations=ToolAnnotations(destructiveHint=True, idempotentHint=True))
def delete_note(note_id: int) -> dict:
    """Permanently delete one note by ID. Get the ID from search_notes."""
    with _db() as conn:
        deleted = conn.execute("DELETE FROM notes WHERE id = ?", (note_id,)).rowcount
    return {"note_id": note_id, "deleted": bool(deleted)}


@mcp.resource("notes://summary")
def notes_summary() -> str:
    """A readable list of all notes, one per line."""
    with _db() as conn:
        rows = conn.execute("SELECT id, title, tag FROM notes ORDER BY id").fetchall()
    if not rows:
        return "No notes have been created yet."
    return "\n".join(f"{r['id']}. {r['title']} [{r['tag']}]" for r in rows)


@mcp.resource("notes://detail/{note_id}")
def note_detail(note_id: int) -> str:
    """The full content of a single note."""
    with _db() as conn:
        row = conn.execute("SELECT * FROM notes WHERE id = ?", (note_id,)).fetchone()
    if not row:
        raise ResourceNotFoundError(f"No note found with id {note_id}.")
    return f"Title: {row['title']}\nTag: {row['tag']}\nCreated: {row['created_at']}\n\n{row['body']}"


@mcp.custom_route("/healthz", methods=["GET"])
async def healthz(request: Request) -> PlainTextResponse:
    return PlainTextResponse("ok")  # the process is alive


@mcp.custom_route("/readyz", methods=["GET"])
async def readyz(request: Request) -> PlainTextResponse:
    with _db() as conn:
        conn.execute("SELECT 1")  # configuration loaded, the database answers
    return PlainTextResponse("ready")


if __name__ == "__main__":
    mcp.run(transport="stdio")

What each part is for:

• ToolError lists every problem, so the model fixes all of them in one retry; an empty search is count: 0, not an error. → ch05
• SearchResult adds an outputSchema, so results carry structuredContent (Appendix A). snippet exists because the host decides when the detail resource is attached. → ch06
• Annotations are honest: deleting a deleted note is a no-op, so delete_note is idempotent. → ch06
• ResourceNotFoundError reaches the client as -32602. → ch05
• /readyz runs SELECT 1 and never calls a slow downstream. → ch07, Appendix J

remote.py

The same server over Streamable HTTP. → ch07

import os

os.environ.setdefault("NOTES_DB_PATH", "/var/lib/notes/notes.db")  # before the import reads it

from server import mcp

if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="127.0.0.1",      # loopback only; a proxy or tunnel faces the network
        port=3000,
        json_response=True,
        stateless_http=True,
    )

On loopback the SDK rejects a foreign Origin with 403. Put TLS and token validation (Chapter 19) in front before exposing it.

test_server.py

A contract test through the in-memory client plus two error-model tests, run with uv run pytest under anyio’s pytest plugin. → ch12

import pytest
from mcp import Client

import server

pytestmark = pytest.mark.anyio


@pytest.fixture(autouse=True)
def fresh_db(tmp_path, monkeypatch):
    monkeypatch.setattr(server, "DB_PATH", str(tmp_path / "notes.db"))


async def test_surface_contract():
    async with Client(server.mcp) as client:
        tools = {t.name: t for t in (await client.list_tools()).tools}
    assert set(tools) == {"create_note", "search_notes", "delete_note"}
    assert tools["search_notes"].annotations.read_only_hint is True
    assert tools["delete_note"].annotations.destructive_hint is True
    assert "items" in tools["search_notes"].output_schema["properties"]


async def test_errors_list_every_problem():
    async with Client(server.mcp) as client:
        result = await client.call_tool("create_note", {"title": " ", "body": ""})
    assert result.is_error
    text = result.content[0].text
    assert "title must not be empty" in text and "body must not be empty" in text


async def test_empty_search_is_not_an_error():
    async with Client(server.mcp) as client:
        await client.call_tool("create_note", {"title": "Plan", "body": "Ship v2"})
        hit = await client.call_tool("search_notes", {"query": "plan"})
        miss = await client.call_tool("search_notes", {"query": "zzz"})
    assert hit.structured_content["count"] == 1
    assert not miss.is_error and miss.structured_content["count"] == 0