Fix iCloud IMAP integration for large mailboxes (#27) (#32)

This commit fixes an issue where iCloud IMAP accounts would report the correct
email count but return empty message lists when fetching emails.

The root cause was that iCloud's IMAP server requires UID-based operations for
large mailboxes and returns metadata-only responses to certain fetch commands.

Changes:
- Switch from sequence-based to UID-based IMAP operations (uid_search, uid fetch)
- Add multiple fetch format attempts (RFC822, BODY[], BODY.PEEK[], etc.) to handle
  different IMAP server implementations
- Detect and handle metadata-only responses from iCloud that contain only
  "FETCH (UID XXXXX)" without actual email content
- Update tests to use UID-based operations

This ensures compatibility with iCloud IMAP accounts while maintaining support
for other IMAP providers.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: Claude <noreply@anthropic.com>

Author: gigq

mcp_email_server/emails/classic.py

@@ -111,12 +111,18 @@ async def get_emails_stream(  # noqa: C901
             await imap.select("INBOX")
 
             search_criteria = self._build_search_criteria(before, since, subject, body, text, from_address, to_address)
-            # Search for messages
-            _, messages = await imap.search(*search_criteria)
             logger.info(f"Get: Search criteria: {search_criteria}")
-            logger.debug(f"Raw messages: {messages}")
-            message_ids = messages[0].split()
-            logger.debug(f"Message IDs: {message_ids}")
+
+            # Search for messages - use UID SEARCH for better compatibility
+            _, messages = await imap.uid_search(*search_criteria)
+
+            # Handle empty or None responses
+            if not messages or not messages[0]:
+                logger.warning("No messages returned from search")
+                message_ids = []
+            else:
+                message_ids = messages[0].split()
+                logger.info(f"Found {len(message_ids)} message IDs")
             start = (page - 1) * page_size
             end = start + page_size
 
@@ -129,20 +135,55 @@ async def get_emails_stream(  # noqa: C901
                     # Convert message_id from bytes to string
                     message_id_str = message_id.decode("utf-8")
 
-                    # Use the string version of the message ID
-                    _, data = await imap.fetch(message_id_str, "RFC822")
+                    # Fetch the email using UID - try different formats for compatibility
+                    data = None
+                    fetch_formats = ["RFC822", "BODY[]", "BODY.PEEK[]", "(BODY.PEEK[])"]
+
+                    for fetch_format in fetch_formats:
+                        try:
+                            _, data = await imap.uid("fetch", message_id_str, fetch_format)
+
+                            if data and len(data) > 0:
+                                # Check if we got actual email content or just metadata
+                                has_content = False
+                                for item in data:
+                                    if (
+                                        isinstance(item, bytes)
+                                        and b"FETCH (" in item
+                                        and b"RFC822" not in item
+                                        and b"BODY" not in item
+                                    ):
+                                        # This is just metadata (like 'FETCH (UID 71998)'), not actual content
+                                        continue
+                                    elif isinstance(item, bytes | bytearray) and len(item) > 100:
+                                        # This looks like email content
+                                        has_content = True
+                                        break
+
+                                if has_content:
+                                    break
+                                else:
+                                    data = None  # Try next format
+
+                        except Exception as e:
+                            logger.debug(f"Fetch format {fetch_format} failed: {e}")
+                            data = None
+
+                    if not data:
+                        logger.error(f"Failed to fetch UID {message_id_str} with any format")
+                        continue
 
                     # Find the email data in the response
                     raw_email = None
 
-                    # The actual email content is in the bytearray at index 1
-                    if len(data) > 1 and isinstance(data[1], bytearray) and len(data[1]) > 0:
+                    # The email content is typically at index 1 as a bytearray
+                    if len(data) > 1 and isinstance(data[1], bytearray):
                         raw_email = bytes(data[1])
                     else:
-                        # Fallback to searching through all items
-                        for _, item in enumerate(data):
+                        # Search through all items for email content
+                        for item in data:
                             if isinstance(item, bytes | bytearray) and len(item) > 100:
-                                # Skip header lines that contain FETCH
+                                # Skip IMAP protocol responses
                                 if isinstance(item, bytes) and b"FETCH" in item:
                                     continue
                                 # This is likely the email content
@@ -220,8 +261,8 @@ async def get_email_count(
             await imap.select("INBOX")
             search_criteria = self._build_search_criteria(before, since, subject, body, text, from_address, to_address)
             logger.info(f"Count: Search criteria: {search_criteria}")
-            # Search for messages and count them
-            _, messages = await imap.search(*search_criteria)
+            # Search for messages and count them - use UID SEARCH for consistency
+            _, messages = await imap.uid_search(*search_criteria)
             return len(messages[0].split())
         finally:
             # Ensure we logout properly

tests/test_email_client.py

@@ -150,7 +150,18 @@ async def test_get_emails_stream(self, email_client):
         mock_imap.login = AsyncMock()
         mock_imap.select = AsyncMock()
         mock_imap.search = AsyncMock(return_value=(None, [b"1 2 3"]))
+        mock_imap.uid_search = AsyncMock(return_value=(None, [b"1 2 3"]))
         mock_imap.fetch = AsyncMock(return_value=(None, [b"HEADER", bytearray(b"EMAIL CONTENT")]))
+        # Create a simple email with headers for testing
+        test_email = b"""From: sender@example.com\r
+To: recipient@example.com\r
+Subject: Test Subject\r
+Date: Mon, 1 Jan 2024 00:00:00 +0000\r
+\r
+This is the email body."""
+        mock_imap.uid = AsyncMock(
+            return_value=(None, [b"1 FETCH (UID 1 RFC822 {%d}" % len(test_email), bytearray(test_email)])
+        )
         mock_imap.logout = AsyncMock()
 
         # Mock IMAP class
@@ -179,8 +190,8 @@ async def test_get_emails_stream(self, email_client):
                     email_client.email_server.user_name, email_client.email_server.password
                 )
                 mock_imap.select.assert_called_once_with("INBOX")
-                mock_imap.search.assert_called_once_with("ALL")
-                assert mock_imap.fetch.call_count == 3
+                mock_imap.uid_search.assert_called_once_with("ALL")
+                assert mock_imap.uid.call_count == 3
                 mock_imap.logout.assert_called_once()
 
     @pytest.mark.asyncio
@@ -194,6 +205,7 @@ async def test_get_email_count(self, email_client):
         mock_imap.login = AsyncMock()
         mock_imap.select = AsyncMock()
         mock_imap.search = AsyncMock(return_value=(None, [b"1 2 3 4 5"]))
+        mock_imap.uid_search = AsyncMock(return_value=(None, [b"1 2 3 4 5"]))
         mock_imap.logout = AsyncMock()
 
         # Mock IMAP class
@@ -207,7 +219,7 @@ async def test_get_email_count(self, email_client):
                 email_client.email_server.user_name, email_client.email_server.password
             )
             mock_imap.select.assert_called_once_with("INBOX")
-            mock_imap.search.assert_called_once_with("ALL")
+            mock_imap.uid_search.assert_called_once_with("ALL")
             mock_imap.logout.assert_called_once()
 
     @pytest.mark.asyncio