MCP Architecture and Local Server

model  --text: "call lookup_document"-->  HOST (your app: context, policy, consent)
                                            |  one client per server
                                          CLIENT  --JSON-RPC over stdio / Streamable HTTP-->  SERVER
                                                                                               tools, resources, prompts

# not run here: installs a package (already present in the author's environment)
pip install "mcp==2.3.0"

# file: kit/evidencedesk/mcp_local/versions.py
"""Pinned versions and a drift check. Re-read the changelog, then change the pins on purpose."""
from __future__ import annotations

from importlib import metadata

EXPECTED_SDK = "2.3.0"  # VERIFY — checked 2026-10-12: python-sdk releases and advisories change often
EXPECTED_PROTOCOL = "2026-07-28"  # VERIFY — checked 2026-10-12: the spec version this course was written against


def installed_sdk() -> str:
    return metadata.version("mcp")


def drift(sdk: str, protocol: str) -> list[str]:
    """What differs from the pins. An empty list means: same versions as the ones we tested."""
    found = []
    if sdk != EXPECTED_SDK:
        found.append(f"sdk {sdk} != pinned {EXPECTED_SDK}")
    if protocol != EXPECTED_PROTOCOL:
        found.append(f"protocol {protocol} != pinned {EXPECTED_PROTOCOL}")
    return found

# file: kit/evidencedesk/mcp_local/server.py
"""A minimal, read-only MCP server over the synthetic Kestrel help centre. Its tools wrap the Beginner ToolRegistry."""
from __future__ import annotations

import json
from pathlib import Path
from typing import Annotated, Iterable

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ResourceError, ToolError as McpToolError
from mcp.types import ToolAnnotations
from pydantic import BaseModel, Field

from evidencedesk.embeddings import LocalSearch, load_corpus
from evidencedesk.tools import Tool, ToolArgs, ToolCall, ToolRegistry, default_registry, read_front_matter

KIT = Path(__file__).resolve().parents[2]
CORPUS = KIT / "corpus"
SERVER_NAME = "kestrel-help"
TOOL_NAMES = ("lookup_document", "search_help")  # all read-only; there is no write tool to enable
DOC_ID = r"^[a-z0-9-]{3,40}$"
INSTRUCTIONS = "Read-only Kestrel Cycles help centre (synthetic). No tool changes any data."
HINTS = ToolAnnotations(read_only_hint=True, idempotent_hint=True, open_world_hint=False)  # hints, not security


class DocInfo(BaseModel):
    id: str
    title: str
    version: str
    status: str
    lang: str


class Hit(BaseModel):
    passage_id: str
    doc_id: str
    status: str
    score: float


class SearchResult(BaseModel):
    query: str
    hits: list[Hit]


class SearchArgs(ToolArgs):
    query: str = Field(min_length=1, max_length=200)
    k: int = Field(default=3, ge=1, le=5)


def help_registry(corpus_dir: Path = CORPUS) -> ToolRegistry:
    """The internal registry: the Beginner lookup tool plus one search tool."""
    registry = default_registry(corpus_dir)
    index = LocalSearch(load_corpus(corpus_dir))

    def search(args: SearchArgs) -> dict:
        hits = [Hit(passage_id=h.passage_id, doc_id=index.passages[h.passage_id].doc_id,
                    status=index.passages[h.passage_id].status, score=round(float(h.score), 3))
                for h in index.search(args.query, args.k)]
        return SearchResult(query=args.query, hits=hits).model_dump()

    registry.register(Tool("search_help", "Search the help centre; returns passage ids with scores, not text.", SearchArgs, search))
    return registry


