Chapter 11: Extended Projects

Patterns are easy to agree with one at a time. The hard part is making every decision at once: which tools, which resources, where state lives, what the model is never allowed to decide, and what you leave out. A server built from a list of individually sensible patterns can still hand the model a choice it should never have had.

The rule for this chapter: every tool is a decision you take away from the model, so ship the smallest surface that covers the user’s intents, and put every irreversible decision behind a human. Both projects below are measured against it.

This chapter builds two complete servers. Not stubs. Not outlines. Each comes with working code, a test against the real MCP surface, and the ways it fails. Project 1, a task manager, runs on Chapter 8’s patterns. Project 2, an operations copilot, assembles Chapter 10’s: server-held state, a human approval gate and a long-running job. Both use the official Python SDK’s MCPServer (version 2.2); the decisions translate directly to TypeScript.

The Process Both Projects Follow

1. Define the user intents. Write down, as plain sentences, what someone will actually ask the model to do. Do this before you design any tools.
2. Pick the smallest tool set that covers them. If two tools are always called together, merge them. If one tool does three unrelated things, split it.
3. Add resources for context the host attaches, and a read tool for anything the model must fetch itself. The host decides when resources are read (Chapter 6), so a resource can’t be the model’s only way to see something.
4. Decide where human approvals belong. Reads are safe. Writes that affect only the user are usually safe. Anything that affects other people or production needs a human in the loop.
5. Set response size budgets. Cap every list, truncate long bodies, and say what was left out.

Project 1: Personal Task Manager

A task manager is the “hello world” of CRUD servers, and that is why it teaches well: CRUD is easy to implement and surprisingly hard to expose well through MCP. Tasks have a title, an optional description, a status (open or completed), a priority (low, normal, high or urgent) and an optional due date, stored in SQLite for one user.

The Surface Design

Step 1, intents: “what’s on my plate today?”, “add a task”, “I finished #12”, “bump the report to urgent”, “move the dentist to Friday”, “run my weekly review”.

Step 2, tools:

Step 3, resources:

Prompts:

Notice what is not here. There is no update_task god-tool with ten optional parameters: the two edits users actually ask for get one narrow tool each. There is no search_tasks with a query language. And list_tasks exists although tasks://all holds the same data, because the host decides when resources are attached. In a client that attaches them only when the user picks one, a model without list_tasks never learns a task_id and can’t complete, reopen or delete anything.

Step 4 is simple here: every write touches only the user’s own list, so none needs confirmation, and delete_task says it can’t be undone. Step 5: list_tasks caps at 50 and says what it omitted.

Implementation

Start with the data layer. Keep it independent of MCP so you can test it separately:

# task_store.py
import sqlite3
from datetime import datetime, timezone
from dataclasses import dataclass
from typing import Optional

@dataclass
class Task:
    id: int
    title: str
    description: Optional[str]
    status: str  # "open" or "completed"
    priority: str  # "low", "normal", "high", "urgent"
    due_date: Optional[str]  # ISO date string
    created_at: str
    completed_at: Optional[str] = None

