Server Instructions in Practice: Spec Versions and Testing

#!/usr/bin/env python3
"""Probe an MCP stdio server and lint its `instructions` field (stdlib only).

Era detection follows the 2026-07-28 stdio "Backward Compatibility" rules:
send `server/discover` first; a DiscoverResult or error -32022 means a modern
server, any other error or a timeout means a legacy server, so fall back to
the `initialize` handshake. The lint rules are engineering heuristics, not
requirements of the specification.

Usage: check_instructions.py [--timeout S] [--max-chars N] -- <server command>
Exit code: 0 = no findings, 1 = warnings, 2 = errors (cannot connect, bad field).
"""
import argparse
import json
import queue
import re
import subprocess
import sys
import threading

MODERN = "2026-07-28"
LEGACY = "2025-11-25"
UNSUPPORTED_VERSION = -32022
CLIENT = {"name": "check_instructions", "version": "0.1"}
STOP = {"the", "a", "an", "to", "for", "of", "by", "in", "on", "is", "are", "and",
        "or", "this", "that", "with", "it", "tool", "tools", "server"}


class Peer:
    """A stdio JSON-RPC peer: one JSON message per line in both directions."""

    def __init__(self, cmd, timeout):
        self.timeout, self.next_id, self.noise = timeout, 0, 0
        self.proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                     stderr=subprocess.DEVNULL, text=True)
        self.lines = queue.Queue()
        threading.Thread(target=self._pump, daemon=True).start()

    def _pump(self):
        for line in self.proc.stdout:
            self.lines.put(line)
        self.lines.put(None)

    def send(self, method, params=None, notify=False):
        msg = {"jsonrpc": "2.0", "method": method}
        if params is not None:
            msg["params"] = params
        if not notify:
            self.next_id += 1
            msg["id"] = self.next_id
        try:
            self.proc.stdin.write(json.dumps(msg) + "\n")
            self.proc.stdin.flush()
        except (BrokenPipeError, OSError):
            pass
        return msg.get("id")

    def call(self, method, params=None):
        rid = self.send(method, params)
        while True:
            try:
                line = self.lines.get(timeout=self.timeout)
            except queue.Empty:
                return {"timeout": True}
            if line is None:
                return {"closed": True}
            try:
                msg = json.loads(line)
            except ValueError:
                self.noise += 1  # stdout must carry MCP messages only
                continue
            if isinstance(msg, dict) and msg.get("id") == rid and ("result" in msg or "error" in msg):
                return msg

    def close(self):
        try:
            self.proc.stdin.close()
            self.proc.wait(timeout=3)
        except Exception:
            self.proc.kill()


def meta(version, capabilities=None):
    return {"io.modelcontextprotocol/protocolVersion": version,
            "io.modelcontextprotocol/clientInfo": CLIENT,
            "io.modelcontextprotocol/clientCapabilities": capabilities or {}}


def handshake(peer, capabilities=None):
    """Return (era, protocol version, result dict) or (None, None, error text)."""
    version = MODERN
    for _ in range(2):
        reply = peer.call("server/discover", {"_meta": meta(version, capabilities)})
        result = reply.get("result")
        if isinstance(result, dict) and "supportedVersions" in result:
            return "modern", version, result
        error = reply.get("error") or {}
        if error.get("code") == UNSUPPORTED_VERSION:  # modern server: never fall back
            usable = [v for v in (error.get("data") or {}).get("supported", []) if v >= MODERN]
            if not usable:
                return None, None, "modern server offers no modern protocol version"
            version = max(usable)
            continue
        break  # any other error, or a timeout: treat as legacy
    reply = peer.call("initialize", {"protocolVersion": LEGACY, "capabilities": capabilities or {},
                                     "clientInfo": CLIENT})
    if not isinstance(reply.get("result"), dict):
        return None, None, "initialize failed: %s" % json.dumps(reply)[:200]
    peer.send("notifications/initialized", notify=True)
    return "legacy", reply["result"].get("protocolVersion"), reply["result"]


def list_tools(peer, era, version, capabilities=None):
    """Return {name: tool object}, or None if tools/list is unavailable."""
    tools, cursor = {}, None
    for _ in range(20):  # pagination guard
        params = {"_meta": meta(version, capabilities)} if era == "modern" else {}
        if cursor:
            params["cursor"] = cursor
        result = peer.call("tools/list", params).get("result")
        if not isinstance(result, dict):
            return tools or None
        for tool in result.get("tools", []):
            tools[tool["name"]] = tool
        cursor = result.get("nextCursor")
        if not cursor:
            break
    return tools


def stem(word):
    """Strip a plural ending: searches -> search, files -> file, reads -> read."""
    if len(word) <= 3:
        return word
    if re.search(r"(ss|ch|sh|x|z)es$", word):
        return word[:-2]
    return word[:-1] if word.endswith("s") and not word.endswith("ss") else word


def words(text):
    return {stem(w) for w in re.findall(r"[a-z0-9]+", text.lower()) if w not in STOP}


def lint(text, tools, max_chars):
    findings = []
    if len(text) > max_chars:
        findings.append(("WARN", "too-long", "%d chars > %d; instructions are not a manual" % (len(text), max_chars)))
    if tools is None:
        return findings
    for ref in sorted(set(re.findall(r"""['"`]([A-Za-z][\w.\-]*\*?)['"`]""", text))):
        known = any(n.startswith(ref[:-1]) for n in tools) if ref.endswith("*") else ref in tools
        if not known:
            findings.append(("WARN", "unknown-reference",
                             "'%s' is not in tools/list (%s)" % (ref, ", ".join(sorted(tools)) or "no tools")))
    for sentence in re.split(r"(?<=[.!?])\s+|\n", text):
        sw = words(sentence)
        if len(sw) < 2:
            continue
        for name, tool in tools.items():
            if len(sw & words(tool.get("description", ""))) / len(sw) >= 0.8:
                findings.append(("WARN", "duplicates-tool-description",
                                 "\"%s\" restates the description of '%s'" % (sentence.strip(), name)))
                break
    return findings


def main():
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    ap.add_argument("--timeout", type=float, default=5.0)
    ap.add_argument("--max-chars", type=int, default=1500)
    ap.add_argument("cmd", nargs=argparse.REMAINDER)
    args = ap.parse_args()
    cmd = args.cmd[1:] if args.cmd[:1] == ["--"] else args.cmd
    if not cmd:
        ap.error("missing server command after --")
    peer = Peer(cmd, args.timeout)
    try:
        era, version, result = handshake(peer)
        if era is None:
            print("[ERROR] %s" % result)
            return 2
        tools = list_tools(peer, era, version)
    finally:
        peer.close()
    extra = " ttlMs=%s cacheScope=%s" % (result.get("ttlMs"), result.get("cacheScope")) if era == "modern" else ""
    print("era: %s (protocol %s)%s" % (era, version, extra))
    print("tools: %s" % (", ".join(sorted(tools)) if tools is not None else "unavailable"))
    findings = []
    text = result.get("instructions")
    if text is None:
        print("instructions: absent (optional field)")
    elif not isinstance(text, str):
        findings.append(("ERROR", "bad-type", "instructions must be a string"))
    else:
        print("instructions: %d chars" % len(text))
        findings += lint(text, tools, args.max_chars)
    if peer.noise:
        findings.append(("WARN", "stdout-noise", "%d non-JSON line(s) on stdout" % peer.noise))
    for level, code, message in findings:
        print("[%s] %s: %s" % (level, code, message))
    if not findings:
        print("[OK] no findings")
    return 2 if any(f[0] == "ERROR" for f in findings) else 1 if findings else 0


if __name__ == "__main__":
    sys.exit(main())

"""Demo server on the official Python SDK (pip install mcp) with server instructions.

DEMO_RENAME_READ=1 registers `read` as `read_file` while the instructions still
say 'read', which simulates tool-name drift after a refactor.
DEMO_PARROT=1 swaps in instructions that restate tool descriptions.
"""
import os

from mcp.server.mcpserver import MCPServer

INSTRUCTIONS = (
    "Call 'authenticate' before 'search' or 'read'. "
    "Use 'search' before 'read' to validate file paths. "
    "Rate limit: 100 requests/minute across all tools."
)
if os.environ.get("DEMO_PARROT"):
    INSTRUCTIONS = "The search tool searches for files. The read tool reads files."

server = MCPServer("demo-files", instructions=INSTRUCTIONS)


@server.tool()
def authenticate(token: str) -> str:
    """Authenticate the session with an API token."""
    return "ok"


@server.tool()
def search(query: str) -> list[str]:
    """Search for files whose names match the query."""
    return []


@server.tool(name="read_file" if os.environ.get("DEMO_RENAME_READ") else "read")
def read(path: str) -> str:
    """Read a file by path."""
    return ""


if __name__ == "__main__":
    server.run("stdio")

"""Minimal legacy-era (initialize handshake) stdio server, stdlib only.

Used to exercise the checker's fallback paths:
  --silent  never answer requests it does not know (the probe must time out)
  --noise   print a banner to stdout before serving (breaks stdio framing)
"""
import json
import sys

SILENT = "--silent" in sys.argv
if "--noise" in sys.argv:
    print("demo-files 0.1 starting...", flush=True)

TOOLS = [{"name": "ping", "description": "Check connectivity.", "inputSchema": {"type": "object"}}]


def reply(rid, result=None, error=None):
    msg = {"jsonrpc": "2.0", "id": rid}
    msg.update({"error": error} if error else {"result": result})
    print(json.dumps(msg), flush=True)


for line in sys.stdin:
    req = json.loads(line)
    method, rid = req.get("method"), req.get("id")
    if method == "initialize":
        reply(rid, {"protocolVersion": "2025-11-25", "capabilities": {"tools": {}},
                    "serverInfo": {"name": "legacy-demo", "version": "0.1"},
                    "instructions": "Call 'ping' once at startup."})
    elif method == "tools/list":
        reply(rid, {"tools": TOOLS})
    elif rid is not None and not SILENT:
        reply(rid, error={"code": -32601, "message": "Method not found"})