skald-connectors/connectors/email/email_mcp_server.py

#!/usr/bin/env python3
"""Generic Email MCP server — IMAP + SMTP over stdio (JSON-RPC 2.0).

Works with ANY email provider (Gmail, Outlook/Office 365, iCloud, Yahoo,
Fastmail, self-hosted, corporate…). No Google Cloud Console, no OAuth: just
standard IMAP for reading/organising and SMTP for sending, with an
app-password. Standard library only — no pip install required.

Capabilities (callable as `mcp__email__<tool>`):
  status               — self-check: IMAP + SMTP login and reachability
  list_messages        — list messages in a folder with a Gmail-like query
  get_message          — read a single message by UID (body text + attachments)
  get_thread           — best-effort thread reconstruction (References/Subject)
  list_folders         — list IMAP folders/mailboxes with total/unread counts
  modify_message        — flag/unflag, mark read/unread, move, archive, delete
  send_message         — send an email (in-thread replies + file attachments)
  get_profile          — configured account + INBOX totals
  create_folder        — create a new IMAP folder/mailbox
  download_attachments — save all attachments from a message to disk

Push notifications: a background watcher emits `event/new_email` the moment a
new message lands in INBOX, using IMAP IDLE when the server advertises it and
falling back to 60s polling otherwise — mirroring the Gmail connector's event.

Configuration is read entirely from environment variables (nothing on disk):
  EMAIL_IMAP_HOST      (required)  e.g. imap.gmail.com
  EMAIL_IMAP_PORT      (default 993, IMAP over SSL)
  EMAIL_SMTP_HOST      (required)  e.g. smtp.gmail.com
  EMAIL_SMTP_PORT      (default 465)
  EMAIL_SMTP_SECURITY  (ssl | starttls | plain; default: ssl if port 465 else starttls)
  EMAIL_USERNAME       (required)  usually the full email address
  EMAIL_PASSWORD       (required)  password or provider app-password
  EMAIL_FROM           (optional)  From address; defaults to EMAIL_USERNAME
"""

from __future__ import annotations

import email
import email.utils
import imaplib
import json
import mimetypes
import os
import re
import select
import smtplib
import ssl
import sys
import threading
import time
from email.header import decode_header, make_header
from email.message import EmailMessage
from html.parser import HTMLParser
from typing import Any, Callable


# Log to stderr so stdout stays clean for JSON-RPC.
def log(msg: str) -> None:
    print(f"[email_mcp] {msg}", file=sys.stderr, flush=True)


# Protects all stdout writes (main request thread + push watcher thread).
_stdout_lock = threading.Lock()


# ── Push notifications ──────────────────────────────────────────────────────────

def _emit_notification(method: str, params: dict) -> None:
    """Write a JSON-RPC notification (no id) to stdout."""
    msg = json.dumps({"jsonrpc": "2.0", "method": method, "params": params})
    with _stdout_lock:
        sys.stdout.write(msg + "\n")
        sys.stdout.flush()


# Re-issue IDLE well within the 29-minute ceiling recommended by RFC 2177.
_IDLE_REFRESH_SECS = 20 * 60
_POLL_INTERVAL_SECS = 60
_RECONNECT_DELAY_SECS = 60
_watch_thread: threading.Thread | None = None


def _start_watching() -> None:
    """Spin up the background push watcher (its own dedicated IMAP connection)."""
    global _watch_thread
    cfg = _get_config()
    if cfg is None:
        log(f"Push watcher disabled: {_init_error}")
        return
    _watch_thread = threading.Thread(target=_watch_loop, daemon=True, name="email-watch")
    _watch_thread.start()


def _watch_loop() -> None:
    """Outer reconnect loop: keep a watcher connection alive forever."""
    cfg = _get_config()
    if cfg is None:
        return
    while True:
        conn = None
        try:
            conn = _connect_imap(cfg)
            conn.select("INBOX", readonly=True)
            last_uid = _inbox_uidnext(conn)
            use_idle = _server_has_idle(conn)
            log(f"Push watcher started (mode={'IDLE' if use_idle else 'poll'}, "
                f"next_uid={last_uid}).")
            while True:
                if use_idle:
                    try:
                        _idle_wait(conn, _IDLE_REFRESH_SECS)
                    except Exception as e:
                        log(f"IDLE failed ({e}); falling back to 60s polling.")
                        use_idle = False
                        time.sleep(_POLL_INTERVAL_SECS)
                else:
                    time.sleep(_POLL_INTERVAL_SECS)
                    try:
                        conn.noop()  # refresh the mailbox view before searching
                    except Exception:
                        raise  # drop to the reconnect loop
                last_uid = _emit_new_since(conn, last_uid)
        except Exception as e:
            log(f"Push watcher connection error: {_format_imap_error(e)}; "
                f"reconnecting in {_RECONNECT_DELAY_SECS}s.")
            if conn is not None:
                try:
                    conn.logout()
                except Exception:
                    pass
            time.sleep(_RECONNECT_DELAY_SECS)


def _server_has_idle(conn: imaplib.IMAP4) -> bool:
    try:
        return any(c.upper() == "IDLE" for c in conn.capabilities)
    except Exception:
        return False


def _inbox_uidnext(conn: imaplib.IMAP4) -> int:
    """UID that will be assigned to the next new message (our 'new mail' cursor)."""
    try:
        typ, data = conn.status("INBOX", "(UIDNEXT)")
        if typ == "OK" and data and data[0]:
            m = re.search(rb"UIDNEXT\s+(\d+)", data[0])
            if m:
                return int(m.group(1))
    except Exception:
        pass
    return 1


def _idle_wait(conn: imaplib.IMAP4, timeout: int) -> None:
    """Enter IMAP IDLE and block until the server reports activity or `timeout`.

    Implemented by hand because imaplib only grew a native idle() in Python 3.13.
    We reuse imaplib's tag counter (_new_tag) so its internal state stays
    consistent, and always send DONE + drain the tagged completion before
    returning so the connection is usable for a follow-up UID SEARCH.
    """
    tag = conn._new_tag()  # type: ignore[attr-defined]
    conn.send(tag + b" IDLE\r\n")
    resp = conn.readline()
    if not resp.lstrip().startswith(b"+"):
        raise RuntimeError(f"server refused IDLE: {resp!r}")
    try:
        ready, _, _ = select.select([conn.sock], [], [], timeout)
        if ready:
            conn.readline()  # consume the untagged EXISTS/RECENT push
    finally:
        conn.send(b"DONE\r\n")
        deadline = time.time() + 10
        while time.time() < deadline:
            line = conn.readline()
            if not line or line.startswith(tag):
                break


