"""Email (send/read) in-process MCP for Jarvis — Namecheap Private Email. Gives Claude a real mailbox: send mail, search the inbox, and read a message — straight over SMTP+IMAP from Ahmed's own mailbox (``EMAIL_ADDRESS``). No API, no OAuth, no third party: sending via SMTP from your own account is ordinary email, so there's nothing to get "banned" for at personal/team volume. Runs IN-PROCESS via the Claude Agent SDK's in-process MCP server (key ``"email"``; Claude sees ``mcp__email__``). Pure standard library (``smtplib`` / ``imaplib`` / ``email``) — no extra dependencies. Config (``.env``): EMAIL_ADDRESS the mailbox, also the From: address EMAIL_PASSWORD its password (or a dedicated app password) EMAIL_SMTP_HOST / EMAIL_SMTP_PORT default smtp.privateemail.com : 465 (SSL) EMAIL_IMAP_HOST / EMAIL_IMAP_PORT default mail.privateemail.com : 993 (SSL) Set ``MCP_EMAIL=0`` to disable. """ from __future__ import annotations import base64 import email import imaplib import os import smtplib from email.header import decode_header, make_header from email.message import EmailMessage from email.utils import getaddresses, parsedate_to_datetime def _addr() -> str: return os.environ.get("EMAIL_ADDRESS", "").strip() def enabled() -> bool: """Whether the email MCP should be exposed to Claude.""" if os.environ.get("MCP_EMAIL", "1") == "0": return False return bool(_addr() and os.environ.get("EMAIL_PASSWORD")) def _text(msg: str) -> dict: """Wrap a plain string as an MCP text content result.""" return {"content": [{"type": "text", "text": msg}]} def _decode(raw) -> str: """Decode an RFC2047 MIME header (subjects/senders) to plain text.""" if raw is None: return "" try: return str(make_header(decode_header(raw))) except Exception: # noqa: BLE001 return str(raw) def _body_of(msg) -> str: """Extract a readable text body from a parsed email message.""" if msg.is_multipart(): for part in msg.walk(): if (part.get_content_type() == "text/plain" and "attachment" not in str(part.get("Content-Disposition"))): try: return part.get_payload(decode=True).decode( part.get_content_charset() or "utf-8", "replace") except Exception: # noqa: BLE001 continue return "(no plain-text body)" try: return msg.get_payload(decode=True).decode( msg.get_content_charset() or "utf-8", "replace") except Exception: # noqa: BLE001 return str(msg.get_payload()) def _send(to: str, subject: str, body: str, cc: str, attachments: list[str] | None = None) -> str: """Blocking SMTP send — called via asyncio.to_thread. `attachments` is an optional list of file paths. Each readable file is added as an application/octet-stream MIME part with ``Content-Disposition: attachment; filename=...`` (EmailMessage upgrades itself to multipart/mixed on the first add_attachment). Missing or unreadable files are SKIPPED and noted in the summary — never fatal. With no attachments the message is built exactly as before. """ host = os.environ.get("EMAIL_SMTP_HOST", "smtp.privateemail.com") port = int(os.environ.get("EMAIL_SMTP_PORT", "465")) msg = EmailMessage() msg["From"] = _addr() msg["To"] = to if cc: msg["Cc"] = cc msg["Subject"] = subject msg.set_content(body) attached: list[str] = [] skipped: list[str] = [] for p in attachments or []: path = os.path.expanduser(str(p).strip()) if not path: continue name = os.path.basename(path.rstrip("/")) or path try: with open(path, "rb") as fh: data = fh.read() except OSError: skipped.append(name) continue msg.add_attachment(data, maintype="application", subtype="octet-stream", filename=name) attached.append(name) rcpts = [a.strip() for a in (to + "," + cc).split(",") if a.strip()] with smtplib.SMTP_SSL(host, port, timeout=30) as s: s.login(_addr(), os.environ["EMAIL_PASSWORD"]) s.send_message(msg, to_addrs=rcpts) out = f"Sent '{subject}' to {to}" + (f" (cc {cc})" if cc else "") if attached: out += f" with {len(attached)} attachment(s): {', '.join(attached)}" if skipped: out += f" — couldn't attach (file missing/unreadable): {', '.join(skipped)}" return out + "." def _imap(): """Open and log into IMAP (SSL). Caller closes it.""" host = os.environ.get("EMAIL_IMAP_HOST", "mail.privateemail.com") port = int(os.environ.get("EMAIL_IMAP_PORT", "993")) m = imaplib.IMAP4_SSL(host, port) m.login(_addr(), os.environ["EMAIL_PASSWORD"]) return m def _search(query: str, limit: int) -> str: """Blocking IMAP search over INBOX — called via asyncio.to_thread. `query` is a plain phrase matched against sender+subject; empty = latest. Returns a numbered list (newest first) with the UID for email_read. """ m = _imap() try: m.select("INBOX") if query.strip(): typ, data = m.uid("search", None, "OR", "FROM", f'"{query}"', "SUBJECT", f'"{query}"') else: typ, data = m.uid("search", None, "ALL") uids = data[0].split() if data and data[0] else [] uids = uids[-limit:][::-1] # newest first if not uids: return "No matching messages." lines = [] for uid in uids: typ, hd = m.uid("fetch", uid, "(BODY.PEEK[HEADER.FIELDS (FROM SUBJECT DATE)])") hdr = email.message_from_bytes(hd[0][1]) if hd and hd[0] else None frm = _decode(hdr.get("From")) if hdr else "?" sub = _decode(hdr.get("Subject")) if hdr else "?" date = (hdr.get("Date") or "") if hdr else "" lines.append(f"[uid {uid.decode()}] {date}\n from: {frm}\n" f" subj: {sub}") return "\n".join(lines) finally: try: m.logout() except Exception: # noqa: BLE001 pass def _read(uid: str) -> str: """Blocking IMAP fetch of one message body — called via asyncio.to_thread.""" m = _imap() try: m.select("INBOX") typ, data = m.uid("fetch", uid.encode(), "(RFC822)") if not data or not data[0]: return f"No message with uid {uid}." msg = email.message_from_bytes(data[0][1]) frm = _decode(msg.get("From")) sub = _decode(msg.get("Subject")) body = _body_of(msg) if len(body) > 4000: body = body[:4000] + "\n…(truncated)" return f"From: {frm}\nSubject: {sub}\n\n{body}" finally: try: m.logout() except Exception: # noqa: BLE001 pass # --------------------------------------------------------------------------- # HUD email card — full message, real attachments, a best-effort THREAD # (voice/mail_view.py turns these into the `email_thread` event). The helpers # above feed Jarvis' VOICE (short, truncated); these feed his SCREEN (whole # message, HTML kept, nothing cut). # # IMAP has no thread ids, so a "thread" is REASSEMBLED from RFC 5322 headers: # the root Message-ID (first entry of References, else In-Reply-To, else the # message's own id) is searched for across INBOX + the Sent folder. Servers # that don't index HEADER searches, or correspondents who break References, # yield a one-message thread — acceptable, never fatal. # # A message is addressed as "|" (msg_id on the wire) so a later # attachment/reply action can find it again; a bare uid means INBOX. # --------------------------------------------------------------------------- _SENT_FOLDERS = ("Sent", "INBOX.Sent", "Sent Items", "Sent Messages") def _split_ident(ident: str) -> tuple[str, str]: """'|' -> (folder, uid). A bare uid means INBOX.""" if "|" in ident: folder, uid = ident.split("|", 1) return (folder or "INBOX"), uid return "INBOX", ident def _fetch(m, folder: str, uid: str): """Select `folder` and fetch one message by UID. None if it's gone.""" typ, _ = m.select(folder) if typ != "OK": return None typ, data = m.uid("fetch", uid.encode(), "(RFC822)") if typ != "OK" or not data or not data[0]: return None return email.message_from_bytes(data[0][1]) def _root_id(msg) -> str: """The thread's root Message-ID: first of References, else In-Reply-To, else the message's own id (it started the thread).""" refs = (msg.get("References") or "").split() if refs: return refs[0].strip() irt = (msg.get("In-Reply-To") or "").strip().split() if irt: return irt[0] return (msg.get("Message-ID") or "").strip() def _thread_uids(m, folder: str, root: str) -> list[str]: """UIDs in `folder` whose References or Message-ID mention the root id.""" typ, _ = m.select(folder) if typ != "OK": return [] try: typ, data = m.uid("search", None, "OR", "HEADER", "REFERENCES", f'"{root}"', "HEADER", "MESSAGE-ID", f'"{root}"') except Exception: # noqa: BLE001 — server may not index HEADER searches return [] if typ != "OK" or not data or not data[0]: return [] return [u.decode() for u in data[0].split()] def _parts_of(msg) -> dict: """Split a parsed message into text body, HTML body, inline (cid:) images and real attachments. Untruncated — the card shows everything. `att_id` is the part's index in walk order: stable for a given message, and all IMAP gives us (there are no Gmail-style attachment ids).""" text, html = "", "" inline: dict[str, dict] = {} atts: list[dict] = [] for idx, part in enumerate(msg.walk()): ctype = part.get_content_type() if ctype.startswith("multipart/"): continue disp = str(part.get("Content-Disposition") or "").lower() fname = _decode(part.get_filename()) cid = (part.get("Content-ID") or "").strip().strip("<>") try: raw = part.get_payload(decode=True) or b"" except Exception: # noqa: BLE001 — a broken part must not kill the card continue if cid and ctype.startswith("image/"): inline[cid] = {"mime": ctype, "b64": base64.b64encode(raw).decode()} continue if fname or "attachment" in disp: atts.append({"att_id": str(idx), "filename": fname or f"part-{idx}", "mime": ctype, "size": len(raw), "path": ""}) continue body = raw.decode(part.get_content_charset() or "utf-8", "replace") if ctype == "text/plain": text += body elif ctype == "text/html": html += body return {"body_text": text, "body_html": html, "inline": inline, "attachments": atts} def _as_msg(folder: str, uid: str, msg) -> dict: """One parsed message in the shape mail_view expects.""" try: ts = parsedate_to_datetime(msg.get("Date")).timestamp() except Exception: # noqa: BLE001 — undated/garbled Date header ts = 0.0 frm = _decode(msg.get("From")) return { "msg_id": f"{folder}|{uid}", "from": frm, "to": _decode(msg.get("To")), "cc": _decode(msg.get("Cc")), "subject": _decode(msg.get("Subject")), "date": msg.get("Date") or "", "ts": ts, # our own address in From = Ahmed wrote it (mail_view double-checks) "outgoing": _addr().lower() in frm.lower(), **_parts_of(msg), } def thread(ident: str) -> dict: """A best-effort THREAD around one message, for the HUD card. Falls back to the single message when the server can't be searched by header.""" folder, uid = _split_ident(ident) m = _imap() try: msg = _fetch(m, folder, uid) if msg is None: raise ValueError(f"no message {ident}") out = [_as_msg(folder, uid, msg)] seen = {f"{folder}|{uid}"} root = _root_id(msg) if root: for fold in ("INBOX", *_SENT_FOLDERS): for u2 in _thread_uids(m, fold, root): key = f"{fold}|{u2}" if key in seen: continue seen.add(key) other = _fetch(m, fold, u2) if other is not None: out.append(_as_msg(fold, u2, other)) out.sort(key=lambda x: x["ts"]) subject = next((x["subject"] for x in out if x["subject"]), "(no subject)") return {"thread_id": out[-1]["msg_id"], "account_addr": _addr(), "subject": subject, "messages": out} finally: try: m.logout() except Exception: # noqa: BLE001 pass def search_uids(query: str, limit: int = 5) -> list[str]: """'|' ids for messages matching a phrase (sender/subject), newest first. Empty query = the latest mail.""" m = _imap() try: m.select("INBOX") if query.strip(): typ, data = m.uid("search", None, "OR", "FROM", f'"{query}"', "SUBJECT", f'"{query}"') else: typ, data = m.uid("search", None, "ALL") uids = data[0].split() if typ == "OK" and data and data[0] else [] return [f"INBOX|{u.decode()}" for u in uids[-limit:][::-1]] finally: try: m.logout() except Exception: # noqa: BLE001 pass def attachment(ident: str, att_id: str, filename: str, thread_id: str = "") -> str: """Download one attachment (by its walk index) to ~/.jarvis/attachments// and return the local path.""" from voice import mail_view # lazy: mail_view imports us back folder, uid = _split_ident(ident) m = _imap() try: msg = _fetch(m, folder, uid) if msg is None: raise ValueError(f"no message {ident}") want = int(att_id) for idx, part in enumerate(msg.walk()): if idx != want: continue raw = part.get_payload(decode=True) or b"" name = filename or _decode(part.get_filename()) or f"part-{idx}" path = mail_view.dest_path(thread_id or ident, name) path.write_bytes(raw) return str(path) raise ValueError(f"no attachment {att_id} on {ident}") finally: try: m.logout() except Exception: # noqa: BLE001 pass def reply(ident: str, body: str, reply_all: bool = False) -> str: """Reply in-thread over SMTP, setting In-Reply-To + References so the correspondent's client threads it. reply_all keeps the other recipients.""" folder, uid = _split_ident(ident) m = _imap() try: orig = _fetch(m, folder, uid) finally: try: m.logout() except Exception: # noqa: BLE001 pass if orig is None: raise ValueError(f"no message {ident}") to = _decode(orig.get("From")) subj = _decode(orig.get("Subject")) if not subj.lower().startswith("re:"): subj = "Re: " + subj msg_id = (orig.get("Message-ID") or "").strip() refs = ((orig.get("References") or "") + " " + msg_id).strip() cc = "" if reply_all: seen = {_addr().lower()} | {a.lower() for _n, a in getaddresses([to]) if a} keep: list[str] = [] for _n, a in getaddresses([_decode(orig.get("To")), _decode(orig.get("Cc"))]): if a and a.lower() not in seen: seen.add(a.lower()) keep.append(a) cc = ", ".join(keep) host = os.environ.get("EMAIL_SMTP_HOST", "smtp.privateemail.com") port = int(os.environ.get("EMAIL_SMTP_PORT", "465")) msg = EmailMessage() msg["From"] = _addr() msg["To"] = to if cc: msg["Cc"] = cc msg["Subject"] = subj if msg_id: msg["In-Reply-To"] = msg_id msg["References"] = refs msg.set_content(body) rcpts = [a for _n, a in getaddresses([to, cc]) if a] with smtplib.SMTP_SSL(host, port, timeout=30) as s: s.login(_addr(), os.environ["EMAIL_PASSWORD"]) s.send_message(msg, to_addrs=rcpts) return f"Replied to {to}" + (f" (cc {cc})" if cc else "") + "." def build_server(): """Build and return the in-process ``email`` MCP server.""" import asyncio from claude_agent_sdk import tool, create_sdk_mcp_server @tool("email_send", "Send an email from Ahmed's mailbox. `to` and `cc` are " "comma-separated addresses; `cc` is optional.", {"to": str, "subject": str, "body": str, "cc": str}) async def email_send(args: dict) -> dict: to = str(args.get("to", "")).strip() if not to: return _text("email_send failed: no recipient.") try: out = await asyncio.to_thread( _send, to, str(args.get("subject", "")), str(args.get("body", "")), str(args.get("cc", "")).strip()) return _text(out) except Exception as e: # noqa: BLE001 return _text(f"email_send failed: {e}") @tool("email_search", "Search the inbox by a phrase matched against sender and subject " "(empty = latest messages). Returns a list with a UID for each; pass " "a UID to email_read to read the full message.", {"query": str, "limit": int}) async def email_search(args: dict) -> dict: try: limit = max(1, min(int(args.get("limit") or 10), 25)) out = await asyncio.to_thread( _search, str(args.get("query", "")), limit) return _text(out) except Exception as e: # noqa: BLE001 return _text(f"email_search failed: {e}") @tool("email_read", "Read the full body of one inbox message by its UID (from " "email_search).", {"uid": str}) async def email_read(args: dict) -> dict: uid = str(args.get("uid", "")).strip() if not uid: return _text("email_read failed: no uid.") try: out = await asyncio.to_thread(_read, uid) return _text(out) except Exception as e: # noqa: BLE001 return _text(f"email_read failed: {e}") return create_sdk_mcp_server( name="email", version="1.0.0", tools=[email_send, email_search, email_read], )