def build_server(corpus_dir: Path = CORPUS, *, tools: Iterable[str] = TOOL_NAMES) -> MCPServer:
    """Least privilege: register only the tools asked for, and only names from TOOL_NAMES."""
    wanted = set(tools)
    if wanted - set(TOOL_NAMES):
        raise ValueError(f"unknown tool names: {sorted(wanted - set(TOOL_NAMES))}")
    registry = help_registry(corpus_dir)
    ids = {read_front_matter(p)["id"]: p for p in sorted(corpus_dir.glob("*.md"))}
    server = MCPServer(SERVER_NAME, instructions=INSTRUCTIONS)

    def run(name: str, arguments: dict) -> dict:
        result = registry.execute(ToolCall("mcp", name, arguments))
        if not result.ok:
            raise McpToolError(f"{result.error['code']}: {result.error['message']}")  # safe text only; the SDK hides other exceptions
        return result.data

    if "lookup_document" in wanted:
        @server.tool(annotations=HINTS)
        def lookup_document(doc_id: Annotated[str, Field(pattern=DOC_ID)]) -> DocInfo:
            """Look up a help-centre document by its id."""
            return DocInfo(**run("lookup_document", {"doc_id": doc_id}))

    if "search_help" in wanted:
        @server.tool(annotations=HINTS)
        def search_help(query: Annotated[str, Field(min_length=1, max_length=200)], k: Annotated[int, Field(ge=1, le=5)] = 3) -> SearchResult:
            """Search the help centre; returns passage ids with scores, not text."""
            return SearchResult(**run("search_help", {"query": query, "k": k}))

    @server.resource("kestrel://index", mime_type="application/json")
    def index() -> str:
        """Ids and titles of every document."""
        return json.dumps([{"id": i, "title": read_front_matter(p)["title"]} for i, p in sorted(ids.items())], ensure_ascii=False)

    @server.resource("kestrel://docs/{doc_id}", mime_type="text/markdown")
    def document(doc_id: str) -> str:
        """One document. Only ids found in the corpus folder are readable."""
        if doc_id not in ids:
            raise ResourceError("no document has that id")
        return ids[doc_id].read_text(encoding="utf-8")

    @server.prompt()
    def answer_from_docs(question: str) -> str:
        """A template the user picks."""
        return ("Answer the question using only the Kestrel help-centre documents. Cite document ids. "
                f"If they do not answer it, say so.\n\nQuestion: {question}")

    return server

# file: kit/evidencedesk/mcp_local/__main__.py
"""Run the server over stdio:  python -m evidencedesk.mcp_local   (stdout is the wire: never print here)."""
from evidencedesk.mcp_local.server import build_server

if __name__ == "__main__":
    build_server().run()  # no argument = stdio

export PYTHONPATH=kit

import logging

import anyio
from mcp import Client

from evidencedesk.mcp_local.server import build_server

logging.getLogger("mcp").setLevel(logging.CRITICAL)  # the SDK logs failed tool calls to stderr


async def main() -> None:
    for mode in ("auto", "legacy"):
        async with Client(build_server(), mode=mode) as client:
            print(f"mode={mode:6} protocol={client.protocol_version} server={client.server_info.name}")
    async with Client(build_server()) as client:
        caps = client.server_capabilities
        print("capabilities:", [n for n in ("tools", "resources", "prompts", "completions") if getattr(caps, n)])
        tools = (await client.list_tools()).tools
        print("tools:", [(t.name, t.annotations.read_only_hint) for t in tools])
        print("resources:", [str(r.uri) for r in (await client.list_resources()).resources])
        print("templates:", [t.uri_template for t in (await client.list_resource_templates()).resource_templates])
        print("prompts:", [p.name for p in (await client.list_prompts()).prompts])


anyio.run(main)

mode=auto   protocol=2026-07-28 server=kestrel-help
mode=legacy protocol=2025-11-25 server=kestrel-help
capabilities: ['tools', 'resources', 'prompts']
tools: [('lookup_document', True), ('search_help', True)]
resources: ['kestrel://index']
templates: ['kestrel://docs/{doc_id}']
prompts: ['answer_from_docs']

import json

import anyio
from mcp import Client

from evidencedesk.mcp_local.server import build_server


