Detect event-loop blocking in tests

Author: Kludex

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,16 @@ 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)
+
             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)

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.