External Tools with MCP: Unit B 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 second of Chapter 14's two practical units. Unit A connected Lucy's agent to a tool in another process and kept the decision to call it inside the application. This unit breaks that connection on purpose, and repairs what each failure exposes.

You bring Unit A's handoff, or the supplied reference, and its answer_for and authorize, which are supplied here in their worked form.

By the end you should be able to:

Reproduce a client that reports a stale reply as a new answer, and prove it from the server's own log.
Implement split_frames, which returns only complete messages and refuses any line beyond the frame limit before it is parsed.
Show a hung server ending at its deadline, and closing the client ending the server's whole process group.
Validate a new tool's arguments against its schema before any call is sent, and state what the client supports and what it does not.

Minutes	Dedicated work	Saved evidence
0–10	Predict wrong-id, stale, oversized and hung outcomes	Failure table
10–25	Reproduce a client that accepts the wrong reply	Failing independent observation
25–45	Construct `split_frames`; pass the visible cases	Learner code and grade table
45–60	Deadlines and cleanup with a hanging and a lingering server	Deadline and process-exit evidence
60–75	Changed-constraint task: validate a new tool's arguments; noisy stderr	Transfer results
75–85	Remove the allowlist and prove the integrated check fails	Negative-control result
85–90	State the supported features and the containment limits	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 from Unit A, with all its failure modes: wrong-id, stale, oversized, stdout-log, stderr-log, hang, notify, old-version and lingering. It logs every call it receives.
Unit A's worked answer_for and authorize, and the message helpers.
Client, which now reads with your split_frames, and accepts an answer function so you can try a careless one.

Run setup on every fresh kernel. Your saved work lives in practical-work/ch14-b.

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 failures

The teaching server can misbehave in four ways. For each one, write down what a client that simply takes the first result it receives would report, and what a careful client should do:

wrong-id: the reply is labeled with the next request's id;
stale: a late copy of the previous call's reply arrives first;
oversized: the reply is 200,000 bytes long;
hang: the server never replies.

python

prediction_notes = {
    "prediction": "For each failure: what a careless client reports, and what a careful one does.",
    "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.",
}

Start from Unit A's handoff

This unit uses the configuration you tested in Unit A. To use your own, replace None with the path to your practical-work/ch14-a/ch14-unit-a-handoff-v1.json. Leave it as None to start from the supplied reference, which is the handoff the worked Unit A produces. Your submission records which you chose.

python

LEARNER_HANDOFF = None

python

import shutil

COURSE_INPUT = COURSE_WORK / "ch14-unit-a-handoff-v1.json"
if LEARNER_HANDOFF is not None:
    learner_input = Path(LEARNER_HANDOFF).expanduser().resolve()
    if not learner_input.is_file():
        raise FileNotFoundError("The selected learner handoff does not exist")
    if learner_input != COURSE_INPUT.resolve():
        shutil.copy2(learner_input, COURSE_INPUT)
    HANDOFF_ORIGIN = "LEARNER_SELECTED"
else:
    reference_encoded = (
        "c-n=KO$&lR7=-Ws3fZ%QiA8iRZ$%xVi^Q_68w!gn`=JN%zxS@12^wB!-g#zr8^?7KNh>CGu2_l=ZUPEuB`1{O(ny"
        "*S#gNylwU8;M!W0^T@btPg=c1lQmyX0sF<B9vi$cL=(_bzW|Er#H2eM`J6az;A8!$++x3R>ziz#fi^}*nkQfNU1f"
        "Be#65N`co=Wq63z&K>iwr^r`#3pM?Cpy3)Yqs6@tPep-6K!k@I$kfgb9|3ay<v*H!_jOL`2;zw^8w)jg5v"
    )
    COURSE_INPUT.write_bytes(zlib.decompress(base64.b85decode(reference_encoded)))
    HANDOFF_ORIGIN = "SUPPLIED_REFERENCE"
print("Starting evidence:", HANDOFF_ORIGIN)

python

handoff = json.loads(COURSE_INPUT.read_text(encoding="utf-8"))
handoff_status = (
    "VERIFIED"
    if handoff.get("status") == "COMPLETED" and handoff.get("unit") == "ch14-a"
    else "INVALID"
)
ALLOWED = set(handoff["allowed"])
EXPECTED_COUNT = handoff["observations"]["word_count"]
print(
    "UNIT_A_HANDOFF",
    handoff_status,
    "| allowed",
    sorted(ALLOWED),
    "| word_count on the test text",
    EXPECTED_COUNT,
)

1. Reproduce a client that accepts the wrong reply

The careless client below takes the first result it receives, whatever its id, and it has no frame limit. Run it against the stale server with two questions. The second question has three words.

python

def careless_answer(request_id, message):
    return message.get("result")


split_frames = split_frames_reference
stale_log = fresh_log("stale")
with Client(
    server_command("stale", stale_log), allowed=ALLOWED, answer=careless_answer
) as careless:
    careless.initialize()
    careless.list_tools()
    first = careless.call_tool("word_count", {"text": "vanilla stock needs review"})
    second = careless.call_tool("word_count", {"text": "one two three"})
print("the client reported:", first, second)
print("the server received:", [call["arguments"]["text"] for call in server_calls(stale_log)])
del split_frames

The client reported 4 for "one two three". The server's log shows it received the right question. The answer was the late copy of the first call's reply, taken because it arrived first. Nothing crashed, and that is the danger: the wrong count looks exactly like a right one.

Unit A's answer_for refuses it, because the late reply carries the first call's id. An exception is the right outcome here. A wrong number that looks right is not.

2. Construct `split_frames`

The careless reader also has no frame limit. It holds whatever arrives until a newline appears, so a server that never sends one can fill memory, and a huge reply is parsed whole before anything checks it.

split_frames(buffer, limit) receives the bytes read so far and the frame limit. It returns a pair (frames, rest): the complete lines in order, without their newlines, and the bytes after the last newline, which wait for the next read. It must raise ValueError:

when a complete line plus its newline is longer than limit;
when the rest still waiting for its newline has already reached limit bytes, so the limit holds before the line ends.

The starter splits correctly and checks nothing.

python

def split_frames(buffer, limit):
    """Complete lines and the waiting remainder; refuse anything beyond `limit`."""
    *frames, rest = buffer.split(b"\n")
    return frames, rest

Hint 1: which lengths to compare

A complete frame occupies its bytes and its newline, so it fits when len(frame) + 1 <= limit. A remainder already at limit bytes cannot become a frame that fits, whatever arrives next.

Hint 2: why check the remainder at all

Without it, a server that sends bytes and never a newline is never refused: every read makes the buffer longer and no complete frame ever appears to check.

python

import copy

FRAME_CASES = [
    ("one frame", (b'{"a":1}\n', 64), ([b'{"a":1}'], b"")),
    ("half a frame waits", (b'{"a"', 64), ([], b'{"a"')),
    ("two frames and a half", (b'{"a":1}\n{"b":2}\n{"c"', 64), ([b'{"a":1}', b'{"b":2}'], b'{"c"')),
    ("a frame exactly at the limit", (b"x" * 15 + b"\n", 16), ([b"x" * 15], b"")),
    ("a complete frame beyond the limit", (b"x" * 16 + b"\n", 16), "ValueError"),
    ("an endless line is refused before it ends", (b"x" * 16, 16), "ValueError"),
]


def grade_frames(candidate, cases):
    rows = []
    for label, arguments, expected in cases:
        supplied = copy.deepcopy(arguments)
        try:
            observed = candidate(*supplied)
            observed = (list(observed[0]), observed[1])
        except Exception as error:
            observed = type(error).__name__
        passed = observed == expected
        rows.append(
            {
                "case": label,
                "expected": expected,
                "observed": observed,
                "status": "PASS" if passed else "FAIL",
            }
        )
    return rows


visible_results = grade_frames(split_frames, FRAME_CASES)
VISIBLE_PASSED = all(r["status"] == "PASS" for r in visible_results)
for visible_row in visible_results:
    print(visible_row["status"], visible_row["case"], "->", str(visible_row["observed"])[:60])
print("VISIBLE_CONTRACT", "PASSED" if VISIBLE_PASSED else "NEEDS_WORK")

3. Run every failure against the repaired client

With your split_frames and Unit A's answer_for, run all the failure modes with a one-second deadline. Before running, predict which end in an exception, and which still return the right count.

python

connected = None
if VISIBLE_PASSED:
    outcomes = {}
    for mode in (
        "normal",
        "wrong-id",
        "stale",
        "oversized",
        "stdout-log",
        "stderr-log",
        "notify",
        "hang",
    ):
        log = fresh_log(mode)
        started = time.monotonic()
        with Client(server_command(mode, log), allowed=ALLOWED, timeout=1.0) as client:
            try:
                client.initialize()
                client.list_tools()
                texts = [client.call_tool("word_count", {"text": "vanilla stock needs review"})]
                texts.append(client.call_tool("word_count", {"text": "one two three"}))
                outcome = "answered " + " and ".join(texts)
            except Exception as error:
                outcome = type(error).__name__
        outcomes[mode] = {
            "outcome": outcome,
            "seconds": round(time.monotonic() - started, 1),
            "server_calls": len(server_calls(log)),
        }
        seen = outcomes[mode]
        print(
            f"{mode:11} {outcome:18} after {seen['seconds']}s; server calls: {seen['server_calls']}"
        )
    assert outcomes["normal"]["outcome"] == f"answered {EXPECTED_COUNT} and 3"
    for mode in ("wrong-id", "stale", "oversized", "stdout-log"):
        assert outcomes[mode]["outcome"] == "ValueError", mode
    assert outcomes["hang"]["outcome"] == "TimeoutError" and outcomes["hang"]["seconds"] < 4
    assert (
        outcomes["stderr-log"]["outcome"]
        == outcomes["notify"]["outcome"]
        == outcomes["normal"]["outcome"]
    )
    connected = outcomes
else:
    print("CONNECTION_NOT_READY — repair split_frames, then run again.")

Every failure ends in an exception, and none of them produces a wrong answer. The hung call ended at its one-second deadline, not when the server chose to answer; the seconds shown also include closing each server, which the next section explains. Diagnostics on stderr and notifications before the answer changed nothing.

A deadline, and an ending

A deadline stops the waiting, but the server is still running. close() shuts it down in the order the stdio transport describes. It closes the server's input and waits a moment, then ends the whole process group with SIGTERM, and finally with SIGKILL. The lingering server starts a grandchild process that would outlive a client that only waits for its own child.

python

lingering_log = fresh_log("lingering")
client = Client(server_command("lingering", lingering_log), allowed=ALLOWED)
client.initialize()
grandchild = int(Path(str(lingering_log) + ".grandchild").read_text())
print("server exit status:", client.close())
time.sleep(0.2)
try:
    os.kill(grandchild, 0)
    grandchild_state = "still running"
except OSError:
    grandchild_state = "gone"
print("the server's grandchild is", grandchild_state)

hung_log = fresh_log("hang-close")
hung = Client(server_command("hang", hung_log), allowed=ALLOWED, timeout=1.0)
hung.initialize()
hung.list_tools()
try:
    hung.call_tool("word_count", {"text": "one"})
except TimeoutError:
    pass
print("hung server exit status:", hung.close())

The healthy server exited with status 0 once its input closed. The grandchild went with its process group. The hung server ignored its closed input and exited with status -15: it was ended by SIGTERM.

4. Save the core report

python

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

Changed-constraint construction: check a new tool's arguments

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

Lucy's operator adds a second server with a new tool, record_delivery, whose schema asks for an order_id string and a whole number of tubs, and nothing else. A model proposes the arguments. Before any call is sent, the client should check them against the schema the server advertised, and refuse arguments that could not be meant.

transfer_problems(schema, arguments) returns a sorted list of problems, and [] when the arguments are valid:

"missing <key>" for each key in required that is absent;
"unexpected <key>" for each key not in properties, when additionalProperties is False;
"<key> must be <type>" for each present key whose value has the wrong type. "string" is str; "integer" is int but not bool; "number" is int or float but not bool; "boolean" is bool.

Do not change the inputs. Write your expected values before you run the table.