def _emit_new_since(conn: imaplib.IMAP4, last_uid: int) -> int:
    """Emit event/new_email for every INBOX message with UID >= last_uid."""
    try:
        typ, data = conn.uid("SEARCH", None, f"UID {last_uid}:*")
    except Exception as e:
        log(f"new-mail search failed: {_format_imap_error(e)}")
        return last_uid
    if typ != "OK" or not data or not data[0]:
        return last_uid
    # "UID n:*" always returns the highest message even if its UID < n, so filter.
    uids = sorted(u for u in (int(x) for x in data[0].split()) if u >= last_uid)
    for uid in uids:
        _fetch_and_emit(conn, uid)
        last_uid = uid + 1
    return last_uid


def _fetch_and_emit(conn: imaplib.IMAP4, uid: int) -> None:
    """Fetch a new message's headers and emit an event/new_email notification."""
    try:
        typ, data = conn.uid(
            "FETCH", str(uid),
            "(BODY.PEEK[HEADER.FIELDS (SUBJECT FROM DATE MESSAGE-ID)])",
        )
        if typ != "OK":
            return
        headers = _parse_header_bytes(_first_literal(data))
        _emit_notification("event/new_email", {
            "message_id": str(uid),
            "folder":     "INBOX",
            "thread_id":  headers.get("message-id", ""),
            "subject":    headers.get("subject", "(no subject)"),
            "from":       headers.get("from", "?"),
            "date":       headers.get("date", "?"),
            "snippet":    "",
        })
        log(f"Notification emitted: new email uid={uid} from {headers.get('from', '?')!r}")
    except Exception as e:
        log(f"Failed to emit notification for uid {uid}: {_format_imap_error(e)}")


# ── Configuration (environment only) ─────────────────────────────────────────────

_config: dict | None = None
_config_loaded = False
_init_error: str | None = None


def _get_config() -> dict | None:
    """Load and validate config from environment variables (once)."""
    global _config, _config_loaded, _init_error
    if _config_loaded:
        return _config
    _config_loaded = True

    missing = [k for k in ("EMAIL_IMAP_HOST", "EMAIL_SMTP_HOST", "EMAIL_USERNAME", "EMAIL_PASSWORD")
               if not os.environ.get(k)]
    if missing:
        _init_error = (
            "Missing required environment variable(s): " + ", ".join(missing) + ". "
            "Set EMAIL_IMAP_HOST, EMAIL_SMTP_HOST, EMAIL_USERNAME and EMAIL_PASSWORD "
            "(plus optional EMAIL_IMAP_PORT / EMAIL_SMTP_PORT / EMAIL_SMTP_SECURITY / EMAIL_FROM)."
        )
        log(_init_error)
        return None

    def _int(name: str, default: int) -> int:
        raw = os.environ.get(name)
        if not raw:
            return default
        try:
            return int(raw)
        except ValueError:
            log(f"Invalid {name}={raw!r}; using default {default}.")
            return default

    smtp_port = _int("EMAIL_SMTP_PORT", 465)
    security = (os.environ.get("EMAIL_SMTP_SECURITY") or "").strip().lower()
    if security not in ("ssl", "starttls", "plain"):
        security = "ssl" if smtp_port == 465 else "starttls"

    username = os.environ["EMAIL_USERNAME"]
    _config = {
        "imap_host": os.environ["EMAIL_IMAP_HOST"],
        "imap_port": _int("EMAIL_IMAP_PORT", 993),
        "smtp_host": os.environ["EMAIL_SMTP_HOST"],
        "smtp_port": smtp_port,
        "smtp_security": security,
        "username": username,
        "password": os.environ["EMAIL_PASSWORD"],
        "from_addr": os.environ.get("EMAIL_FROM") or username,
    }
    log(f"Config loaded for {username} (imap {_config['imap_host']}:{_config['imap_port']}, "
        f"smtp {_config['smtp_host']}:{smtp_port}/{security}).")
    return _config


# ── IMAP connection (request thread) ─────────────────────────────────────────────

# Single connection reused by the request-handling thread. The main loop reads
# stdin sequentially and dispatches synchronously, so this is only ever touched
# by one thread; the push watcher keeps its OWN separate connection.
_imap_conn: imaplib.IMAP4 | None = None


def _connect_imap(cfg: dict) -> imaplib.IMAP4_SSL:
    """Open and authenticate a fresh IMAP-over-SSL connection (raises on failure)."""
    conn = imaplib.IMAP4_SSL(cfg["imap_host"], cfg["imap_port"],
                             ssl_context=ssl.create_default_context())
    conn.login(cfg["username"], cfg["password"])
    return conn


def _imap() -> imaplib.IMAP4 | None:
    """Return a healthy IMAP connection for the request thread, or None (+ _init_error)."""
    global _imap_conn, _init_error
    cfg = _get_config()
    if cfg is None:
        return None
    if _imap_conn is not None:
        try:
            _imap_conn.noop()
            return _imap_conn
        except Exception:
            try:
                _imap_conn.logout()
            except Exception:
                pass
            _imap_conn = None
    try:
        _imap_conn = _connect_imap(cfg)
        return _imap_conn
    except Exception as e:
        _init_error = _format_imap_error(e)
        log(_init_error)
        return None


def _select(conn: imaplib.IMAP4, folder: str, readonly: bool = True) -> tuple[bool, str]:
    """SELECT a folder; return (ok, error_message)."""
    typ, data = conn.select(_quote_mailbox(folder), readonly=readonly)
    if typ != "OK":
        detail = data[0].decode("utf-8", "replace") if data and data[0] else "unknown error"
        return False, f"Error: cannot open folder {folder!r}: {detail}"
    return True, ""


def _quote_mailbox(name: str) -> str:
    """Quote a mailbox name for IMAP if it contains spaces/specials."""
    if name and re.fullmatch(r"[A-Za-z0-9_./\-]+", name):
        return name
    return '"' + name.replace("\\", "\\\\").replace('"', '\\"') + '"'


# ── Error mapping ────────────────────────────────────────────────────────────────

