External Tools with MCP: Unit A Exercises

Student edition · 90 minutes of dedicated work · 2026-09-28

Runs on Google Colab or any Python 3.12+ Jupyter kernel. It needs no packages beyond the standard library, no model and no network.

This is the first of Chapter 14's two practical units, each a complete ninety-minute session.

Unit A connects Lucy's agent to a tool that runs in another process, over the Model Context Protocol's stdio transport, and keeps the decision to call it inside the application.
Unit B breaks the connection on purpose: wrong replies, endless lines, a hung server, and repairs what the failures expose.

You bring JSON values, exceptions and the typed dispatcher from earlier chapters. No JSON-RPC, subprocess or MCP knowledge is assumed. Requests, responses, notifications, framing and the handshake are introduced below, before you build anything.

By the end you should be able to:

Explain how a request's id binds exactly one response to it, and why a reply with another id must be refused.
Exchange newline-framed JSON with a child process, keeping its diagnostics out of the protocol.
Implement answer_for, which accepts a reply only for its own request, and authorize, which allows a call only when the server advertised the tool and the application allowed it.
Show, from the server's own log, that an advertised but forbidden tool was never invoked.

Minutes	Dedicated work	Saved evidence
0–10	Match printed requests and responses by identity	Written associations
10–25	One JSON line to a child process; a frame split across reads	Observed outputs
25–50	Construct `answer_for` and `authorize`; pass the visible cases	Learner code and grade table
50–70	Handshake, discovery and one allowed call against the teaching server	Server call log
70–85	Changed-constraint task: offer the model only allowed tools	Transfer results
85–90	Explain protocol versus permission and save evidence	Retained submission

These times are planning estimates, not measured completion times. Run All only checks that the notebook executes; the unfinished student functions deliberately report NEEDS_WORK. Keep your first attempt before opening the worked edition.

Run the self-contained setup

The unit needs only Python's standard library. The collapsed cell below creates your work folder and defines the supplied parts of the unit:

The teaching server: a small MCP server written to a file and started as a child process. It advertises two tools: word_count, which Lucy's agent may call, and place_purchase, which it may not. It writes every call it receives to a log of its own, so you can check what happened.
Messages and frames: request_message, notification_message, encode_frame, FrameReader and parse_frame.
The handshake checks: check_initialize and check_tools.
Client, a bounded stdio client. It calls your answer_for for every reply and your authorize before every tool call.

Run setup on every fresh kernel. Your saved work lives in practical-work/ch14-a. Restarting a kernel clears variables, not saved files.

Supplied setup, teaching server and client

The complete supplied setup cell is in the downloadable notebook. Run it there before attempting the cells below. This web edition omits only that compressed setup payload.

Commit to a prediction before the examples

Lucy's agent sent three requests, with ids 1, 2 and 3. Four lines came back, in this order:

{"jsonrpc": "2.0", "method": "notifications/message", "params": {"data": "working"}}
{"jsonrpc": "2.0", "id": 2, "result": {"tools": []}}
{"jsonrpc": "2.0", "id": 1, "result": {"protocolVersion": "2025-06-18"}}
{"jsonrpc": "2.0", "id": 4, "result": {"content": [{"type": "text", "text": "4"}]}}

Before you run anything below, write down which line answers which request, and which lines answer none of them.

python

prediction_notes = {
    "prediction": "Write which line answers requests 1, 2 and 3, and which answer none.",
    "reason": "Name the rule behind that prediction.",
    "falsifier": "Name an observation that would prove the explanation wrong.",
    "revision": "After execution, explain what changed in your understanding.",
}

A request, and the reply that answers it

A request carries an id. Exactly one response with the same id answers it. A notification carries no id, and nothing answers it. Arrival order proves nothing: only the id binds a reply to its request. Line 4 answers a request that was never sent, so it answers none of ours. Request 3 is still waiting.

python

arrived = [
    {"jsonrpc": "2.0", "method": "notifications/message", "params": {"data": "working"}},
    {"jsonrpc": "2.0", "id": 2, "result": {"tools": []}},
    {"jsonrpc": "2.0", "id": 1, "result": {"protocolVersion": "2025-06-18"}},
    {"jsonrpc": "2.0", "id": 4, "result": {"content": [{"type": "text", "text": "4"}]}},
]
for request_id in (1, 2, 3):
    answers = [
        line for line, message in enumerate(arrived, start=1) if message.get("id") == request_id
    ]
    print(f"request {request_id} is answered by line {answers[0] if answers else 'none yet'}")
print(
    "no request:",
    [line for line, message in enumerate(arrived, start=1) if message.get("id") not in (1, 2, 3)],
)

One JSON line to a child process

