test: interaction-model end-to-end suite with a requirements manifest (#2691)

pyproject.toml

@@ -78,7 +78,7 @@ dev = [
     # We add mcp[cli,ws] so `uv sync` considers the extras.
     "mcp[cli,ws]",
     "pyright>=1.1.400",
-    "pytest>=8.3.4",
+    "pytest>=8.4.0",
     "ruff>=0.8.5",
     "trio>=0.26.2",
     "pytest-flakefinder>=1.1.0",
@@ -193,6 +193,9 @@ strict-no-cover = { git = "https://github.com/pydantic/strict-no-cover" }
 [tool.pytest.ini_options]
 log_cli = true
 xfail_strict = true
+markers = [
+    "requirement(id): links a test to the entry in tests/interaction/_requirements.py it exercises",
+]
 addopts = """
     --color=yes
     --capture=fd

src/mcp/client/auth/oauth2.py

@@ -360,10 +360,10 @@ class OAuthClientProvider(httpx.Auth):
         auth_code, returned_state = await self.context.callback_handler()
 
         if returned_state is None or not secrets.compare_digest(returned_state, state):
-            raise OAuthFlowError(f"State parameter mismatch: {returned_state} != {state}")  # pragma: no cover
+            raise OAuthFlowError(f"State parameter mismatch: {returned_state} != {state}")
 
         if not auth_code:
-            raise OAuthFlowError("No authorization code received")  # pragma: no cover
+            raise OAuthFlowError("No authorization code received")
 
         # Return auth code and code verifier for token exchange
         return auth_code, pkce_params.code_verifier
@@ -452,7 +452,7 @@ class OAuthClientProvider(httpx.Auth):
 
         return httpx.Request("POST", token_url, data=refresh_data, headers=headers)
 
-    async def _handle_refresh_response(self, response: httpx.Response) -> bool:  # pragma: no cover
+    async def _handle_refresh_response(self, response: httpx.Response) -> bool:
         """Handle token refresh response. Returns True if successful."""
         if response.status_code != 200:
             logger.warning(f"Token refresh failed: {response.status_code}")
@@ -468,12 +468,12 @@ class OAuthClientProvider(httpx.Auth):
             await self.context.storage.set_tokens(token_response)
 
             return True
-        except ValidationError:
+        except ValidationError:  # pragma: no cover
             logger.exception("Invalid refresh response")
             self.context.clear_tokens()
             return False
 
-    async def _initialize(self) -> None:  # pragma: no cover
+    async def _initialize(self) -> None:
         """Load stored tokens and client info."""
         self.context.current_tokens = await self.context.storage.get_tokens()
         self.context.client_info = await self.context.storage.get_client_info()
@@ -507,17 +507,17 @@ class OAuthClientProvider(httpx.Auth):
         """HTTPX auth flow integration."""
         async with self.context.lock:
             if not self._initialized:
-                await self._initialize()  # pragma: no cover
+                await self._initialize()
 
             # Capture protocol version from request headers
             self.context.protocol_version = request.headers.get(MCP_PROTOCOL_VERSION)
 
             if not self.context.is_token_valid() and self.context.can_refresh_token():
                 # Try to refresh token
-                refresh_request = await self._refresh_token()  # pragma: no cover
-                refresh_response = yield refresh_request  # pragma: no cover
+                refresh_request = await self._refresh_token()
+                refresh_response = yield refresh_request
 
-                if not await self._handle_refresh_response(refresh_response):  # pragma: no cover
+                if not await self._handle_refresh_response(refresh_response):
                     # Refresh failed, need full re-authentication
                     self._initialized = False
 
@@ -612,7 +612,7 @@ class OAuthClientProvider(httpx.Auth):
                     # Step 5: Perform authorization and complete token exchange
                     token_response = yield await self._perform_authorization()
                     await self._handle_token_response(token_response)
-                except Exception:  # pragma: no cover
+                except Exception:
                     logger.exception("OAuth flow error")
                     raise
 

src/mcp/client/client.py

@@ -305,4 +305,4 @@ class Client:
     async def send_roots_list_changed(self) -> None:
         """Send a notification that the roots list has changed."""
         # TODO(Marcelo): Currently, there is no way for the server to handle this. We should add support.
-        await self.session.send_roots_list_changed()  # pragma: no cover
+        await self.session.send_roots_list_changed()

src/mcp/client/session.py

@@ -74,7 +74,7 @@ async def _default_elicitation_callback(
     context: RequestContext[ClientSession],
     params: types.ElicitRequestParams,
 ) -> types.ElicitResult | types.ErrorData:
-    return types.ErrorData(  # pragma: no cover
+    return types.ErrorData(
         code=types.INVALID_REQUEST,
         message="Elicitation not supported",
     )
@@ -337,9 +337,7 @@ class ClientSession(
             from jsonschema import SchemaError, ValidationError, validate
 
             if result.structured_content is None:
-                raise RuntimeError(
-                    f"Tool {name} has an output schema but did not return structured content"
-                )  # pragma: no cover
+                raise RuntimeError(f"Tool {name} has an output schema but did not return structured content")
             try:
                 validate(result.structured_content, output_schema)
             except ValidationError as e:
@@ -408,7 +406,7 @@ class ClientSession(
 
         return result
 
-    async def send_roots_list_changed(self) -> None:  # pragma: no cover
+    async def send_roots_list_changed(self) -> None:
         """Send a roots/list_changed notification."""
         await self.send_notification(types.RootsListChangedNotification())
 
@@ -449,7 +447,7 @@ class ClientSession(
                     client_response = ClientResponse.validate_python(response)
                     await responder.respond(client_response)
 
-            case types.PingRequest():  # pragma: no cover
+            case types.PingRequest():
                 with responder:
                     return await responder.respond(types.EmptyResult())
 

src/mcp/client/streamable_http.py

@@ -210,7 +210,7 @@ class StreamableHTTPTransport:
                     # Stream ended normally (server closed) - reset attempt counter
                     attempt = 0
 
-            except Exception:  # pragma: lax no cover
+            except Exception:
                 logger.debug("GET stream error", exc_info=True)
                 attempt += 1
 
@@ -267,8 +267,8 @@ class StreamableHTTPTransport:
                 logger.debug("Received 202 Accepted")
                 return
 
-            if response.status_code == 404:  # pragma: no branch
-                if isinstance(message, JSONRPCRequest):  # pragma: no branch
+            if response.status_code == 404:
+                if isinstance(message, JSONRPCRequest):
                     error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated")
                     session_message = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data))
                     await ctx.read_stream_writer.send(session_message)
@@ -492,17 +492,17 @@ class StreamableHTTPTransport:
 
     async def terminate_session(self, client: httpx.AsyncClient) -> None:
         """Terminate the session by sending a DELETE request."""
-        if not self.session_id:  # pragma: lax no cover
-            return
+        if not self.session_id:
+            return  # pragma: no cover
 
         try:
             headers = self._prepare_headers()
             response = await client.delete(self.url, headers=headers)
 
-            if response.status_code == 405:  # pragma: lax no cover
+            if response.status_code == 405:
                 logger.debug("Server does not allow session termination")
-            elif response.status_code not in (200, 204):  # pragma: lax no cover
-                logger.warning(f"Session termination failed: {response.status_code}")
+            elif response.status_code not in (200, 204):
+                logger.warning(f"Session termination failed: {response.status_code}")  # pragma: no cover
         except Exception as exc:  # pragma: no cover
             logger.warning(f"Session termination failed: {exc}")
 

src/mcp/server/auth/handlers/authorize.py

@@ -117,7 +117,7 @@ class AuthorizationHandler:
                     pass
 
             # the error response MUST contain the state specified by the client, if any
-            if state is None:  # pragma: no cover
+            if state is None:
                 # make last-ditch effort to load state
                 state = best_effort_extract_string("state", params)
 

src/mcp/server/auth/middleware/bearer_auth.py

@@ -95,7 +95,7 @@ class RequireAuthMiddleware:
         """Send an authentication error response with WWW-Authenticate header."""
         # Build WWW-Authenticate header value
         www_auth_parts = [f'error="{error}"', f'error_description="{description}"']
-        if self.resource_metadata_url:  # pragma: no cover
+        if self.resource_metadata_url:
             www_auth_parts.append(f'resource_metadata="{self.resource_metadata_url}"')
 
         www_authenticate = f"Bearer {', '.join(www_auth_parts)}"

src/mcp/server/lowlevel/server.py

@@ -349,12 +349,12 @@ class Server(Generic[LifespanResultT]):
         Raises:
             RuntimeError: If called before streamable_http_app() has been called.
         """
-        if self._session_manager is None:  # pragma: no cover
-            raise RuntimeError(
+        if self._session_manager is None:
+            raise RuntimeError(  # pragma: no cover
                 "Session manager can only be accessed after calling streamable_http_app(). "
                 "The session manager is created lazily to avoid unnecessary initialization."
             )
-        return self._session_manager  # pragma: no cover
+        return self._session_manager
 
     async def run(
         self,
@@ -513,7 +513,7 @@ class Server(Generic[LifespanResultT]):
                     if raise_exceptions:  # pragma: no cover
                         raise err
                     response = types.ErrorData(code=0, message=str(err))
-            else:  # pragma: no cover
+            else:
                 response = types.ErrorData(code=types.METHOD_NOT_FOUND, message="Method not found")
 
             if isinstance(response, types.ErrorData) and span is not None:
@@ -603,7 +603,7 @@ class Server(Generic[LifespanResultT]):
         required_scopes: list[str] = []
 
         # Set up auth if configured
-        if auth:  # pragma: no cover
+        if auth:
             required_scopes = auth.required_scopes or []
 
             # Add auth middleware if token verifier is available
@@ -629,10 +629,10 @@ class Server(Generic[LifespanResultT]):
                 )
 
         # Set up routes with or without auth
-        if token_verifier:  # pragma: no cover
+        if token_verifier:
             # Determine resource metadata URL
             resource_metadata_url = None
-            if auth and auth.resource_server_url:
+            if auth and auth.resource_server_url:  # pragma: no branch
                 # Build compliant metadata URL for WWW-Authenticate header
                 resource_metadata_url = build_resource_metadata_url(auth.resource_server_url)
 
@@ -652,7 +652,7 @@ class Server(Generic[LifespanResultT]):
             )
 
         # Add protected resource metadata endpoint if configured as RS
-        if auth and auth.resource_server_url:  # pragma: no cover
+        if auth and auth.resource_server_url:
             routes.extend(
                 create_protected_resource_routes(
                     resource_url=auth.resource_server_url,

src/mcp/server/mcpserver/context.py

@@ -94,7 +94,7 @@ class Context(BaseModel, Generic[LifespanContextT, RequestT]):
         """
         progress_token = self.request_context.meta.get("progress_token") if self.request_context.meta else None
 
-        if progress_token is None:  # pragma: no cover
+        if progress_token is None:
             return
 
         await self.request_context.session.send_progress_notification(
@@ -242,7 +242,7 @@ class Context(BaseModel, Generic[LifespanContextT, RequestT]):
             This is a no-op if not using StreamableHTTP transport with event_store.
             The callback is only available when event_store is configured.
         """
-        if self._request_context and self._request_context.close_sse_stream:  # pragma: no cover
+        if self._request_context and self._request_context.close_sse_stream:  # pragma: no branch
             await self._request_context.close_sse_stream()
 
     async def close_standalone_sse_stream(self) -> None:

src/mcp/server/mcpserver/prompts/base.py

@@ -185,5 +185,5 @@ class Prompt(BaseModel):
                     raise ValueError(f"Could not convert prompt result to message: {msg}")
 
             return messages
-        except Exception as e:  # pragma: no cover
+        except Exception as e:
             raise ValueError(f"Error rendering prompt {self.name}: {e}")

src/mcp/server/mcpserver/server.py

@@ -244,7 +244,7 @@ class MCPServer(Generic[LifespanResultT]):
         Raises:
             RuntimeError: If called before streamable_http_app() has been called.
         """
-        return self._lowlevel_server.session_manager  # pragma: no cover
+        return self._lowlevel_server.session_manager
 
     @overload
     def run(self, transport: Literal["stdio"] = ...) -> None: ...

src/mcp/server/session.py

@@ -223,7 +223,7 @@ class ServerSession(
             related_request_id,
         )
 
-    async def send_resource_updated(self, uri: str | AnyUrl) -> None:  # pragma: no cover
+    async def send_resource_updated(self, uri: str | AnyUrl) -> None:
         """Send a resource updated notification."""
         await self.send_notification(
             types.ResourceUpdatedNotification(
@@ -447,7 +447,7 @@ class ServerSession(
             metadata=ServerMessageMetadata(related_request_id=related_request_id),
         )
 
-    async def send_ping(self) -> types.EmptyResult:  # pragma: no cover
+    async def send_ping(self) -> types.EmptyResult:
         """Send a ping request."""
         return await self.send_request(
             types.PingRequest(),
@@ -479,11 +479,11 @@ class ServerSession(
         """Send a resource list changed notification."""
         await self.send_notification(types.ResourceListChangedNotification())
 
-    async def send_tool_list_changed(self) -> None:  # pragma: no cover
+    async def send_tool_list_changed(self) -> None:
         """Send a tool list changed notification."""
         await self.send_notification(types.ToolListChangedNotification())
 
-    async def send_prompt_list_changed(self) -> None:  # pragma: no cover
+    async def send_prompt_list_changed(self) -> None:
         """Send a prompt list changed notification."""
         await self.send_notification(types.PromptListChangedNotification())
 

src/mcp/server/sse.py

@@ -116,15 +116,15 @@ class SseServerTransport:
         logger.debug(f"SseServerTransport initialized with endpoint: {endpoint}")
 
     @asynccontextmanager
-    async def connect_sse(self, scope: Scope, receive: Receive, send: Send):  # pragma: no cover
-        if scope["type"] != "http":
+    async def connect_sse(self, scope: Scope, receive: Receive, send: Send):
+        if scope["type"] != "http":  # pragma: no cover
             logger.error("connect_sse received non-HTTP request")
             raise ValueError("connect_sse can only handle HTTP requests")
 
         # Validate request headers for DNS rebinding protection
         request = Request(scope, receive)
         error_response = await self._security.validate_request(request, is_post=False)
-        if error_response:
+        if error_response:  # pragma: no cover
             await error_response(scope, receive, send)
             raise ValueError("Request validation failed")
 
@@ -179,6 +179,7 @@ class SseServerTransport:
                 await EventSourceResponse(content=sse_stream_reader, data_sender_callable=sse_writer)(
                     scope, receive, send
                 )
+                await sse_stream_reader.aclose()
                 await read_stream_writer.aclose()
                 await write_stream_reader.aclose()
                 self._read_stream_writers.pop(session_id, None)
@@ -190,13 +191,13 @@ class SseServerTransport:
             logger.debug("Yielding read and write streams")
             yield (read_stream, write_stream)
 
-    async def handle_post_message(self, scope: Scope, receive: Receive, send: Send) -> None:  # pragma: no cover
+    async def handle_post_message(self, scope: Scope, receive: Receive, send: Send) -> None:
         logger.debug("Handling POST message")
         request = Request(scope, receive)
 
         # Validate request headers for DNS rebinding protection
         error_response = await self._security.validate_request(request, is_post=True)
-        if error_response:
+        if error_response:  # pragma: no cover
             return await error_response(scope, receive, send)
 
         session_id_param = request.query_params.get("session_id")
@@ -225,7 +226,7 @@ class SseServerTransport:
         try:
             message = types.jsonrpc_message_adapter.validate_json(body, by_name=False)
             logger.debug(f"Validated client message: {message}")
-        except ValidationError as err:
+        except ValidationError as err:  # pragma: no cover
             logger.exception("Failed to parse message")
             response = Response("Could not parse message", status_code=400)
             await response(scope, receive, send)

src/mcp/server/streamable_http.py

@@ -179,7 +179,7 @@ class StreamableHTTPServerTransport:
         """Check if this transport has been explicitly terminated."""
         return self._terminated
 
-    def close_sse_stream(self, request_id: RequestId) -> None:  # pragma: no cover
+    def close_sse_stream(self, request_id: RequestId) -> None:
         """Close SSE connection for a specific request without terminating the stream.
 
         This method closes the HTTP connection for the specified request, triggering
@@ -198,11 +198,11 @@ class StreamableHTTPServerTransport:
             the disconnect.
         """
         writer = self._sse_stream_writers.pop(request_id, None)
-        if writer:
+        if writer:  # pragma: no branch
             writer.close()
 
         # Also close and remove request streams
-        if request_id in self._request_streams:
+        if request_id in self._request_streams:  # pragma: no branch
             send_stream, receive_stream = self._request_streams.pop(request_id)
             send_stream.close()
             receive_stream.close()
@@ -242,7 +242,7 @@ class StreamableHTTPServerTransport:
         # Only provide close callbacks when client supports resumability
         if self._event_store and protocol_version >= "2025-11-25":
 
-            async def close_stream_callback() -> None:  # pragma: no cover
+            async def close_stream_callback() -> None:
                 self.close_sse_stream(request_id)
 
             async def close_standalone_stream_callback() -> None:  # pragma: no cover
@@ -293,7 +293,7 @@ class StreamableHTTPServerTransport:
     ) -> Response:
         """Create an error response with a simple string message."""
         response_headers = {"Content-Type": CONTENT_TYPE_JSON}
-        if headers:  # pragma: no cover
+        if headers:
             response_headers.update(headers)
 
         if self.mcp_session_id:
@@ -320,10 +320,10 @@ class StreamableHTTPServerTransport:
     ) -> Response:
         """Create a JSON response from a JSONRPCMessage."""
         response_headers = {"Content-Type": CONTENT_TYPE_JSON}
-        if headers:  # pragma: lax no cover
-            response_headers.update(headers)
+        if headers:
+            response_headers.update(headers)  # pragma: no cover
 
-        if self.mcp_session_id:  # pragma: lax no cover
+        if self.mcp_session_id:
             response_headers[MCP_SESSION_ID_HEADER] = self.mcp_session_id
 
         return Response(
@@ -344,7 +344,7 @@ class StreamableHTTPServerTransport:
         }
 
         # If an event ID was provided, include it
-        if event_message.event_id:  # pragma: no cover
+        if event_message.event_id:
             event_data["id"] = event_message.event_id
 
         return event_data
@@ -374,7 +374,7 @@ class StreamableHTTPServerTransport:
             await error_response(scope, receive, send)
             return
 
-        if self._terminated:  # pragma: no cover
+        if self._terminated:
             # If the session has been terminated, return 404 Not Found
             response = self._create_error_response(
                 "Not Found: Session has been terminated",
@@ -389,7 +389,7 @@ class StreamableHTTPServerTransport:
             await self._handle_get_request(request, send)
         elif request.method == "DELETE":
             await self._handle_delete_request(request, send)
-        else:  # pragma: no cover
+        else:
             await self._handle_unsupported_request(request, send)
 
     def _check_accept_headers(self, request: Request) -> tuple[bool, bool]:
@@ -421,7 +421,7 @@ class StreamableHTTPServerTransport:
         has_json, has_sse = self._check_accept_headers(request)
         if self.is_json_response_enabled:
             # For JSON-only responses, only require application/json
-            if not has_json:  # pragma: lax no cover
+            if not has_json:  # pragma: no cover
                 response = self._create_error_response(
                     "Not Acceptable: Client must accept application/json",
                     HTTPStatus.NOT_ACCEPTABLE,
@@ -469,7 +469,7 @@ class StreamableHTTPServerTransport:
 
             try:
                 message = jsonrpc_message_adapter.validate_python(raw_message, by_name=False)
-            except ValidationError as e:  # pragma: no cover
+            except ValidationError as e:
                 response = self._create_error_response(
                     f"Validation error: {str(e)}",
                     HTTPStatus.BAD_REQUEST,
@@ -495,7 +495,7 @@ class StreamableHTTPServerTransport:
                         )
                         await response(scope, receive, send)
                         return
-            elif not await self._validate_request_headers(request, send):  # pragma: no cover
+            elif not await self._validate_request_headers(request, send):
                 return
 
             # For notifications and responses only, return 202 Accepted
@@ -579,7 +579,7 @@ class StreamableHTTPServerTransport:
                 # Store writer reference so close_sse_stream() can close it
                 self._sse_stream_writers[request_id] = sse_stream_writer
 
-                async def sse_writer():  # pragma: lax no cover
+                async def sse_writer():
                     # Get the request ID from the incoming request message
                     try:
                         async with sse_stream_writer, request_stream_reader:
@@ -595,10 +595,10 @@ class StreamableHTTPServerTransport:
                                 # If response, remove from pending streams and close
                                 if isinstance(event_message.message, JSONRPCResponse | JSONRPCError):
                                     break
-                    except anyio.ClosedResourceError:
+                    except anyio.ClosedResourceError:  # pragma: lax no cover
                         # Expected when close_sse_stream() is called
                         logger.debug("SSE stream closed by close_sse_stream()")
-                    except Exception:
+                    except Exception:  # pragma: lax no cover
                         logger.exception("Error in SSE writer")
                     finally:
                         logger.debug("Closing SSE writer")
@@ -628,14 +628,14 @@ class StreamableHTTPServerTransport:
                         # Then send the message to be processed by the server
                         session_message = self._create_session_message(message, request, request_id, protocol_version)
                         await writer.send(session_message)
-                except Exception:  # pragma: no cover
+                except Exception:  # pragma: lax no cover
                     logger.exception("SSE response error")
                     await sse_stream_writer.aclose()
                     await self._clean_up_memory_streams(request_id)
                 finally:
                     await sse_stream_reader.aclose()
 
-        except Exception as err:  # pragma: no cover
+        except Exception as err:
             logger.exception("Error handling POST request")
             response = self._create_error_response(
                 f"Error handling POST request: {err}",
@@ -643,9 +643,9 @@ class StreamableHTTPServerTransport:
                 INTERNAL_ERROR,
             )
             await response(scope, receive, send)
-            if writer:
+            if writer:  # pragma: no cover
                 await writer.send(Exception(err))
-            return
+            return  # pragma: no cover
 
     async def _handle_get_request(self, request: Request, send: Send) -> None:
         """Handle GET request to establish SSE.
@@ -661,7 +661,7 @@ class StreamableHTTPServerTransport:
         # Validate Accept header - must include text/event-stream
         _, has_sse = self._check_accept_headers(request)
 
-        if not has_sse:  # pragma: no cover
+        if not has_sse:
             response = self._create_error_response(
                 "Not Acceptable: Client must accept text/event-stream",
                 HTTPStatus.NOT_ACCEPTABLE,
@@ -673,7 +673,7 @@ class StreamableHTTPServerTransport:
             return
 
         # Handle resumability: check for Last-Event-ID header
-        if last_event_id := request.headers.get(LAST_EVENT_ID_HEADER):  # pragma: no cover
+        if last_event_id := request.headers.get(LAST_EVENT_ID_HEADER):
             await self._replay_events(last_event_id, request, send)
             return
 
@@ -683,11 +683,11 @@ class StreamableHTTPServerTransport:
             "Content-Type": CONTENT_TYPE_SSE,
         }
 
-        if self.mcp_session_id:
+        if self.mcp_session_id:  # pragma: no branch
             headers[MCP_SESSION_ID_HEADER] = self.mcp_session_id
 
         # Check if we already have an active GET stream
-        if GET_STREAM_KEY in self._request_streams:  # pragma: no cover
+        if GET_STREAM_KEY in self._request_streams:
             response = self._create_error_response(
                 "Conflict: Only one SSE stream is allowed per session",
                 HTTPStatus.CONFLICT,
@@ -707,7 +707,7 @@ class StreamableHTTPServerTransport:
 
                 async with sse_stream_writer, standalone_stream_reader:
                     # Process messages from the standalone stream
-                    async for event_message in standalone_stream_reader:  # pragma: lax no cover
+                    async for event_message in standalone_stream_reader:
                         # For the standalone stream, we handle:
                         # - JSONRPCNotification (server sends notifications to client)
                         # - JSONRPCRequest (server sends requests to client)
@@ -716,8 +716,8 @@ class StreamableHTTPServerTransport:
                         # Send the message via SSE
                         event_data = self._create_event_data(event_message)
                         await sse_stream_writer.send(event_data)
-            except Exception:  # pragma: no cover
-                logger.exception("Error in standalone SSE writer")
+            except Exception:
+                logger.exception("Error in standalone SSE writer")  # pragma: no cover
             finally:
                 logger.debug("Closing standalone SSE writer")
                 await self._clean_up_memory_streams(GET_STREAM_KEY)
@@ -775,7 +775,7 @@ class StreamableHTTPServerTransport:
         request_stream_keys = list(self._request_streams.keys())
 
         # Close all request streams asynchronously
-        for key in request_stream_keys:  # pragma: lax no cover
+        for key in request_stream_keys:
             await self._clean_up_memory_streams(key)
 
         # Clear the request streams dictionary immediately
@@ -793,13 +793,13 @@ class StreamableHTTPServerTransport:
             # During cleanup, we catch all exceptions since streams might be in various states
             logger.debug(f"Error closing streams: {e}")
 
-    async def _handle_unsupported_request(self, request: Request, send: Send) -> None:  # pragma: no cover
+    async def _handle_unsupported_request(self, request: Request, send: Send) -> None:
         """Handle unsupported HTTP methods."""
         headers = {
             "Content-Type": CONTENT_TYPE_JSON,
             "Allow": "GET, POST, DELETE",
         }
-        if self.mcp_session_id:
+        if self.mcp_session_id:  # pragma: no branch
             headers[MCP_SESSION_ID_HEADER] = self.mcp_session_id
 
         response = self._create_error_response(
@@ -809,7 +809,7 @@ class StreamableHTTPServerTransport:
         )
         await response(request.scope, request.receive, send)
 
-    async def _validate_request_headers(self, request: Request, send: Send) -> bool:  # pragma: lax no cover
+    async def _validate_request_headers(self, request: Request, send: Send) -> bool:
         if not await self._validate_session(request, send):
             return False
         if not await self._validate_protocol_version(request, send):
@@ -818,7 +818,7 @@ class StreamableHTTPServerTransport:
 
     async def _validate_session(self, request: Request, send: Send) -> bool:
         """Validate the session ID in the request."""
-        if not self.mcp_session_id:  # pragma: no cover
+        if not self.mcp_session_id:
             # If we're not using session IDs, return True
             return True
 
@@ -826,7 +826,7 @@ class StreamableHTTPServerTransport:
         request_session_id = self._get_session_id(request)
 
         # If no session ID provided but required, return error
-        if not request_session_id:  # pragma: no cover
+        if not request_session_id:
             response = self._create_error_response(
                 "Bad Request: Missing session ID",
                 HTTPStatus.BAD_REQUEST,
@@ -851,11 +851,11 @@ class StreamableHTTPServerTransport:
         protocol_version = request.headers.get(MCP_PROTOCOL_VERSION_HEADER)
 
         # If no protocol version provided, assume default version
-        if protocol_version is None:  # pragma: no cover
+        if protocol_version is None:
             protocol_version = DEFAULT_NEGOTIATED_VERSION
 
         # Check if the protocol version is supported
-        if protocol_version not in SUPPORTED_PROTOCOL_VERSIONS:  # pragma: no cover
+        if protocol_version not in SUPPORTED_PROTOCOL_VERSIONS:
             supported_versions = ", ".join(SUPPORTED_PROTOCOL_VERSIONS)
             response = self._create_error_response(
                 f"Bad Request: Unsupported protocol version: {protocol_version}. "
@@ -867,14 +867,14 @@ class StreamableHTTPServerTransport:
 
         return True
 
-    async def _replay_events(self, last_event_id: str, request: Request, send: Send) -> None:  # pragma: no cover
+    async def _replay_events(self, last_event_id: str, request: Request, send: Send) -> None:
         """Replays events that would have been sent after the specified event ID.
 
         Only used when resumability is enabled.
         """
         event_store = self._event_store
         if not event_store:
-            return
+            return  # pragma: no cover
 
         try:
             headers = {
@@ -883,7 +883,7 @@ class StreamableHTTPServerTransport:
                 "Content-Type": CONTENT_TYPE_SSE,
             }
 
-            if self.mcp_session_id:
+            if self.mcp_session_id:  # pragma: no branch
                 headers[MCP_SESSION_ID_HEADER] = self.mcp_session_id
 
             # Get protocol version from header (already validated in _validate_protocol_version)
@@ -904,7 +904,7 @@ class StreamableHTTPServerTransport:
                         stream_id = await event_store.replay_events_after(last_event_id, send_event)
 
                         # If stream ID not in mapping, create it
-                        if stream_id and stream_id not in self._request_streams:
+                        if stream_id and stream_id not in self._request_streams:  # pragma: no branch
                             # Register SSE writer so close_sse_stream() can close it
                             self._sse_stream_writers[stream_id] = sse_stream_writer
 
@@ -921,10 +921,10 @@ class StreamableHTTPServerTransport:
                                     event_data = self._create_event_data(event_message)
 
                                     await sse_stream_writer.send(event_data)
-                except anyio.ClosedResourceError:
+                except anyio.ClosedResourceError:  # pragma: lax no cover
                     # Expected when close_sse_stream() is called
                     logger.debug("Replay SSE stream closed by close_sse_stream()")
-                except Exception:
+                except Exception:  # pragma: lax no cover
                     logger.exception("Error in replay sender")
 
             # Create and start EventSourceResponse
@@ -936,13 +936,13 @@ class StreamableHTTPServerTransport:
 
             try:
                 await response(request.scope, request.receive, send)
-            except Exception:
+            except Exception:  # pragma: lax no cover
                 logger.exception("Error in replay response")
             finally:
                 await sse_stream_writer.aclose()
                 await sse_stream_reader.aclose()
 
-        except Exception:
+        except Exception:  # pragma: lax no cover
             logger.exception("Error replaying events")
             response = self._create_error_response(
                 "Error replaying events",
@@ -993,7 +993,7 @@ class StreamableHTTPServerTransport:
                         if isinstance(message, JSONRPCResponse | JSONRPCError) and message.id is not None:
                             target_request_id = str(message.id)
                         # Extract related_request_id from meta if it exists
-                        elif (  # pragma: no cover
+                        elif (
                             session_message.metadata is not None
                             and isinstance(
                                 session_message.metadata,
@@ -1009,7 +1009,7 @@ class StreamableHTTPServerTransport:
                         # regardless of whether a client is connected
                         # messages will be replayed on the re-connect
                         event_id = None
-                        if self._event_store:  # pragma: lax no cover
+                        if self._event_store:
                             event_id = await self._event_store.store_event(request_stream_id, message)
                             logger.debug(f"Stored {event_id} from {request_stream_id}")
 
@@ -1020,14 +1020,14 @@ class StreamableHTTPServerTransport:
                             except (anyio.BrokenResourceError, anyio.ClosedResourceError):  # pragma: no cover
                                 # Stream might be closed, remove from registry
                                 self._request_streams.pop(request_stream_id, None)
-                        else:  # pragma: no cover
+                        else:
                             logger.debug(
                                 f"""Request stream {request_stream_id} not found
                                 for message. Still processing message as the client
                                 might reconnect and replay."""
                             )
                 except anyio.ClosedResourceError:
-                    if self._terminated:
+                    if self._terminated:  # pragma: lax no cover
                         logger.debug("Read stream closed by client")
                     else:
                         logger.exception("Unexpected closure of read stream in message router")
@@ -1041,7 +1041,7 @@ class StreamableHTTPServerTransport:
                 # Yield the streams for the caller to use
                 yield read_stream, write_stream
             finally:
-                for stream_id in list(self._request_streams.keys()):  # pragma: lax no cover
+                for stream_id in list(self._request_streams.keys()):
                     await self._clean_up_memory_streams(stream_id)
                 self._request_streams.clear()
 

src/mcp/server/streamable_http_manager.py

@@ -173,7 +173,7 @@ class StreamableHTTPSessionManager:
                         self.app.create_initialization_options(),
                         stateless=True,
                     )
-                except Exception:  # pragma: no cover
+                except Exception:  # pragma: lax no cover
                     logger.exception("Stateless session crashed")
 
         # Assert task group is not None for type checking

src/mcp/server/transport_security.py

@@ -40,19 +40,19 @@ class TransportSecurityMiddleware:
         # If not specified, disable DNS rebinding protection by default for backwards compatibility
         self.settings = settings or TransportSecuritySettings(enable_dns_rebinding_protection=False)
 
-    def _validate_host(self, host: str | None) -> bool:  # pragma: no cover
+    def _validate_host(self, host: str | None) -> bool:
         """Validate the Host header against allowed values."""
-        if not host:
+        if not host:  # pragma: no cover
             logger.warning("Missing Host header in request")
             return False
 
         # Check exact match first
-        if host in self.settings.allowed_hosts:
+        if host in self.settings.allowed_hosts:  # pragma: no cover
             return True
 
         # Check wildcard port patterns
         for allowed in self.settings.allowed_hosts:
-            if allowed.endswith(":*"):
+            if allowed.endswith(":*"):  # pragma: no branch
                 # Extract base host from pattern
                 base_host = allowed[:-2]
                 # Check if the actual host starts with base host and has a port
@@ -62,19 +62,19 @@ class TransportSecurityMiddleware:
         logger.warning(f"Invalid Host header: {host}")
         return False
 
-    def _validate_origin(self, origin: str | None) -> bool:  # pragma: no cover
+    def _validate_origin(self, origin: str | None) -> bool:
         """Validate the Origin header against allowed values."""
         # Origin can be absent for same-origin requests
-        if not origin:
+        if not origin:  # pragma: no cover
             return True
 
         # Check exact match first
-        if origin in self.settings.allowed_origins:
+        if origin in self.settings.allowed_origins:  # pragma: no cover
             return True
 
         # Check wildcard port patterns
         for allowed in self.settings.allowed_origins:
-            if allowed.endswith(":*"):
+            if allowed.endswith(":*"):  # pragma: no branch
                 # Extract base origin from pattern
                 base_origin = allowed[:-2]
                 # Check if the actual origin starts with base origin and has a port
@@ -103,14 +103,14 @@ class TransportSecurityMiddleware:
         if not self.settings.enable_dns_rebinding_protection:
             return None
 
-        # Validate Host header  # pragma: no cover
-        host = request.headers.get("host")  # pragma: no cover
-        if not self._validate_host(host):  # pragma: no cover
-            return Response("Invalid Host header", status_code=421)  # pragma: no cover
+        # Validate Host header
+        host = request.headers.get("host")
+        if not self._validate_host(host):
+            return Response("Invalid Host header", status_code=421)
 
-        # Validate Origin header  # pragma: no cover
-        origin = request.headers.get("origin")  # pragma: no cover
-        if not self._validate_origin(origin):  # pragma: no cover
-            return Response("Invalid Origin header", status_code=403)  # pragma: no cover
+        # Validate Origin header
+        origin = request.headers.get("origin")
+        if not self._validate_origin(origin):
+            return Response("Invalid Origin header", status_code=403)
 
-        return None  # pragma: no cover
+        return None

src/mcp/shared/auth.py

@@ -91,9 +91,9 @@ class OAuthClientMetadata(BaseModel):
         requested_scopes = requested_scope.split(" ")
         allowed_scopes = [] if self.scope is None else self.scope.split(" ")
         for scope in requested_scopes:
-            if scope not in allowed_scopes:  # pragma: no branch
+            if scope not in allowed_scopes:
                 raise InvalidScopeError(f"Client was not registered with scope {scope}")
-        return requested_scopes  # pragma: no cover
+        return requested_scopes
 
     def validate_redirect_uri(self, redirect_uri: AnyUrl | None) -> AnyUrl:
         if redirect_uri is not None:

src/mcp/shared/session.py

@@ -451,7 +451,7 @@ class BaseSession(
                     try:
                         await stream.send(JSONRPCError(jsonrpc="2.0", id=id, error=error))
                         await stream.aclose()
-                    except Exception:  # pragma: no cover
+                    except Exception:  # pragma: lax no cover
                         # Stream might already be closed
                         pass
                 self._response_streams.clear()

tests/interaction/README.md

@@ -0,0 +1,228 @@
+# Interaction-model test suite
+
+This suite enumerates the MCP interaction model as end-to-end tests: one test per piece of
+functionality, asserting the full client↔server round trip through the public API. It exists to
+pin the SDK's observable behaviour — every request type, every notification direction, every
+error plane — so that internal rewrites of the send/receive path can be proven equivalent by
+running the suite before and after.
+
+```bash
+uv run --frozen pytest tests/interaction/
+```
+
+The whole suite is in-process and event-driven — including the streamable HTTP, SSE, and OAuth
+flows — with a single subprocess test for stdio.
+
+## Ground rules
+
+- **Public API only.** Tests drive a `Client` connected to a `Server` or `MCPServer`. Nothing
+  reaches into session internals, so the suite keeps working when those internals change.
+  `ClientSession` is used directly only for behaviours `Client` cannot express (skipping
+  initialization, requesting a non-default protocol version).
+- **Pin current behaviour.** Every test passes against the current `main`, including behaviours
+  that diverge from the specification. A failing or xfailed test proves nothing about whether a
+  rewrite preserved behaviour; a passing test that pins the wrong output exactly does. Known
+  divergences are recorded as data on the requirement (see below), not worked around in the test.
+- **Spec-mandated assertions, not implementation quirks.** Error *codes* are asserted against
+  the constants in `mcp.types`; error *message strings* are pinned only where they are the
+  SDK's own deliberate output.
+- **No sleeps, no real I/O.** Concurrency is coordinated with `anyio.Event`; every wait that
+  could hang is bounded by `anyio.fail_after(5)`. The HTTP and OAuth tests drive the Starlette
+  app in-process through the suite's streaming ASGI bridge (`transports/_bridge.py`), which
+  delivers each response chunk as the server produces it — full duplex, but still no sockets,
+  threads, or subprocesses anywhere outside the one stdio test.
+
+## Layout
+
+```text
+tests/interaction/
+  _requirements.py      the requirements manifest (see below)
+  _helpers.py           shared type aliases + the wire-recording transport
+  _connect.py           the transport-parametrized connection factories
+  conftest.py           the connect fixture (the transport matrix)
+  test_coverage.py      enforces the manifest ↔ test contract
+  lowlevel/             one file per feature area, against the low-level Server
+  mcpserver/            the same feature areas in MCPServer's natural idiom
+  transports/           behaviour specific to one transport (sessions, resumability, framing)
+  auth/                 OAuth flows against an in-process authorization server
+```
+
+The two server APIs produce genuinely different wire output for the same conceptual feature
+(`MCPServer` generates schemas, converts exceptions to `isError` results, attaches structured
+content), so they get parallel directories with mirrored file names rather than one parametrized
+test body — each directory pins its flavour's true output exactly.
+
+### The transport matrix
+
+Transport-agnostic tests take the `connect` fixture instead of constructing `Client(server)`
+directly, and therefore run once per transport: over the in-memory transport, over the server's
+real streamable HTTP app driven in-process through the streaming bridge, and over the legacy SSE
+transport the same way. A test connects with `async with connect(server, ...) as client:` and
+asserts the same output on every leg, because the transport is not supposed to change observable
+behaviour. Tests that are tied to one transport do not use the fixture: the wire-recording tests
+(their seam is the in-memory stream pair), the bare-`ClientSession` lifecycle tests, the
+real-clock timeout tests (the timeout machinery is transport-independent and must not race
+transport latency), and everything under `transports/`, which pins behaviour only observable on
+that transport.
+
+A transport conformance test in `transports/` speaks raw `httpx` against the mounted ASGI app
+**only** when its assertion is about HTTP semantics that `Client` cannot observe — status codes,
+response headers, SSE event fields, which stream a message travels on. Any other behaviour is
+asserted through a `Client`, connected to the mounted app via `client_via_http(http)` so several
+clients can share one session manager.
+
+## The requirements manifest
+
+`_requirements.py` maps every behaviour the suite covers to the reason it must hold:
+
+```python
+"tools:call:content:text": Requirement(
+    source=f"{SPEC_BASE_URL}/server/tools#text-content",
+    behavior="tools/call delivers arguments to the tool handler and returns its text content.",
+),
+```
+
+- **`source`** is a deep link into the MCP specification for externally mandated behaviour,
+  the literal string `"sdk"` for behaviour the SDK chose where the spec is silent, or
+  `"issue:#n"` for a regression lock.
+- **`behavior`** describes the *required* behaviour — what the specification (or the SDK's own
+  contract) says should happen. Tests always pin the SDK's current behaviour; where that falls
+  short of `behavior`, the gap is recorded as data rather than hidden in the test.
+- **`divergence`** records that gap for entries whose tests pin the divergent current behaviour.
+- **`deferred`** marks a behaviour that is tracked but has no test in this suite, with a precise
+  reason: the SDK does not implement it, the negative cannot be observed, the assertion is
+  schema-level rather than interaction-level, the feature is experimental (tasks), or the test
+  would require real-time waits the suite refuses.
+- **`transports`** names the transports a behaviour applies to; omitted means transport-independent.
+- **`issue`** carries the tracking link for a recorded gap once one is filed.
+
+Tests link themselves to the manifest with a decorator:
+
+```python
+@requirement("tools:call:content:text")
+async def test_call_tool_returns_text_content() -> None: ...
+```
+
+`test_coverage.py` enforces the contract in both directions: every non-deferred requirement must
+be exercised by at least one test, every deferred requirement by none, and an unknown ID fails at
+import time. A behaviour without a manifest entry cannot be silently half-tested, and a manifest
+entry without a test cannot be silently aspirational.
+
+### The divergence lifecycle
+
+1. A test reveals that the SDK does not do what the spec says. The test pins what the SDK
+   *actually does* and a `Divergence(note=..., issue=...)` goes on the requirement.
+2. When the behaviour is eventually fixed, the pinned test fails. Whoever makes the change finds
+   the divergence note explaining that the old behaviour was a known gap, re-pins the test to the
+   spec-correct output, and deletes the `Divergence`.
+3. An empty divergence list means the SDK is spec-conformant on every behaviour the suite covers.
+
+A requirement may carry both `divergence` and `deferred`: the divergence records that the SDK falls
+short of the spec, and the deferral records why no test pins it (typically because the divergent
+behaviour cannot be driven through the public API). Divergence alone implies a test pins the
+divergent behaviour; divergence plus deferred means the gap is known but unpinned.
+
+This is also the triage key for any rewrite: a test that fails on the new code path either has a
+divergence note (the rewrite accidentally fixed a known gap — decide whether to keep the fix) or
+it does not (the rewrite broke something that was correct — fix the rewrite).
+
+### When a new spec revision is released
+
+1. Update `SPEC_REVISION` and walk the new revision's changelog.
+2. For each changed interaction, find its requirements (the IDs use the wire method strings the
+   changelog speaks in), re-audit the tests against the new text, and update `source` links and
+   assertions where behaviour legitimately changed.
+3. New interactions get new requirements and new tests; removed interactions get their
+   requirements deleted along with their tests.
+4. A behaviour that is correct under both revisions needs no change beyond the `source` link.
+
+## Writing a test
+
+The shortest complete example of the conventions:
+
+```python
+@requirement("tools:call:content:text")
+async def test_call_tool_returns_text_content() -> None:
+    """Arguments reach the tool handler; its content comes back as the call result."""
+
+    async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
+        assert params.name == "add"
+        assert params.arguments is not None
+        return CallToolResult(content=[TextContent(text=str(params.arguments["a"] + params.arguments["b"]))])
+
+    server = Server("adder", on_call_tool=call_tool)
+
+    async with Client(server) as client:
+        result = await client.call_tool("add", {"a": 2, "b": 3})
+
+    assert result == snapshot(CallToolResult(content=[TextContent(text="5")]))
+```
+
+- **The server is defined inside the test** (or in a small fixture at the top of the file when
+  several tests genuinely share it). The whole observable behaviour fits on one screen.
+- **Test names are behaviour sentences** — they state the observable outcome, not the feature
+  being poked. Docstrings add the one or two sentences of context a reviewer needs, including
+  whether the assertion is spec-mandated, SDK-defined, or a known divergence.
+- **Handlers assert their dispatch identity first** (`assert params.name == "add"`), proving the
+  request that arrived is the request the test sent.
+- **The result proves the round trip.** Server-side observations travel back to the test through
+  the protocol itself (a tool returns what it saw) or through a closure-captured list; the test
+  asserts after the call returns.
+- **Order within a test**: server handlers → server construction → client callbacks → connect →
+  act → assert. The test reads in the order the conversation happens.
+- A registered handler or tool that a test never invokes gets a `raise NotImplementedError` body
+  so it cannot silently become load-bearing.
+- A test that needs a peer no real `Server` or `Client` can play (a server that answers initialize
+  with an unsupported version, a client that sends malformed params) plays that side of the wire by
+  hand over `create_client_server_memory_streams()`. This scripted-peer pattern is the suite's only
+  way to drive behaviour the typed API cannot produce, and the docstring of every such test says so.
+
+Stack a second `@requirement` decorator only when a test's natural assertions incidentally prove
+another behaviour — one capabilities snapshot proving four `*:capability:declared` entries, one
+input-schema identity check proving each preserved keyword. Do not build a test around covering
+many requirements at once; if the assertions would be separate, write separate tests.
+
+### Choosing an assertion
+
+| The property under test is… | Assert with |
+|---|---|
+| the result of a transformation (arguments → output, exception → error result) | `result == snapshot(...)` of the full object, so any field the implementation adds or drops fails the test |
+| pass-through of an opaque value (`_meta`, cursors) | identity against the same variable that was sent — a snapshot of a pass-through value only matches the input because a human checked two literals correspond |
+| an error | `pytest.raises(MCPError)` and a snapshot of `exc.value.error` when the message is the SDK's own; a plain `==` on `.code` against the `mcp.types` constant when it is not |
+| third-party output embedded in a result (validation messages) | the stable prefix only — never pin text that changes with a dependency upgrade |
+
+### Notifications and concurrency
+
+The client's receive loop dispatches each incoming message to completion before reading the next,
+and the in-memory transport delivers everything on one ordered stream. Together these guarantee
+that every notification a server handler emits before its response reaches the client callback
+before the originating request returns — so tests collect notifications into a plain list and
+assert after the call, with no synchronisation. The exceptions:
+
+- a notification not triggered by a request the test is awaiting needs an `anyio.Event` set in
+  the receiving handler and awaited under `anyio.fail_after(5)`;
+- the ordering guarantee does not survive transports that split messages across streams (the
+  streamable HTTP standalone GET stream) — see `transports/test_streamable_http.py`.
+
+### Coverage
+
+CI requires 100% line and branch coverage, including `tests/`, and `strict-no-cover` fails the
+build if a line marked `# pragma: no cover` is ever executed. When a new test starts covering a
+pragma'd line in `src/`, delete the pragma in the same change. Do not add new `# type: ignore` or
+`# noqa` comments; restructure instead. Two pragmas are sanctioned in this suite's test code, both
+for known-upstream tracer bugs and only after restructuring has been tried: `# pragma: no branch`
+on a `with`/`async with` line whose only fault is coverage.py mis-tracing the exit arc of a nested
+async context (reserve it for shapes that cannot collapse — a sync `with` adjacent to an
+`async with`); and `# pragma: lax no cover` on a single statement that 3.11's tracer drops because
+the preceding `async with` unwinds via `coro.throw()` (python/cpython#106749, wontfix on 3.11) —
+this hits any test that must run statements after a `ClientSession`/`streamable_http_client` exits
+but still inside an outer `async with`, and no restructure can avoid it.
+
+A handful of `# pragma: lax no cover` markers in `src/` cover teardown exception handlers whose
+execution is timing-dependent under the in-process HTTP bridge — the POST-stream and
+stateless-session `except Exception` handlers in `server/streamable_http*.py`, the `_terminated`
+check in `message_router`, and the response-stream double-close guard in
+`BaseSession._receive_loop`. `strict-no-cover` does not check `lax` lines; do not promote them to
+strict `no cover` without first making the teardown ordering deterministic. The suite also relies
+on a one-line `src/mcp/server/sse.py` fix (`sse_stream_reader.aclose()`) that closes a stream the
+SSE leg would otherwise leak.