class TaskStore:
    def __init__(self, db_path: str = "tasks.db"):
        # The SDK runs sync tools in worker threads.
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.conn.row_factory = sqlite3.Row
        self._init_schema()

    def _init_schema(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS tasks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                description TEXT,
                status TEXT NOT NULL DEFAULT 'open',
                priority TEXT NOT NULL DEFAULT 'normal',
                due_date TEXT,
                created_at TEXT NOT NULL,
                completed_at TEXT
            )
        """)
        self.conn.commit()

    def create(self, title: str, description: str = None,
               priority: str = "normal", due_date: str = None) -> Task:
        self._check(priority, due_date)
        now = datetime.now(timezone.utc).isoformat()
        cursor = self.conn.execute(
            "INSERT INTO tasks (title, description, priority, due_date, created_at) "
            "VALUES (?, ?, ?, ?, ?)",
            (title, description, priority, due_date, now)
        )
        self.conn.commit()
        return self._get(cursor.lastrowid)

    def set_priority(self, task_id: int, priority: str) -> Task:
        self._check(priority, None)
        return self._update(task_id, "priority", priority)

    def set_due_date(self, task_id: int, due_date: str | None) -> Task:
        self._check("normal", due_date)
        return self._update(task_id, "due_date", due_date)

    def complete(self, task_id: int) -> Task:
        task = self._get(task_id)
        if task.status == "completed":
            raise ValueError(f"Task {task_id} is already completed")
        now = datetime.now(timezone.utc).isoformat()
        self.conn.execute(
            "UPDATE tasks SET status = 'completed', completed_at = ? WHERE id = ?",
            (now, task_id)
        )
        self.conn.commit()
        return self._get(task_id)

    def reopen(self, task_id: int) -> Task:
        task = self._get(task_id)
        if task.status == "open":
            raise ValueError(f"Task {task_id} is already open")
        self.conn.execute(
            "UPDATE tasks SET status = 'open', completed_at = NULL WHERE id = ?",
            (task_id,)
        )
        self.conn.commit()
        return self._get(task_id)

    def delete(self, task_id: int) -> None:
        self._get(task_id)  # Raises if missing
        self.conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
        self.conn.commit()

    def open_tasks(self, due_date: str = None) -> list[Task]:
        if due_date:
            rows = self.conn.execute(
                "SELECT * FROM tasks WHERE status = 'open' AND due_date <= ? "
                "ORDER BY due_date, CASE priority WHEN 'urgent' THEN 0 WHEN 'high' THEN 1 WHEN 'normal' THEN 2 ELSE 3 END",
                (due_date,)
            ).fetchall()
        else:
            rows = self.conn.execute(
                "SELECT * FROM tasks WHERE status = 'open' "
                "ORDER BY CASE priority WHEN 'urgent' THEN 0 WHEN 'high' THEN 1 WHEN 'normal' THEN 2 ELSE 3 END, due_date"
            ).fetchall()
        return [self._row_to_task(r) for r in rows]

    def summary(self) -> dict:
        rows = self.conn.execute("""
            SELECT
                COUNT(*) FILTER (WHERE status = 'open') as open_count,
                COUNT(*) FILTER (WHERE status = 'completed') as completed_count,
                COUNT(*) FILTER (WHERE status = 'open'
                    AND due_date < date('now')) as overdue_count
            FROM tasks
        """).fetchone()
        return {
            "open": rows["open_count"],
            "completed": rows["completed_count"],
            "overdue": rows["overdue_count"]
        }

    def _check(self, priority: str, due_date: str | None) -> None:
        if priority not in ("low", "normal", "high", "urgent"):
            raise ValueError(f"Invalid priority: {priority}. "
                             "Use low, normal, high or urgent.")
        if due_date:
            try:
                datetime.strptime(due_date, "%Y-%m-%d")
            except ValueError:
                raise ValueError(f"Invalid date format: {due_date}. Use YYYY-MM-DD.")

    def _update(self, task_id: int, column: str, value) -> Task:
        self._get(task_id)  # Raises if missing
        # column comes from this class, never from the caller
        self.conn.execute(f"UPDATE tasks SET {column} = ? WHERE id = ?",
                          (value, task_id))
        self.conn.commit()
        return self._get(task_id)

    def _get(self, task_id: int) -> Task:
        row = self.conn.execute(
            "SELECT * FROM tasks WHERE id = ?", (task_id,)
        ).fetchone()
        if not row:
            raise ValueError(f"Task {task_id} not found")
        return self._row_to_task(row)

    def _row_to_task(self, row) -> Task:
        return Task(**dict(row))

The data layer validates inputs, raises clear exceptions, and knows nothing about MCP. That means you can unit-test every edge case (invalid priorities, duplicate completions, missing IDs) without starting a server.

Now the MCP layer. It should be thin:

# server.py
from datetime import date
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from task_store import TaskStore

mcp = MCPServer("TaskManager")
store = TaskStore()

def run(op, *args):
    """Turn the store's ValueErrors into tool errors the model can read."""
    try:
        return op(*args)
    except ValueError as e:
        raise ToolError(str(e)) from e

# --- Tools ---

@mcp.tool()
def list_tasks(due_by: str = "") -> str:
    """List open tasks with their IDs, most urgent first (max 50).
    due_by (YYYY-MM-DD) limits the list to tasks due on or before that date."""
    tasks = run(store.open_tasks, due_by or None)
    if not tasks:
        return "No open tasks."
    lines = [f"#{t.id} [{t.priority}] {t.title}"
             + (f" (due {t.due_date})" if t.due_date else "") for t in tasks[:50]]
    if len(tasks) > 50:
        lines.append(f"... {len(tasks) - 50} more; narrow with due_by.")
    return "\n".join(lines)

@mcp.tool()
def create_task(title: str, description: str = "",
                priority: str = "normal", due_date: str = "") -> str:
    """Create a new task. Priority: low, normal, high, urgent.
    Due date format: YYYY-MM-DD."""
    task = run(store.create, title, description or None,
               priority, due_date or None)
    return f"Created task #{task.id}: {task.title} [{task.priority}]"

@mcp.tool()
def set_priority(task_id: int, priority: str) -> str:
    """Change a task's priority: low, normal, high, urgent."""
    task = run(store.set_priority, task_id, priority)
    return f"Task #{task.id} is now {task.priority}"

@mcp.tool()
def set_due_date(task_id: int, due_date: str = "") -> str:
    """Set a task's due date (YYYY-MM-DD). Empty string clears it."""
    task = run(store.set_due_date, task_id, due_date or None)
    return f"Task #{task.id} due {task.due_date or 'never'}"

@mcp.tool()
def complete_task(task_id: int) -> str:
    """Mark an open task as completed."""
    task = run(store.complete, task_id)
    return f"Completed task #{task.id}: {task.title}"

@mcp.tool()
def reopen_task(task_id: int) -> str:
    """Reopen a previously completed task."""
    task = run(store.reopen, task_id)
    return f"Reopened task #{task.id}: {task.title}"

@mcp.tool()
def delete_task(task_id: int) -> str:
    """Permanently delete a task. Cannot be undone."""
    run(store.delete, task_id)
    return f"Deleted task #{task_id}"

# --- Resources: context the host can attach ---

@mcp.resource("tasks://today")
def today_tasks() -> str:
    """Open tasks due today or overdue."""
    tasks = store.open_tasks(due_date=date.today().isoformat())
    if not tasks:
        return "No tasks due today or overdue."
    lines = []
    for t in tasks:
        overdue = " ⚠️ OVERDUE" if t.due_date and t.due_date < date.today().isoformat() else ""
        lines.append(f"#{t.id} [{t.priority}] {t.title}{overdue}")
    return "\n".join(lines)

@mcp.resource("tasks://all")
def all_open_tasks() -> str:
    """All open tasks grouped by priority."""
    tasks = store.open_tasks()
    if not tasks:
        return "No open tasks."
    grouped = {}
    for t in tasks:
        grouped.setdefault(t.priority, []).append(t)
    lines = []
    for priority in ("urgent", "high", "normal", "low"):
        group = grouped.get(priority, [])
        if group:
            lines.append(f"\n## {priority.upper()}")
            for t in group:
                due = f" (due {t.due_date})" if t.due_date else ""
                lines.append(f"  #{t.id}: {t.title}{due}")
    return "\n".join(lines)

@mcp.resource("tasks://summary")
def task_summary() -> str:
    """Quick summary: open, completed, and overdue counts."""
    s = store.summary()
    return (
        f"Open: {s['open']} | Completed: {s['completed']} | "
        f"Overdue: {s['overdue']}"
    )

# --- Prompts ---

@mcp.prompt()
def weekly_review() -> str:
    """Structured weekly review. Reads the full task list, identifies
    stale items, and suggests reprioritization."""
    return (
        "Please review my current tasks. For each open task:\n"
        "1. Is it still relevant? If not, suggest deleting it.\n"
        "2. Is the priority still correct?\n"
        "3. Does it need a due date or a due date change?\n"
        "4. Are any tasks blocked or waiting on something?\n\n"
        "Start by calling list_tasks, then walk through your "
        "recommendations. Apply the ones I approve with set_priority, "
        "set_due_date or delete_task."
    )

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

Every tool calls the store through run, which turns ValueError into ToolError. MCPServer passes the message through only for ToolError; any other exception reaches the model as a bare “Error executing tool complete_task”, which gives it nothing to correct. With run, a missing task comes back as an isError: true result reading “Task 42 not found”, and the model can call list_tasks and retry. And every prompt instruction maps to a tool that exists: the review asks about priority and due dates because set_priority and set_due_date can act on the answers.

Testing Both Layers

Test the data layer with ordinary unit tests:

# test_task_store.py
import pytest
from task_store import TaskStore

@pytest.fixture
def store(tmp_path):
    return TaskStore(str(tmp_path / "test.db"))

def test_complete_already_completed_raises(store):
    task = store.create("Test task")
    store.complete(task.id)
    with pytest.raises(ValueError, match="already completed"):
        store.complete(task.id)

def test_invalid_priority_raises(store):
    with pytest.raises(ValueError, match="Invalid priority"):
        store.create("Bad task", priority="critical")

Then test the MCP surface with the SDK’s in-memory client. Client accepts the server object itself and dispatches to it in-process, with no subprocess and no transport:

# test_server.py
import pytest
from mcp import Client
import server

@pytest.mark.anyio
async def test_errors_reach_the_model(tmp_path, monkeypatch):
    monkeypatch.setattr(server, "store", server.TaskStore(str(tmp_path / "t.db")))
    async with Client(server.mcp) as client:
        await client.call_tool("create_task", {"title": "Ship ch11"})
        listed = await client.call_tool("list_tasks", {})
        assert "#1 [normal] Ship ch11" in listed.content[0].text
        missing = await client.call_tool("complete_task", {"task_id": 42})
        assert missing.is_error and "Task 42 not found" in missing.content[0].text

This test proves the ToolError translation, and it is what caught the bug behind the check_same_thread=False comment: the SDK runs sync tools in worker threads, and a default SQLite connection refuses any thread but its creator’s. The unit tests can’t see that.

For a manual pass, let the Inspector launch the server (a stdio server started on its own just waits on stdin):

npx @modelcontextprotocol/inspector uv run server.py

Create a task, list it, complete it, and check tasks://summary.

Failure Modes

The god-tool trap. A single manage_tasks tool with an action parameter sounds efficient. In practice models pick actions inconsistently, the schema becomes a wall of conditionally required parameters, and errors can’t say which operation failed.

Phantom task IDs. Models sometimes invent IDs they have never seen. A “Task 42 not found” error lets the model self-correct. A vague “operation failed” causes retry loops.

Date format drift. Without explicit format guidance in the description, models will send “March 15”, “next Tuesday” or “3/15/2026”. Validate strictly and put the expected format in the error message, so the model can retry once.

Stale reads. A resource the host attached at the start of the conversation is a snapshot. After a batch of mutations, the model should call list_tasks again rather than trust what is already in its context.

SQLite beyond one user. One connection shared across worker threads is fine for one user’s single-statement writes. For many users, move to PostgreSQL rather than stretching SQLite.

Project 2: Operations Copilot

The second project points a model at production. An on-call engineer asks “why is checkout slow?”, and the copilot pulls health and errors, keeps an incident timeline, runs a diagnostic runbook in the background and, if a human agrees, restarts one stateless service. Operators come to trust a copilot through excellent diagnostics, not through maximum power. Every sharp edge in this project is state, time or consequence, which is Chapter 10’s territory.

The Surface Design

Step 1, intents: “what’s unhealthy?”, “what is checkout-web throwing?”, “start an incident”, “note that p95 recovered”, “run the database latency checks”, “restart checkout-web”.

