Apply ruff formatting

Author: pja-ant

src/mcp/client/sse.py

@@ -20,18 +20,19 @@
 
 def remove_request_params(url: str) -> str:
     return urljoin(url, urlparse(url).path)
-        
+
+
 async def compliant_aiter_sse(event_source: EventSource) -> AsyncIterator[ServerSentEvent]:
     """
     Safely iterate over SSE events, working around httpx issue where U+2028 and U+2029
     are incorrectly treated as newlines, breaking SSE stream parsing.
-    
+
     This function replaces event_source.aiter_sse() to handle these Unicode characters
     correctly by processing the raw byte stream and only splitting on actual newlines.
-    
+
     Args:
         event_source: The EventSource to iterate over
-        
+
     Yields:
         ServerSentEvent objects parsed from the stream
     """
@@ -44,10 +45,10 @@ async def compliant_aiter_sse(event_source: EventSource) -> AsyncIterator[Server
     # Note: this is tricky, because we could have a "\r" at the end of a chunk and not yet
     # know if the next chunk starts with a "\n" or not.
     skip_leading_lf = False
-    
+
     async for chunk in event_source.response.aiter_bytes():
         buffer += chunk
-        
+
         while len(buffer) != 0:
             if skip_leading_lf and buffer.startswith(b"\n"):
                 buffer = buffer[1:]
@@ -63,21 +64,21 @@ async def compliant_aiter_sse(event_source: EventSource) -> AsyncIterator[Server
                 break
 
             line_bytes = buffer[:pos]
-            buffer = buffer[pos + 1:]
+            buffer = buffer[pos + 1 :]
 
             # If we have a CR first, skip any LF immediately after (may be in next chunk)
-            skip_leading_lf = (pos == cr)
+            skip_leading_lf = pos == cr
 
-            line = line_bytes.decode('utf-8', errors='replace')
+            line = line_bytes.decode("utf-8", errors="replace")
             sse = decoder.decode(line)
             if sse is not None:
                 yield sse
-    
+
     # Process any remaining data in buffer
     if buffer:
         assert b"\n" not in buffer
         assert b"\r" not in buffer
-        line = buffer.decode('utf-8', errors='replace')
+        line = buffer.decode("utf-8", errors="replace")
         sse = decoder.decode(line)
         if sse is not None:
             yield sse

tests/client/test_sse_unicode.py

@@ -16,146 +16,146 @@ def create_mock_event_source(data_chunks: list[bytes]) -> EventSource:
     event_source = MagicMock(spec=EventSource)
     response = AsyncMock()
     event_source.response = response
-    
+
     async def mock_aiter_bytes() -> AsyncIterator[bytes]:
         for chunk in data_chunks:
             yield chunk
-    
+
     response.aiter_bytes = mock_aiter_bytes
     return event_source
 
 
 async def test_compliant_aiter_sse_handles_unicode_line_separators():
     """Test that compliant_aiter_sse correctly handles U+2028 and U+2029 characters."""
-    
+
     # Simulate SSE data with U+2028 in JSON content
     # The server sends: event: message\ndata: {"text":"Hello\u2028World"}\n\n
     test_data = [
-        b'event: message\n',
+        b"event: message\n",
         b'data: {"text":"Hello',
-        b'\xe2\x80\xa8',  # UTF-8 encoding of U+2028
+        b"\xe2\x80\xa8",  # UTF-8 encoding of U+2028
         b'World"}\n',
-        b'\n',
+        b"\n",
     ]
-    
+
     event_source = create_mock_event_source(test_data)
-    
+
     # Collect events
     events = [event async for event in compliant_aiter_sse(event_source)]
-    
+
     # Should receive one message event
     assert len(events) == 1
     assert events[0].event == "message"
     # The U+2028 should be preserved in the data
-    assert '\u2028' in events[0].data
+    assert "\u2028" in events[0].data
     assert events[0].data == '{"text":"Hello\u2028World"}'
 
 
 async def test_compliant_aiter_sse_handles_paragraph_separator():
     """Test that compliant_aiter_sse correctly handles U+2029 (PARAGRAPH SEPARATOR)."""
-    
+
     # Simulate SSE data with U+2029
     test_data = [
-        b'event: test\ndata: Line1',
-        b'\xe2\x80\xa9',  # UTF-8 encoding of U+2029
-        b'Line2\n\n',
+        b"event: test\ndata: Line1",
+        b"\xe2\x80\xa9",  # UTF-8 encoding of U+2029
+        b"Line2\n\n",
     ]
-    
+
     event_source = create_mock_event_source(test_data)
-    
+
     events = [event async for event in compliant_aiter_sse(event_source)]
-    
+
     assert len(events) == 1
     assert events[0].event == "test"
     # U+2029 should be preserved, not treated as a newline
-    assert '\u2029' in events[0].data
-    assert events[0].data == 'Line1\u2029Line2'
+    assert "\u2029" in events[0].data
+    assert events[0].data == "Line1\u2029Line2"
 
 
 async def test_compliant_aiter_sse_handles_crlf():
     """Test that compliant_aiter_sse correctly handles \\r\\n line endings."""
-    
+
     # Simulate SSE data with CRLF line endings
     test_data = [
-        b'event: message\r\n',
-        b'data: test data\r\n',
-        b'\r\n',
+        b"event: message\r\n",
+        b"data: test data\r\n",
+        b"\r\n",
     ]
-    
+
     event_source = create_mock_event_source(test_data)
-    
+
     events = [event async for event in compliant_aiter_sse(event_source)]
-    
+
     assert len(events) == 1
     assert events[0].event == "message"
     assert events[0].data == "test data"
 
 
 async def test_compliant_aiter_sse_handles_split_utf8():
     """Test that compliant_aiter_sse handles UTF-8 characters split across chunks."""
-    
+
     # Split a UTF-8 emoji (🎉 = \xf0\x9f\x8e\x89) across chunks
     test_data = [
-        b'event: message\n',
-        b'data: Party ',
-        b'\xf0\x9f',  # First half of emoji
-        b'\x8e\x89',  # Second half of emoji
-        b' time!\n\n',
+        b"event: message\n",
+        b"data: Party ",
+        b"\xf0\x9f",  # First half of emoji
+        b"\x8e\x89",  # Second half of emoji
+        b" time!\n\n",
     ]
-    
+
     event_source = create_mock_event_source(test_data)
-    
+
     events = [event async for event in compliant_aiter_sse(event_source)]
-    
+
     assert len(events) == 1
     assert events[0].event == "message"
     assert events[0].data == "Party 🎉 time!"
 
 
 async def test_compliant_aiter_sse_handles_multiple_events():
     """Test that compliant_aiter_sse correctly handles multiple SSE events."""
-    
+
     # Multiple events with problematic Unicode
     test_data = [
-        b'event: first\ndata: Hello\xe2\x80\xa8World\n\n',
-        b'event: second\ndata: Test\xe2\x80\xa9Data\n\n',
-        b'data: No event name\n\n',
+        b"event: first\ndata: Hello\xe2\x80\xa8World\n\n",
+        b"event: second\ndata: Test\xe2\x80\xa9Data\n\n",
+        b"data: No event name\n\n",
     ]
-    
+
     event_source = create_mock_event_source(test_data)
-    
+
     events = [event async for event in compliant_aiter_sse(event_source)]
-    
+
     assert len(events) == 3
-    
+
     assert events[0].event == "first"
-    assert '\u2028' in events[0].data
-    
+    assert "\u2028" in events[0].data
+
     assert events[1].event == "second"
-    assert '\u2029' in events[1].data
-    
+    assert "\u2029" in events[1].data
+
     # Default event type is "message"
     assert events[2].event == "message"
     assert events[2].data == "No event name"
 
 
 async def test_compliant_aiter_sse_handles_split_crlf():
     """Test that \r at end of chunk followed by \n in next chunk is treated as one newline."""
-    
+
     # Test case where \r is at the end of one chunk and \n starts the next
     # This should be treated as a single CRLF line ending, not two separate newlines
     test_data = [
-        b'event: test\r',  # \r at end of chunk
-        b'\ndata: line1\r',  # \n at start of next chunk, then another \r at end
-        b'\ndata: line2\n\n',  # \n at start, completing the CRLF
+        b"event: test\r",  # \r at end of chunk
+        b"\ndata: line1\r",  # \n at start of next chunk, then another \r at end
+        b"\ndata: line2\n\n",  # \n at start, completing the CRLF
     ]
-    
+
     event_source = create_mock_event_source(test_data)
-    
+
     events = [event async for event in compliant_aiter_sse(event_source)]
-    
+
     # Should get exactly one event with both data lines
     assert len(events) == 1
     assert events[0].event == "test"
     # The SSE decoder concatenates multiple data fields with \n
-    assert events[0].data == "line1\nline2"
+    assert events[0].data == "line1\nline2"

tests/issues/test_1356_sse_parsing_line_separator.py

@@ -78,9 +78,7 @@ async def handle_sse(request: Request) -> Response:
 def run_problematic_server(server_port: int) -> None:
     """Run the problematic Unicode test server."""
     app = make_problematic_server_app()
-    server = uvicorn.Server(
-        config=uvicorn.Config(app=app, host="127.0.0.1", port=server_port, log_level="error")
-    )
+    server = uvicorn.Server(config=uvicorn.Config(app=app, host="127.0.0.1", port=server_port, log_level="error"))
     server.run()
 
 
@@ -123,7 +121,7 @@ def problematic_server(problematic_server_port: int) -> Generator[str, None, Non
 
 async def test_json_parsing_with_problematic_unicode(problematic_server: str) -> None:
     """Test that special Unicode characters like U+2028 are handled properly.
-    
+
     This test reproduces issue #1356 where special Unicode characters
     cause JSON parsing to fail and the raw exception is sent to the stream,
     preventing proper error handling.
@@ -137,20 +135,20 @@ async def test_json_parsing_with_problematic_unicode(problematic_server: str) ->
 
             # Call the tool that returns problematic Unicode
             # This should succeed and not hang
-            
+
             # Use a timeout to detect if we're hanging
             with anyio.fail_after(5):  # 5 second timeout
                 try:
                     response = await session.call_tool("get_problematic_unicode", {})
-                    
+
                     # If we get here, the Unicode was handled properly
                     assert len(response.content) == 1
                     text_content = response.content[0]
                     assert hasattr(text_content, "text"), f"Response doesn't have text: {text_content}"
-                    
+
                     expected = "This text contains a line separator\u2028character that may break JSON parsing"
                     assert text_content.text == expected, f"Expected: {expected!r}, Got: {text_content.text!r}"
-                    
+
                 except McpError:
                     pytest.fail("Unexpected error with tool call")
                 except TimeoutError: