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",
     )
@@ -337,9 +337,7 @@ async def _validate_tool_result(self, name: str, result: types.CallToolResult) -
             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 @@ 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 +447,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/client/streamable_http.py

@@ -210,7 +210,7 @@ async def handle_get_stream(self, client: httpx.AsyncClient, read_stream_writer:
                     # 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
 
@@ -492,17 +492,17 @@ async def handle_request_async():
 
     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/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(
@@ -237,7 +237,7 @@ async def close_sse_stream(self) -> None:
             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 @@ async def render(
                     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 @@ 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)