Steps 2 and 3, tools and resources:

Both resources have a tool twin (check_service with no argument, and the timeline add_note returns). Step 4 is the design: exactly one tool changes production, and only for allowlisted services, after a human confirms, at most once per ten minutes per service. Step 5: five error groups over at most 60 minutes, because a model reading 500 stack traces reasons worse than one reading the top five.

Implementation

platform is your adapter to whatever holds the truth: metrics, the error tracker, the orchestrator. TokenBucket is Chapter 9’s limiter, imported unchanged rather than rewritten.

# ops_server.py
import asyncio, secrets, time
from typing import Annotated
from pydantic import BaseModel
from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.elicitation import ElicitationResult
from mcp.server.mcpserver import Elicit, MCPServer, Resolve
from mcp.server.mcpserver.exceptions import ToolError
from rate_limit import TokenBucket          # Chapter 9, unchanged
from platform_api import platform           # your metrics/orchestrator adapter

mcp = MCPServer("OpsCopilot")
RESTARTABLE = {"checkout-web", "search-worker"}   # stateless; safe to bounce
RUNBOOKS = {"db-latency": ["connections", "slow_queries", "replication_lag"]}
INCIDENT_TTL = 4 * 3600
incidents: dict[str, dict] = {}
jobs: dict[str, dict] = {}
restart_limits: dict[str, TokenBucket] = {}

def caller() -> str:
    token = get_access_token()          # identity never comes from an argument
    return token.subject if token else "local-operator"   # stdio: one operator

def incident(incident_id: str) -> dict:
    inc = incidents.get(incident_id)
    if inc is None or inc["owner"] != caller():
        raise ToolError(f"No incident {incident_id}. Call open_incident first.")
    if time.time() > inc["expires"]:
        raise ToolError(f"Incident {incident_id} expired. Open a new one.")
    inc["expires"] = time.time() + INCIDENT_TTL
    return inc

def log(inc: dict, entry: str) -> None:
    inc["timeline"].append(f"{time.strftime('%H:%M:%S')} {entry}")

# --- Diagnostics: read-only, cheap, and the reason operators trust this ---

def health_lines(services) -> str:
    return "\n".join(f"{s.name}: {s.status}, p95 {s.p95_ms} ms, "
                     f"{s.error_rate:.1%} errors, last deploy {s.last_deploy}"
                     for s in services)

@mcp.resource("ops://services")
def services_overview() -> str:
    """Health of every service, one line each."""
    return health_lines(platform.services())

@mcp.tool()
def check_service(service: str = "") -> str:
    """Health, p95 latency, error rate and last deploy for one service,
    or for every service when service is empty."""
    if not service:
        return health_lines(platform.services())
    s = platform.service(service)
    if s is None:
        raise ToolError(f"Unknown service {service}. "
                        "Call check_service with no argument for the list.")
    return health_lines([s])

@mcp.tool()
def recent_errors(service: str, minutes: int = 15) -> str:
    """Top 5 error messages for a service in the last N minutes (max 60)."""
    groups = platform.errors(service, min(minutes, 60))[:5]
    return "\n".join(f"{g.count}x {g.message}" for g in groups) or "No errors."

# --- Incident state: server-minted ids, owned by the caller ---

@mcp.tool()
def open_incident(summary: str) -> str:
    """Start an incident. Returns the incident_id the other tools need."""
    incident_id = "inc_" + secrets.token_hex(4)
    incidents[incident_id] = {"owner": caller(), "summary": summary,
                              "timeline": [], "expires": time.time() + INCIDENT_TTL}
    log(incidents[incident_id], f"opened: {summary}")
    return incident_id

@mcp.tool()
def add_note(incident_id: str, note: str) -> str:
    """Append a finding or decision to the incident timeline.
    Returns the whole timeline."""
    inc = incident(incident_id)
    log(inc, note)
    return "\n".join(inc["timeline"])

@mcp.resource("ops://incidents/{incident_id}")
def incident_timeline(incident_id: str) -> str:
    """The incident's timeline, oldest first."""
    return "\n".join(incident(incident_id)["timeline"])

# --- One narrow action, behind a human ---

class Confirm(BaseModel):
    confirm: bool

def ask_restart(incident_id: str, service: str) -> Elicit[Confirm]:
    # A resolver may run on every retry round (Chapter 10): checks only, no effects.
    incident(incident_id)
    if service not in RESTARTABLE:
        raise ToolError(f"{service} is not restartable from here. "
                        f"Allowed: {', '.join(sorted(RESTARTABLE))}.")
    return Elicit(f"Rolling restart of {service} for {incident_id}?", Confirm)

@mcp.tool()
async def restart_service(
    incident_id: str, service: str,
    answer: Annotated[ElicitationResult[Confirm], Resolve(ask_restart)],
) -> str:
    """Rolling restart of a stateless service (checkout-web or
    search-worker). A human confirms every restart."""
    # The body runs once, after the human has answered.
    inc = incident(incident_id)
    if answer.action != "accept" or not answer.data.confirm:
        log(inc, f"restart of {service} declined")
        return f"Restart of {service} not confirmed. Nothing changed."
    bucket = restart_limits.setdefault(service, TokenBucket(rate=1 / 600, burst=1))
    if not await bucket.acquire():
        raise ToolError(f"{service} was restarted in the last 10 minutes. "
                        "Check it before restarting again.")
    await platform.rolling_restart(service)
    log(inc, f"restarted {service} (confirmed by {caller()})")
    return f"Rolling restart of {service} started."

# --- A long-running runbook: return a handle, own the polling pace ---

@mcp.tool()
async def start_runbook(incident_id: str, runbook: str) -> dict:
    """Run a read-only diagnostic runbook (db-latency). Returns a job_id;
    poll runbook_status after poll_after_seconds."""
    inc = incident(incident_id)
    if runbook not in RUNBOOKS:
        raise ToolError(f"Unknown runbook {runbook}. Known: {', '.join(RUNBOOKS)}.")
    job_id = "job_" + secrets.token_hex(4)
    job = jobs[job_id] = {"owner": caller(), "status": "working",
                          "done": [], "steps": RUNBOOKS[runbook]}

    async def run():
        try:
            for step in job["steps"]:
                job["done"].append(f"{step}: {await platform.run_check(step)}")
            job["status"] = "completed"
        except Exception as e:
            job["status"], job["error"] = "failed", str(e)
        log(inc, f"runbook {runbook} {job['status']}")

    job["task"] = asyncio.create_task(run())
    return {"job_id": job_id, "status": "working", "poll_after_seconds": 10}

@mcp.tool()
def runbook_status(job_id: str) -> dict:
    """Progress and findings of a runbook job."""
    job = jobs.get(job_id)
    if job is None or job["owner"] != caller():
        raise ToolError(f"No job {job_id}.")
    out = {"status": job["status"], "findings": job["done"],
           "progress": f"{len(job['done'])}/{len(job['steps'])}"}
    if job["status"] == "working":
        out["poll_after_seconds"] = 10
    if "error" in job:
        out["error"] = job["error"]
    return out

The Server Mints the Handles, and the Handles Have Owners

The model can’t invent an incident. open_incident mints inc_ plus eight random hex characters, binds the id to the caller, and gives it a four-hour TTL that each use extends. The caller comes from the access token, never from a parameter; over stdio there is one operator. A wrong, foreign or expired id gets an error that names the next step, never a fresh empty incident. Job handles follow the same rules.

The Restart Runs Once, However Many Times the Handler Runs

Under 2026-07-28, elicitation is a retry: the server returns InputRequiredResult, the client asks the human and calls the tool again with the answer (Chapter 10). The SDK’s resolvers run that loop for you. A parameter annotated Resolve(ask_restart) is filled by running ask_restart first; when it returns Elicit(...), the framework asks and injects the answer. The resolver may run on every round, and the body runs once. A traced call shows ask_restart, the elicitation, ask_restart again, then the body, and one restart. So the resolver only checks, and every effect sits in the body. Calling ctx.elicit() in the body instead fails on a 2026-07-28 connection: there is no back-channel to send the question on.