def _format_imap_error(e: Exception) -> str:
    """Map an IMAP/socket exception into an actionable Error: string."""
    text = str(e).strip()
    low = text.lower()
    if isinstance(e, imaplib.IMAP4.error):
        if "authentication" in low or "login" in low or "credentials" in low or "auth" in low:
            return ("Error: IMAP login was rejected. Check EMAIL_USERNAME / EMAIL_PASSWORD — "
                    "most providers require an app-specific password (not your normal login "
                    "password) and IMAP to be enabled in account settings.")
        return f"Error: IMAP error: {text}"
    if isinstance(e, (TimeoutError, ConnectionError, OSError)):
        return (f"Error: could not reach the IMAP server ({text}). Check EMAIL_IMAP_HOST / "
                "EMAIL_IMAP_PORT and your network.")
    return f"Error: IMAP call failed: {text}"


def _format_smtp_error(e: Exception) -> str:
    text = str(e).strip()
    if isinstance(e, smtplib.SMTPAuthenticationError):
        return ("Error: SMTP login was rejected. Check EMAIL_USERNAME / EMAIL_PASSWORD — an "
                "app-specific password is usually required for SMTP too.")
    if isinstance(e, smtplib.SMTPException):
        return f"Error: SMTP error: {text}"
    if isinstance(e, (TimeoutError, ConnectionError, OSError)):
        return (f"Error: could not reach the SMTP server ({text}). Check EMAIL_SMTP_HOST / "
                "EMAIL_SMTP_PORT / EMAIL_SMTP_SECURITY.")
    return f"Error: SMTP call failed: {text}"


def _unavailable() -> str:
    """Error string for tools when config/connection isn't ready (no double 'Error:')."""
    msg = _init_error or "Email connector is not configured."
    return msg if msg.startswith("Error:") else f"Error: {msg}"


def _status_report(icon: str, label: str, kind: str, description: str,
                   steps: list[str] | None = None) -> str:
    lines = [f"Status: {label} {icon} ({kind})", description]
    if steps:
        lines.append("")
        lines.append("What to do:")
        for i, s in enumerate(steps, 1):
            lines.append(f"{i}. {s}")
    return "\n".join(lines)


# ── MIME / parsing helpers ───────────────────────────────────────────────────────

def _decode_hdr(value: str | None) -> str:
    """Decode a possibly MIME-encoded header (=?utf-8?...?=) to a plain string."""
    if not value:
        return ""
    try:
        return str(make_header(decode_header(value)))
    except Exception:
        return value


def _first_literal(data: Any) -> bytes:
    """Return the first literal ({N}-prefixed) byte payload from a FETCH response."""
    if not data:
        return b""
    for item in data:
        if isinstance(item, tuple) and len(item) >= 2 and item[1] is not None:
            return item[1]
    return b""


def _parse_header_bytes(raw: bytes) -> dict[str, str]:
    """Parse RFC822 header bytes into a lowercased, MIME-decoded dict."""
    if not raw:
        return {}
    msg = email.message_from_bytes(raw)
    out: dict[str, str] = {}
    for key in msg.keys():
        out[key.lower()] = _decode_hdr(msg.get(key))
    return out


def _part_text(part: email.message.Message) -> str:
    payload = part.get_payload(decode=True)
    if payload is None:
        return ""
    charset = part.get_content_charset() or "utf-8"
    try:
        return payload.decode(charset, errors="replace")
    except (LookupError, TypeError):
        return payload.decode("utf-8", errors="replace")


def _is_attachment(part: email.message.Message) -> bool:
    disp = str(part.get("Content-Disposition") or "").lower()
    if "attachment" in disp:
        return True
    return bool(part.get_filename())


def _extract_body(msg: email.message.Message) -> tuple[str, bool]:
    """Return (text, from_html). Prefer text/plain; fall back to text/html→text."""
    if msg.is_multipart():
        plain = html = ""
        for part in msg.walk():
            if part.is_multipart() or _is_attachment(part):
                continue
            ctype = part.get_content_type()
            if ctype == "text/plain" and not plain:
                plain = _part_text(part)
            elif ctype == "text/html" and not html:
                html = _part_text(part)
        if plain:
            return plain, False
        if html:
            return _html_to_text(html), True
        return "", False
    text = _part_text(msg)
    if msg.get_content_type() == "text/html":
        return _html_to_text(text), True
    return text, False


def _iter_attachments(msg: email.message.Message):
    for part in msg.walk():
        if part.is_multipart():
            continue
        if _is_attachment(part):
            yield part


class _HTMLTextExtractor(HTMLParser):
    """Collect readable text from HTML, skipping scripts/styles, block newlines."""

    _SKIP = {"script", "style", "head"}
    _BLOCK = {"p", "div", "br", "li", "tr", "h1", "h2", "h3", "h4", "h5", "h6"}

    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self._chunks: list[str] = []
        self._skip_depth = 0

    def handle_starttag(self, tag: str, attrs: Any) -> None:
        if tag in self._SKIP:
            self._skip_depth += 1
        elif tag == "br":
            self._chunks.append("\n")

    def handle_endtag(self, tag: str) -> None:
        if tag in self._SKIP and self._skip_depth:
            self._skip_depth -= 1
        elif tag in self._BLOCK:
            self._chunks.append("\n")

    def handle_data(self, data: str) -> None:
        if not self._skip_depth:
            self._chunks.append(data)

    def get_text(self) -> str:
        return re.sub(r"\n{3,}", "\n\n", "".join(self._chunks)).strip()


def _html_to_text(html_str: str) -> str:
    try:
        parser = _HTMLTextExtractor()
        parser.feed(html_str)
        return parser.get_text()
    except Exception:
        return html_str


# ── Gmail-like search → IMAP SEARCH criteria ─────────────────────────────────────

def _imap_date(value: str) -> str | None:
    """Convert YYYY-MM-DD (or DD-Mon-YYYY) into an IMAP date (01-Jan-2024)."""
    value = value.strip()
    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%d-%b-%Y"):
        try:
            return time.strftime("%d-%b-%Y", time.strptime(value, fmt))
        except ValueError:
            continue
    return None


def _q(s: str) -> str:
    """IMAP quoted-string. imaplib does NOT quote SEARCH args, so we must — any
    value with a space (TEXT/SUBJECT phrases) is otherwise split into atoms."""
    return '"' + s.replace("\\", "\\\\").replace('"', '\\"') + '"'


def _build_search(query: str, unread_only: bool) -> list[str]:
    """Turn a Gmail-ish query into IMAP SEARCH tokens.

    Supported: from:x  to:x  subject:x  since:YYYY-MM-DD  before:YYYY-MM-DD
    unread / is:unread / read / is:read  has:attachment  and free text (→ TEXT).
    """
    criteria: list[str] = []
    free: list[str] = []
    for tok in (query or "").split():
        low = tok.lower()
        if low in ("unread", "is:unread"):
            criteria += ["UNSEEN"]
        elif low in ("read", "is:read"):
            criteria += ["SEEN"]
        elif low in ("has:attachment", "is:attachment"):
            criteria += ["KEYWORD", "attachment"]  # best-effort; not all servers honour it
        elif low.startswith("from:"):
            criteria += ["FROM", _q(tok[5:])]
        elif low.startswith("to:"):
            criteria += ["TO", _q(tok[3:])]
        elif low.startswith("subject:"):
            criteria += ["SUBJECT", _q(tok[8:])]
        elif low.startswith("since:"):
            d = _imap_date(tok[6:])
            if d:
                criteria += ["SINCE", d]
        elif low.startswith("before:"):
            d = _imap_date(tok[7:])
            if d:
                criteria += ["BEFORE", d]
        else:
            free.append(tok)
    if free:
        criteria += ["TEXT", _q(" ".join(free))]
    if unread_only and "UNSEEN" not in criteria:
        criteria += ["UNSEEN"]
    return criteria or ["ALL"]


# ── Tool implementations ─────────────────────────────────────────────────────────

def _email_status(args: dict | None = None) -> str:
    """Self-check: IMAP login + SMTP login both succeed."""
    cfg = _get_config()
    if cfg is None:
        return _status_report("❌", "NOT_CONFIGURED", "action needed",
            _init_error or "Configuration is incomplete.",
            ["Set the required environment variables: EMAIL_IMAP_HOST, EMAIL_SMTP_HOST, "
             "EMAIL_USERNAME, EMAIL_PASSWORD.",
             "Use an app-specific password if your provider requires one (Gmail, iCloud, Yahoo…)."])

    # IMAP probe.
    conn = _imap()
    if conn is None:
        return _status_report("❌", "IMAP_ERROR", "action needed",
            _init_error or "Could not connect to IMAP.",
            ["Verify EMAIL_IMAP_HOST / EMAIL_IMAP_PORT and that IMAP is enabled for the account.",
             "Verify EMAIL_USERNAME / EMAIL_PASSWORD (app password may be required)."])
    try:
        conn.select("INBOX", readonly=True)
    except Exception as e:
        return _status_report("❌", "IMAP_ERROR", "action needed",
            _format_imap_error(e), ["Check IMAP settings and credentials."])

    # SMTP probe (connect + login, then quit).
    try:
        smtp = _open_smtp(cfg)
        smtp.quit()
    except Exception as e:
        return _status_report("⚠️", "IMAP_OK_SMTP_ERROR", "partial",
            f"IMAP works, but SMTP login failed: {_format_smtp_error(e)}",
            ["Reading works; sending will not until SMTP is fixed.",
             "Check EMAIL_SMTP_HOST / EMAIL_SMTP_PORT / EMAIL_SMTP_SECURITY and the password."])

    return _status_report("✅", "READY", "ok",
        "Email integration is operational: IMAP and SMTP both authenticate. All tools "
        "(list/get/thread/folders/modify/send/download) are usable.\n"
        f"Account: {cfg['username']} (IMAP {cfg['imap_host']}, SMTP {cfg['smtp_host']})")


def _fetch_summary(conn: imaplib.IMAP4, uid: int) -> str:
    """One-message summary line for list_messages (headers + read/flag state)."""
    typ, data = conn.uid(
        "FETCH", str(uid),
        "(FLAGS BODY.PEEK[HEADER.FIELDS (SUBJECT FROM DATE)])",
    )
    if typ != "OK":
        return f"- uid {uid} (error fetching)"
    flags = imaplib.ParseFlags(data[0][0]) if data and isinstance(data[0], tuple) else ()
    flag_names = {f.decode("ascii", "replace").lstrip("\\").lower() for f in flags}
    headers = _parse_header_bytes(_first_literal(data))
    unread = "seen" not in flag_names
    marks = []
    if unread:
        marks.append("UNREAD")
    if "flagged" in flag_names:
        marks.append("★")
    mark_str = (" [" + ", ".join(marks) + "]") if marks else ""
    return (f"- {headers.get('subject', '(no subject)')}{mark_str}\n"
            f"  From: {headers.get('from', '?')} | Date: {headers.get('date', '?')} | UID: {uid}")


def _email_list_messages(args: dict) -> str:
    conn = _imap()
    if conn is None:
        return _unavailable()

    folder = args.get("folder", "INBOX")
    query = args.get("query", "")
    unread_only = bool(args.get("unread_only", False))
    max_results = min(int(args.get("max_results", 20) or 20), 50)

    ok, err = _select(conn, folder, readonly=True)
    if not ok:
        return err

    criteria = _build_search(query, unread_only)
    try:
        typ, data = conn.uid("SEARCH", None, *criteria)
    except Exception as e:
        return _format_imap_error(e)
    if typ != "OK":
        return f"Error: IMAP SEARCH failed in {folder!r}."

    uids = [int(x) for x in data[0].split()] if data and data[0] else []
    if not uids:
        return f"No messages found in {folder!r}."

    # Newest first, capped.
    uids = sorted(uids, reverse=True)[:max_results]
    lines = [f"Messages in {folder!r} ({len(uids)} shown, newest first):"]
    for uid in uids:
        try:
            lines.append(_fetch_summary(conn, uid))
        except Exception as e:
            lines.append(f"- uid {uid} (error: {e})")
    lines.append("\nUse get_message with the UID (and folder if not INBOX) to read a message.")
    return "\n".join(lines)


def _fetch_full(conn: imaplib.IMAP4, uid: int) -> email.message.Message | None:
    typ, data = conn.uid("FETCH", str(uid), "(BODY.PEEK[])")
    if typ != "OK":
        return None
    raw = _first_literal(data)
    if not raw:
        return None
    return email.message_from_bytes(raw)


def _email_get_message(args: dict) -> str:
    conn = _imap()
    if conn is None:
        return _unavailable()

    uid = args.get("message_id")
    if not uid:
        return "Error: Missing required parameter 'message_id' (the message UID)."
    folder = args.get("folder", "INBOX")
    include_body = args.get("include_body", True)

    ok, err = _select(conn, folder, readonly=True)
    if not ok:
        return err

    try:
        msg = _fetch_full(conn, int(uid))
    except ValueError:
        return f"Error: message_id must be a numeric UID, got {uid!r}."
    except Exception as e:
        return _format_imap_error(e)
    if msg is None:
        return f"Error: message UID {uid} not found in folder {folder!r}."

    lines = [
        f"UID: {uid}",
        f"Folder: {folder}",
        f"From: {_decode_hdr(msg.get('From'))}",
        f"To: {_decode_hdr(msg.get('To'))}",
        f"Date: {_decode_hdr(msg.get('Date'))}",
        f"Subject: {_decode_hdr(msg.get('Subject')) or '(no subject)'}",
        f"Message-ID: {msg.get('Message-ID', '?')}",
    ]

    attachments = [_decode_hdr(p.get_filename()) for p in _iter_attachments(msg)]
    if attachments:
        lines.append(f"Attachments: {', '.join(a for a in attachments if a)}")

    if include_body:
        body_text, from_html = _extract_body(msg)
        if body_text:
            label = "--- Body (converted from HTML) ---" if from_html else "--- Body ---"
            lines.append("\n" + label)
            if len(body_text) > 10000:
                lines.append(body_text[:10000] + "\n... [truncated at 10000 chars]")
            else:
                lines.append(body_text)
        else:
            lines.append("\n(no text body found)")

    return "\n".join(lines)


def _email_get_thread(args: dict) -> str:
    """Best-effort thread: group messages by shared Message-ID references / subject."""
    conn = _imap()
    if conn is None:
        return _unavailable()

    uid = args.get("message_id")
    if not uid:
        return "Error: Missing required parameter 'message_id' (a message UID in the thread)."
    folder = args.get("folder", "INBOX")

    ok, err = _select(conn, folder, readonly=True)
    if not ok:
        return err

    try:
        seed = _fetch_full(conn, int(uid))
    except ValueError:
        return f"Error: message_id must be a numeric UID, got {uid!r}."
    except Exception as e:
        return _format_imap_error(e)
    if seed is None:
        return f"Error: message UID {uid} not found in folder {folder!r}."

    msg_id = (seed.get("Message-ID") or "").strip()
    subject = _decode_hdr(seed.get("Subject"))
    base_subject = re.sub(r"(?i)^\s*(re|fwd|fw|r|aw|antw|sv)\s*:\s*", "", subject).strip()

    # Collect candidate UIDs: same base subject, plus anything referencing this Message-ID.
    found: set[int] = {int(uid)}
    try:
        if base_subject:
            typ, data = conn.uid("SEARCH", None, "SUBJECT", _q(base_subject))
            if typ == "OK" and data and data[0]:
                found.update(int(x) for x in data[0].split())
        if msg_id:
            for field in ("In-Reply-To", "References"):
                typ, data = conn.uid("SEARCH", None, "HEADER", field, _q(msg_id))
                if typ == "OK" and data and data[0]:
                    found.update(int(x) for x in data[0].split())
    except Exception as e:
        log(f"thread search partial failure: {_format_imap_error(e)}")

    ordered = sorted(found)
    lines = [f"Thread (best-effort) around UID {uid} in {folder!r} — {len(ordered)} message(s):"]
    if base_subject:
        lines.insert(1, f"Subject: {base_subject}")
    for muid in ordered:
        try:
            typ, data = conn.uid(
                "FETCH", str(muid),
                "(BODY.PEEK[HEADER.FIELDS (SUBJECT FROM DATE)])",
            )
            h = _parse_header_bytes(_first_literal(data))
            lines.append(f"\n[UID {muid}] From: {h.get('from', '?')} | Date: {h.get('date', '?')}")
            lines.append(f"    Subject: {h.get('subject', '(no subject)')}")
        except Exception as e:
            lines.append(f"\n[UID {muid}] (error: {e})")
    return "\n".join(lines)


def _email_list_folders(args: dict) -> str:
    conn = _imap()
    if conn is None:
        return _unavailable()

    try:
        typ, data = conn.list()
    except Exception as e:
        return _format_imap_error(e)
    if typ != "OK" or not data:
        return "No folders found."

    lines = ["Folders:"]
    for raw in data:
        if raw is None:
            continue
        name = _parse_list_mailbox(raw)
        if not name:
            continue
        counts = ""
        try:
            typ2, st = conn.status(_quote_mailbox(name), "(MESSAGES UNSEEN)")
            if typ2 == "OK" and st and st[0]:
                total = re.search(rb"MESSAGES\s+(\d+)", st[0])
                unseen = re.search(rb"UNSEEN\s+(\d+)", st[0])
                counts = (f" — {int(total.group(1)) if total else '?'} total, "
                          f"{int(unseen.group(1)) if unseen else '?'} unread")
        except Exception:
            pass
        lines.append(f"- {name}{counts}")
    return "\n".join(lines)


def _parse_list_mailbox(raw: bytes) -> str:
    """Extract the mailbox name from a LIST response line."""
    text = raw.decode("utf-8", "replace") if isinstance(raw, bytes) else str(raw)
    # Format: (\HasNoChildren) "/" "Folder Name"  — name is the last token, maybe quoted.
    m = re.search(r'"(?:[^"\\]|\\.)*"\s*$', text)
    if m:
        return m.group(0)[1:-1].replace('\\"', '"').replace("\\\\", "\\")
    parts = text.rsplit(" ", 1)
    return parts[-1].strip() if parts else ""


# Friendly flag name → IMAP system flag.
_FLAG_MAP = {
    "read": "\\Seen", "seen": "\\Seen",
    "flagged": "\\Flagged", "starred": "\\Flagged", "star": "\\Flagged",
    "answered": "\\Answered", "draft": "\\Draft", "deleted": "\\Deleted",
}


def _map_flags(names: Any) -> list[str]:
    if isinstance(names, str):
        names = [names]
    out = []
    for n in names or []:
        out.append(_FLAG_MAP.get(str(n).strip().lower(), n))
    return out


def _expunge_uid(conn: imaplib.IMAP4, uid: str) -> None:
    """Expunge one message. Uses UID EXPUNGE (UIDPLUS) so we don't remove other
    \\Deleted messages in the folder; falls back to a full EXPUNGE otherwise."""
    if any(c.upper() == "UIDPLUS" for c in conn.capabilities):
        try:
            conn.uid("EXPUNGE", uid)
            return
        except Exception:
            pass
    conn.expunge()


