refactor: minimal targeted pragmas instead of blanket lax no cover

Remove the lazy approach of downgrading 36 pragmas to `lax no cover`
and instead do the work: strip pragmas entirely, run coverage, apply
minimal annotations to only what's actually uncovered.

Results:
  - 69 `# pragma: no cover` removed entirely — code is fully covered,
    pragmas were subprocess-era lies (coverage.py couldn't see into
    the subprocess, now it sees into the thread)
  + 21 `# pragma: no cover` — genuinely unhit error paths (empty Host
    header, invalid UUID, defensive except Exception handlers, unused
    test tool branches)
  + 20 `# pragma: no branch` — one-sided conditionals where tests only
    exercise one path (if self.mcp_session_id:, if ctx.close_sse_stream:)
  + 2  `# pragma: lax no cover` — timing-race shutdown handlers where
    coverage depends on cancellation-vs-closed-stream ordering.
    Confirmed nondeterministic (2/3 runs pass, 1/3 fail strict-no-cover)

Net pragma delta: -26. Net coverage: +287 stmts, +110 branches now
tracked instead of excluded.

Notable findings along the way:
  - session.py send_resource_updated is 100% covered — pragma removed
  - except clauses are "covered" by exception-type matching walk even
    when they don't match; the pragma goes on the body, not the except
  - transport_security _validate_host/_validate_origin went from
    whole-function excluded to just the 3 untested branches excluded

Author: maxisbey

src/mcp/server/session.py

@@ -222,7 +222,7 @@ async def send_log_message(
             related_request_id,
         )
 