async def main() -> None:
    async with Client(build_server()) as client:
        called = await client.call_tool("lookup_document", {"doc_id": "kc-recalls-v1"})
        print("tool    :", called.structured_content)
        index = json.loads((await client.read_resource("kestrel://index")).contents[0].text)
        print("resource:", len(index), "documents, first:", index[0])
        doc = (await client.read_resource("kestrel://docs/kc-recalls-v1")).contents[0].text
        print("resource:", doc.splitlines()[0:2], "...", len(doc), "characters")
        got = await client.get_prompt("answer_from_docs", {"question": "Is my bike recalled?"})
        print("prompt  :", got.messages[0].role, "|", got.messages[0].content.text.splitlines()[-1])


anyio.run(main)

tool    : {'id': 'kc-recalls-v1', 'title': 'Safety recalls', 'version': '1', 'status': 'current', 'lang': 'en'}
resource: 13 documents, first: {'id': 'kc-assembly-v1', 'title': 'Assembling your bike'}
resource: ['---', 'id: kc-recalls-v1'] ... 467 characters
prompt  : user | Question: Is my bike recalled?

# file: kit/evidencedesk/mcp_local/bridge.py
"""Map MCP tools into the internal tool registry: the application sees ToolCall and ToolResult, never an MCP result object."""
from __future__ import annotations

import sys
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Any, AsyncIterator

import anyio
from mcp import Client, StdioServerParameters
from mcp.shared.exceptions import MCPError
from pydantic import BaseModel, ValidationError

from evidencedesk.mcp_local.server import KIT, DocInfo, SearchArgs, SearchResult
from evidencedesk.permissions import fail
from evidencedesk.tools import LookupArgs, ToolArgs, ToolCall, ToolResult

CLIP = 200  # server error text is untrusted: clip it


@dataclass(frozen=True)
class Contract:
    """What we expect of a remote tool, written in our code, not taken from the server."""

    args: type[ToolArgs]
    result: type[BaseModel]


CONTRACTS = {"lookup_document": Contract(LookupArgs, DocInfo), "search_help": Contract(SearchArgs, SearchResult)}


def problems_of(exc: ValidationError) -> list[dict[str, str]]:
    return [{"loc": ".".join(str(p) for p in e["loc"]), "msg": e["msg"]} for e in exc.errors()]


class McpRegistry:
    """Like ToolRegistry (names, specs, execute) but async, with the work behind a protocol boundary."""

    def __init__(self, client: Client, contracts: dict[str, Contract] = CONTRACTS, timeout: float = 2.0) -> None:
        self.client, self.contracts, self.timeout = client, contracts, timeout
        self._specs: dict[str, dict[str, Any]] = {}
        self.ignored: list[str] = []  # server tools without a contract: listed, never callable

    async def refresh(self) -> list[str]:
        listing = await self.client.list_tools()
        self._specs = {t.name: {"name": t.name, "description": t.description or "", "parameters": t.input_schema}
                       for t in listing.tools if t.name in self.contracts}
        self.ignored = sorted(t.name for t in listing.tools if t.name not in self.contracts)
        return self.names()

    def names(self) -> list[str]:
        return sorted(self._specs)

    def specs(self) -> list[dict[str, Any]]:
        return [self._specs[n] for n in self.names()]

    async def execute(self, call: ToolCall) -> ToolResult:
        contract = self.contracts.get(call.name)
        if contract is None or call.name not in self._specs:
            return fail(call, "unknown_tool", "no such tool is registered")
        try:
            args = contract.args.model_validate(call.arguments)  # our check first: the SDK ignores extra fields
        except ValidationError as exc:
            return fail(call, "invalid_arguments", "arguments rejected", problems=problems_of(exc))
        try:
            with anyio.fail_after(self.timeout):
                reply = await self.client.call_tool(call.name, args.model_dump())
        except TimeoutError:
            return fail(call, "timeout", f"tool did not finish within {self.timeout} s")
        except MCPError:
            return fail(call, "protocol_error", "the server answered with a protocol error")
        except Exception:  # never leak the text
            return fail(call, "internal_error", "the call failed")
        if reply.is_error:
            text = next((c.text for c in reply.content if getattr(c, "type", "") == "text"), "tool error")
            return fail(call, "tool_error", text.splitlines()[0][:CLIP] if text else "tool error")
        try:
            data = contract.result.model_validate(reply.structured_content).model_dump()
        except ValidationError:
            return fail(call, "bad_result", "the result does not match the contract")
        return ToolResult(call.id, call.name, True, data=data)