The gate fails closed. A client without elicitation gets “Client did not declare the form elicitation capability”, and nothing restarts. There is deliberately no stage-and-confirm fallback: for a production action, “the model promised it asked” isn’t approval. Declines are logged, and a second restart within ten minutes is refused by Chapter 9’s bucket (rate=1/600, burst=1), because a restart that didn’t help once rarely helps twice.

The Server Sets the Polling Pace

The runbook takes minutes, so start_runbook returns a handle at once, with poll_after_seconds: 10. Without it, models either poll in a tight loop or poll once, see “working” and give up. runbook_status repeats the field until the job ends. The states (working, completed, failed) are the Tasks extension’s (Chapter 10), so moving to native Tasks changes the envelope, not the logic.

Testing the Gate

In-process, an elicitation callback plays the human and a fake platform records restarts:

async def approve(context, params):
    return types.ElicitResult(action="accept", content={"confirm": True})

async with Client(ops_server.mcp, elicitation_callback=approve) as c:
    inc = (await c.call_tool("open_incident", {"summary": "checkout slow"})).content[0].text
    await c.call_tool("restart_service", {"incident_id": inc, "service": "checkout-web"})
    again = await c.call_tool("restart_service", {"incident_id": inc, "service": "checkout-web"})
    assert again.is_error and platform.restarts == ["checkout-web"]

Repeat with a callback answering {"confirm": False} and with a service off the allowlist; platform.restarts must not change.

Failure Modes

The restart that ran twice. Put the restart in the resolver, or before an elicitation in the body, and every retry round repeats it. The test above fails loudly if that happens.

Allowlist drift. A service on RESTARTABLE gains a local cache or leader election, and a “safe” bounce now drops work. Review the allowlist with each service’s deploy config, not with the copilot’s code.

Handles that die with the process. incidents and jobs live in memory, so a deploy wipes timelines and orphans jobs. In production, keep both in Redis or your database, and persist runbook findings step by step.

Diagnostics that page through everything. An uncapped recent_errors returns hundreds of near-identical stack traces, and the model latches onto whichever one comes first. Group errors, cap the groups and state the window.

Failure Modes That Cut Across All Projects

Tool description rot. Descriptions written during initial development become inaccurate as behavior changes. The model trusts the description, not the implementation. Review descriptions during every behavior change and include them in your test assertions.

State leaking between sessions. If your server holds state for a caller (an incident, a job, a cached identity), key it to the authenticated caller and check that key on every read. A server that caches User A’s incident and then serves it to User B is a security incident, not a performance optimization.

Testing only the happy path. The Inspector will show you that create_task works. It won’t show you what happens when the task ID doesn’t exist, the human declines, or the handler runs twice. Write in-memory client tests for those paths before you connect a real model.

Implementation Checklist

• The user intents are written down, and every tool maps to at least one of them
• Every constrained parameter lists its valid values in the tool description
• The data layer has unit tests, and the MCP surface has an in-memory Client test that runs in CI
• Expected failures raise ToolError with a message that names the next step; empty results are not errors
• Anything the model must read on its own has a read tool, not only a resource
• Every list has a cap, and the response says what it left out
• Identity comes from the request context, and cross-call ids are minted by the server
• Every production action is allowlisted, confirmed by a human in the tool body, and rate-limited

What Comes Next

Both projects keep this chapter’s rule. The task manager takes every composition decision away from the model with seven narrow tools. The copilot has seven tools too, and only one can touch production: after a human says yes, and never twice in ten minutes. In-memory tests prove the error path, the gate and the handler that runs twice. They can’t prove that a model picks the right tool, reads a description correctly, or keeps polling. Chapter 12 builds the testing strategy for those failures, the ones that appear only when a real model drives a real server.