fix: remove unnecessary pragma: no cover from src/ files

Remove or adjust coverage pragmas in src/ files that were flagged
by strict-no-cover as covering already-tested code.

Changes:
- Remove pragmas from lines/blocks that are fully covered by tests
- Change '# pragma: no cover' to '# pragma: no branch' on lines
  where the condition itself is executed but one branch isn't taken
- Move pragmas from except clause headers to their uncovered bodies
  where the except IS matched but the handler body isn't reached
- Add targeted pragmas to genuinely uncovered code paths (error
  handlers, rarely-hit branches, cleanup paths)

Files modified: 16 src/ files across cli, client, server, and shared

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 branch
+            if pkg:
                 cmd.extend(["--with", pkg])
 
     # Add mcp run command

src/mcp/client/auth/oauth2.py

@@ -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 branch
+                    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_group.py

@@ -223,22 +223,22 @@ async def disconnect_from_server(self, session: mcp.ClientSession) -> None:
                 )
             )
 
-        if session_known_for_components:  # pragma: no branch
+        if session_known_for_components:
             component_names = self._sessions.pop(session)  # Pop from _sessions tracking
 
             # Remove prompts associated with the session.
             for name in component_names.prompts:
-                if name in self._prompts:  # pragma: no branch
+                if name in self._prompts:
                     del self._prompts[name]
             # Remove resources associated with the session.
             for name in component_names.resources:
-                if name in self._resources:  # pragma: no branch
+                if name in self._resources:
                     del self._resources[name]
             # Remove tools associated with the session.
             for name in component_names.tools:
-                if name in self._tools:  # pragma: no branch
+                if name in self._tools:
                     del self._tools[name]
-                if name in self._tool_to_session:  # pragma: no branch
+                if name in self._tool_to_session:
                     del self._tool_to_session[name]
 
         # Clean up the session's resources via its dedicated exit stack

src/mcp/client/sse.py

@@ -119,10 +119,10 @@ 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
+                        except SSEError as sse_exc:
                             logger.exception("Encountered SSE exception")
                             raise sse_exc
-                        except Exception as exc:  # pragma: no cover
+                        except Exception as exc:
                             logger.exception("Error in sse_reader")
                             await read_stream_writer.send(exc)
                         finally:
@@ -143,7 +143,7 @@ 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
+                        except Exception:
                             logger.exception("Error in post_writer")
                         finally:
                             await write_stream.aclose()

src/mcp/client/streamable_http.py

@@ -201,8 +201,8 @@ async def handle_get_stream(self, client: httpx.AsyncClient, read_stream_writer:
                     attempt = 0
 
             except Exception as exc:
-                logger.debug(f"GET stream error: {exc}")
-                attempt += 1
+                logger.debug(f"GET stream error: {exc}")  # pragma: no cover
+                attempt += 1  # pragma: no cover
 
             if attempt >= MAX_RECONNECTION_ATTEMPTS:  # pragma: no cover
                 logger.debug(f"GET stream max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded")
@@ -334,7 +334,7 @@ async def _handle_sse_response(
                     await response.aclose()
                     return  # Normal completion, no reconnect needed
         except Exception as e:
-            logger.debug(f"SSE stream ended: {e}")
+            logger.debug(f"SSE stream ended: {e}")  # pragma: no cover
 
         # Stream ended without response - reconnect if we received an event with ID
         if last_event_id is not None:  # pragma: no branch
@@ -474,18 +474,18 @@ 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:
+        if not self.session_id:  # pragma: no branch
             return
 
         try:
             headers = self._prepare_headers()
             response = await client.delete(self.url, headers=headers)
 
-            if response.status_code == 405:
+            if response.status_code == 405:  # pragma: no branch
                 logger.debug("Server does not allow session termination")
-            elif response.status_code not in (200, 204):
+            elif response.status_code not in (200, 204):  # pragma: no branch
                 logger.warning(f"Session termination failed: {response.status_code}")
-        except Exception as exc:
+        except Exception as exc:  # pragma: no cover
             logger.warning(f"Session termination failed: {exc}")
 
     def get_session_id(self) -> str | None:

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 branch
+            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

@@ -96,7 +96,7 @@ async def authenticate_request(self, request: Request) -> OAuthClientInformation
 
         # If client from the store expects a secret, validate that the request provides
         # that secret
-        if client.client_secret:  # pragma: no branch
+        if client.client_secret:
             if not request_client_secret:
                 raise AuthenticationError("Client secret is required")
 

src/mcp/server/experimental/task_context.py

@@ -317,7 +317,7 @@ async def elicit_url(
             response_data = await resolver.wait()
             await self._store.update_task(self.task_id, status=TASK_STATUS_WORKING)
             return ElicitResult.model_validate(response_data)
-        except anyio.get_cancelled_exc_class():  # pragma: no cover
+        except anyio.get_cancelled_exc_class():
             await self._store.update_task(self.task_id, status=TASK_STATUS_WORKING)
             raise
 
@@ -486,7 +486,7 @@ async def elicit_as_task(
             await self._store.update_task(self.task_id, status=TASK_STATUS_WORKING)
             return result
 
-        except anyio.get_cancelled_exc_class():  # pragma: no cover
+        except anyio.get_cancelled_exc_class():
             await self._store.update_task(self.task_id, status=TASK_STATUS_WORKING)
             raise
 

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: