Topic 6: Module 3 lab
4 min read·22 Sept 2026
The Lab runs the whole module in one script against a temporary copy of the notes: it checks the definitions (names, annotations, approval, cost), measures selection accuracy with the stand-in, lets a scripted model recover from a validation error, retries a side effect and recovers, and changes the tool list while listening.
python
"""Module 3 lab: define, describe, measure, break, and change the notes tools in one run."""
import json
import logging
import re
import shutil
import tempfile
from pathlib import Path
import anyio
from mcp import Client
from mcp.client.subscriptions import ToolsListChanged
from mcp.shared.tool_name_validation import validate_tool_name
from examples.m03_dynamic_tools import build_dynamic_server
from examples.m03_tool_cost import as_model_sees_it
from examples.m03_tool_selection_eval import evaluate
from notes_assistant.host import needs_approval
from notes_assistant.store import NoteStore
logging.basicConfig(level=logging.WARNING)
def scripted_model(last_error: str | None) -> dict:
"""A scripted stand-in for the model: asks for 50 hits, then obeys the limit named in the error."""
if last_error is None:
return {"query": "sleep", "limit": 50}
match = re.search(r"less than or equal to (\d+)", last_error)
return {"query": "sleep", "limit": int(match.group(1))} if match else {"query": "sleep"}
async def main() -> None:
folder = Path(tempfile.mkdtemp()) / "notes"
shutil.copytree("notes", folder)
async with Client(build_dynamic_server(NoteStore(folder))) as client:
tools = (await client.list_tools()).tools
print("1. Definitions")
for tool in tools:
hints = tool.annotations.model_dump(exclude_none=True) if tool.annotations else {}
print(f" {tool.name:<22} name valid={validate_tool_name(tool.name).is_valid} needs approval={needs_approval(tool)} {hints}")
chars = len(json.dumps([as_model_sees_it(t) for t in tools]))
print(f" definition cost: {chars} chars, about {chars // 4} tokens (chars / 4 estimate)")
print("2. Selection accuracy (keyword stand-in, not a model)")
await evaluate("merged", use_llm=False, verbose=False)
print("3. Self-correction from a validation error (scripted stand-in model)")
error = None
for attempt in (1, 2):
args = scripted_model(error)
result = await client.call_tool("search_notes", args)
error = result.content[0].text if result.is_error else None
outcome = f"error: {error.splitlines()[2].strip()}" if error else f"{len(result.structured_content['hits'])} hits"
print(f" attempt {attempt} {args} -> {outcome}")
print("4. Side effects and retries")
note = {"title": "Nap study consent form", "body": "Draft sent to Priya on 14 September.", "tags": ["sleep"]}
for attempt in (1, 2):
result = await client.call_tool("create_note", note)
print(f" create_note attempt {attempt}: is_error={result.is_error} {result.content[0].text if result.is_error else result.structured_content['uri']}")
found = await client.call_tool("search_notes", {"query": "consent form", "limit": 1})
print(f" recovery: search_notes finds {found.structured_content['hits'][0]['uri']}")
print("5. Dynamic tool list")
async with client.listen(tools_list_changed=True) as sub:
await client.call_tool("enable_meeting_tools", {})
with anyio.fail_after(5):
async for event in sub:
if isinstance(event, ToolsListChanged):
names = [t.name for t in (await client.list_tools()).tools]
print(f" tools/list_changed received; tools now: {names}")
break
shutil.rmtree(folder.parent)
if __name__ == "__main__":
anyio.run(main)Code explained