def _email_modify_message(args: dict) -> str:
    """Add/remove flags, mark read/unread, move to another folder, or delete."""
    conn = _imap()
    if conn is None:
        return _unavailable()

    uid = args.get("message_id")
    if not uid:
        return "Error: Missing required parameter 'message_id' (the message UID)."
    folder = args.get("folder", "INBOX")
    add_flags = _map_flags(args.get("add_flags"))
    remove_flags = _map_flags(args.get("remove_flags"))
    move_to = args.get("move_to_folder")

    # Convenience booleans.
    if args.get("mark_read"):
        add_flags.append("\\Seen")
    if args.get("mark_unread"):
        remove_flags.append("\\Seen")

    ok, err = _select(conn, folder, readonly=False)
    if not ok:
        return err

    changes = []
    try:
        if add_flags:
            conn.uid("STORE", str(uid), "+FLAGS", "(" + " ".join(add_flags) + ")")
            changes.append(f"added flags {add_flags}")
        if remove_flags:
            conn.uid("STORE", str(uid), "-FLAGS", "(" + " ".join(remove_flags) + ")")
            changes.append(f"removed flags {remove_flags}")

        if move_to:
            # Prefer server-side MOVE; fall back to COPY + \Deleted + EXPUNGE.
            # Gate MOVE on both the server capability AND imaplib knowing the verb
            # (older Python builds lack MOVE in imaplib.Commands).
            can_move = ("MOVE" in imaplib.Commands) and any(c.upper() == "MOVE" for c in conn.capabilities)
            if can_move:
                conn.uid("MOVE", str(uid), _quote_mailbox(move_to))
            else:
                conn.uid("COPY", str(uid), _quote_mailbox(move_to))
                conn.uid("STORE", str(uid), "+FLAGS", "(\\Deleted)")
                _expunge_uid(conn, str(uid))
            changes.append(f"moved to {move_to!r}")
        elif "\\Deleted" in add_flags:
            _expunge_uid(conn, str(uid))
            changes.append("expunged")
    except Exception as e:
        return _format_imap_error(e)

    if not changes:
        return ("Nothing to do: pass add_flags/remove_flags (e.g. 'read', 'flagged', 'deleted'), "
                "mark_read/mark_unread, or move_to_folder.")
    return f"✅ Message UID {uid} in {folder!r}: {'; '.join(changes)}"


_MAX_ATTACHMENT_BYTES = 25 * 1024 * 1024


def _open_smtp(cfg: dict) -> smtplib.SMTP:
    """Open and authenticate an SMTP connection per the configured security mode."""
    ctx = ssl.create_default_context()
    if cfg["smtp_security"] == "ssl":
        smtp: smtplib.SMTP = smtplib.SMTP_SSL(cfg["smtp_host"], cfg["smtp_port"], context=ctx, timeout=30)
    else:
        smtp = smtplib.SMTP(cfg["smtp_host"], cfg["smtp_port"], timeout=30)
        smtp.ehlo()
        if cfg["smtp_security"] == "starttls":
            smtp.starttls(context=ctx)
            smtp.ehlo()
    smtp.login(cfg["username"], cfg["password"])
    return smtp


def _append_to_sent(conn: imaplib.IMAP4, raw: bytes) -> None:
    """Best-effort: save a sent message into the account's Sent folder."""
    sent = None
    try:
        typ, data = conn.list()
        if typ == "OK":
            for line in data or []:
                if line and b"\\Sent" in line:
                    sent = _parse_list_mailbox(line)
                    break
    except Exception:
        pass
    candidates = [sent] if sent else []
    candidates += ["Sent", "Sent Items", "Sent Mail", "[Gmail]/Sent Mail", "INBOX.Sent"]
    for name in candidates:
        if not name:
            continue
        try:
            typ, _ = conn.append(_quote_mailbox(name), "(\\Seen)",
                                 imaplib.Time2Internaldate(time.time()), raw)
            if typ == "OK":
                return
        except Exception:
            continue


def _email_send_message(args: dict) -> str:
    conn_cfg = _get_config()
    if conn_cfg is None:
        return _unavailable()

    to = args.get("to")
    if not to:
        return "Error: Missing required parameter 'to'."
    subject = args.get("subject", "")
    body_text = args.get("body", "")
    cc = args.get("cc")
    bcc = args.get("bcc")
    in_reply_to = args.get("in_reply_to")
    attachments = args.get("attachments") or []
    if isinstance(attachments, str):
        attachments = [attachments]

    # Resolve attachment paths (absolute or relative to this connector's folder).
    root = os.path.dirname(os.path.abspath(__file__))
    resolved: list[str] = []
    total = 0
    for raw_path in attachments:
        path = raw_path if os.path.isabs(raw_path) else os.path.join(root, raw_path)
        if not os.path.isfile(path):
            return f"Error: attachment not found: {raw_path}"
        total += os.path.getsize(path)
        resolved.append(path)
    if total > _MAX_ATTACHMENT_BYTES:
        return (f"Error: attachments total ~{total // (1024 * 1024)} MB, over the "
                f"{_MAX_ATTACHMENT_BYTES // (1024 * 1024)} MB limit.")

    msg = EmailMessage()
    msg["From"] = conn_cfg["from_addr"]
    msg["To"] = to
    if cc:
        msg["Cc"] = cc
    if bcc:
        msg["Bcc"] = bcc
    msg["Subject"] = subject
    msg["Date"] = email.utils.formatdate(localtime=True)
    msg["Message-ID"] = email.utils.make_msgid()
    if in_reply_to:
        ref = in_reply_to if in_reply_to.startswith("<") else f"<{in_reply_to}>"
        msg["In-Reply-To"] = ref
        msg["References"] = ref
    msg.set_content(body_text)

    for path in resolved:
        ctype, encoding = mimetypes.guess_type(path)
        if ctype is None or encoding is not None:
            ctype = "application/octet-stream"
        maintype, subtype = ctype.split("/", 1)
        try:
            with open(path, "rb") as f:
                data = f.read()
        except Exception as e:
            return f"Error: could not read attachment {path}: {e}"
        msg.add_attachment(data, maintype=maintype, subtype=subtype,
                           filename=os.path.basename(path))

    # Recipient list includes Cc/Bcc for the envelope.
    recipients = [a.strip() for a in re.split(r"[,;]", to) if a.strip()]
    for extra in (cc, bcc):
        if extra:
            recipients += [a.strip() for a in re.split(r"[,;]", extra) if a.strip()]

    try:
        smtp = _open_smtp(conn_cfg)
    except Exception as e:
        return _format_smtp_error(e)
    try:
        smtp.send_message(msg, from_addr=conn_cfg["from_addr"], to_addrs=recipients)
    except Exception as e:
        return _format_smtp_error(e)
    finally:
        try:
            smtp.quit()
        except Exception:
            pass

    # Best-effort save-to-Sent (many providers don't do this for SMTP sends).
    conn = _imap()
    if conn is not None:
        try:
            _append_to_sent(conn, msg.as_bytes())
        except Exception:
            pass

    suffix = (f" ({len(resolved)} attachment{'s' if len(resolved) != 1 else ''})"
              if resolved else "")
    return f"✅ Message sent to {to}{suffix}."