-    async def send_resource_updated(self, uri: str | AnyUrl) -> None:  # pragma: lax no cover
+    async def send_resource_updated(self, uri: str | AnyUrl) -> None:
         """Send a resource updated notification."""
         await self.send_notification(
             types.ResourceUpdatedNotification(

src/mcp/server/sse.py

@@ -116,8 +116,8 @@ 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: lax 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")
 
@@ -195,7 +195,7 @@ 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: lax no cover
+    async def handle_post_message(self, scope: Scope, receive: Receive, send: Send) -> None:
         logger.debug("Handling POST message")
         request = Request(scope, receive)
 
@@ -205,15 +205,15 @@ async def handle_post_message(self, scope: Scope, receive: Receive, send: Send)
             return await error_response(scope, receive, send)
 
         session_id_param = request.query_params.get("session_id")
-        if session_id_param is None:
+        if session_id_param is None:  # pragma: no cover
             logger.warning("Received request without session_id")
             response = Response("session_id is required", status_code=400)
             return await response(scope, receive, send)
 
         try:
             session_id = UUID(hex=session_id_param)
             logger.debug(f"Parsed session ID: {session_id}")
-        except ValueError:
+        except ValueError:  # pragma: no cover
             logger.warning(f"Received invalid session ID: {session_id_param}")
             response = Response("Invalid session ID", status_code=400)
             return await response(scope, receive, send)
@@ -230,7 +230,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

@@ -177,7 +177,7 @@ def is_terminated(self) -> bool:
         """Check if this transport has been explicitly terminated."""
         return self._terminated
 
-    def close_sse_stream(self, request_id: RequestId) -> None:  # pragma: lax no cover
+    def close_sse_stream(self, request_id: RequestId) -> None:
         """Close SSE connection for a specific request without terminating the stream.
 
         This method closes the HTTP connection for the specified request, triggering
@@ -200,12 +200,12 @@ def close_sse_stream(self, request_id: RequestId) -> None:  # pragma: lax no cov
             writer.close()
 
         # Also close and remove request streams
-        if request_id in self._request_streams:
+        if request_id in self._request_streams:  # pragma: no branch
             send_stream, receive_stream = self._request_streams.pop(request_id)
             send_stream.close()
             receive_stream.close()
 
-    def close_standalone_sse_stream(self) -> None:  # pragma: lax no cover
+    def close_standalone_sse_stream(self) -> None:
         """Close the standalone GET SSE stream, triggering client reconnection.
 
         This method closes the HTTP connection for the standalone GET stream used
@@ -240,10 +240,10 @@ def _create_session_message(
         # Only provide close callbacks when client supports resumability
         if self._event_store and protocol_version >= "2025-11-25":
 
-            async def close_stream_callback() -> None:  # pragma: lax no cover
+            async def close_stream_callback() -> None:
                 self.close_sse_stream(request_id)
 
-            async def close_standalone_stream_callback() -> None:  # pragma: lax no cover
+            async def close_standalone_stream_callback() -> None:
                 self.close_standalone_sse_stream()
 
             metadata = ServerMessageMetadata(
@@ -291,7 +291,7 @@ def _create_error_response(
     ) -> Response:
         """Create an error response with a simple string message."""
         response_headers = {"Content-Type": CONTENT_TYPE_JSON}
-        if headers:  # pragma: lax no cover
+        if headers:
             response_headers.update(headers)
 
         if self.mcp_session_id:
@@ -342,7 +342,7 @@ def _create_event_data(self, event_message: EventMessage) -> dict[str, str]:
         }
 
         # If an event ID was provided, include it
-        if event_message.event_id:  # pragma: lax no cover
+        if event_message.event_id:
             event_data["id"] = event_message.event_id
 
         return event_data
@@ -372,7 +372,7 @@ async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> No
             await error_response(scope, receive, send)
             return
 
-        if self._terminated:  # pragma: lax no cover
+        if self._terminated:
             # If the session has been terminated, return 404 Not Found
             response = self._create_error_response(
                 "Not Found: Session has been terminated",
@@ -387,7 +387,7 @@ async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> No
             await self._handle_get_request(request, send)
         elif request.method == "DELETE":
             await self._handle_delete_request(request, send)
-        else:  # pragma: lax no cover
+        else:
             await self._handle_unsupported_request(request, send)
 
     def _check_accept_headers(self, request: Request) -> tuple[bool, bool]:
@@ -467,7 +467,7 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
 
             try:
                 message = jsonrpc_message_adapter.validate_python(raw_message, by_name=False)
-            except ValidationError as e:  # pragma: lax no cover
+            except ValidationError as e:
                 response = self._create_error_response(
                     f"Validation error: {str(e)}",
                     HTTPStatus.BAD_REQUEST,
@@ -493,7 +493,7 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                         )
                         await response(scope, receive, send)
                         return
-            elif not await self._validate_request_headers(request, send):  # pragma: lax no cover
+            elif not await self._validate_request_headers(request, send):
                 return
 
             # For notifications and responses only, return 202 Accepted
@@ -633,7 +633,7 @@ async def sse_writer():  # pragma: lax no cover
                 finally:
                     await sse_stream_reader.aclose()
 
-        except Exception as err:  # pragma: lax no cover
+        except Exception as err:  # pragma: no cover
             logger.exception("Error handling POST request")
             response = self._create_error_response(
                 f"Error handling POST request: {err}",
@@ -659,19 +659,19 @@ async def _handle_get_request(self, request: Request, send: Send) -> None:
         # Validate Accept header - must include text/event-stream
         _, has_sse = self._check_accept_headers(request)
 
-        if not has_sse:  # pragma: lax no cover
+        if not has_sse:
             response = self._create_error_response(
                 "Not Acceptable: Client must accept text/event-stream",
                 HTTPStatus.NOT_ACCEPTABLE,
             )
             await response(request.scope, request.receive, send)
             return
 
-        if not await self._validate_request_headers(request, send):  # pragma: lax no cover
+        if not await self._validate_request_headers(request, send):
             return
 
         # Handle resumability: check for Last-Event-ID header
-        if last_event_id := request.headers.get(LAST_EVENT_ID_HEADER):  # pragma: lax no cover
+        if last_event_id := request.headers.get(LAST_EVENT_ID_HEADER):
             await self._replay_events(last_event_id, request, send)
             return
 
@@ -681,11 +681,11 @@ async def _handle_get_request(self, request: Request, send: Send) -> None:
             "Content-Type": CONTENT_TYPE_SSE,
         }
 
-        if self.mcp_session_id:
+        if self.mcp_session_id:  # pragma: no branch
             headers[MCP_SESSION_ID_HEADER] = self.mcp_session_id
 
         # Check if we already have an active GET stream
-        if GET_STREAM_KEY in self._request_streams:  # pragma: lax no cover
+        if GET_STREAM_KEY in self._request_streams:
             response = self._create_error_response(
                 "Conflict: Only one SSE stream is allowed per session",
                 HTTPStatus.CONFLICT,
@@ -714,8 +714,12 @@ async def standalone_sse_writer():
                         # Send the message via SSE
                         event_data = self._create_event_data(event_message)
                         await sse_stream_writer.send(event_data)
-            except Exception:  # pragma: lax no cover
-                logger.exception("Error in standalone SSE writer")
+            except Exception:
+                # Timing race: if task cancellation reaches this writer before it
+                # hits the closed stream, we get CancelledError (not caught here).
+                # If the closed-stream error fires first, this logs. Neither order is
+                # a bug — both are valid shutdown sequences.
+                logger.exception("Error in standalone SSE writer")  # pragma: lax no cover
             finally:
                 logger.debug("Closing standalone SSE writer")
                 await self._clean_up_memory_streams(GET_STREAM_KEY)
@@ -791,13 +795,13 @@ async def terminate(self) -> None:
             # During cleanup, we catch all exceptions since streams might be in various states
             logger.debug(f"Error closing streams: {e}")
 
-    async def _handle_unsupported_request(self, request: Request, send: Send) -> None:  # pragma: lax no cover
+    async def _handle_unsupported_request(self, request: Request, send: Send) -> None:
         """Handle unsupported HTTP methods."""
         headers = {
             "Content-Type": CONTENT_TYPE_JSON,
             "Allow": "GET, POST, DELETE",
         }
-        if self.mcp_session_id:
+        if self.mcp_session_id:  # pragma: no branch
             headers[MCP_SESSION_ID_HEADER] = self.mcp_session_id
 
         response = self._create_error_response(
@@ -824,7 +828,7 @@ async def _validate_session(self, request: Request, send: Send) -> bool:
         request_session_id = self._get_session_id(request)
 
         # If no session ID provided but required, return error
-        if not request_session_id:  # pragma: lax no cover
+        if not request_session_id:
             response = self._create_error_response(
                 "Bad Request: Missing session ID",
                 HTTPStatus.BAD_REQUEST,
@@ -849,11 +853,11 @@ async def _validate_protocol_version(self, request: Request, send: Send) -> bool
         protocol_version = request.headers.get(MCP_PROTOCOL_VERSION_HEADER)
 
         # If no protocol version provided, assume default version
-        if protocol_version is None:  # pragma: lax no cover
+        if protocol_version is None:
             protocol_version = DEFAULT_NEGOTIATED_VERSION
 
         # Check if the protocol version is supported
-        if protocol_version not in SUPPORTED_PROTOCOL_VERSIONS:  # pragma: lax no cover
+        if protocol_version not in SUPPORTED_PROTOCOL_VERSIONS:
             supported_versions = ", ".join(SUPPORTED_PROTOCOL_VERSIONS)
             response = self._create_error_response(
                 f"Bad Request: Unsupported protocol version: {protocol_version}. "
@@ -865,13 +869,13 @@ async def _validate_protocol_version(self, request: Request, send: Send) -> bool
 
         return True
 
-    async def _replay_events(self, last_event_id: str, request: Request, send: Send) -> None:  # pragma: lax no cover
+    async def _replay_events(self, last_event_id: str, request: Request, send: Send) -> None:
         """Replays events that would have been sent after the specified event ID.
 
         Only used when resumability is enabled.
         """
         event_store = self._event_store
-        if not event_store:
+        if not event_store:  # pragma: no cover
             return
 
         try:
@@ -881,7 +885,7 @@ async def _replay_events(self, last_event_id: str, request: Request, send: Send)
                 "Content-Type": CONTENT_TYPE_SSE,
             }
 
-            if self.mcp_session_id:
+            if self.mcp_session_id:  # pragma: no branch
                 headers[MCP_SESSION_ID_HEADER] = self.mcp_session_id
 
             # Get protocol version from header (already validated in _validate_protocol_version)
@@ -902,7 +906,7 @@ async def send_event(event_message: EventMessage) -> None:
                         stream_id = await event_store.replay_events_after(last_event_id, send_event)
 
                         # If stream ID not in mapping, create it
-                        if stream_id and stream_id not in self._request_streams:
+                        if stream_id and stream_id not in self._request_streams:  # pragma: no branch
                             # Register SSE writer so close_sse_stream() can close it
                             self._sse_stream_writers[stream_id] = sse_stream_writer
 
@@ -921,9 +925,9 @@ async def send_event(event_message: EventMessage) -> None:
                                     await sse_stream_writer.send(event_data)
                 except anyio.ClosedResourceError:
                     # Expected when close_sse_stream() is called
-                    logger.debug("Replay SSE stream closed by close_sse_stream()")
+                    logger.debug("Replay SSE stream closed by close_sse_stream()")  # pragma: no cover
                 except Exception:
-                    logger.exception("Error in replay sender")
+                    logger.exception("Error in replay sender")  # pragma: no cover
 
             # Create and start EventSourceResponse
             response = EventSourceResponse(
@@ -934,13 +938,13 @@ async def send_event(event_message: EventMessage) -> None:
 
             try:
                 await response(request.scope, request.receive, send)
-            except Exception:
+            except Exception:  # pragma: no cover
                 logger.exception("Error in replay response")
             finally:
                 await sse_stream_writer.aclose()
                 await sse_stream_reader.aclose()
 
-        except Exception:
+        except Exception:  # pragma: no cover
             logger.exception("Error replaying events")
             response = self._create_error_response(
                 "Error replaying events",
@@ -991,7 +995,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: lax no cover
+                        elif (
                             session_message.metadata is not None
                             and isinstance(
                                 session_message.metadata,
@@ -1015,10 +1019,16 @@ async def message_router():
                             try:
                                 # Send both the message and the event ID
                                 await self._request_streams[request_stream_id][0].send(EventMessage(message, event_id))
-                            except (anyio.BrokenResourceError, anyio.ClosedResourceError):  # pragma: lax no cover
+                            except (  # pragma: lax no cover
+                                # Timing race: if cancellation reaches this coroutine
+                                # before send() hits the closed stream, CancelledError
+                                # propagates instead. Either shutdown sequence is valid.
+                                anyio.BrokenResourceError,
+                                anyio.ClosedResourceError,
+                            ):
                                 # Stream might be closed, remove from registry
                                 self._request_streams.pop(request_stream_id, None)
-                        else:  # pragma: lax no cover
+                        else:
                             logger.debug(
                                 f"""Request stream {request_stream_id} not found
                                 for message. Still processing message as the client