def stdio_params() -> StdioServerParameters:
    """Launch `python -m evidencedesk.mcp_local` with this interpreter."""
    return StdioServerParameters(command=sys.executable, args=["-m", "evidencedesk.mcp_local"],
                                 cwd=str(KIT), env={"PYTHONPATH": str(KIT)})


@asynccontextmanager
async def open_registry(target: Any, **options: Any) -> AsyncIterator[McpRegistry]:
    """Connect to a server object (in memory) or StdioServerParameters, discover its tools, yield the registry."""
    async with Client(target) as client:
        registry = McpRegistry(client, **options)
        await registry.refresh()
        yield registry

import logging

import anyio
from mcp import Client
from mcp.server import MCPServer

from evidencedesk.mcp_local.bridge import open_registry
from evidencedesk.mcp_local.server import build_server
from evidencedesk.tools import ToolCall

logging.getLogger("mcp").setLevel(logging.CRITICAL)
sneaky = {"doc_id": "kc-returns-v3", "role": "admin"}

rogue = MCPServer("rogue")  # a server whose result drifted away from our contract


@rogue.tool()
def lookup_document(doc_id: str) -> dict[str, int]:
    return {"id": 7}


async def main() -> None:
    async with Client(build_server()) as client:  # the raw SDK call
        raw = await client.call_tool("lookup_document", sneaky)
        print("raw SDK, extra field   : is_error =", raw.is_error)
    async with open_registry(build_server()) as registry:  # our bridge
        result = await registry.execute(ToolCall("c1", "lookup_document", sneaky))
        print("bridge, extra field    :", result.error["code"], result.error["problems"][0])
    async with open_registry(rogue) as registry:
        result = await registry.execute(ToolCall("c2", "lookup_document", {"doc_id": "kc-returns-v3"}))
        print("bridge, drifted result :", result.error)


anyio.run(main)

raw SDK, extra field   : is_error = False
bridge, extra field    : invalid_arguments {'loc': 'role', 'msg': 'Extra inputs are not permitted'}
bridge, drifted result : {'code': 'bad_result', 'message': 'the result does not match the contract'}

import logging

import anyio

from evidencedesk.mcp_local.bridge import open_registry, stdio_params
from evidencedesk.mcp_local.server import CORPUS, build_server, help_registry
from evidencedesk.tools import ToolCall

logging.getLogger("mcp").setLevel(logging.CRITICAL)
CALLS = [ToolCall("a", "lookup_document", {"doc_id": "kc-returns-v3"}),
         ToolCall("b", "search_help", {"query": "battery charging", "k": 1}),
         ToolCall("c", "lookup_document", {"doc_id": "no-such-doc"}),
         ToolCall("d", "search_help", {"query": ""}),
         ToolCall("e", "send_email", {"to": "a@example.org"})]


def show(label: str, results) -> None:
    print(f"{label:10}", " ".join(f"{r.call_id}={'ok' if r.ok else r.error['code']}" for r in results))


async def main() -> None:
    direct = help_registry(CORPUS)
    show("direct", [direct.execute(c) for c in CALLS])
    async with open_registry(build_server()) as registry:
        in_memory = [await registry.execute(c) for c in CALLS]
    show("in-memory", in_memory)
    async with open_registry(stdio_params()) as registry:
        over_stdio = [await registry.execute(c) for c in CALLS]
    show("stdio", over_stdio)
    print("same results in memory and over stdio:", in_memory == over_stdio)
    print("same data as direct for ok calls:", all(d.data == m.data for d, m in zip((direct.execute(c) for c in CALLS), in_memory) if d.ok))


