Mini-project 4: build an MCP client harness

Reading a protocol specification tells you what the messages look like. Connecting to a real server tells you what it is actually like to work with one.

In this chapter you build both ends. First a small MCP server exposing the same inventory tools from Chapter 3, so you have something you fully control to test against. Then a client harness — a reusable class that connects to any MCP server, discovers its tools, translates them into the shape your model API wants, calls them, and handles every category of failure.

Then you point that same harness at a real third-party server and watch it work unchanged. That moment is the whole point of a protocol.

Setup

mkdir -p mcp-project && cd mcp-project
pip install "mcp[cli]"

A note on versions before you write a line. The stable Python SDK is mcp 1.x — the code here was verified against 1.27.0 — and it implements the session-based protocol revisions. The 2.0.0b1 beta implements the 2026-07-28 stateless revision and renames FastMCP to MCPServer. Check what you have:

python -c "import importlib.metadata as m; print(m.version('mcp'))"

The client concepts below are identical across both. Where the API differs, it is called out.

Part 1: the server

Create inventory_server.py.

from __future__ import annotations
from typing import Literal

from mcp.server.fastmcp import FastMCP
from mcp.types import ToolAnnotations

mcp = FastMCP("inventory")

INVENTORY = {
    "SKU-1001": {"name": "Astro Zoom Trainers", "warehouse": "AMS", "on_hand": 42},
    "SKU-1002": {"name": "Cirrus Rain Shell",   "warehouse": "AMS", "on_hand": 0},
    "SKU-2001": {"name": "Tidal Swim Goggles",  "warehouse": "SIN", "on_hand": 7},
}


@mcp.tool(
    title="Find product SKUs by name",
    annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True),
)
def find_sku(query: str, limit: int = 3) -> list[dict]:
    """Find product SKUs whose name matches a shopper's words.

    Args:
        query: Words from the product name, e.g. "rain shell".
        limit: Maximum number of matches to return.
    """
    q = query.lower()
    return [{"sku": k, "name": v["name"]}
            for k, v in INVENTORY.items() if q in v["name"].lower()][:limit]


@mcp.tool(
    title="Check shippable stock for one SKU",
    annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True),
)
def check_stock(sku: str, warehouse: Literal["AMS", "SIN", "any"] = "any") -> dict:
    """Report how many units of one SKU are available to ship today.

    Args:
        sku: The exact SKU identifier, e.g. "SKU-1001". Get it from find_sku.
        warehouse: Which warehouse to check. "any" checks all of them.
    """
    row = INVENTORY.get(sku)
    if row is None:
        raise ValueError(
            f"No product with SKU {sku!r}. Ask the shopper for the product name, "
            "call find_sku to get a valid SKU, then try again."
        )
    if warehouse != "any" and row["warehouse"] != warehouse:
        return {"sku": sku, "on_hand": 0, "warehouse": warehouse, "shippable": False}
    return {"sku": sku, "on_hand": row["on_hand"], "warehouse": row["warehouse"],
            "shippable": row["on_hand"] > 0}


@mcp.resource("inventory://warehouses")
def warehouses() -> str:
    """The list of warehouse codes this server knows about."""
    return "AMS (Amsterdam), SIN (Singapore)"


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

This should look extremely familiar. The SDK does exactly what you built by hand in Chapter 3 — it reads the signature, reads the docstring, and generates the inputSchema. Literal[...] becomes an enum here too. Having written that machinery yourself, you now know precisely what this decorator is and is not doing for you.

Three things specific to MCP.

annotations=ToolAnnotations(...) sets the behavior hints. Read-only, non-destructive, idempotent — all true here. Remember from Chapter 4 that these are hints a client may choose to trust, not enforcement.

raise ValueError(...) is how you produce a tool error. The SDK catches it and returns a result with isError: true and your message as text. That message is a prompt, so it tells the model what to do next.

@mcp.resource("inventory://warehouses") exposes a resource under a URI. Small, but it lets you exercise resources/list in the client.

The critical operational rule: on stdio, stdout is the protocol stream. A stray print() in a tool corrupts the connection and produces a baffling parse error. Log to stderr.

Part 2: the harness

Create mcp_harness.py.

Connecting

from __future__ import annotations
import asyncio, json, sys
from contextlib import AsyncExitStack

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import CallToolResult, TextContent


class McpHarness:
    def __init__(self, command: str, args: list[str], name: str = "server"):
        self.params = StdioServerParameters(command=command, args=args)
        self.name = name
        self._stack = AsyncExitStack()
        self.session: ClientSession | None = None
        self.tools: dict = {}

    async def __aenter__(self):
        read, write = await self._stack.enter_async_context(stdio_client(self.params))
        self.session = await self._stack.enter_async_context(ClientSession(read, write))
        info = await self.session.initialize()
        print(f"[connected] {info.serverInfo.name} v{info.serverInfo.version} "
              f"protocol={info.protocolVersion}")
        await self.refresh_tools()
        return self

    async def __aexit__(self, *exc):
        await self._stack.aclose()

The AsyncExitStack is the detail worth pausing on. Both stdio_client and ClientSession are async context managers, and nesting them with async with inside __aenter__ does not work — the block would exit immediately and tear down the connection. AsyncExitStack lets you enter them and hold them open until __aexit__, which is what makes the harness usable as one clean async with.

initialize() performs the handshake and returns the server’s identity, protocol version, and capabilities. Under the 2026-07-28 revision this call is gone — requests are self-contained and capability discovery is the optional server/discover. On the 1.x SDK you still call it. It is a good place to log what you connected to, which you will want the first time a server behaves unexpectedly.

Discovering and translating tools

    async def refresh_tools(self):
        result = await self.session.list_tools()
        self.tools = {t.name: t for t in result.tools}
        return self.tools

    def as_model_specs(self) -> list[dict]:
        """Translate MCP tool definitions into the shape a model API wants."""
        return [
            {
                "name": f"{self.name}__{t.name}",
                "description": (t.description or "").strip(),
                "input_schema": t.inputSchema,
            }
            for t in self.tools.values()
        ]

as_model_specs is the bridge, and it is smaller than people expect. An MCP Tool already carries a name, a description, and a JSON Schema. An Anthropic tool declaration wants a name, a description, and a JSON Schema. The translation is a rename of one field.

The {server}__{tool} prefix is not cosmetic. Connect two servers and both will have a search tool, and you need to know which one the model meant. Namespace at the boundary.

The refresh_tools method exists because tool lists can change at runtime. Under the current spec you can also honor the ttlMs and cacheScope hints returned by tools/list and cache accordingly instead of re-fetching.

Reading results

    @staticmethod
    def flatten(result: CallToolResult) -> str:
        if result.structuredContent is not None:
            return json.dumps(result.structuredContent, ensure_ascii=False)
        parts = []
        for block in result.content:
            if isinstance(block, TextContent):
                parts.append(block.text)
            else:
                parts.append(f"[{block.type} content omitted]")
        return "\n".join(parts)

Structured content wins when present — it is validated against the tool’s output schema and is unambiguous. Otherwise walk the content blocks.

Do not silently drop non-text blocks. Images, audio, and embedded resources are real, and a placeholder in the transcript is far better for debugging than a mysteriously empty result. Chapter 4’s warning applies here: resource links point at content you would then fetch, and fetching from an untrusted server is how arbitrary text gets into your model’s context.

Calling, with the full failure ladder

    async def call(self, name: str, args: dict, timeout_s: float = 15.0) -> dict:
        short = name.split("__", 1)[-1]
        if short not in self.tools:
            return {"ok": False, "error": f"Unknown tool {short!r}. "
                                          f"Available: {', '.join(sorted(self.tools))}."}
        try:
            result = await asyncio.wait_for(
                self.session.call_tool(short, args), timeout=timeout_s)
        except asyncio.TimeoutError:
            return {"ok": False, "error": f"{short} timed out after {timeout_s:g}s."}
        except Exception as e:                      # JSON-RPC level failure
            return {"ok": False,
                    "error": f"protocol error calling {short}: {type(e).__name__}: {e}"}
        text = self.flatten(result)
        if result.isError:
            return {"ok": False, "error": text}     # tool-level failure
        return {"ok": True, "value": text}

Four distinct failure categories, deliberately separated.

Unknown tool, caught locally before any network traffic. Cheaper than a round trip, and the message lists the real names so the model can recover.

Timeout. The protocol does not impose one. A hung server will hang your agent forever unless you wrap the call.

Protocol error — a JSON-RPC error response, a transport failure, a crashed server. This is usually a client-side problem to handle, not something to hand to the model verbatim.

Tool error — isError: true, meaning the call arrived and the tool declined. This one does go to the model, because it is the tool talking, and per Chapter 1 it should contain a recovery instruction.

Note the return shape: {"ok": bool, "value" | "error": str}. It is the same uniform contract your Chapter 3 dispatcher produced. That is what lets you drop MCP tools into an existing agent loop without touching the loop.

The main function

async def main():
    async with McpHarness(sys.executable, ["inventory_server.py"], name="inv") as h:
        print("\n== tools ==")
        for spec in h.as_model_specs():
            print(f"- {spec['name']}  required={spec['input_schema'].get('required', [])}")
            print(f"    {spec['description'].splitlines()[0]}")

        print("\n== resources ==")
        for r in (await h.session.list_resources()).resources:
            print(f"- {r.uri}  ({r.name})")

        print("\n== calls ==")
        for name, args in [
            ("inv__find_sku",    {"query": "rain shell"}),
            ("inv__check_stock", {"sku": "SKU-1002"}),
            ("inv__check_stock", {"sku": "SKU-2001", "warehouse": "SIN"}),
            ("inv__check_stock", {"sku": "SKU-9999"}),        # tool-level error
            ("inv__check_stock", {"sku": 1001}),              # schema violation
            ("inv__chekc_stock", {"sku": "SKU-1001"}),        # unknown tool
        ]:
            out = await h.call(name, args)
            body = out.get("value") or out["error"]
            print(f"[{'ok ' if out['ok'] else 'ERR'}] {name} {json.dumps(args)}\n      {body[:150]}")


if __name__ == "__main__":
    asyncio.run(main())

sys.executable rather than "python" — it launches the server with the same interpreter, so it sees the same installed packages. This saves a surprising amount of confusion.

Part 3: run it

python mcp_harness.py

Real output:

[connected] inventory v1.27.0 protocol=2025-11-25

== tools ==
- inv__find_sku  required=['query']
    Find product SKUs whose name matches a shopper's words.
- inv__check_stock  required=['sku']
    Report how many units of one SKU are available to ship today.

== resources ==
- inventory://warehouses  (warehouses)

== calls ==
[ok ] inv__find_sku {"query": "rain shell"}
      {"result": [{"sku": "SKU-1002", "name": "Cirrus Rain Shell"}]}
[ok ] inv__check_stock {"sku": "SKU-1002"}
      {"sku": "SKU-1002", "on_hand": 0, "warehouse": "AMS", "shippable": false}
[ok ] inv__check_stock {"sku": "SKU-2001", "warehouse": "SIN"}
      {"sku": "SKU-2001", "on_hand": 7, "warehouse": "SIN", "shippable": true}
[ERR] inv__check_stock {"sku": "SKU-9999"}
      Error executing tool check_stock: No product with SKU 'SKU-9999'. Ask the shopper for the
      product name, call find_sku to get a valid SKU, then try aga
[ERR] inv__check_stock {"sku": 1001}
      Error executing tool check_stock: 1 validation error for check_stockArguments
      sku
        Input should be a valid string [type=string_type, input_value=1001
[ERR] inv__chekc_stock {"sku": "SKU-1001"}
      Unknown tool 'chekc_stock'. Available: check_stock, find_sku.

Two details worth noticing in that output.

find_sku returns {"result": [...]} while check_stock returns its dict directly. The SDK wraps non-object return values — a list, an int, a string — in a result key, because structured content must be a JSON object. If you want a predictable top-level shape, return a dict from your tools.

The type-error message comes from Pydantic, and it is verbose and rather internal. It is technically actionable but it is not the sentence you would have written. For anything user-facing, validate explicitly and raise a ValueError with your own wording.

Part 4: point it at a real server

Now the payoff. Your harness was written against your own server, but it was written against the protocol. Point it somewhere else without changing a line.

The reference filesystem server ships on npm and needs no setup beyond having Node installed.

import asyncio, json
from mcp_harness import McpHarness

async def main():
    async with McpHarness(
        "npx", ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/sc"], name="fs"
    ) as h:
        print("\n== tools ==")
        for s in h.as_model_specs():
            print(f"- {s['name']}  required={s['input_schema'].get('required', [])}")
        print("\n== calls ==")
        for name, args in [
            ("fs__list_directory",  {"path": "/tmp/sc"}),
            ("fs__read_text_file",  {"path": "/tmp/sc/inventory_server.py", "head": 3}),
            ("fs__read_text_file",  {"path": "/etc/passwd"}),   # outside the sandbox
        ]:
            out = await h.call(name, args, timeout_s=30)
            body = out.get("value") or out["error"]
            print(f"[{'ok ' if out['ok'] else 'ERR'}] {name} {json.dumps(args)}\n      {body[:200]}")

asyncio.run(main())

Real output:

Secure MCP Filesystem Server running on stdio
[connected] secure-filesystem-server v0.2.0 protocol=2025-11-25
Client does not support MCP Roots, using allowed directories set from server args: [ '/tmp/sc' ]

== tools ==
- fs__read_file  required=['path']
- fs__read_text_file  required=['path']
- fs__read_media_file  required=['path']
- fs__read_multiple_files  required=['paths']
- fs__write_file  required=['path', 'content']
- fs__edit_file  required=['path', 'edits']
- fs__create_directory  required=['path']
- fs__list_directory  required=['path']
- fs__list_directory_with_sizes  required=['path']
- fs__directory_tree  required=['path']
- fs__move_file  required=['source', 'destination']
- fs__search_files  required=['path', 'pattern']
- fs__get_file_info  required=['path']
- fs__list_allowed_directories  required=[]

== calls ==
[ok ] fs__list_directory {"path": "/tmp/sc"}
      {"content": "[DIR] __pycache__\n[FILE] demo.py\n[FILE] inventory_server.py\n[FILE] mcp_harness.py\n[FILE] real.py\n[FILE] toolkit.py"}
[ok ] fs__read_text_file {"path": "/tmp/sc/inventory_server.py", "head": 3}
      {"content": "\"\"\"A minimal MCP server over stdio. Run: python inventory_server.py\"\"\"\nfrom __future__ import annotations\n"}
[ERR] fs__read_text_file {"path": "/etc/passwd"}
      Access denied - path outside allowed directories: /etc/passwd not in /tmp/sc

A Python client, a TypeScript server, launched by npx, and the harness did not change. That is \(N + M\) instead of \(N \times M\), working in front of you.

Three things in that output are worth reading carefully.

Fourteen tools from one server. That is roughly 2,000 tokens of schema in your context on every single request. Connect four servers like this and tool definitions dominate your prompt before the user says a word. This is the context-bloat problem from Chapter 4, and here it is measured rather than asserted.

Six of those fourteen tools mutate the filesystem. write_file, edit_file, move_file, create_directory. Your agent just acquired the ability to overwrite files. Whether that is fine depends entirely on what else the agent can be persuaded to do — which is exactly why hosts allowlist tool names rather than accepting whatever a server advertises.

The sandbox held. /etc/passwd was refused with a clear message. Note that the enforcement came from a directory argument the server was launched with, not from anything the protocol guaranteed — and note the log line saying the client does not support Roots, which is the deprecated capability that once nominally handled this. Sandboxing is the server’s job and always was.

Extensions worth doing

Multiple servers. Wrap several harnesses in a MultiHarness that merges as_model_specs() and routes call() by the name prefix. This is what real hosts do.

Wire it to the Chapter 3 loop. as_model_specs() already emits the right shape and call() already returns the right contract. Merge MCP tools with local @tool functions in one registry and the model cannot tell the difference — nor should it.

An allowlist. Give the harness an allowed_tools set and filter in refresh_tools. Log loudly when a server advertises something outside it, because that is the dynamic-capability-injection risk from Chapter 4 becoming visible.

Definition pinning. Hash each tool’s name plus inputSchema on first connection. On reconnect, compare. If a description changed under you, stop and alert — a changed description is a changed prompt.

Streamable HTTP. Swap stdio_client for streamablehttp_client and point at a remote server. Everything downstream of the transport is unchanged.

MCP Inspector. Run npx @modelcontextprotocol/inspector python inventory_server.py and click through your own server. It shows raw JSON-RPC traffic, which is the fastest way to understand a server that is misbehaving.

What you should be able to do now

Write a working MCP server with typed, documented tools, correct annotations, and tool errors that instruct the model — while keeping stdout clean for the protocol.
Build a reusable client harness that connects over stdio, discovers tools, and holds the connection open with AsyncExitStack.
Translate MCP tool definitions into your model API’s declaration format and namespace them so multiple servers can coexist.
Distinguish and handle all four failure categories — unknown tool, timeout, protocol error, tool error — and return a uniform result contract to your agent loop.
Point the same harness at an unmodified third-party server, inventory what it exposes, and assess its context cost and mutation surface before letting an agent use it.

Production Agent Engineering — A Practical Guide