python

def transfer_problems(schema, arguments):
    raise NotImplementedError("List missing, unexpected and wrongly typed arguments")

python

DELIVERY = {
    "type": "object",
    "properties": {"order_id": {"type": "string"}, "tubs": {"type": "integer"}},
    "required": ["order_id", "tubs"],
    "additionalProperties": False,
}
TRANSFER_CASES = [
    ("valid arguments", [DELIVERY, {"order_id": "A-1043", "tubs": 20}], []),
    ("a missing key", [DELIVERY, {"order_id": "A-1043"}], ["missing tubs"]),
    (
        "an unexpected key",
        [DELIVERY, {"order_id": "A-1043", "tubs": 20, "price": 900}],
        ["unexpected price"],
    ),
    (
        "tubs given as text",
        [DELIVERY, {"order_id": "A-1043", "tubs": "20"}],
        ["tubs must be integer"],
    ),
    (
        "True is not a number of tubs",
        [DELIVERY, {"order_id": "A-1043", "tubs": True}],
        ["tubs must be integer"],
    ),
    (
        "several problems, sorted",
        [DELIVERY, {"tubs": 2.5, "note": "x"}],
        ["missing order_id", "tubs must be integer", "unexpected note"],
    ),
]


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, "->", actual)
    return observations


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

Now use your checker on the teaching server's own schema, over a server that writes diagnostics to stderr. A model proposes a number where the schema wants text; the client refuses it without a call. Then the valid call goes through.

python

if TRANSFER_PASSED and VISIBLE_PASSED:
    noisy_log = fresh_log("stderr-transfer")
    with Client(server_command("stderr-log", noisy_log), allowed=ALLOWED) as client:
        client.initialize()
        schema = client.list_tools()["word_count"]["inputSchema"]
        for proposed in ({"text": 5}, {"text": "vanilla stock needs review"}):
            problems = transfer_problems(schema, proposed)
            print(proposed, "->", problems or client.call_tool("word_count", proposed))
    print("the server received", len(server_calls(noisy_log)), "call(s)")

The negative control: remove the allowlist

A check that nothing can fail proves nothing. Replace authorize with one that allows anything the server advertises, and call the purchase. The server's own log is the independent witness.

python

worked_authorize = authorize


def authorize(name, catalog, allowed):
    return name in catalog


control_log = fresh_log("negative-control")
with Client(server_command("normal", control_log), allowed=ALLOWED) as client:
    client.initialize()
    client.list_tools()
    client.call_tool("place_purchase", {"flavor": "vanilla", "tubs": 100})
authorize = worked_authorize
purchases = [call for call in server_calls(control_log) if call["name"] == "place_purchase"]
NEGATIVE_CONTROL = "FAILED_AS_EXPECTED" if purchases else "NOT_DETECTED"
print("purchases the server received:", purchases)
print("NEGATIVE_CONTROL", NEGATIVE_CONTROL)

Without the allowlist, the purchase the description asked for reached the server, and its log shows it. With the allowlist back, Unit A's integration check fails for exactly this reason.

What this client supports, and what it does not

It supports:

the stdio transport of MCP 2025-06-18;
one request at a time, with integer ids;
bounded, unpaginated tool lists;
text tool results;
skipped notifications;
byte and time bounds;
ending the server's process group.

It does not support:

the Streamable HTTP transport or authorization;
prompts, resources, sampling or subscriptions;
concurrent requests, cancellation or progress;
paginated lists;
tool lists that change while connected.

A successful handshake is not containment. Starting the server's executable gave it the same access as any program you run. Chapter 15 adds the operating-system boundary the protocol does not provide.

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.

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-b",
    "planned_minutes": 90,
    "starting_evidence": HANDOFF_ORIGIN,
    "prediction": prediction_notes,
    "explanation": explanation_notes,
    "core_report": exercise_report,
    "failures": connected,
    "negative_control": NEGATIVE_CONTROL,
    "transfer": transfer_observations,
    "explanation_review": "HUMAN_REVIEW_REQUIRED",
}
submission_path = COURSE_WORK / "ch14-b-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-b",
            "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.