Detect event-loop blocking in tests

pyproject.toml

@@ -77,6 +77,8 @@ dev = [
     "strict-no-cover",
     "logfire>=3.0.0",
     "opentelemetry-sdk>=1.39.1",
+    # BlockBuster minor releases may add breaking detection rules.
+    "blockbuster>=1.5.26,<1.6",
 ]
 docs = [
     # Zensical is the Material team's successor to MkDocs; it natively

src/mcp/client/session.py

@@ -12,6 +12,7 @@
 import anyio
 import anyio.abc
 import anyio.lowlevel
+import anyio.to_thread
 import mcp_types as types
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from mcp_types import (
@@ -1133,12 +1134,18 @@ async def validate_tool_result(self, name: str, result: types.CallToolResult) ->
             logger.warning(f"Tool {name} not listed by server, cannot validate any structured content")
 
         if output_schema is not None:
+            if result.structured_content is None:
+                raise RuntimeError(f"Tool {name} has an output schema but did not return structured content")
+            validator = self._tool_output_validators.get(name)
+            if validator is None:
+                # First compilation lazily reads jsonschema's bundled schemas.
+                validator = await anyio.to_thread.run_sync(self._output_schema_validator, name, output_schema)
+                if _same_schema(self._tool_output_schemas.get(name), output_schema):
+                    self._tool_output_validators[name] = validator
+
             from jsonschema import exceptions as jsonschema_exceptions
             from referencing.exceptions import Unresolvable
 
-            if result.structured_content is None:
-                raise RuntimeError(f"Tool {name} has an output schema but did not return structured content")
-            validator = self._output_schema_validator(name, output_schema)
             # `best_match` picks the same error the previous `jsonschema.validate()` call raised,
             # so the message a caller sees is unchanged. It is untyped upstream.
             errors = validator.iter_errors(result.structured_content)
@@ -1169,18 +1176,13 @@ def _output_schema_validator(self, name: str, output_schema: dict[str, Any]) ->
         from jsonschema.validators import validator_for
         from referencing import Registry
 
-        if (validator := self._tool_output_validators.get(name)) is not None:
-            return validator
-
         validator_cls = validator_for(output_schema)
         try:
             validator_cls.check_schema(output_schema)
         except SchemaError as e:
             raise RuntimeError(f"Invalid schema for tool {name}: {e}")
         # An explicit empty registry: `$ref`s resolve within the schema document and the bundled metaschemas.
-        validator = validator_cls(output_schema, registry=Registry())
-        self._tool_output_validators[name] = validator
-        return validator
+        return validator_cls(output_schema, registry=Registry())
 
     async def list_prompts(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListPromptsResult:
         """Send a prompts/list request.

tests/client/test_session_promotions.py

@@ -1,7 +1,11 @@
 """`dispatch_input_request` and `validate_tool_result` are public `ClientSession` API."""
 
+import anyio
+import anyio.from_thread
 import mcp_types as types
 import pytest
+from jsonschema.protocols import Validator
+from jsonschema.validators import validator_for
 from mcp_types import (
     CallToolResult,
     ErrorData,
@@ -125,3 +129,46 @@ async def on_list_tools(ctx: ServerRequestContext, params: PaginatedRequestParam
         await client.session.list_tools()
         with pytest.raises(RuntimeError, match="Invalid structured content returned by tool t"):
             await client.session.validate_tool_result("t", integer_result)
+
+
+@pytest.mark.anyio
+async def test_schema_change_during_compilation_does_not_cache_the_old_validator(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """SDK-defined: a concurrent tool relisting cannot leave a stale compiled validator cached."""
+    schemas = [
+        {"type": "object", "properties": {"x": {"type": "integer"}}, "required": ["x"]},
+        {"type": "object", "properties": {"x": {"type": "string"}}, "required": ["x"]},
+    ]
+
+    async def on_list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
+        return ListToolsResult(tools=[Tool(name="t", input_schema={"type": "object"}, output_schema=schemas.pop(0))])
+
+    compilation_started = anyio.Event()
+    continue_compilation = anyio.Event()
+
+    def paused_validator_for(schema: dict[str, object], default: type[Validator] | None = None) -> type[Validator]:
+        validator = validator_for(schema) if default is None else validator_for(schema, default=default)
+        if not compilation_started.is_set():
+            anyio.from_thread.run_sync(compilation_started.set)
+            anyio.from_thread.run(continue_compilation.wait)
+        return validator
+
+    monkeypatch.setattr("jsonschema.validators.validator_for", paused_validator_for)
+    server = Server("test-server", on_list_tools=on_list_tools)
+    async with Client(server) as client:
+        await client.session.list_tools()
+        with anyio.fail_after(5):
+            async with anyio.create_task_group() as task_group:
+                task_group.start_soon(
+                    client.session.validate_tool_result,
+                    "t",
+                    CallToolResult(content=[], structured_content={"x": 1}),
+                )
+                await compilation_started.wait()
+                try:
+                    await client.session.list_tools()
+                finally:
+                    continue_compilation.set()
+
+        await client.session.validate_tool_result("t", CallToolResult(content=[], structured_content={"x": "yes"}))

tests/conftest.py

@@ -1,7 +1,9 @@
 import os
 from collections.abc import AsyncIterator, Iterator
 
+import httpcore2 as _httpcore2
 import pytest
+from blockbuster import BlockBuster
 
 # OpenTelemetry's `set_tracer_provider` is set-once per process, so the suite
 # uses a single span-capture mechanism: logfire's `capfire` fixture (its
@@ -17,12 +19,34 @@
 
 import mcp.shared._otel  # noqa: E402
 
+# Load httpx2's lazy default transport before BlockBuster starts.
+del _httpcore2
+
 
 @pytest.fixture(scope="session")
 def anyio_backend() -> str:
     return "asyncio"
 
 
+_BLOCKBUSTER = BlockBuster(["mcp", "mcp_types"])
+# Coverage reads source files while collecting data.
+_BLOCKBUSTER.functions["os.stat"].can_block_in("coverage/python.py", "get_python_source")
+_BLOCKBUSTER.functions["io.BufferedReader.read"].can_block_in("coverage/python.py", "read_python_source")
+# These public synchronous conversions read the media file by design.
+_BLOCKBUSTER.functions["io.BufferedReader.read"].can_block_in(
+    "mcp/server/mcpserver/utilities/types.py", ("to_image_content", "to_audio_content")
+)
+
+
+@pytest.fixture(autouse=True)
+def blockbuster() -> Iterator[BlockBuster]:
+    try:
+        _BLOCKBUSTER.activate()
+        yield _BLOCKBUSTER
+    finally:
+        _BLOCKBUSTER.deactivate()
+
+
 @pytest.fixture(scope="module", autouse=True)
 async def _module_runner_lease(anyio_backend: str) -> AsyncIterator[None]:
     """Share one event loop across each module's tests instead of one per test.