def _email_get_profile(args: dict) -> str:
    cfg = _get_config()
    if cfg is None:
        return _unavailable()
    conn = _imap()
    if conn is None:
        return _unavailable()

    inbox_total = inbox_unread = "?"
    folder_count = "?"
    try:
        typ, st = conn.status("INBOX", "(MESSAGES UNSEEN)")
        if typ == "OK" and st and st[0]:
            m = re.search(rb"MESSAGES\s+(\d+)", st[0])
            u = re.search(rb"UNSEEN\s+(\d+)", st[0])
            inbox_total = int(m.group(1)) if m else "?"
            inbox_unread = int(u.group(1)) if u else "?"
        typ2, data = conn.list()
        if typ2 == "OK" and data:
            folder_count = sum(1 for x in data if x)
    except Exception as e:
        return _format_imap_error(e)

    return (f"Account: {cfg['username']}\n"
            f"From address: {cfg['from_addr']}\n"
            f"IMAP: {cfg['imap_host']}:{cfg['imap_port']}\n"
            f"SMTP: {cfg['smtp_host']}:{cfg['smtp_port']} ({cfg['smtp_security']})\n"
            f"Folders: {folder_count}\n"
            f"INBOX: {inbox_total} total, {inbox_unread} unread")


def _email_create_folder(args: dict) -> str:
    conn = _imap()
    if conn is None:
        return _unavailable()
    name = args.get("name")
    if not name:
        return "Error: Missing required parameter 'name'."
    try:
        typ, data = conn.create(_quote_mailbox(name))
    except Exception as e:
        return _format_imap_error(e)
    if typ != "OK":
        detail = data[0].decode("utf-8", "replace") if data and data[0] else "unknown error"
        return f"Error: could not create folder {name!r}: {detail}"
    return f"✅ Folder {name!r} created."


def _email_download_attachments(args: dict) -> str:
    conn = _imap()
    if conn is None:
        return _unavailable()
    uid = args.get("message_id")
    if not uid:
        return "Error: Missing required parameter 'message_id' (the message UID)."
    folder = args.get("folder", "INBOX")
    default_folder = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                                  "data", "email_attachments")
    dest = args.get("dest_folder") or default_folder

    ok, err = _select(conn, folder, readonly=True)
    if not ok:
        return err
    try:
        msg = _fetch_full(conn, int(uid))
    except ValueError:
        return f"Error: message_id must be a numeric UID, got {uid!r}."
    except Exception as e:
        return _format_imap_error(e)
    if msg is None:
        return f"Error: message UID {uid} not found in folder {folder!r}."

    parts = list(_iter_attachments(msg))
    if not parts:
        return "No attachments found."
    os.makedirs(dest, exist_ok=True)

    saved = []
    for part in parts:
        filename = _decode_hdr(part.get_filename()) or "attachment.bin"
        payload = part.get_payload(decode=True)
        if payload is None:
            saved.append(f"- {filename}: empty payload")
            continue
        safe_name = os.path.basename(filename)
        path = os.path.join(dest, safe_name)
        try:
            with open(path, "wb") as f:
                f.write(payload)
        except Exception as e:
            saved.append(f"- {safe_name}: ERROR writing file: {e}")
            continue
        saved.append(f"- {os.path.abspath(path)} ({len(payload)} bytes)")
    return "\n".join(["✅ Attachments downloaded:"] + saved)


# ── Tool manifest ────────────────────────────────────────────────────────────────

TOOLS = [
    {
        "name": "status",
        "title": "Status",
        "description": (
            "Self-check that the email integration is operational: verifies IMAP and SMTP both "
            "authenticate with the configured credentials. Call this first whenever another email "
            "tool fails, or to give the user a quick yes/no on whether email is usable right now."
        ),
        "inputSchema": {"type": "object", "properties": {}},
    },
    {
        "name": "list_messages",
        "title": "List Messages",
        "description": (
            "List messages in a folder (default INBOX), newest first. The optional 'query' supports "
            "a Gmail-like mini-syntax: from:x, to:x, subject:x, since:YYYY-MM-DD, before:YYYY-MM-DD, "
            "'unread'/'read', 'has:attachment', and free text (matched against headers+body). "
            "Returns subject, sender, date, read/flag state and the message UID; pass the UID (and "
            "the same folder) to get_message / modify_message."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "folder": {"type": "string", "description": "IMAP folder to list (default 'INBOX'). See list_folders."},
                "query": {"type": "string", "description": "Gmail-like query, e.g. 'from:john unread since:2024-01-01'. Empty = all."},
                "unread_only": {"type": "boolean", "description": "Shortcut to only return unread messages (default false)."},
                "max_results": {"type": "integer", "description": "Max messages to return (default 20, max 50)."},
            },
        },
    },
    {
        "name": "get_message",
        "title": "Get Message",
        "description": (
            "Get the full content of a message by its UID, including body text (truncated at 10000 "
            "chars; HTML-only emails are converted to readable text). Attachment filenames are "
            "listed — download them with download_attachments. Pass 'folder' if the message is not "
            "in INBOX (a UID is only unique within one folder)."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "message_id": {"type": "string", "description": "The message UID (as returned by list_messages)."},
                "folder": {"type": "string", "description": "Folder the message is in (default 'INBOX')."},
                "include_body": {"type": "boolean", "description": "Include the full body text (default true)."},
            },
            "required": ["message_id"],
        },
    },
    {
        "name": "get_thread",
        "title": "Get Thread",
        "description": (
            "Best-effort thread reconstruction for a message UID: gathers messages in the same "
            "folder that share the Message-ID reference chain (In-Reply-To/References) or the same "
            "base subject. IMAP has no native threads, so this is heuristic."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "message_id": {"type": "string", "description": "A message UID that belongs to the thread."},
                "folder": {"type": "string", "description": "Folder to search (default 'INBOX')."},
            },
            "required": ["message_id"],
        },
    },
    {
        "name": "list_folders",
        "title": "List Folders",
        "description": "List all IMAP folders/mailboxes with total and unread message counts. Use to discover folder names for the other tools.",
        "inputSchema": {"type": "object", "properties": {}},
    },
    {
        "name": "modify_message",
        "title": "Modify Message",
        "description": (
            "Change a message's state: add/remove flags, mark read/unread, move to another folder, "
            "archive or delete. Friendly flag names: 'read'/'seen', 'flagged'/'starred', 'answered', "
            "'deleted'. mark_read=true marks as read; move_to_folder='Archive' archives (moves); "
            "add_flags=['deleted'] deletes (and expunges). Pass 'folder' if not INBOX."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "message_id": {"type": "string", "description": "The message UID to modify."},
                "folder": {"type": "string", "description": "Folder the message is in (default 'INBOX')."},
                "add_flags": {"type": ["string", "array"], "items": {"type": "string"},
                              "description": "Flag(s) to add: 'read', 'flagged', 'answered', 'deleted'. String or array."},
                "remove_flags": {"type": ["string", "array"], "items": {"type": "string"},
                                 "description": "Flag(s) to remove (e.g. 'read' to mark unread)."},
                "mark_read": {"type": "boolean", "description": "Convenience: mark the message as read."},
                "mark_unread": {"type": "boolean", "description": "Convenience: mark the message as unread."},
                "move_to_folder": {"type": "string", "description": "Move the message to this folder (archive = move to your archive folder)."},
            },
            "required": ["message_id"],
        },
    },
    {
        "name": "send_message",
        "title": "Send Message",
        "description": (
            "Send an email via SMTP. Supports in-thread replies by passing in_reply_to (the "
            "Message-ID of the message being replied to, with or without angle brackets), which sets "
            "the In-Reply-To/References headers. Attach files by passing local paths in "
            "'attachments'; if any path is missing the email is NOT sent. The message is also "
            "best-effort saved to the account's Sent folder."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "to": {"type": "string", "description": "Recipient address(es), comma-separated."},
                "subject": {"type": "string", "description": "Subject line."},
                "body": {"type": "string", "description": "Plain-text body."},
                "cc": {"type": "string", "description": "CC address(es) (optional)."},
                "bcc": {"type": "string", "description": "BCC address(es) (optional)."},
                "in_reply_to": {"type": "string", "description": "Message-ID being replied to, for correct threading (optional)."},
                "attachments": {"type": "array", "items": {"type": "string"},
                                "description": "File path(s) to attach (absolute or relative to the connector folder). Total ~25 MB."},
            },
            "required": ["to", "subject", "body"],
        },
    },
    {
        "name": "get_profile",
        "title": "Get Profile",
        "description": "Show the configured account: username, From address, IMAP/SMTP servers, folder count, and INBOX total/unread.",
        "inputSchema": {"type": "object", "properties": {}},
    },
    {
        "name": "create_folder",
        "title": "Create Folder",
        "description": "Create a new IMAP folder/mailbox. Fails if it already exists.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "name": {"type": "string", "description": "Name of the new folder (e.g. 'Archive', 'Receipts')."},
            },
            "required": ["name"],
        },
    },
    {
        "name": "download_attachments",
        "title": "Download Attachments",
        "description": (
            "Download all attachments from a message to a local folder (default data/email_attachments/). "
            "Returns the absolute path and size of each saved file. Pass 'folder' if the message is not in INBOX."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "message_id": {"type": "string", "description": "The message UID to download attachments from."},
                "folder": {"type": "string", "description": "Folder the message is in (default 'INBOX')."},
                "dest_folder": {"type": "string", "description": "Local folder to save into (default data/email_attachments/)."},
            },
            "required": ["message_id"],
        },
    },
]


# ── JSON-RPC dispatch ────────────────────────────────────────────────────────────

TOOL_DISPATCH: dict[str, Callable[[dict], str]] = {
    "status":               _email_status,
    "list_messages":        _email_list_messages,
    "get_message":          _email_get_message,
    "get_thread":           _email_get_thread,
    "list_folders":         _email_list_folders,
    "modify_message":       _email_modify_message,
    "send_message":         _email_send_message,
    "get_profile":          _email_get_profile,
    "create_folder":        _email_create_folder,
    "download_attachments": _email_download_attachments,
}


def _ok(req_id: Any, result: Any) -> str:
    return json.dumps({"jsonrpc": "2.0", "id": req_id, "result": result})


def _text_result(req_id: Any, text: str, is_error: bool = False) -> str:
    payload: dict = {
        "jsonrpc": "2.0",
        "id": req_id,
        "result": {"content": [{"type": "text", "text": text}]},
    }
    if is_error:
        payload["result"]["isError"] = True
    return json.dumps(payload)


def handle_request(msg: dict) -> str | None:
    method = msg.get("method", "")
    req_id = msg.get("id")

    if method == "initialize":
        return _ok(req_id, {
            "protocolVersion": "2024-11-05",
            "capabilities": {"tools": {}},
            "serverInfo": {"name": "email", "version": "1.0.0"},
        })

    if method == "notifications/initialized":
        return None

    if method == "tools/list":
        return _ok(req_id, {"tools": TOOLS})

    if method == "tools/call":
        params = msg.get("params", {})
        tool_name = params.get("name", "")
        tool_args = params.get("arguments", {}) or {}
        handler = TOOL_DISPATCH.get(tool_name)
        if handler is None:
            return _text_result(req_id, f"Error: Unknown tool: {tool_name}", is_error=True)
        try:
            text = handler(tool_args)
            return _text_result(req_id, text, is_error=text.startswith("Error:"))
        except Exception as e:
            log(f"Unhandled exception in tool '{tool_name}': {e}")
            return _text_result(req_id, f"Error: Internal error in tool '{tool_name}': {e}", is_error=True)

    return json.dumps({
        "jsonrpc": "2.0",
        "id": req_id,
        "error": {"code": -32601, "message": f"Method not found: {method}"},
    })


# ── Main loop ────────────────────────────────────────────────────────────────────

def main() -> None:
    log("Starting Email MCP server")
    # Validate config and start the background push watcher (best-effort).
    _start_watching()
    try:
        for line in sys.stdin:
            line = line.strip()
            if not line:
                continue
            try:
                msg = json.loads(line)
            except json.JSONDecodeError as e:
                log(f"Invalid JSON input: {e}")
                continue
            resp = handle_request(msg)
            if resp is not None:
                with _stdout_lock:
                    sys.stdout.write(resp + "\n")
                    sys.stdout.flush()
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    main()