feat: Add verify_ssl support for IMAP connections

Author: crionuke

README.md

@@ -73,6 +73,7 @@ You can also configure the email server using environment variables, which is pa
 | `MCP_EMAIL_SERVER_IMAP_HOST`                  | IMAP server host                                  | -             | Yes      |
 | `MCP_EMAIL_SERVER_IMAP_PORT`                  | IMAP server port                                  | `993`         | No       |
 | `MCP_EMAIL_SERVER_IMAP_SSL`                   | Enable IMAP SSL                                   | `true`        | No       |
+| `MCP_EMAIL_SERVER_IMAP_VERIFY_SSL`            | Verify IMAP SSL certificates (disable for self-signed) | `true`        | No       |
 | `MCP_EMAIL_SERVER_SMTP_HOST`                  | SMTP server host                                  | -             | Yes      |
 | `MCP_EMAIL_SERVER_SMTP_PORT`                  | SMTP server port                                  | `465`         | No       |
 | `MCP_EMAIL_SERVER_SMTP_SSL`                   | Enable SMTP SSL                                   | `true`        | No       |
@@ -162,6 +163,7 @@ If you're using a local mail server with self-signed certificates (like ProtonMa
       "command": "uvx",
       "args": ["mcp-email-server@latest", "stdio"],
       "env": {
+        "MCP_EMAIL_SERVER_IMAP_VERIFY_SSL": "false",
         "MCP_EMAIL_SERVER_SMTP_VERIFY_SSL": "false"
       }
     }
@@ -176,6 +178,9 @@ Or in TOML configuration:
 account_name = "protonmail"
 # ... other settings ...
 
+[emails.incoming]
+verify_ssl = false
+
 [emails.outgoing]
 verify_ssl = false
 ```

mcp_email_server/config.py

@@ -102,6 +102,7 @@ def init(
         imap_password: str | None = None,
         imap_port: int = 993,
         imap_ssl: bool = True,
+        imap_verify_ssl: bool = True,
         smtp_port: int = 465,
         smtp_ssl: bool = True,
         smtp_start_ssl: bool = False,
@@ -121,6 +122,7 @@ def init(
                 host=imap_host,
                 port=imap_port,
                 use_ssl=imap_ssl,
+                verify_ssl=imap_verify_ssl,
             ),
             outgoing=EmailServer(
                 user_name=smtp_user_name or user_name,
@@ -148,6 +150,7 @@ def from_env(cls) -> EmailSettings | None:
         - MCP_EMAIL_SERVER_IMAP_HOST
         - MCP_EMAIL_SERVER_IMAP_PORT (default: 993)
         - MCP_EMAIL_SERVER_IMAP_SSL (default: true)
+        - MCP_EMAIL_SERVER_IMAP_VERIFY_SSL (default: true)
         - MCP_EMAIL_SERVER_SMTP_HOST
         - MCP_EMAIL_SERVER_SMTP_PORT (default: 465)
         - MCP_EMAIL_SERVER_SMTP_SSL (default: true)
@@ -185,6 +188,7 @@ def from_env(cls) -> EmailSettings | None:
                 imap_host=imap_host,
                 imap_port=int(os.getenv("MCP_EMAIL_SERVER_IMAP_PORT", "993")),
                 imap_ssl=_parse_bool_env(os.getenv("MCP_EMAIL_SERVER_IMAP_SSL"), True),
+                imap_verify_ssl=_parse_bool_env(os.getenv("MCP_EMAIL_SERVER_IMAP_VERIFY_SSL"), True),
                 smtp_host=smtp_host,
                 smtp_port=int(os.getenv("MCP_EMAIL_SERVER_SMTP_PORT", "465")),
                 smtp_ssl=_parse_bool_env(os.getenv("MCP_EMAIL_SERVER_SMTP_SSL"), True),

mcp_email_server/emails/classic.py

@@ -79,8 +79,8 @@ async def _send_imap_id(imap: aioimaplib.IMAP4 | aioimaplib.IMAP4_SSL) -> None:
         logger.warning(f"IMAP ID command failed: {e!s}")
 
 
-def _create_smtp_ssl_context(verify_ssl: bool) -> ssl.SSLContext | None:
-    """Create SSL context for SMTP connections.
+def _create_ssl_context(verify_ssl: bool) -> ssl.SSLContext | None:
+    """Create SSL context for SMTP/IMAP connections.
 
     Returns None for default verification, or permissive context
     for self-signed certificates when verify_ssl=False.
@@ -93,20 +93,29 @@ def _create_smtp_ssl_context(verify_ssl: bool) -> ssl.SSLContext | None:
     return ctx
 
 
+# Backwards-compatible alias
+_create_smtp_ssl_context = _create_ssl_context
+
+
 class EmailClient:
     def __init__(self, email_server: EmailServer, sender: str | None = None):
         self.email_server = email_server
         self.sender = sender or email_server.user_name
 
         self.imap_class = aioimaplib.IMAP4_SSL if self.email_server.use_ssl else aioimaplib.IMAP4
+        self.imap_ssl_context = _create_ssl_context(self.email_server.verify_ssl)
 
         self.smtp_use_tls = self.email_server.use_ssl
         self.smtp_start_tls = self.email_server.start_ssl
         self.smtp_verify_ssl = self.email_server.verify_ssl
 
+    def _imap_connect(self) -> aioimaplib.IMAP4_SSL | aioimaplib.IMAP4:
+        """Create a new IMAP connection with the configured SSL context."""
+        return self.imap_class(self.email_server.host, self.email_server.port, ssl_context=self.imap_ssl_context)
+
     def _get_smtp_ssl_context(self) -> ssl.SSLContext | None:
         """Get SSL context for SMTP connections based on verify_ssl setting."""
-        return _create_smtp_ssl_context(self.smtp_verify_ssl)
+        return _create_ssl_context(self.smtp_verify_ssl)
 
     @staticmethod
     def _parse_recipients(email_message) -> list[str]:
@@ -403,7 +412,7 @@ async def get_email_count(
         flagged: bool | None = None,
         answered: bool | None = None,
     ) -> int:
-        imap = self.imap_class(self.email_server.host, self.email_server.port)
+        imap = self._imap_connect()
         try:
             # Wait for the connection to be established
             await imap._client_task
@@ -449,7 +458,7 @@ async def get_emails_metadata_stream(
         flagged: bool | None = None,
         answered: bool | None = None,
     ) -> AsyncGenerator[dict[str, Any], None]:
-        imap = self.imap_class(self.email_server.host, self.email_server.port)
+        imap = self._imap_connect()
         try:
             # Wait for the connection to be established
             await imap._client_task
@@ -563,7 +572,7 @@ async def _fetch_email_with_formats(self, imap, email_id: str) -> list | None:
         return None
 
     async def get_email_body_by_id(self, email_id: str, mailbox: str = "INBOX") -> dict[str, Any] | None:
-        imap = self.imap_class(self.email_server.host, self.email_server.port)
+        imap = self._imap_connect()
         try:
             # Wait for the connection to be established
             await imap._client_task
@@ -618,7 +627,7 @@ async def download_attachment(
         Returns:
             A dictionary with download result information.
         """
-        imap = self.imap_class(self.email_server.host, self.email_server.port)
+        imap = self._imap_connect()
         try:
             await imap._client_task
             await imap.wait_hello_from_server()
@@ -854,7 +863,8 @@ async def append_to_sent(
             True if successfully saved, False otherwise
         """
         imap_class = aioimaplib.IMAP4_SSL if incoming_server.use_ssl else aioimaplib.IMAP4
-        imap = imap_class(incoming_server.host, incoming_server.port)
+        imap_ssl_context = _create_ssl_context(incoming_server.verify_ssl)
+        imap = imap_class(incoming_server.host, incoming_server.port, ssl_context=imap_ssl_context)
 
         # Common Sent folder names across different providers
         sent_folder_candidates = [
@@ -928,7 +938,7 @@ async def append_to_sent(
 
     async def delete_emails(self, email_ids: list[str], mailbox: str = "INBOX") -> tuple[list[str], list[str]]:
         """Delete emails by their UIDs. Returns (deleted_ids, failed_ids)."""
-        imap = self.imap_class(self.email_server.host, self.email_server.port)
+        imap = self._imap_connect()
         deleted_ids = []
         failed_ids = []