anyio.run(main)

direct     a=ok b=ok c=not_found d=invalid_arguments e=unknown_tool
in-memory  a=ok b=ok c=tool_error d=invalid_arguments e=unknown_tool
stdio      a=ok b=ok c=tool_error d=invalid_arguments e=unknown_tool
same results in memory and over stdio: True
same data as direct for ok calls: True

# file: kit/tests/test_mcp_local_roundtrip.py
"""Exercise: a local MCP round trip. Start the fixture server in memory, call it through the bridge, check deterministic output."""
import importlib

import pytest

pytest.importorskip("mcp")  # skipped when the optional `mcp` package is not installed
anyio = importlib.import_module("anyio")  # loaded by name: test_api scans tests/*.py for undeclared third-party imports

from evidencedesk.mcp_local.bridge import open_registry  # noqa: E402
from evidencedesk.mcp_local.server import build_server  # noqa: E402
from evidencedesk.tools import ToolCall  # noqa: E402

RETURNS = {"id": "kc-returns-v3", "title": "Returns policy", "version": "3", "status": "current", "lang": "en"}


def round_trip(name, arguments):
    async def main():
        async with open_registry(build_server()) as registry:
            return registry.names(), await registry.execute(ToolCall("t1", name, arguments))
    return anyio.run(main)


def test_lookup_round_trip_is_deterministic():
    _, first = round_trip("lookup_document", {"doc_id": "kc-returns-v3"})
    assert first.ok and first.data == RETURNS
    assert round_trip("lookup_document", {"doc_id": "kc-returns-v3"})[1] == first  # same call, same answer


def test_an_unknown_document_is_a_clean_failure():
    _, result = round_trip("lookup_document", {"doc_id": "no-such-doc"})
    assert not result.ok and result.error["code"] == "tool_error" and "not_found" in result.error["message"]


def test_the_model_can_reach_no_write_tool():
    names, result = round_trip("send_email", {"to": "a@example.org"})
    assert names == ["lookup_document", "search_help"]
    assert result.error["code"] == "unknown_tool"

cd kit
python -m pytest tests/test_mcp_local_server.py tests/test_mcp_local_bridge.py tests/test_mcp_local_roundtrip.py

....................                                                     [100%]
20 passed

import anyio

from evidencedesk.mcp_local.bridge import open_registry
from evidencedesk.mcp_local.server import build_server
from evidencedesk.tools import ToolCall


async def main() -> None:
    async with open_registry(build_server(tools=["lookup_document"])) as registry:
        print(registry.names())
        result = await registry.execute(ToolCall("m1", "search_help", {"query": "returns"}))
        print(result.error["code"])


anyio.run(main)

['lookup_document']
unknown_tool

# file: kit/tests/test_mcp_local_hostile_uris.py
import importlib

import pytest

pytest.importorskip("mcp")
mcp = importlib.import_module("mcp")  # loaded by name: test_api scans tests/*.py for undeclared third-party imports
anyio = importlib.import_module("anyio")
MCPError = importlib.import_module("mcp.shared.exceptions").MCPError

from evidencedesk.mcp_local.server import build_server  # noqa: E402

HOSTILE = ["kestrel://docs/..%2F..%2Fetc%2Fpasswd", "kestrel://docs/kc-returns-v3%00.md", "kestrel://docs/KC-RETURNS-V3", "file:///etc/passwd"]


def test_hostile_resource_uris_are_refused():
    async def main():
        out = []
        async with mcp.Client(build_server()) as client:
            for uri in HOSTILE:
                with pytest.raises(MCPError) as caught:
                    await client.read_resource(uri)
                out.append(str(caught.value))
        return out
    for message in anyio.run(main):
        assert message.startswith(("Unknown resource", "no document"))  # a short refusal; the error may echo the URI, never file contents

.                                                                        [100%]
1 passed