Removed now unneeded "pragma: no cover"

Author: aabmass

src/mcp/server/lowlevel/server.py

@@ -739,9 +739,7 @@ async def _handle_request(
                 request_data = None
                 close_sse_stream_cb = None
                 close_standalone_sse_stream_cb = None
-                if message.message_metadata is not None and isinstance(
-                    message.message_metadata, ServerMessageMetadata
-                ):  # pragma: no cover
+                if message.message_metadata is not None and isinstance(message.message_metadata, ServerMessageMetadata):
                     request_data = message.message_metadata.request_context
                     close_sse_stream_cb = message.message_metadata.close_sse_stream
                     close_standalone_sse_stream_cb = message.message_metadata.close_standalone_sse_stream

src/mcp/server/streamable_http.py

@@ -91,7 +91,7 @@ async def store_event(self, stream_id: StreamId, message: JSONRPCMessage | None)
         Returns:
             The generated event ID for the stored event
         """
-        pass  # pragma: no cover
+        pass
 
     @abstractmethod
     async def replay_events_after(
@@ -108,7 +108,7 @@ async def replay_events_after(
         Returns:
             The stream ID of the replayed events
         """
-        pass  # pragma: no cover
+        pass
 
 
 class StreamableHTTPServerTransport:
@@ -175,7 +175,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: 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
@@ -203,7 +203,7 @@ def close_sse_stream(self, request_id: RequestId) -> None:  # pragma: no cover
             send_stream.close()
             receive_stream.close()
 
-    def close_standalone_sse_stream(self) -> None:  # pragma: 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
@@ -238,10 +238,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: no cover
+            async def close_stream_callback() -> None:
                 self.close_sse_stream(request_id)
 
-            async def close_standalone_stream_callback() -> None:  # pragma: no cover
+            async def close_standalone_stream_callback() -> None:
                 self.close_standalone_sse_stream()
 
             metadata = ServerMessageMetadata(
@@ -289,7 +289,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: no cover
+        if headers:
             response_headers.update(headers)
 
         if self.mcp_session_id:
@@ -328,11 +328,11 @@ def _create_json_response(
             headers=response_headers,
         )
 
-    def _get_session_id(self, request: Request) -> str | None:  # pragma: no cover
+    def _get_session_id(self, request: Request) -> str | None:
         """Extract the session ID from request headers."""
         return request.headers.get(MCP_SESSION_ID_HEADER)
 
-    def _create_event_data(self, event_message: EventMessage) -> dict[str, str]:  # pragma: no cover
+    def _create_event_data(self, event_message: EventMessage) -> dict[str, str]:
         """Create event data dictionary from an EventMessage."""
         event_data = {
             "event": "message",
@@ -352,7 +352,7 @@ async def _clean_up_memory_streams(self, request_id: RequestId) -> None:
                 # Close the request stream
                 await self._request_streams[request_id][0].aclose()
                 await self._request_streams[request_id][1].aclose()
-            except Exception:  # pragma: no cover
+            except Exception:
                 # During cleanup, we catch all exceptions since streams might be in various states
                 logger.debug("Error closing memory streams - may already be closed")
             finally:
@@ -370,7 +370,7 @@ async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> No
             await error_response(scope, receive, send)
             return
 
-        if self._terminated:  # pragma: 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",
@@ -381,11 +381,11 @@ async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> No
 
         if request.method == "POST":
             await self._handle_post_request(scope, request, receive, send)
-        elif request.method == "GET":  # pragma: no cover
+        elif request.method == "GET":
             await self._handle_get_request(request, send)
-        elif request.method == "DELETE":  # pragma: no cover
+        elif request.method == "DELETE":
             await self._handle_delete_request(request, send)
-        else:  # pragma: no cover
+        else:
             await self._handle_unsupported_request(request, send)
 
     def _check_accept_headers(self, request: Request) -> tuple[bool, bool]:
@@ -430,15 +430,15 @@ async def _validate_accept_header(self, request: Request, scope: Scope, send: Se
     async def _handle_post_request(self, scope: Scope, request: Request, receive: Receive, send: Send) -> None:  # noqa: PLR0915
         """Handle POST requests containing JSON-RPC messages."""
         writer = self._read_stream_writer
-        if writer is None:  # pragma: no cover
+        if writer is None:
             raise ValueError("No read stream writer available. Ensure connect() is called first.")
         try:
             # Validate Accept header
             if not await self._validate_accept_header(request, scope, send):
                 return
 
             # Validate Content-Type
-            if not self._check_content_type(request):  # pragma: no cover
+            if not self._check_content_type(request):
                 response = self._create_error_response(
                     "Unsupported Media Type: Content-Type must be application/json",
                     HTTPStatus.UNSUPPORTED_MEDIA_TYPE,
@@ -458,7 +458,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: no cover
+            except ValidationError as e:
                 response = self._create_error_response(
                     f"Validation error: {str(e)}",
                     HTTPStatus.BAD_REQUEST,
@@ -470,7 +470,7 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
             # Check if this is an initialization request
             is_initialization_request = isinstance(message, JSONRPCRequest) and message.method == "initialize"
 
-            if is_initialization_request:  # pragma: no cover
+            if is_initialization_request:
                 # Check if the server already has an established session
                 if self.mcp_session_id:
                     # Check if request has a session ID
@@ -484,11 +484,11 @@ 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: no cover
+            elif not await self._validate_request_headers(request, send):
                 return
 
             # For notifications and responses only, return 202 Accepted
-            if not isinstance(message, JSONRPCRequest):  # pragma: no cover
+            if not isinstance(message, JSONRPCRequest):
                 # Create response object and send it
                 response = self._create_json_response(
                     None,
@@ -535,23 +535,23 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                             response_message = event_message.message
                             break
                         # For notifications and request, keep waiting
-                        else:  # pragma: no cover
+                        else:
                             logger.debug(f"received: {event_message.message.method}")
 
                     # At this point we should have a response
                     if response_message:
                         # Create JSON response
                         response = self._create_json_response(response_message)
                         await response(scope, receive, send)
-                    else:  # pragma: no cover
+                    else:
                         # This shouldn't happen in normal operation
                         logger.error("No response message received before stream closed")
                         response = self._create_error_response(
                             "Error processing request: No response received",
                             HTTPStatus.INTERNAL_SERVER_ERROR,
                         )
                         await response(scope, receive, send)
-                except Exception:  # pragma: no cover
+                except Exception:
                     logger.exception("Error processing JSON response")
                     response = self._create_error_response(
                         "Error processing request",
@@ -561,7 +561,7 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                     await response(scope, receive, send)
                 finally:
                     await self._clean_up_memory_streams(request_id)
-            else:  # pragma: no cover
+            else:
                 # Create SSE stream
                 sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[dict[str, str]](0)
 
@@ -624,7 +624,7 @@ async def sse_writer():
                 finally:
                     await sse_stream_reader.aclose()
 
-        except Exception as err:  # pragma: no cover
+        except Exception as err:
             logger.exception("Error handling POST request")
             response = self._create_error_response(
                 f"Error handling POST request: {err}",
@@ -636,7 +636,7 @@ async def sse_writer():
                 await writer.send(Exception(err))
             return
 
-    async def _handle_get_request(self, request: Request, send: Send) -> None:  # pragma: no cover
+    async def _handle_get_request(self, request: Request, send: Send) -> None:
         """Handle GET request to establish SSE.
 
         This allows the server to communicate to the client without the client
@@ -728,7 +728,7 @@ async def standalone_sse_writer():
             await sse_stream_writer.aclose()
             await sse_stream_reader.aclose()
 
-    async def _handle_delete_request(self, request: Request, send: Send) -> None:  # pragma: no cover
+    async def _handle_delete_request(self, request: Request, send: Send) -> None:
         """Handle DELETE requests for explicit session termination."""
         # Validate session ID
         if not self.mcp_session_id:
@@ -778,11 +778,11 @@ async def terminate(self) -> None:
                 await self._write_stream_reader.aclose()
             if self._write_stream is not None:  # pragma: no branch
                 await self._write_stream.aclose()
-        except Exception as e:  # pragma: no cover
+        except Exception as e:
             # 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: no cover
+    async def _handle_unsupported_request(self, request: Request, send: Send) -> None:
         """Handle unsupported HTTP methods."""
         headers = {
             "Content-Type": CONTENT_TYPE_JSON,
@@ -798,14 +798,14 @@ async def _handle_unsupported_request(self, request: Request, send: Send) -> Non
         )
         await response(request.scope, request.receive, send)
 
-    async def _validate_request_headers(self, request: Request, send: Send) -> bool:  # pragma: no cover
+    async def _validate_request_headers(self, request: Request, send: Send) -> bool:
         if not await self._validate_session(request, send):
             return False
         if not await self._validate_protocol_version(request, send):
             return False
         return True
 
-    async def _validate_session(self, request: Request, send: Send) -> bool:  # pragma: no cover
+    async def _validate_session(self, request: Request, send: Send) -> bool:
         """Validate the session ID in the request."""
         if not self.mcp_session_id:
             # If we're not using session IDs, return True
@@ -834,7 +834,7 @@ async def _validate_session(self, request: Request, send: Send) -> bool:  # prag
 
         return True
 
-    async def _validate_protocol_version(self, request: Request, send: Send) -> bool:  # pragma: no cover
+    async def _validate_protocol_version(self, request: Request, send: Send) -> bool:
         """Validate the protocol version header in the request."""
         # Get the protocol version from the request headers
         protocol_version = request.headers.get(MCP_PROTOCOL_VERSION_HEADER)
@@ -856,7 +856,7 @@ 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: 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.
         """
@@ -982,7 +982,7 @@ async def message_router():
                             # send it there
                             target_request_id = response_id
                         # Extract related_request_id from meta if it exists
-                        elif (  # pragma: no cover
+                        elif (
                             session_message.metadata is not None
                             and isinstance(
                                 session_message.metadata,
@@ -1006,13 +1006,13 @@ 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 (  # pragma: no cover
+                            except (
                                 anyio.BrokenResourceError,
                                 anyio.ClosedResourceError,
                             ):
                                 # Stream might be closed, remove from registry
                                 self._request_streams.pop(request_stream_id, None)
-                        else:  # pragma: no cover
+                        else:
                             logger.debug(
                                 f"""Request stream {request_stream_id} not found
                                 for message. Still processing message as the client
@@ -1043,6 +1043,6 @@ async def message_router():
                     await read_stream.aclose()
                     await write_stream_reader.aclose()
                     await write_stream.aclose()
-                except Exception as e:  # pragma: no cover
+                except Exception as e:
                     # During cleanup, we catch all exceptions since streams might be in various states
                     logger.debug(f"Error closing streams: {e}")

src/mcp/server/streamable_http_manager.py

@@ -161,7 +161,7 @@ async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STA
                         self.app.create_initialization_options(),
                         stateless=True,
                     )
-                except Exception:  # pragma: no cover
+                except Exception:
                     logger.exception("Stateless session crashed")
 
         # Assert task group is not None for type checking
@@ -181,7 +181,7 @@ async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: S
         request_mcp_session_id = request.headers.get(MCP_SESSION_ID_HEADER)
 
         # Existing session case
-        if request_mcp_session_id is not None and request_mcp_session_id in self._server_instances:  # pragma: no cover
+        if request_mcp_session_id is not None and request_mcp_session_id in self._server_instances:
             transport = self._server_instances[request_mcp_session_id]
             logger.debug("Session already exists, handling request directly")
             await transport.handle_request(scope, receive, send)
@@ -261,5 +261,5 @@ class StreamableHTTPASGIApp:
     def __init__(self, session_manager: StreamableHTTPSessionManager):
         self.session_manager = session_manager
 
-    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:  # pragma: no cover
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         await self.session_manager.handle_request(scope, receive, send)