MCP's stdio transport starts the server as a child process. The client writes each message to the child's standard input as one line of JSON and reads replies from its standard output. Diagnostics belong on standard error, which is never read as protocol. Here is the whole idea with a three-line child that answers one request:

python

ECHO = (
    "import json, sys\n"
    "message = json.loads(sys.stdin.readline())\n"
    "print('a diagnostic, on stderr', file=sys.stderr)\n"
    "reply = {'jsonrpc': '2.0', 'id': message['id'], 'result': {'echo': message['method']}}\n"
    "print(json.dumps(reply))\n"
)
exchange = subprocess.run(
    [sys.executable, "-c", ECHO],
    input=encode_frame(request_message(7, "tools/list", {}), limit=1024),
    capture_output=True,
    timeout=10,
)
print("stdout:", exchange.stdout)
print("stderr:", exchange.stderr)

A frame split across two reads

A pipe carries bytes, not messages. One read can return half a message, or one and a half. FrameReader keeps a buffer and returns only complete lines, keeping the rest for the next read. Predict what each feed returns.

python

frame = encode_frame(request_message(41, "tools/list", {}), limit=1024)
reader = FrameReader(limit=1024)
print(reader.feed(frame[:20]))
print(reader.feed(frame[20:] + b'{"jsonrpc":"2.0"'))
print("waiting in the buffer:", reader.buffer)

1. Construct `answer_for`

answer_for(request_id, message) decides what one arrived message means for the request we are waiting for. It must:

return None for a notification: a message with a method and no id;
raise ValueError for a reply whose id is not exactly the integer request_id. In a client that asks one thing at a time, another id means the stream is confused, and accepting it would attach one request's evidence to another. Note that True == 1 in Python, but True is not an id;
raise RuntimeError for a reply that carries an error;
raise ValueError when a matching reply has no result object;
otherwise return the result dictionary.

The starter returns whatever result arrives first. Run the visible cases to see where it fails, then repair it.

python

def answer_for(request_id, message):
    """The result `message` gives request `request_id`; None for a notification."""
    return message.get("result")

2. Construct `authorize`

authorize(name, catalog, allowed) decides whether the client may send a call. catalog is the dictionary of tools the server advertised; allowed is the set of names the operator configured. Return True only when the tool was advertised and is allowed; return False otherwise. Do not change either input.

The starter treats discovery as permission: if the server advertised it, it may be called.

python

def authorize(name, catalog, allowed):
    """May the application call `name`?"""
    return name in catalog

Hint 1: the order of the checks in answer_for

A notification has no id, so check for it first. Then compare ids: type(x) is int rejects True, "3" and 3.0, which isinstance(x, int) or == would let through.

Hint 2: why not skip a wrong id and keep reading

A client that sends one request at a time can never legitimately see another id. Skipping it would hide a confused stream, or a late reply to a call we gave up on, and the next read might then attach it to the wrong question.

Hint 3: whose decision authorize is

The server decides what it offers. The operator decides what Lucy's agent may use. A tool must pass both: a name the operator allowed but this server never advertised cannot be called here either.

python

import copy


def reply(identity, result=None, error=None):
    message = {"jsonrpc": "2.0", "id": identity}
    if error is not None:
        message["error"] = error
    elif result is not None:
        message["result"] = result
    return message


CATALOG = {"word_count": {"name": "word_count"}, "place_purchase": {"name": "place_purchase"}}
ANSWER_CASES = [
    ("its own reply", 3, reply(3, {"content": []}), {"content": []}),
    ("another id is refused", 3, reply(4, {"content": []}), "ValueError"),
    ("a notification is skipped", 3, {"jsonrpc": "2.0", "method": "notifications/message"}, None),
    ("an error is raised", 3, reply(3, error={"code": -32602, "message": "bad"}), "RuntimeError"),
    ("True is not the id 1", 1, reply(True, {"content": []}), "ValueError"),
    ("a reply without a result", 3, reply(3), "ValueError"),
]
AUTHORIZE_CASES = [
    ("advertised and allowed", ("word_count", CATALOG, {"word_count"}), True),
    ("advertised, not allowed", ("place_purchase", CATALOG, {"word_count"}), False),
    ("allowed, not advertised", ("stock_level", CATALOG, {"word_count", "stock_level"}), False),
    ("nothing allowed", ("word_count", CATALOG, set()), False),
]


def grade_unit(answer, allow):
    rows = []
    for label, request_id, message, expected in ANSWER_CASES:
        supplied = copy.deepcopy(message)
        try:
            observed = answer(request_id, supplied)
        except Exception as error:
            observed = type(error).__name__
        passed = observed == expected and supplied == message
        rows.append(
            {
                "case": label,
                "expected": expected,
                "observed": observed,
                "status": "PASS" if passed else "FAIL",
            }
        )
    for label, arguments, expected in AUTHORIZE_CASES:
        supplied = copy.deepcopy(arguments)
        try:
            observed = allow(*supplied)
        except Exception as error:
            observed = type(error).__name__
        passed = observed is expected and supplied == arguments
        rows.append(
            {
                "case": label,
                "expected": expected,
                "observed": observed,
                "status": "PASS" if passed else "FAIL",
            }
        )
    return rows


visible_results = grade_unit(answer_for, authorize)
VISIBLE_PASSED = all(r["status"] == "PASS" for r in visible_results)
for visible_row in visible_results:
    print(visible_row["status"], visible_row["case"], "->", visible_row["observed"])
print("VISIBLE_CONTRACT", "PASSED" if VISIBLE_PASSED else "NEEDS_WORK")

3. Connect Lucy's agent to the teaching server

Once the visible cases pass, the cell below uses your two functions with the supplied Client:

It starts the teaching server, completes the handshake and lists the advertised tools.
It calls word_count on "vanilla stock needs review".
It tries to call place_purchase, which the server advertises, and whose description tells the model to ALWAYS call it.
It reads the server's own log of the calls it received.
It starts a second server that claims an older protocol version, and checks that nothing is listed or called.

Before running it, predict the count, and what the server's log will contain.

python

connected = None
if VISIBLE_PASSED:
    log = COURSE_WORK / "calls.jsonl"
    log.unlink(missing_ok=True)
    with Client(server_command("normal", log), allowed={"word_count"}) as client:
        hello = client.initialize()
        catalog = client.list_tools()
        count = client.call_tool("word_count", {"text": "vanilla stock needs review"})
        try:
            client.call_tool("place_purchase", {"flavor": "vanilla", "tubs": 100})
            purchase = "called"
        except PermissionError:
            purchase = "refused"
    old_log = COURSE_WORK / "old-version-calls.jsonl"
    old_log.unlink(missing_ok=True)
    with Client(server_command("old-version", old_log), allowed={"word_count"}) as old:
        try:
            old.initialize()
            old_version = "accepted"
        except ValueError:
            old_version = "refused"
    connected = {
        "protocolVersion": hello["protocolVersion"],
        "advertised": sorted(catalog),
        "word_count": count,
        "purchase": purchase,
        "server_calls": [call["name"] for call in server_calls(log)],
        "old_version": old_version,
        "old_version_calls": len(server_calls(old_log)),
    }
    print(json.dumps(connected, indent=1))
    assert count == "4" and purchase == "refused" and connected["server_calls"] == ["word_count"]
    assert old_version == "refused" and connected["old_version_calls"] == 0
else:
    print("CONNECTION_NOT_READY — repair answer_for and authorize, then run again.")

The count is four, from a process you did not write. The purchase was refused inside the client, before a byte went to the server, and the server's own log proves it: the only call it received is word_count. The description that shouted ALWAYS changed nothing, because the client never asks the description for permission.

4. Save the handoff

Unit B breaks this connection on purpose. The handoff records the configuration you tested and what the server's log showed.

python

ARTIFACT_PATH = Path("ch14-unit-a-handoff-v1.json")
artifact_status = "NOT_WRITTEN"
if connected is not None:
    handoff = {
        "unit": "ch14-a",
        "status": "COMPLETED",
        "protocolVersion": connected["protocolVersion"],
        "advertised": connected["advertised"],
        "allowed": ["word_count"],
        "observations": connected,
    }
    ARTIFACT_PATH.write_text(json.dumps(handoff, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    artifact_status = "WRITTEN"
    print(ARTIFACT_PATH)
else:
    print("HANDOFF_NOT_WRITTEN")

Exit ticket

Answer three questions:

Why must a reply labeled 4 be refused while you wait for request 3, even if its content looks right?
Which two parties decide whether a tool may be called, and which one decides for Lucy's agent?
Which observation proves the purchase never reached the server, and why is it stronger than the client's own report?

python

exercise_report = {
    "unit": "ch14-a",
    "attempted": 1,
    "completed": int(VISIBLE_PASSED),
    "failed": int(not VISIBLE_PASSED),
    "skipped": 0,
    "connection": "PASSED" if connected else "NOT_READY",
    "handoff": artifact_status,
}
print("EXERCISE_REPORT=" + json.dumps(exercise_report, sort_keys=True))

Changed-constraint construction: offer the model only allowed tools

Allow fifteen minutes: three to predict, eight to implement and trace, and four for a case of your own.

authorize stops a forbidden call. It is better still if the model never hears of the tool: every tool offered costs prompt tokens, and a tool the model cannot use is a distraction. Chapter 14 measured both costs.

transfer_offer(catalog, allowed) receives the advertised catalog and the allowed names. Return, sorted by name, one function-calling description for each tool that is both advertised and allowed:

python

{"type": "function", "function": {"name": ..., "description": ..., "parameters": <its inputSchema>}}

Use an empty string for a missing description. Do not change the inputs.

Write your expected values before you run the table.

python

def transfer_offer(catalog, allowed):
    raise NotImplementedError(
        "Keep advertised AND allowed tools, sorted by name, in function-calling shape"
    )

python

SCHEMA = {"type": "object", "properties": {"text": {"type": "string"}}, "required": ["text"]}
TOOLS = {
    "word_count": {"name": "word_count", "description": "Count words.", "inputSchema": SCHEMA},
    "place_purchase": {
        "name": "place_purchase",
        "description": "Buy.",
        "inputSchema": {"type": "object"},
    },
    "char_count": {"name": "char_count", "inputSchema": SCHEMA},
}


def offer(name, description):
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": TOOLS[name]["inputSchema"],
        },
    }


