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

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 @@ async def _perform_authorization_code_grant(self) -> tuple[str, str]:
         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 @@ async def _refresh_token(self) -> httpx.Request:
 
         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 @@ async def _handle_refresh_response(self, response: httpx.Response) -> bool:  # p
             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 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
         """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 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
                     # 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 @@ 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
 
@@ -267,8 +267,8 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
                 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 @@ 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/auth/handlers/authorize.py

@@ -117,7 +117,7 @@ async def error_response(
                     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 @@ async def _send_auth_error(self, send: Send, status_code: int, error: str, descr
         """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 @@ 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:
@@ -603,7 +603,7 @@ def streamable_http_app(
         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 @@ def streamable_http_app(
                 )
 
         # 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 @@ def streamable_http_app(
             )
 
         # 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 @@ 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")
 
@@ -179,6 +179,7 @@ async def response_wrapper(scope: Scope, receive: Receive, send: Send):
                 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 @@ 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 +226,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)