Use lax no cover for timing-dependent exception paths; fix ruff format

Change 5 pragma annotations from strict "no cover" to "lax no cover"
since these exception handlers may or may not fire depending on
cancellation timing. Fix ruff-format in test file.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: wiggzz

src/mcp/server/streamable_http.py

@@ -626,14 +626,14 @@ async def sse_writer():  # pragma: lax no cover
                         # Then send the message to be processed by the server
                         session_message = self._create_session_message(message, request, request_id, protocol_version)
                         await writer.send(session_message)
-                except Exception:  # pragma: no cover
+                except Exception:  # pragma: lax no cover
                     logger.exception("SSE response error")
                     await sse_stream_writer.aclose()
                     await self._clean_up_memory_streams(request_id)
                 finally:
                     await sse_stream_reader.aclose()
 
-        except Exception as err:  # pragma: no cover
+        except Exception as err:  # pragma: lax no cover
             logger.exception("Error handling POST request")
             response = self._create_error_response(
                 f"Error handling POST request: {err}",
@@ -816,7 +816,7 @@ async def _validate_request_headers(self, request: Request, send: Send) -> bool:
 
     async def _validate_session(self, request: Request, send: Send) -> bool:
         """Validate the session ID in the request."""
-        if not self.mcp_session_id:  # pragma: no cover
+        if not self.mcp_session_id:  # pragma: lax no cover
             # If we're not using session IDs, return True
             return True
 

src/mcp/server/streamable_http_manager.py

@@ -178,7 +178,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:  # pragma: lax no cover
                     logger.exception("Stateless session crashed")
 
         # Use a request-scoped task group instead of the global one.

src/mcp/shared/session.py

@@ -423,7 +423,7 @@ async def _receive_loop(self) -> None:
                     try:
                         await stream.send(JSONRPCError(jsonrpc="2.0", id=id, error=error))
                         await stream.aclose()
-                    except Exception:  # pragma: no cover
+                    except Exception:  # pragma: lax no cover
                         # Stream might already be closed
                         pass
                 self._response_streams.clear()

tests/server/test_streamable_http_manager.py

@@ -312,12 +312,8 @@ async def handle_call_tool(ctx: ServerRequestContext, params: Any) -> CallToolRe
         session_manager = app._session_manager
 
         async def make_and_abandon_tool_call():
-            async with httpx.AsyncClient(
-                transport=transport, base_url=f"http://{host}", timeout=30.0
-            ) as http_client:
-                async with Client(
-                    streamable_http_client(f"http://{host}/mcp", http_client=http_client)
-                ) as client:
+            async with httpx.AsyncClient(transport=transport, base_url=f"http://{host}", timeout=30.0) as http_client:
+                async with Client(streamable_http_client(f"http://{host}/mcp", http_client=http_client)) as client:
                     # Start tool call — this will block until tool completes
                     # We'll cancel it from outside to simulate disconnect
                     await client.call_tool("slow_tool", {})
@@ -343,10 +339,7 @@ async def make_and_abandon_tool_call():
         await anyio.sleep(0.1)
         leaked = len(session_manager._task_group._tasks)
 
-    assert leaked == 0, (
-        f"Expected 0 lingering tasks but found {leaked}. "
-        f"Stateless request tasks are leaking after client disconnect."
-    )
+    assert leaked == 0, f"Expected 0 lingering tasks but found {leaked}. Stateless request tasks are leaking after client disconnect."
 
 
 @pytest.mark.anyio