feat(connectors): generic IMAP email connector for any provider (#518)

frontend/src/types/connectors.ts

@@ -120,6 +120,29 @@ export const SOURCE_CATALOG: ConnectorMeta[] = [
       { name: 'password', placeholder: 'App password (xxxx xxxx xxxx xxxx)', type: 'password' },
     ],
   },
+  {
+    connector_id: 'imap',
+    display_name: 'Email (IMAP)',
+    auth_type: 'oauth',
+    category: 'communication',
+    icon: 'Mail',
+    color: 'text-amber-400',
+    description: 'Any IMAP mailbox',
+    unitLabel: 'emails',
+    steps: [
+      {
+        label: 'Generate an app password with your email provider (IMAP must be enabled), then enter your email address and the app password below. The server is detected from your email domain.',
+      },
+    ],
+    troubleshooting: [
+      'Most providers require an app password rather than your normal password.',
+      'Connects over TLS on port 993; providers that only offer STARTTLS on port 143 are not supported yet.',
+    ],
+    inputFields: [
+      { name: 'email', placeholder: 'you@example.com', type: 'text' },
+      { name: 'password', placeholder: 'App password', type: 'password' },
+    ],
+  },
   {
     connector_id: 'slack',
     display_name: 'Slack',

src/openjarvis/cli/serve.py

@@ -295,7 +295,12 @@ def serve(
                     from openjarvis.core.registry import ToolRegistry
                     from openjarvis.tools._stubs import BaseTool
 
-                    _DEFAULT_TOOLS = {"think", "calculator", "web_search"}
+                    _DEFAULT_TOOLS = {
+                        "think",
+                        "calculator",
+                        "web_search",
+                        "knowledge_search",
+                    }
                     configured = config.agent.tools
                     if configured:
                         if isinstance(configured, list):
@@ -319,7 +324,12 @@ def serve(
                         if isinstance(tool_cls, type) and issubclass(
                             tool_cls, BaseTool
                         ):
-                            tools.append(tool_cls())
+                            if name == "knowledge_search":
+                                from openjarvis.connectors.store import KnowledgeStore
+
+                                tools.append(tool_cls(store=KnowledgeStore()))
+                            else:
+                                tools.append(tool_cls())
                         elif isinstance(tool_cls, BaseTool):
                             tools.append(tool_cls)
 
@@ -399,7 +409,12 @@ def serve(
                         from openjarvis.core.registry import ToolRegistry
                         from openjarvis.tools._stubs import BaseTool
 
-                        _DEFAULT_TOOLS = {"think", "calculator", "web_search"}
+                        _DEFAULT_TOOLS = {
+                            "think",
+                            "calculator",
+                            "web_search",
+                            "knowledge_search",
+                        }
                         configured = config.agent.tools
                         if configured:
                             if isinstance(configured, list):
@@ -422,7 +437,14 @@ def serve(
                                 continue
                             _tcls = ToolRegistry.get(_tname)
                             if isinstance(_tcls, type) and issubclass(_tcls, BaseTool):
-                                _channel_tools.append(_tcls())
+                                if _tname == "knowledge_search":
+                                    from openjarvis.connectors.store import (
+                                        KnowledgeStore,
+                                    )
+
+                                    _channel_tools.append(_tcls(store=KnowledgeStore()))
+                                else:
+                                    _channel_tools.append(_tcls())
                             elif isinstance(_tcls, BaseTool):
                                 _channel_tools.append(_tcls)
 

src/openjarvis/connectors/__init__.py

@@ -73,6 +73,11 @@
 except ImportError:
     pass
 
+try:
+    import openjarvis.connectors.imap  # noqa: F401
+except ImportError:
+    pass
+
 try:
     import openjarvis.connectors.gcalendar  # noqa: F401
 except ImportError:

src/openjarvis/connectors/gmail_imap.py

@@ -14,6 +14,7 @@
 from datetime import datetime
 from email.header import decode_header
 from email.utils import parsedate_to_datetime
+from imaplib import IMAP4
 from typing import Iterator, List, Optional
 
 from openjarvis.connectors._stubs import BaseConnector, Document, SyncStatus
@@ -27,17 +28,26 @@
 _DEFAULT_CREDENTIALS_PATH = str(DEFAULT_CONFIG_DIR / "connectors" / "gmail_imap.json")
 
 
-def _decode_subject(raw: str) -> str:
-    """Decode a possibly-encoded email subject header."""
+def _decode_header(raw: object) -> str:
+    """Decode an email header (str or Header) to a safe plain string.
+
+    Real mail can carry raw 8-bit bytes in headers, which makes ``msg.get()``
+    return an ``email.header.Header`` (not JSON serializable) and yields bogus
+    charsets like ``unknown-8bit``. Coerce to a string, replacing what can't be
+    decoded.
+    """
     if not raw:
         return ""
-    decoded_parts = decode_header(raw)
-    return "".join(
-        part.decode(enc or "utf-8", errors="replace")
-        if isinstance(part, bytes)
-        else part
-        for part, enc in decoded_parts
-    )
+    out = []
+    for part, enc in decode_header(raw if isinstance(raw, str) else str(raw)):
+        if isinstance(part, bytes):
+            try:
+                out.append(part.decode(enc or "utf-8", errors="replace"))
+            except LookupError:
+                out.append(part.decode("utf-8", errors="replace"))
+        else:
+            out.append(part)
+    return "".join(out)
 
 
 def _extract_text_body(msg: email_lib.message.Message) -> str:
@@ -139,6 +149,14 @@ def handle_callback(self, code: str) -> None:
                 {"email": "", "password": code.strip()},
             )
 
+    def _open_imap(self) -> "imaplib.IMAP4_SSL":
+        """Open an authenticated, INBOX-selected IMAP connection."""
+        em, pw = self._resolve_credentials()
+        imap = imaplib.IMAP4_SSL(self._imap_host)
+        imap.login(em, pw)
+        imap.select("INBOX", readonly=True)
+        return imap
+
     def sync(
         self,
         *,
@@ -149,15 +167,12 @@ def sync(
         if not em or not pw:
             return
 
-        imap = imaplib.IMAP4_SSL(self._imap_host)
         try:
-            imap.login(em, pw)
-        except imaplib.IMAP4.error as exc:
+            imap = self._open_imap()
+        except (IMAP4.error, OSError) as exc:
             logger.error("IMAP login failed: %s", exc)
             return
 
-        imap.select("INBOX", readonly=True)
-
         # Always SEARCH ALL. IMAP has no native cursor that survives a
         # server restart, so applying the SyncEngine's ``since`` filter
         # during a partial backfill would silently skip the older
@@ -178,22 +193,59 @@ def sync(
             ordered = ordered[: self._max_messages]
         synced = 0
 
-        for mid in ordered:
+        # A dropped TLS connection mid-backfill (e.g. "[SSL: BAD_LENGTH]") is
+        # transient. Reconnect and retry the message instead of aborting the
+        # whole sync, so a large inbox finishes without manual "Retry Sync".
+        # OSError covers ssl.SSLError and socket errors; IMAP4.abort is raised
+        # when the server drops the connection.
+        max_reconnects = 5
+        consecutive_reconnects = 0
+        index = 0
+        while index < len(ordered):
+            mid = ordered[index]
             try:
                 _, msg_data = imap.fetch(mid, "(RFC822)")
                 raw = msg_data[0][1]
                 msg = email_lib.message_from_bytes(raw)
+            except (IMAP4.abort, OSError) as exc:
+                if consecutive_reconnects >= max_reconnects:
+                    logger.error(
+                        "IMAP sync stopping after %d reconnect attempts: %s",
+                        consecutive_reconnects,
+                        exc,
+                    )
+                    break
+                consecutive_reconnects += 1
+                logger.warning(
+                    "IMAP connection dropped (%s); reconnecting %d/%d",
+                    exc,
+                    consecutive_reconnects,
+                    max_reconnects,
+                )
+                try:
+                    imap.logout()
+                except Exception:
+                    pass
+                try:
+                    imap = self._open_imap()
+                except (IMAP4.error, OSError) as reconnect_exc:
+                    logger.error("IMAP reconnect failed: %s", reconnect_exc)
+                    break
+                continue
             except Exception:
+                index += 1
                 continue
 
-            subject = _decode_subject(msg.get("Subject", ""))
-            sender = msg.get("From", "")
-            to = msg.get("To", "")
+            subject = _decode_header(msg.get("Subject", ""))
+            sender = _decode_header(msg.get("From", ""))
+            to = _decode_header(msg.get("To", ""))
             body = _extract_text_body(msg)
             timestamp = _parse_date(msg)
-            message_id = msg.get("Message-ID", mid.decode())
+            message_id = str(msg.get("Message-ID", mid.decode()))
 
             synced += 1
+            index += 1
+            consecutive_reconnects = 0
             yield Document(
                 doc_id=f"gmail:{message_id}",
                 source="gmail",
@@ -203,14 +255,17 @@ def sync(
                 author=sender,
                 participants=[a.strip() for a in (to or "").split(",") if a.strip()],
                 timestamp=timestamp,
-                thread_id=msg.get("In-Reply-To", ""),
+                thread_id=_decode_header(msg.get("In-Reply-To", "")),
                 url="https://mail.google.com/mail/u/0/#inbox",
                 metadata={
                     "message_id": message_id,
                 },
             )
 
-        imap.logout()
+        try:
+            imap.logout()
+        except Exception:
+            pass
         self._items_synced = synced
 
     def sync_status(self) -> SyncStatus:

src/openjarvis/connectors/imap.py

@@ -0,0 +1,98 @@
+"""Generic IMAP email connector for any provider."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Iterator, Optional
+
+from openjarvis.connectors._stubs import Document
+from openjarvis.connectors.gmail_imap import GmailIMAPConnector
+from openjarvis.connectors.oauth import load_tokens, save_tokens
+from openjarvis.core.config import DEFAULT_CONFIG_DIR
+from openjarvis.core.registry import ConnectorRegistry
+
+_DEFAULT_CREDENTIALS_PATH = str(DEFAULT_CONFIG_DIR / "connectors" / "imap.json")
+
+_PROVIDER_HOSTS = {
+    "gmail.com": "imap.gmail.com",
+    "googlemail.com": "imap.gmail.com",
+    "outlook.com": "outlook.office365.com",
+    "hotmail.com": "outlook.office365.com",
+    "live.com": "outlook.office365.com",
+    "icloud.com": "imap.mail.me.com",
+    "me.com": "imap.mail.me.com",
+    "yahoo.com": "imap.mail.yahoo.com",
+    "fastmail.com": "imap.fastmail.com",
+    "aol.com": "imap.aol.com",
+    "zoho.com": "imap.zoho.com",
+    "gmx.com": "imap.gmx.com",
+}
+
+
+def resolve_imap_host(email_address: str) -> str:
+    """Resolve the IMAP host for an email address from its domain."""
+    domain = email_address.rsplit("@", 1)[-1].lower() if "@" in email_address else ""
+    if not domain:
+        return ""
+    return _PROVIDER_HOSTS.get(domain, f"imap.{domain}")
+
+
+@ConnectorRegistry.register("imap")
+class IMAPConnector(GmailIMAPConnector):
+    """Generic IMAP connector for any email provider."""
+
+    connector_id = "imap"
+    display_name = "Email (IMAP)"
+    _default_imap_host = ""
+
+    def __init__(
+        self,
+        email_address: str = "",
+        app_password: str = "",
+        credentials_path: str = "",
+        *,
+        imap_host: str = "",
+        max_messages: Optional[int] = None,
+    ) -> None:
+        super().__init__(
+            email_address,
+            app_password,
+            credentials_path or _DEFAULT_CREDENTIALS_PATH,
+            imap_host=imap_host,
+            max_messages=max_messages,
+        )
+
+    def auth_url(self) -> str:
+        return ""
+
+    def handle_callback(self, code: str) -> None:
+        # code format: "email:password"
+        if ":" in code:
+            email_addr, password = code.split(":", 1)
+            email_addr, password = email_addr.strip(), password.strip()
+        else:
+            email_addr, password = "", code.strip()
+        save_tokens(
+            self._credentials_path,
+            {
+                "email": email_addr,
+                "password": password,
+                "imap_host": resolve_imap_host(email_addr),
+            },
+        )
+
+    def sync(
+        self,
+        *,
+        since: Optional[datetime] = None,
+        cursor: Optional[str] = None,
+    ) -> Iterator[Document]:
+        if not self._imap_host:
+            tokens = load_tokens(self._credentials_path) or {}
+            self._imap_host = tokens.get("imap_host", "") or resolve_imap_host(
+                tokens.get("email", "")
+            )
+        for doc in super().sync(since=since, cursor=cursor):
+            doc.source = self.connector_id
+            doc.doc_id = doc.doc_id.replace("gmail:", f"{self.connector_id}:", 1)
+            yield doc