The fastest way to internalize a protocol is to put a program on the other end of it. In this lesson you build a complete MCP server around a small but honest system — a personal task store backed by SQLite — and take it through the full builder loop: define tools, test them interactively in the MCP Inspector, plug the server into a commercial host, and finally connect it to the bare agent loop you wrote in Agent Architectures: Loops, Planning & Memory. The same server, unchanged, serves all three hosts. That is the M + N promise from Lesson 1 made physical.
One note on the code: MCP servers are real processes speaking over stdin/stdout, so they cannot run in this page's in-browser Python. Every block is complete and correct — build the file locally as you read, and run the Inspector step yourself.
Resist the demo trap of wrapping a dictionary. A server over real storage forces the questions that make MCP interesting: what happens on double-completion, how do you phrase errors so a model can recover, which operations are dangerous. SQLite from the standard library is enough:
# store.py — the system under the server. No MCP here at all.
import sqlite3
from pathlib import Path
DB = Path.home() / ".tasks.db"
def _conn() -> sqlite3.Connection:
conn = sqlite3.connect(DB)
conn.row_factory = sqlite3.Row
conn.execute(
"CREATE TABLE IF NOT EXISTS tasks ("
" id INTEGER PRIMARY KEY,"
" title TEXT NOT NULL,"
" due TEXT," # ISO date or NULL
" status TEXT NOT NULL DEFAULT 'open'" # open | done
")"
)
return conn
def add(title: str, due: str | None) -> int:
with _conn() as c:
return c.execute("INSERT INTO tasks (title, due) VALUES (?, ?)",
(title, due)).lastrowid
def list_by_status(status: str, limit: int) -> list[dict]:
with _conn() as c:
rows = c.execute("SELECT * FROM tasks WHERE status = ? "
"ORDER BY id LIMIT ?", (status, limit)).fetchall()
return [dict(r) for r in rows]
def set_status(task_id: int, status: str) -> bool:
with _conn() as c:
return c.execute("UPDATE tasks SET status = ? WHERE id = ?",
(status, task_id)).rowcount == 1
def delete(task_id: int) -> bool:
with _conn() as c:
return c.execute("DELETE FROM tasks WHERE id = ?",
(task_id,)).rowcount == 1
Keeping the store MCP-free is deliberate: the protocol layer you write next is an adapter, exactly one of the N adapters from the M + N arithmetic, and it should stay thin enough that you could swap SQLite for Postgres without touching a tool definition.
The official Python SDK (package mcp) includes
FastMCP, a decorator-style server API that reads Python type
hints and docstrings and compiles them into the protocol-level
tool schemas — the same trick your hand-rolled tool layer did
with JSON Schema, now standardized. Set up and write the first
tool:
uv init tasks-server && cd tasks-server
uv add "mcp[cli]"
# server.py
from mcp.server.fastmcp import FastMCP
import store
mcp = FastMCP("tasks")
@mcp.tool()
def add_task(title: str, due: str | None = None) -> str:
'''Create a new task in the user's personal task list.
Use this when the user asks to remember, track, or schedule
something. Args: title is a short imperative phrase ('Renew the
TLS cert'); due is an optional ISO date like '2026-08-15' — omit
it rather than guessing a date the user did not give.
Returns a confirmation containing the new task's numeric id.
'''
task_id = store.add(title, due)
return f"Created task {task_id}: {title}" + (f" (due {due})" if due else "")
if __name__ == "__main__":
mcp.run() # stdio transport by default
Run uv run server.py and it appears to hang. It is
not hanging — it is a stdio server, silently waiting for a
JSON-RPC initialize on stdin. From the type hints
FastMCP generated an input schema (title: required
string, due: optional string), and from the
docstring it took the description. Nothing else was required:
the handshake from Lesson 2, capability declaration, and message
routing all come with the SDK.
Before adding the remaining tools, absorb the most
disproportionate fact in MCP engineering: your docstrings
are prompt engineering. Every connected host injects
your tool names, descriptions, and schemas into the model's
context; they are the only interface the model has to your
server. A vague description — '''Lists tasks.''' —
produces wrong calls at runtime, on someone else's host, where
you cannot debug. Study the docstring on list_tasks
below: it does four jobs, saying when to reach for the
tool, defining every argument's domain, committing to an output
format the model can parse, and redirecting the model away from
misuse. The rest of the set — read, complete, delete:
@mcp.tool()
def list_tasks(status: str = "open", limit: int = 20) -> str:
'''List tasks filtered by status.
Use this to answer questions about what is pending or recently
finished. Args: status is 'open' (default) or 'done'; limit caps
results at 50 (default 20). Returns one task per line as
'id | title | due | status', or 'No tasks.' if none match.
'''
if status not in ("open", "done"):
return "Error: status must be 'open' or 'done'."
rows = store.list_by_status(status, min(limit, 50))
if not rows:
return "No tasks."
return "\n".join(
f"{r['id']} | {r['title']} | {r['due'] or '-'} | {r['status']}"
for r in rows)
@mcp.tool()
def complete_task(task_id: int) -> str:
'''Mark a single task as done, by numeric id.
Get ids from list_tasks first — never guess an id. Completing
an already-done task succeeds and says so. Returns confirmation
or an error naming the missing id.
'''
if store.set_status(task_id, "done"):
return f"Task {task_id} marked done."
return f"Error: no task with id {task_id}. Use list_tasks to see valid ids."
@mcp.tool()
def delete_task(task_id: int) -> str:
'''Permanently delete a single task, by numeric id.
Irreversible. Prefer complete_task unless the user explicitly
asks to delete or remove. Returns confirmation or an error
naming the missing id.
'''
if store.delete(task_id):
return f"Task {task_id} deleted."
return f"Error: no task with id {task_id}. Use list_tasks to see valid ids."
Note the error style: failures are returned as readable strings
(in-band, isError content — Lesson 2's distinction),
and each error tells the model what to do next
(use list_tasks to see valid ids). That one recovery
hint measurably reduces flailing: the model's next action is a
listing, not three more guessed ids.
Four of these tools are not equally dangerous, and MCP gives you vocabulary to say so. Annotations are structured hints on each tool describing its behavior, which hosts use to shape UX — auto-approving read-only calls, demanding confirmation for destructive ones:
| Annotation | Default | Meaning |
|---|---|---|
readOnlyHint | false | Tool does not modify its environment |
destructiveHint | true | Updates may be irreversible (meaningful only when not read-only) |
idempotentHint | false | Repeating a call with the same args has no additional effect |
openWorldHint | true | Tool touches the open world (web, third parties) rather than a closed system |
from mcp.types import ToolAnnotations
@mcp.tool(annotations=ToolAnnotations(
readOnlyHint=True, openWorldHint=False))
def list_tasks(status: str = "open", limit: int = 20) -> str:
...
@mcp.tool(annotations=ToolAnnotations(
destructiveHint=False, idempotentHint=True, openWorldHint=False))
def complete_task(task_id: int) -> str:
...
@mcp.tool(annotations=ToolAnnotations(
destructiveHint=True, idempotentHint=True, openWorldHint=False))
def delete_task(task_id: int) -> str:
...
Reasoning through the choices: listing is read-only over a closed
system. Completing is a write, but reversible and idempotent —
marking done twice changes nothing. Deleting is destructive and
(per-id) idempotent. add_task keeps defaults except
openWorldHint=False: it is not idempotent —
the same call twice creates two tasks — which tells a host it
must not blindly retry on timeout.
You never debug a server through a chat model — too many moving parts. The MCP Inspector is the protocol's standard test bench: a local web app that acts as a bare host, showing you exactly what any client sees. One command, no install:
npx @modelcontextprotocol/inspector uv run server.py
Your browser opens the Inspector UI (served on
localhost:6274). Click Connect and
walk Lesson 2's protocol live: the initialize
exchange with negotiated version and capabilities, then a
Tools tab where List Tools
shows your four tools with the schemas FastMCP generated. Now
exercise the edges, because this is the cheapest place to find
bugs:
add_task with only a title — the optional
due should be genuinely optional.complete_task with id 9999 — you should get
your recovery-hint error as tool content, not a protocol
error or a stack trace.list_tasks with status="urgent"
— the validation message should come back in-band.
Make the Inspector a reflex: every server change gets thirty
seconds here before it gets near a model. It also surfaces the
stdout bug instantly — add a stray print, reconnect,
and watch the connection fail with a parse error.
Now give the server to a real host. For Claude Desktop, add an
entry to claude_desktop_config.json (Settings →
Developer); Cursor and VS Code use the same shape in their own
config files:
{
"mcpServers": {
"tasks": {
"command": "uv",
"args": ["run", "--directory", "/absolute/path/to/tasks-server", "server.py"]
}
}
}
Restart the app. The host spawns your server as a subprocess,
runs the handshake, and merges your tools into the model's
toolbox under the tasks namespace. Ask 'what's still
open on my list?' and watch the pieces cooperate: the model
selects list_tasks (guided by your description), the
host asks permission on first use, the call crosses stdio as the
exact frames you saw in the Inspector. Then try 'clean up
everything I've finished' and notice the host treating
delete_task more cautiously — your annotations at
work.
The same SDK contains the client side, which turns the agent loop you built two courses ago into a genuine MCP host. Where you previously hand-registered Python functions into a tool registry, you now discover tools at runtime:
# agent_host.py — your loop from 'Agent Architectures', MCP edition
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
server = StdioServerParameters(
command="uv",
args=["run", "--directory", "/absolute/path/to/tasks-server", "server.py"],
)
async def main() -> None:
async with stdio_client(server) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# Discover tools; name/description/inputSchema map
# one-to-one onto your LLM API's tool format.
tools = await session.list_tools()
specs = [{"name": t.name, "description": t.description,
"input_schema": t.inputSchema} for t in tools.tools]
# Inside your agent loop, when the model emits a tool call:
result = await session.call_tool(
"add_task", {"title": "Review MCP lesson"})
print(result.content[0].text) # -> Created task 1: ...
asyncio.run(main())
Look at what disappeared from your old architecture: the tool registry, the schema definitions, the dispatch table. Your agent now asks the server what exists and forwards calls. Point the same loop at any of the thousands of published MCP servers — GitHub, Postgres, a browser — and your agent gains those capabilities with zero new integration code. You have crossed to the M side of M + N.
Check: your add_task tool occasionally times out on
a slow disk, and you notice the host retrying the call — users end
up with duplicate tasks. Which server-side change most directly
addresses this?
You now own the complete minimal loop of MCP server engineering. Before you call any server done, run this list: storage layer separate from protocol layer; every docstring states when to use the tool, argument domains, and return format; every failure returns an in-band error with a recovery hint; annotations truthful for all four hints; zero writes to stdout; edge cases exercised in the Inspector; verified against at least one real host. Next lesson the server grows the rest of the primitive vocabulary — resources, prompts, progress, and mid-call conversations with the user.