[v1.x] Server revalidates every tools/call with one-shot jsonschema.validate(), re-checking the schema each time #3676

import asyncio, time
import jsonschema
from jsonschema import exceptions, validators
from pydantic import BaseModel
from mcp import types
from mcp.server.fastmcp import FastMCP
from mcp.server.lowlevel import server as lowlevel

class Owner(BaseModel):
    address: str
    since_block: int

class Record(BaseModel):
    name: str
    value: str
    status: str
    owner: Owner | None = None

mcp = FastMCP("repro")

@mcp.tool()
def read_record(name: str) -> Record:
    return Record(name=name, value="{}", status="confirmed",
                  owner=Owner(address="x", since_block=1))

async def main(n=500):
    schema = {t.name: t for t in await mcp.list_tools()}["read_record"].outputSchema
    instance = read_record("a").model_dump(mode="json")

    t = time.perf_counter()
    for _ in range(n):
        jsonschema.validate(instance=instance, schema=schema)
    one_shot = (time.perf_counter() - t) / n

    cls = validators.validator_for(schema)
    cls.check_schema(schema)
    v = cls(schema)
    t = time.perf_counter()
    for _ in range(n):
        exceptions.best_match(v.iter_errors(instance))
    cached = (time.perf_counter() - t) / n

    handler = mcp._mcp_server.request_handlers[types.CallToolRequest]
    req = types.CallToolRequest(method="tools/call",
        params=types.CallToolRequestParams(name="read_record", arguments={"name": "a"}))
    await handler(req)
    t = time.perf_counter()
    for _ in range(n):
        await handler(req)
    e2e = (time.perf_counter() - t) / n

    class Cached:  # same semantics as jsonschema.validate, validator reused
        ValidationError = jsonschema.ValidationError
        def validate(self, instance, schema):
            if (e := exceptions.best_match(v.iter_errors(instance))):
                raise e
    lowlevel.jsonschema = Cached()
    t = time.perf_counter()
    for _ in range(n):
        await handler(req)
    e2e_cached = (time.perf_counter() - t) / n

    print(f"outputSchema check: one-shot {one_shot*1e3:.2f} ms, cached {cached*1e3:.3f} ms")
    print(f"tools/call handler: as shipped {e2e*1e3:.2f} ms, cached {e2e_cached*1e3:.2f} ms")

asyncio.run(main())