test: interaction-model end-to-end test suite with a requirements manifest

pyproject.toml

@@ -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/client.py

@@ -305,4 +305,4 @@ async def list_tools(self, *, cursor: str | None = None, meta: RequestParamsMeta
     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",
     )
@@ -408,7 +408,7 @@ async def list_tools(self, *, params: types.PaginatedRequestParams | None = None
 
         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 +449,7 @@ async def _received_request(self, responder: RequestResponder[types.ServerReques
                     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/server/lowlevel/server.py

@@ -349,12 +349,12 @@ def session_manager(self) -> StreamableHTTPSessionManager:
         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 @@ async def _handle_request(
                     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:

src/mcp/server/mcpserver/context.py

@@ -94,7 +94,7 @@ async def report_progress(self, progress: float, total: float | None = None, mes
         """
         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(

src/mcp/server/mcpserver/server.py

@@ -244,7 +244,7 @@ def session_manager(self) -> StreamableHTTPSessionManager:
         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 @@ async def send_log_message(
             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 @@ async def elicit_url(
             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 @@ async def send_resource_list_changed(self) -> None:
         """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 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
         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")
 
@@ -190,13 +190,13 @@ async def response_wrapper(scope: Scope, receive: Receive, send: Send):
             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 +225,7 @@ async def handle_post_message(self, scope: Scope, receive: Receive, send: Send)
         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

@@ -374,7 +374,7 @@ async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> No
             await error_response(scope, receive, send)
             return
 
-        if self._terminated:  # pragma: no cover
+        if self._terminated:  # pragma: lax no cover
             # If the session has been terminated, return 404 Not Found
             response = self._create_error_response(
                 "Not Found: Session has been terminated",
@@ -635,7 +635,7 @@ async def sse_writer():  # pragma: lax no cover
                 finally:
                     await sse_stream_reader.aclose()
 
-        except Exception as err:  # pragma: no cover
+        except Exception as err:  # pragma: lax no cover
             logger.exception("Error handling POST request")
             response = self._create_error_response(
                 f"Error handling POST request: {err}",
@@ -818,7 +818,7 @@ async def _validate_request_headers(self, request: Request, send: Send) -> bool:
 
     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
 
@@ -993,7 +993,7 @@ async def message_router():
                         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,