fix: remove unnecessary pragmas from src/ (batch 1) and use lax for flaky tests

- Remove unnecessary '# pragma: no cover' from 15 src/ files where
  lines are confirmed covered by CI
- Switch flaky test pragmas to '# pragma: lax no cover' for lines that
  are sometimes covered and sometimes not across CI runs

Author: maxisbey

src/mcp/cli/claude.py

@@ -72,7 +72,7 @@ def update_claude_config(
         )
 
     config_file = config_dir / "claude_desktop_config.json"
-    if not config_file.exists():  # pragma: no cover
+    if not config_file.exists():
         try:
             config_file.write_text("{}")
         except Exception:

src/mcp/cli/cli.py

@@ -77,7 +77,7 @@ def _build_uv_command(
 
     if with_packages:
         for pkg in with_packages:
-            if pkg:  # pragma: no cover
+            if pkg:
                 cmd.extend(["--with", pkg])
 
     # Add mcp run command

src/mcp/client/auth/oauth2.py

@@ -192,7 +192,7 @@ def prepare_token_auth(
             headers = {}  # pragma: no cover
 
         if not self.client_info:
-            return data, headers  # pragma: no cover
+            return data, headers
 
         auth_method = self.client_info.token_endpoint_auth_method
 
@@ -418,7 +418,7 @@ async def _refresh_token(self) -> httpx.Request:
             raise OAuthTokenError("No client info available")  # pragma: no cover
 
         if self.context.oauth_metadata and self.context.oauth_metadata.token_endpoint:
-            token_url = str(self.context.oauth_metadata.token_endpoint)  # pragma: no cover
+            token_url = str(self.context.oauth_metadata.token_endpoint)
         else:
             auth_base_url = self.context.get_authorization_base_url(self.context.server_url)
             token_url = urljoin(auth_base_url, "/token")
@@ -534,7 +534,7 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
                     )
 
                     # Step 2: Discover OAuth Authorization Server Metadata (OASM) (with fallback for legacy servers)
-                    for url in asm_discovery_urls:  # pragma: no cover
+                    for url in asm_discovery_urls:
                         oauth_metadata_request = create_oauth_metadata_request(url)
                         oauth_metadata_response = yield oauth_metadata_request
 

src/mcp/client/session.py

@@ -250,7 +250,7 @@ async def set_logging_level(
         meta: RequestParamsMeta | None = None,
     ) -> types.EmptyResult:
         """Send a logging/setLevel request."""
-        return await self.send_request(  # pragma: no cover
+        return await self.send_request(
             types.SetLevelRequest(params=types.SetLevelRequestParams(level=level, _meta=meta)),
             types.EmptyResult,
         )
@@ -285,14 +285,14 @@ async def read_resource(self, uri: str, *, meta: RequestParamsMeta | None = None
 
     async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
         """Send a resources/subscribe request."""
-        return await self.send_request(  # pragma: no cover
+        return await self.send_request(
             types.SubscribeRequest(params=types.SubscribeRequestParams(uri=uri, _meta=meta)),
             types.EmptyResult,
         )
 
     async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
         """Send a resources/unsubscribe request."""
-        return await self.send_request(  # pragma: no cover
+        return await self.send_request(
             types.UnsubscribeRequest(params=types.UnsubscribeRequestParams(uri=uri, _meta=meta)),
             types.EmptyResult,
         )
@@ -344,7 +344,7 @@ async def _validate_tool_result(self, name: str, result: types.CallToolResult) -
             try:
                 validate(result.structured_content, output_schema)
             except ValidationError as e:
-                raise RuntimeError(f"Invalid structured content returned by tool {name}: {e}")  # pragma: no cover
+                raise RuntimeError(f"Invalid structured content returned by tool {name}: {e}")
             except SchemaError as e:  # pragma: no cover
                 raise RuntimeError(f"Invalid schema for tool {name}: {e}")  # pragma: no cover
 

src/mcp/client/session_group.py

@@ -223,7 +223,7 @@ async def disconnect_from_server(self, session: mcp.ClientSession) -> None:
                 )
             )
 
-        if session_known_for_components:  # pragma: no cover
+        if session_known_for_components:
             component_names = self._sessions.pop(session)  # Pop from _sessions tracking
 
             # Remove prompts associated with the session.

src/mcp/client/sse.py

@@ -119,12 +119,12 @@ async def sse_reader(task_status: TaskStatus[str] = anyio.TASK_STATUS_IGNORED):
                                         await read_stream_writer.send(session_message)
                                     case _:  # pragma: no cover
                                         logger.warning(f"Unknown SSE event: {sse.event}")  # pragma: no cover
-                        except SSEError as sse_exc:  # pragma: no cover
-                            logger.exception("Encountered SSE exception")  # pragma: no cover
-                            raise sse_exc  # pragma: no cover
-                        except Exception as exc:  # pragma: no cover
-                            logger.exception("Error in sse_reader")  # pragma: no cover
-                            await read_stream_writer.send(exc)  # pragma: no cover
+                        except SSEError as sse_exc:
+                            logger.exception("Encountered SSE exception")
+                            raise sse_exc
+                        except Exception as exc:
+                            logger.exception("Error in sse_reader")
+                            await read_stream_writer.send(exc)
                         finally:
                             await read_stream_writer.aclose()
 
@@ -143,8 +143,8 @@ async def post_writer(endpoint_url: str):
                                     )
                                     response.raise_for_status()
                                     logger.debug(f"Client message sent successfully: {response.status_code}")
-                        except Exception:  # pragma: no cover
-                            logger.exception("Error in post_writer")  # pragma: no cover
+                        except Exception:
+                            logger.exception("Error in post_writer")
                         finally:
                             await write_stream.aclose()
 

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

@@ -178,7 +178,7 @@ async def handle(self, request: Request):
                 except TokenError as e:
                     return self.response(TokenErrorResponse(error=e.error, error_description=e.error_description))
 
-            case RefreshTokenRequest():  # pragma: no cover
+            case RefreshTokenRequest():
                 refresh_token = await self.provider.load_refresh_token(client_info, token_request.refresh_token)
                 if refresh_token is None or refresh_token.client_id != token_request.client_id:
                     # if token belongs to different client, pretend it doesn't exist

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

@@ -13,7 +13,7 @@
 
 class AuthenticationError(Exception):
     def __init__(self, message: str):
-        self.message = message  # pragma: no cover
+        self.message = message
 
 
 class ClientAuthenticator:
@@ -98,13 +98,13 @@ async def authenticate_request(self, request: Request) -> OAuthClientInformation
         # that secret
         if client.client_secret:  # pragma: no branch
             if not request_client_secret:
-                raise AuthenticationError("Client secret is required")  # pragma: no cover
+                raise AuthenticationError("Client secret is required")
 
             # hmac.compare_digest requires that both arguments are either bytes or a `str` containing
             # only ASCII characters. Since we do not control `request_client_secret`, we encode both
             # arguments to bytes.
             if not hmac.compare_digest(client.client_secret.encode(), request_client_secret.encode()):
-                raise AuthenticationError("Invalid client_secret")  # pragma: no cover
+                raise AuthenticationError("Invalid client_secret")
 
             if client.client_secret_expires_at and client.client_secret_expires_at < int(time.time()):
                 raise AuthenticationError("Client secret has expired")  # pragma: no cover

src/mcp/server/fastmcp/resources/types.py

@@ -124,7 +124,7 @@ class FileResource(Resource):
 
     @pydantic.field_validator("path")
     @classmethod
-    def validate_absolute_path(cls, path: Path) -> Path:  # pragma: no cover
+    def validate_absolute_path(cls, path: Path) -> Path:
         """Ensure path is absolute."""
         if not path.is_absolute():
             raise ValueError("Path must be absolute")

src/mcp/server/fastmcp/server.py

@@ -873,7 +873,7 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send):  # pragma: no
                     app=RequireAuthMiddleware(sse.handle_post_message, required_scopes, resource_metadata_url),
                 )
             )
-        else:  # pragma: no cover
+        else:
             # Auth is disabled, no need for RequireAuthMiddleware
             # Since handle_sse is an ASGI app, we need to create a compatible endpoint
             async def sse_endpoint(request: Request) -> Response: