"""Run beside result_fixture.py; official schemas are downloaded once and pinned.""" import copy import hashlib import json import pathlib import subprocess import sys import urllib.request from jsonschema import Draft202012Validator, ValidationError from referencing import Registry from referencing.exceptions import NoSuchResource, Unresolvable import result_fixture as f SPEC_COMMIT = "ccaf9102ec5eb656eab50ca8333ef49dcd259c26" HASHES = { "2026-07-28": "ef70b61f99b6d2e5e3b46863822eab08dff6a45bedc7a08914e0e5b133f40203", "2025-11-25": "268a5f82ba70fd7e4b6dc4aa1e64f116f74b4d0edcb69dc046829c79dd4e97e7", } def load_schema(version): path = pathlib.Path("schema-" + version + ".json") if not path.exists(): url = ("https://raw.githubusercontent.com/modelcontextprotocol/" f"modelcontextprotocol/{SPEC_COMMIT}/schema/{version}/schema.json") with urllib.request.urlopen(url, timeout=30) as response: path.write_bytes(response.read()) raw = path.read_bytes() assert hashlib.sha256(raw).hexdigest() == HASHES[version] return json.loads(raw) def deny_network(uri): raise NoSuchResource(ref=uri) REGISTRY = Registry(retrieve=deny_network) SCHEMA = load_schema(f.VERSION) OLD = load_schema("2025-11-25") def wire(definition, value, schema=SCHEMA): validator = Draft202012Validator( {**schema, "$ref": "#/$defs/" + definition}, registry=REGISTRY) validator.validate(value) def accept(tool, result): wire("CallToolResult", result) if result["resultType"] != "complete": raise ValueError("Not a complete result") if result.get("isError"): raise ValueError("Tool execution failed") if "outputSchema" in tool: if "structuredContent" not in result: raise ValueError("Missing structuredContent") Draft202012Validator(tool["outputSchema"], registry=REGISTRY).validate( result["structuredContent"]) return result CASES = [] def check(label, run): run() CASES.append({"name": label, "status": "PASS"}) def rejected(run): try: run() except (ValueError, ValidationError, Unresolvable): return raise AssertionError("Expected rejection") def require(condition): if not condition: raise AssertionError("Condition failed") def main(): proc = subprocess.Popen([sys.executable, str(pathlib.Path(f.__file__))], stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True) serial = 0 transcript = [] def rpc(method, **params): nonlocal serial serial += 1 req = {"jsonrpc": "2.0", "id": serial, "method": method, "params": {**params, "_meta": { "io.modelcontextprotocol/protocolVersion": f.VERSION, "io.modelcontextprotocol/clientCapabilities": {}, "io.modelcontextprotocol/clientInfo": { "name": "result-probe", "version": "1.0.0"}}}} request_types = {"server/discover": "DiscoverRequest", "tools/list": "ListToolsRequest", "tools/call": "CallToolRequest", "resources/list": "ListResourcesRequest", "resources/read": "ReadResourceRequest"} wire(request_types[method], req) proc.stdin.write(json.dumps(req) + "\n") proc.stdin.flush() response = json.loads(proc.stdout.readline()) require(response["id"] == serial) transcript.append({"request": req, "response": response}) return response try: discovered = rpc("server/discover")["result"] check("discover-wire", lambda: wire("DiscoverResult", discovered)) listed = rpc("tools/list")["result"] check("tools-list-wire", lambda: wire("ListToolsResult", listed)) for entry in listed["tools"]: wire("Tool", entry) check("four-definitions", lambda: require(len(listed["tools"]) == 4)) results = {} for tool in listed["tools"]: mode = tool["name"].removeprefix("search_") result = rpc("tools/call", name=tool["name"], arguments={"q": "MCP"})["result"] results[mode] = result check(mode + "-wire-and-contract", lambda t=tool, r=result: accept(t, r)) check("text-body-equal", lambda: require( json.loads(results["text"]["content"][0]["text"]) == f.DATA)) check("structured-body-equal", lambda: require( results["structured"]["structuredContent"] == f.DATA)) check("structured-duplicate-equal", lambda: require( json.loads(results["structured"]["content"][0]["text"]) == results["structured"]["structuredContent"])) check("embedded-body-equal", lambda: require( json.loads(results["embedded"]["content"][0]["resource"]["text"]) == f.DATA)) check("no-implicit-read", lambda: require( not any(x["request"]["method"] == "resources/read" for x in transcript))) resources = rpc("resources/list")["result"] check("resource-list-wire", lambda: wire("ListResourcesResult", resources)) check("link-unlisted", lambda: require(resources["resources"] == [])) link = results["link"]["content"][0] read = rpc("resources/read", uri=link["uri"])["result"] check("resource-read-wire", lambda: wire("ReadResourceResult", read)) check("link-body-equal-after-read", lambda: require( json.loads(read["contents"][0]["text"]) == f.DATA)) check("one-explicit-read", lambda: require(sum( x["request"]["method"] == "resources/read" for x in transcript) == 1)) missing = rpc("resources/read", uri="search://results/missing") check("missing-uri-error", lambda: require(missing["error"]["code"] == -32602)) error = rpc("tools/call", name="search_structured", arguments={"q": "other"})["result"] check("tool-error-wire", lambda: wire("CallToolResult", error)) check("tool-error-not-success", lambda: rejected(lambda: accept(f.TOOLS[1], error))) bad = copy.deepcopy(results["structured"]) bad["structuredContent"]["total"] = "2" check("bad-data-still-valid-wire", lambda: wire("CallToolResult", bad)) check("bad-data-rejected-by-output-schema", lambda: rejected(lambda: accept(f.TOOLS[1], bad))) check("schema-with-text-only-rejected", lambda: rejected( lambda: accept(f.TOOLS[1], results["text"]))) missing_content = {"resultType": "complete", "structuredContent": f.DATA} check("missing-content-rejected", lambda: rejected(lambda: wire("CallToolResult", missing_content))) check("empty-content-accepted", lambda: accept(f.TOOLS[1], { "resultType": "complete", "content": [], "structuredContent": f.DATA})) bad_link = copy.deepcopy(results["link"]) del bad_link["content"][0]["name"] check("link-without-name-rejected", lambda: rejected(lambda: wire("CallToolResult", bad_link))) bad_embedded = copy.deepcopy(results["embedded"]) bad_embedded["content"][0]["resource"]["text"] = "not-json" check("embedded-invalid-json-valid-wire", lambda: wire("CallToolResult", bad_embedded)) check("embedded-json-parse-rejected", lambda: rejected(lambda: json.loads( bad_embedded["content"][0]["resource"]["text"]))) bad_total = copy.deepcopy(results["structured"]) bad_total["structuredContent"]["total"] = 999 check("semantic-error-passes-schema", lambda: accept(f.TOOLS[1], bad_total)) check("semantic-error-custom-check-rejected", lambda: rejected(lambda: require_count(bad_total))) check("duplicate-drift-custom-check-rejected", lambda: rejected(lambda: require_duplicate(bad_total))) unknown = {"resultType": "unknown", "content": []} check("unknown-result-type-structurally-valid", lambda: wire("CallToolResult", unknown)) check("unknown-result-type-prose-check-rejected", lambda: rejected(lambda: accept(f.TOOLS[0], unknown))) no_cache = copy.deepcopy(read) del no_cache["ttlMs"] check("read-without-cache-field-rejected", lambda: rejected(lambda: wire("ReadResourceResult", no_cache))) for value in [[], "", 0, False, None]: result = {"resultType": "complete", "content": [], "structuredContent": value} label = json.dumps(value) check("modern-json-value-" + label, lambda r=result: wire("CallToolResult", r)) check("legacy-rejects-" + label, lambda r=result: rejected( lambda: wire("CallToolResult", r, schema=OLD))) foreign_ref = {"$ref": "https://invalid.example/never-fetch.json"} check("external-ref-rejected-without-fetch", lambda: rejected( lambda: Draft202012Validator(foreign_ref, registry=REGISTRY).validate({}))) # Server has not silently appended a generated result to its catalog. after = rpc("resources/list")["result"] check("catalog-still-empty", lambda: require(after["resources"] == [])) finally: proc.stdin.close() proc.wait(timeout=5) pathlib.Path("transcript.json").write_text(json.dumps(transcript, indent=2)) print(json.dumps({"cases": len(CASES), "passed": len(CASES), "rpc_count": len(transcript), "results": CASES}, indent=2)) def require_count(result): data = result["structuredContent"] if data["total"] != len(data["hits"]): # Only for this non-paginated fixture. raise ValueError("Wrong total") def require_duplicate(result): if json.loads(result["content"][0]["text"]) != result["structuredContent"]: raise ValueError("Diverging duplicate") if __name__ == "__main__": main()