TRANSFER_CASES = [
    ("only the allowed tool", [TOOLS, {"word_count"}], [offer("word_count", "Count words.")]),
    (
        "an allowed name the server lacks",
        [TOOLS, {"word_count", "stock_level"}],
        [offer("word_count", "Count words.")],
    ),
    ("nothing allowed", [TOOLS, set()], []),
    (
        "sorted by name",
        [TOOLS, {"word_count", "char_count"}],
        [offer("char_count", ""), offer("word_count", "Count words.")],
    ),
    ("a missing description is empty", [TOOLS, {"char_count"}], [offer("char_count", "")]),
]


def run_transfer(candidate, cases):
    observations = []
    for label, arguments, expected in cases:
        supplied = copy.deepcopy(arguments)
        try:
            actual = candidate(*supplied)
        except NotImplementedError:
            actual = {"unfinished": True}
        except Exception as error:
            actual = {"raises": type(error).__name__}
        passed = actual == expected and supplied == arguments
        observations.append(
            {"case": label, "expected": expected, "observed": actual, "passed": passed}
        )
        print("PASS" if passed else "NEEDS_WORK", label)
    return observations


transfer_observations = run_transfer(transfer_offer, TRANSFER_CASES)
TRANSFER_PASSED = all(r["passed"] for r in transfer_observations)
print("TRANSFER_STATUS", "PASS" if TRANSFER_PASSED else "NEEDS_WORK")

Design a counterexample and retrieve the mechanism

Add one case with an expected outcome you worked out independently, and rerun the driver. Then, in a temporary copy, offer every advertised tool instead, and name the tool the model could now choose that the client would refuse.

Save your evidence and explain the result

Fill in the prediction notes and your explanation before saving. Include:

the exact observed value, and the input that caused it;
your code's invocation point;
one failed hypothesis;
the strongest claim the evidence still cannot support.

The teaching server is scripted and local. A successful handshake proves the protocol works, not that the server is safe to run: starting its executable gave it the same access as any program you start. Chapter 15 adds the operating-system boundary MCP does not provide.

python

explanation_notes = {
    "causal_trace": "Explain the input, learner invocation and observed result.",
    "failed_hypothesis": "Describe a prediction the evidence changed.",
    "remaining_limit": "Name the guarantee not established by this experiment.",
}
course_submission = {
    "unit": "ch14-a",
    "planned_minutes": 90,
    "starting_evidence": globals().get("HANDOFF_ORIGIN", "INDEPENDENT_UNIT_A"),
    "prediction": prediction_notes,
    "explanation": explanation_notes,
    "core_report": exercise_report,
    "transfer": transfer_observations,
    "explanation_review": "HUMAN_REVIEW_REQUIRED",
}
submission_path = COURSE_WORK / "ch14-a-submission-v1.json"
submission_path.write_text(
    json.dumps(course_submission, indent=2, sort_keys=True), encoding="utf-8"
)
print("Saved evidence:", submission_path)
print(
    "COURSE_REPORT="
    + json.dumps(
        {
            "unit": "ch14-a",
            "transfer_passed": TRANSFER_PASSED,
            "starting_evidence": course_submission["starting_evidence"],
            "edition": "student",
        },
        sort_keys=True,
    )
)

Keep building with Prof Rod

Found this material through a colleague, classroom or shared download? Get the complete book at profrod.ai/book and join the Prof Rod learner community. Bring one result, one question or one failure you learned from. Share this resource with another learner and keep its source links with it so they can find the full course and future updates.