docs: assert CORS behavior, not starlette internals; drop pragmas the docs tests cover

The lowest-direct CI cells failed on tests/docs_src/test_asgi.py reaching into
starlette's `Middleware` internals — `.kwargs` does not exist on the oldest
supported starlette. The test now drives the app over `httpx.ASGITransport`
and asserts what the page promises a browser: the preflight allows
GET/POST/DELETE and a cross-origin response exposes `Mcp-Session-Id`.

The locked cells failed at `strict-no-cover`: the docs tests execute 21 lines
marked `# pragma: no cover` — the `Image`/`Audio` helper paths, the
elicitation cancel arm, custom Starlette routes on the low-level app,
`Context.request_context` outside a request, and the
token_verifier-without-auth `ValueError`. Those markers are no longer true,
so they are removed; the full suite still reports 100% under branch coverage.

Author: maxisbey

src/mcp/server/elicitation.py

@@ -113,7 +113,7 @@ async def elicit_with_validation(
         return AcceptedElicitation(data=validated_data)
     elif result.action == "decline":
         return DeclinedElicitation()
-    elif result.action == "cancel":  # pragma: no cover
+    elif result.action == "cancel":
         return CancelledElicitation()
     else:  # pragma: no cover
         # This should never happen, but handle it just in case

src/mcp/server/lowlevel/server.py

@@ -561,7 +561,7 @@ def streamable_http_app(
                 )
             )
 
-        if custom_starlette_routes:  # pragma: no cover
+        if custom_starlette_routes:
             routes.extend(custom_starlette_routes)
 
         return Starlette(

src/mcp/server/mcpserver/context.py

@@ -82,7 +82,7 @@ def mcp_server(self) -> MCPServer:
     @property
     def request_context(self) -> ServerRequestContext[LifespanContextT, RequestT]:
         """Access to the underlying request context."""
-        if self._request_context is None:  # pragma: no cover
+        if self._request_context is None:
             raise ValueError("Context is not available outside of a request")
         return self._request_context
 

src/mcp/server/mcpserver/server.py

@@ -192,7 +192,7 @@ def __init__(
                 raise ValueError("Cannot specify both auth_server_provider and token_verifier")
             if not auth_server_provider and not token_verifier:  # pragma: no cover
                 raise ValueError("Must specify either auth_server_provider or token_verifier when auth is enabled")
-        elif auth_server_provider or token_verifier:  # pragma: no cover
+        elif auth_server_provider or token_verifier:
             raise ValueError("Cannot specify auth_server_provider or token_verifier without auth settings")
 
         self._auth_server_provider = auth_server_provider
@@ -821,15 +821,15 @@ async def health_check(request: Request) -> Response:
             ```
         """
 
-        def decorator(  # pragma: no cover
+        def decorator(
             func: Callable[[Request], Awaitable[Response]],
         ) -> Callable[[Request], Awaitable[Response]]:
             self._custom_starlette_routes.append(
                 Route(path, endpoint=func, methods=methods, name=name, include_in_schema=include_in_schema)
             )
             return func
 
-        return decorator  # pragma: no cover
+        return decorator
 
     async def run_stdio_async(self) -> None:
         """Run the server using stdio transport."""

src/mcp/server/mcpserver/utilities/types.py

@@ -27,7 +27,7 @@ def __init__(
 
     def _get_mime_type(self) -> str:
         """Get MIME type from format or guess from file extension."""
-        if self._format:  # pragma: no cover
+        if self._format:
             return f"image/{self._format.lower()}"
 
         if self.path:
@@ -39,14 +39,14 @@ def _get_mime_type(self) -> str:
                 ".gif": "image/gif",
                 ".webp": "image/webp",
             }.get(suffix, "application/octet-stream")
-        return "image/png"  # pragma: no cover  # default for raw binary data
+        return "image/png"  # default for raw binary data
 
     def to_image_content(self) -> ImageContent:
         """Convert to MCP ImageContent."""
         if self.path:
             with open(self.path, "rb") as f:
                 data = base64.b64encode(f.read()).decode()
-        elif self.data is not None:  # pragma: no cover
+        elif self.data is not None:
             data = base64.b64encode(self.data).decode()
         else:  # pragma: no cover
             raise ValueError("No image data available")
@@ -73,7 +73,7 @@ def __init__(
 
     def _get_mime_type(self) -> str:
         """Get MIME type from format or guess from file extension."""
-        if self._format:  # pragma: no cover
+        if self._format:
             return f"audio/{self._format.lower()}"
 
         if self.path:
@@ -86,14 +86,14 @@ def _get_mime_type(self) -> str:
                 ".aac": "audio/aac",
                 ".m4a": "audio/mp4",
             }.get(suffix, "application/octet-stream")
-        return "audio/wav"  # pragma: no cover  # default for raw binary data
+        return "audio/wav"  # default for raw binary data
 
     def to_audio_content(self) -> AudioContent:
         """Convert to MCP AudioContent."""
         if self.path:
             with open(self.path, "rb") as f:
                 data = base64.b64encode(f.read()).decode()
-        elif self.data is not None:  # pragma: no cover
+        elif self.data is not None:
             data = base64.b64encode(self.data).decode()
         else:  # pragma: no cover
             raise ValueError("No audio data available")

tests/docs_src/test_asgi.py

@@ -124,11 +124,21 @@ async def test_streamable_http_path_moves_the_endpoint_to_the_mount_prefix() ->
 
 
 async def test_cors_exposes_the_session_id_header() -> None:
-    """tutorial005: the CORS middleware exposes `Mcp-Session-Id` and allows the three MCP methods."""
+    """tutorial005: the browser origin gets the three MCP methods and can read `Mcp-Session-Id`."""
     (middleware,) = tutorial005.app.user_middleware
     assert middleware.cls is CORSMiddleware
-    assert middleware.kwargs["expose_headers"] == ["Mcp-Session-Id"]
-    assert middleware.kwargs["allow_methods"] == ["GET", "POST", "DELETE"]
+    transport = httpx.ASGITransport(app=tutorial005.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1") as http:
+        preflight = await http.options(
+            "/mcp",
+            headers={"Origin": "https://app.example.com", "Access-Control-Request-Method": "POST"},
+        )
+        assert preflight.status_code == 200
+        assert preflight.headers["access-control-allow-methods"] == "GET, POST, DELETE"
+
+        response = await http.get("/not-the-endpoint", headers={"Origin": "https://app.example.com"})
+        assert response.headers["access-control-allow-origin"] == "https://app.example.com"
+        assert response.headers["access-control-expose-headers"] == "Mcp-Session-Id"
 
 
 async def test_custom_route_lands_next_to_the_mcp_endpoint() -> None: