#!/usr/bin/env python3 """ ============================================================================== MCP server: EMAILY (vsechny schranky importovane z Microsoft Graph) Hybridni dotaz nad: - PostgreSQL 192.168.1.76 db=MongoEmaily tabulka=emails (fulltext tsvector - subject + sender + recipients + attachments + body, GIN index, ts_headline, ts_rank) - MongoDB 192.168.1.76 db=emaily kolekce= (puvodni dokumenty z parse_emails_graph_v1.3.py: headers, body_html, recipients[], attachments[], ...) Source: U:\\janssen\\EmailsImport\\enrich_fulltext_emails_v1.0.py Spusteni: python mcp_emaily.py (stdio MCP) ============================================================================== """ from __future__ import annotations import sys import traceback from datetime import datetime, timezone, timedelta from typing import Optional, Union import psycopg from mcp.server.fastmcp import FastMCP from pymongo import MongoClient MONGO_URI = "mongodb://192.168.1.76:27017" MONGO_DB = "emaily" PG_DSN = ("host=192.168.1.76 port=5432 dbname=MongoEmaily " "user=vladimir.buzalka password=Vlado7309208104++") DEFAULT_BODY_CHARS = 8000 MAX_BODY_CHARS = 200_000 SKIP_COLLECTIONS = {"attachments_index", "sync_state"} def log(msg: str) -> None: print(msg, file=sys.stderr, flush=True) try: mongo = MongoClient(MONGO_URI, serverSelectionTimeoutMS=5000) mongo.admin.command("ping") log(f"Mongo OK ({MONGO_URI})") except Exception as e: log(f"Mongo connection failed: {e}") sys.exit(1) try: _t = psycopg.connect(PG_DSN, connect_timeout=10) _t.close() log("Postgres OK") except Exception as e: log(f"Postgres connection failed: {e}") sys.exit(1) def pg_conn(): return psycopg.connect(PG_DSN, connect_timeout=10) def serialize(obj): if isinstance(obj, datetime): return obj.isoformat() if isinstance(obj, bytes): return obj.decode("utf-8", errors="replace") if isinstance(obj, dict): return {k: serialize(v) for k, v in obj.items()} if isinstance(obj, list): return [serialize(v) for v in obj] return obj def normalize_mailbox(mailbox: Optional[Union[str, list]]) -> Optional[list[str]]: if mailbox is None or mailbox == "" or mailbox == []: return None if isinstance(mailbox, str): return [mailbox] return list(mailbox) def parse_since(s: Optional[str]) -> Optional[datetime]: if not s: return None try: if "T" in s: return datetime.fromisoformat(s.replace("Z", "+00:00")) return datetime.strptime(s, "%Y-%m-%d").replace(tzinfo=timezone.utc) except Exception as e: raise ValueError(f"Bad date {s!r}: {e}") # --- MCP -------------------------------------------------------------------- mcp = FastMCP("emaily") @mcp.tool() def ping() -> dict: """Quick health check. Reports Mongo + Postgres connectivity, total mailboxes, PG indexed emails count, ok/error breakdown. """ try: info = mongo.admin.command("buildInfo") mailboxes = [c for c in mongo[MONGO_DB].list_collection_names() if c not in SKIP_COLLECTIONS] mongo_counts = {} for mb in mailboxes: mongo_counts[mb] = mongo[MONGO_DB][mb].estimated_document_count() with pg_conn() as pg, pg.cursor() as cur: cur.execute("SELECT mailbox, ok, count(*) FROM emails " "GROUP BY mailbox, ok ORDER BY mailbox, ok") rows = cur.fetchall() pg_summary: dict = {} for mb, ok, c in rows: pg_summary.setdefault(mb, {})[("ok" if ok else "error")] = c return { "status": "ok", "mongo_version": info.get("version"), "mailboxes": mailboxes, "mongo_email_count": mongo_counts, "pg_indexed_per_mailbox": pg_summary, } except Exception as e: log(traceback.format_exc()) return {"status": "error", "error": str(e)} @mcp.tool() def list_mailboxes() -> dict: """Overview of all mailboxes — totals, indexed coverage, earliest/latest received_at, top senders by volume. Use to understand the corpus before searching. """ out = {} try: mailboxes = [c for c in mongo[MONGO_DB].list_collection_names() if c not in SKIP_COLLECTIONS] for mb in mailboxes: with pg_conn() as pg, pg.cursor() as cur: cur.execute(""" SELECT count(*) FILTER (WHERE ok) AS ok, count(*) AS total, min(received_at) AS first_at, max(received_at) AS last_at, count(*) FILTER (WHERE has_attachments) AS with_att FROM emails WHERE mailbox = %s """, (mb,)) ok, total, first_at, last_at, with_att = cur.fetchone() cur.execute(""" SELECT sender_email, count(*) c FROM emails WHERE mailbox = %s AND sender_email IS NOT NULL GROUP BY sender_email ORDER BY c DESC LIMIT 5 """, (mb,)) top_senders = [{"email": s, "count": c} for s, c in cur.fetchall()] out[mb] = { "indexed_ok": ok, "indexed_total": total, "with_attachments": with_att, "first_received": serialize(first_at), "last_received": serialize(last_at), "top_senders": top_senders, } return {"mailboxes": out} except Exception as e: log(traceback.format_exc()) return {"error": str(e)} @mcp.tool() def search( query: str, mailbox: Optional[Union[str, list]] = None, since: Optional[str] = None, until: Optional[str] = None, folder_contains: Optional[str] = None, sender_contains: Optional[str] = None, has_attachments: Optional[bool] = None, limit: int = 20, ) -> dict: """PRIMARY TOOL — fulltext search across all indexed emails. Index includes: subject, sender (email + name), recipients (to/cc), attachment filenames, AND full body text. query: websearch_to_tsquery syntax: invoice payment -> AND "lot expiration" -> phrase SAE OR "serious adverse" -> OR urgent -newsletter -> exclude mailbox: one mailbox string or list (e.g. "vbuzalka@its.jnj.com"). None = all. since/until: ISO date "YYYY-MM-DD" on received_at folder_contains: substring match against folder_path (case-insensitive) sender_contains: substring match against sender_email OR sender_name (case-insensitive) has_attachments: True / False / None (any) limit: max 100 Returns ranked results with `snippet` showing matches highlighted as <<...>>. Use `read_email` to fetch full body of any hit. """ try: mboxes = normalize_mailbox(mailbox) since_dt = parse_since(since) until_dt = parse_since(until) limit = min(max(1, limit), 100) sql = """ WITH q AS ( SELECT websearch_to_tsquery('soubory'::regconfig, %(query)s) AS tsq ) SELECT e.id, e.mailbox, e.message_id, e.conversation_id, e.folder_path, e.subject, e.sender_email, e.sender_name, e.to_addrs, e.cc_addrs, e.received_at, e.sent_at, e.is_read, e.has_attachments, e.attachment_count, e.attachments_summary, e.body_length, e.body_source, ts_rank(e.tsv, q.tsq) AS rank, ts_headline('soubory'::regconfig, left(coalesce(e.body, e.subject), 200000), q.tsq, 'MaxFragments=3, MinWords=4, MaxWords=18, ' 'StartSel=<<, StopSel=>>, FragmentDelimiter= ... ') AS snippet FROM emails e, q WHERE e.tsv @@ q.tsq AND e.ok = TRUE AND (%(mboxes)s::text[] IS NULL OR e.mailbox = ANY(%(mboxes)s::text[])) AND (%(since)s::timestamptz IS NULL OR e.received_at >= %(since)s::timestamptz) AND (%(until)s::timestamptz IS NULL OR e.received_at < %(until)s::timestamptz) AND (%(folder)s::text IS NULL OR e.folder_path ILIKE %(folder_like)s) AND (%(sender)s::text IS NULL OR e.sender_email ILIKE %(sender_like)s OR e.sender_name ILIKE %(sender_like)s) AND (%(has_att)s::boolean IS NULL OR e.has_attachments = %(has_att)s::boolean) ORDER BY rank DESC, e.received_at DESC NULLS LAST LIMIT %(limit)s """ params = { "query": query, "mboxes": mboxes, "since": since_dt, "until": until_dt, "folder": folder_contains, "folder_like": f"%{folder_contains}%" if folder_contains else None, "sender": sender_contains, "sender_like": f"%{sender_contains}%" if sender_contains else None, "has_att": has_attachments, "limit": limit, } with pg_conn() as pg, pg.cursor() as cur: cur.execute(sql, params) cols = [c.name for c in cur.description] rows = [dict(zip(cols, r)) for r in cur.fetchall()] results = [] for r in rows: results.append({ "mailbox": r["mailbox"], "message_id": r["message_id"], "conversation_id": r["conversation_id"], "folder": r["folder_path"], "subject": r["subject"], "from": (f"{r['sender_name']} <{r['sender_email']}>" if r["sender_name"] else r["sender_email"]), "to": r["to_addrs"], "cc": r["cc_addrs"], "received_at": serialize(r["received_at"]), "is_read": r["is_read"], "has_attachments": r["has_attachments"], "attachment_count": r["attachment_count"], "attachments": r["attachments_summary"], "body_length": r["body_length"], "body_source": r["body_source"], "rank": round(float(r["rank"]), 5), "snippet": (r["snippet"] or "").strip(), }) return { "query": query, "filters": {"mailbox": mboxes, "since": since, "until": until, "folder_contains": folder_contains, "sender_contains": sender_contains, "has_attachments": has_attachments, "limit": limit}, "count": len(results), "results": results, "tip": "Use read_email(mailbox=..., message_id=...) for full body or thread.", } except Exception as e: log(traceback.format_exc()) return {"error": str(e), "query": query} @mcp.tool() def read_email( message_id: Optional[str] = None, mailbox: Optional[str] = None, offset: int = 0, length: int = DEFAULT_BODY_CHARS, around_match: Optional[str] = None, include_html: bool = False, ) -> dict: """Read one email — full plain text body + metadata. Identify by `message_id` (Internet Message-ID, the _id in Mongo). `mailbox` narrows the lookup if the same Message-ID appears in multiple mailboxes (e.g. you got copies in both work and personal accounts). offset, length: slice the body. length max 200000. around_match: case-insensitive substring; returns up to 3 windows of ~1000 chars centered on matches, instead of a flat slice. include_html: also return raw body_html from Mongo (typically large — only if you really need the original markup). """ if not message_id: return {"error": "Provide message_id."} try: length = min(max(1, length), MAX_BODY_CHARS) sql = """ SELECT id, mailbox, message_id, graph_id, conversation_id, folder_path, subject, sender_email, sender_name, to_addrs, cc_addrs, bcc_addrs, sent_at, received_at, modified_at, is_read, is_draft, has_attachments, attachment_count, attachments_summary, body, body_length, body_source, extractor_version, extracted_at, ok, error FROM emails WHERE message_id = %s """ params = [message_id] if mailbox: sql += " AND mailbox = %s" params.append(mailbox) sql += " LIMIT 1" with pg_conn() as pg, pg.cursor() as cur: cur.execute(sql, params) row = cur.fetchone() cols = [c.name for c in cur.description] if not row: return {"error": "Email not found.", "message_id": message_id, "mailbox": mailbox} rec = dict(zip(cols, row)) body = rec.get("body") or "" if around_match and body: needle = around_match.lower() hay = body.lower() windows = [] start = 0 while len(windows) < 3: pos = hay.find(needle, start) if pos < 0: break lo = max(0, pos - 400) hi = min(len(body), pos + 600) windows.append({"offset": lo, "text": body[lo:hi]}) start = pos + len(needle) body_out = None slice_info = {"mode": "around_match", "match": around_match, "windows_found": len(windows), "windows": windows} else: body_out = body[offset:offset + length] slice_info = { "mode": "slice", "offset": offset, "length_returned": len(body_out), "has_more": offset + length < len(body), "next_offset": offset + length if offset + length < len(body) else None, } out = { "mailbox": rec["mailbox"], "message_id": rec["message_id"], "conversation_id": rec["conversation_id"], "folder": rec["folder_path"], "subject": rec["subject"], "from": (f"{rec['sender_name']} <{rec['sender_email']}>" if rec["sender_name"] else rec["sender_email"]), "to": rec["to_addrs"], "cc": rec["cc_addrs"], "bcc": rec["bcc_addrs"], "received_at": serialize(rec["received_at"]), "sent_at": serialize(rec["sent_at"]), "is_read": rec["is_read"], "is_draft": rec["is_draft"], "has_attachments": rec["has_attachments"], "attachment_count": rec["attachment_count"], "attachments": rec["attachments_summary"], "body_length": rec["body_length"], "body_source": rec["body_source"], "extractor_version": rec["extractor_version"], "ok": rec["ok"], "error": rec["error"], } if body_out is not None: out["body"] = body_out out["slice"] = slice_info if include_html: mdoc = mongo[MONGO_DB][rec["mailbox"]].find_one( {"_id": rec["message_id"]}, {"body_html": 1, "attachments": 1}) if mdoc: out["body_html"] = mdoc.get("body_html") out["attachments_detail"] = mdoc.get("attachments") return out except Exception as e: log(traceback.format_exc()) return {"error": str(e)} @mcp.tool() def by_sender( sender: str, mailbox: Optional[Union[str, list]] = None, since: Optional[str] = None, has_attachments: Optional[bool] = None, limit: int = 30, ) -> dict: """List emails from a specific sender (substring match on sender_email or sender_name, case-insensitive). Use for "what did X send me" or "all newsletters from Y". Returned sorted by received_at DESC. """ try: mboxes = normalize_mailbox(mailbox) since_dt = parse_since(since) limit = min(max(1, limit), 200) sql = """ SELECT mailbox, message_id, subject, sender_email, sender_name, to_addrs, folder_path, received_at, has_attachments, attachment_count, attachments_summary, body_length FROM emails WHERE ok = TRUE AND (sender_email ILIKE %(s)s OR sender_name ILIKE %(s)s) AND (%(mboxes)s::text[] IS NULL OR mailbox = ANY(%(mboxes)s::text[])) AND (%(since)s::timestamptz IS NULL OR received_at >= %(since)s::timestamptz) AND (%(has_att)s::boolean IS NULL OR has_attachments = %(has_att)s::boolean) ORDER BY received_at DESC NULLS LAST LIMIT %(limit)s """ with pg_conn() as pg, pg.cursor() as cur: cur.execute(sql, {"s": f"%{sender}%", "mboxes": mboxes, "since": since_dt, "has_att": has_attachments, "limit": limit}) cols = [c.name for c in cur.description] rows = [dict(zip(cols, r)) for r in cur.fetchall()] for r in rows: r["received_at"] = serialize(r["received_at"]) return {"sender_match": sender, "count": len(rows), "results": rows} except Exception as e: log(traceback.format_exc()) return {"error": str(e)} @mcp.tool() def recent_emails( mailbox: Optional[Union[str, list]] = None, days: int = 7, folder_contains: Optional[str] = None, has_attachments: Optional[bool] = None, limit: int = 30, ) -> dict: """List recent emails (by received_at). Use for "what came in today/this week". days=0 to ignore time window (just top-N newest). """ try: mboxes = normalize_mailbox(mailbox) limit = min(max(1, limit), 200) since_dt = None if days and days > 0: since_dt = datetime.now(timezone.utc) - timedelta(days=days) sql = """ SELECT mailbox, message_id, subject, sender_email, sender_name, folder_path, received_at, has_attachments, attachment_count, attachments_summary, body_length, is_read FROM emails WHERE ok = TRUE AND (%(mboxes)s::text[] IS NULL OR mailbox = ANY(%(mboxes)s::text[])) AND (%(since)s::timestamptz IS NULL OR received_at >= %(since)s::timestamptz) AND (%(folder)s::text IS NULL OR folder_path ILIKE %(folder_like)s) AND (%(has_att)s::boolean IS NULL OR has_attachments = %(has_att)s::boolean) ORDER BY received_at DESC NULLS LAST LIMIT %(limit)s """ with pg_conn() as pg, pg.cursor() as cur: cur.execute(sql, { "mboxes": mboxes, "since": since_dt, "folder": folder_contains, "folder_like": f"%{folder_contains}%" if folder_contains else None, "has_att": has_attachments, "limit": limit, }) cols = [c.name for c in cur.description] rows = [dict(zip(cols, r)) for r in cur.fetchall()] for r in rows: r["received_at"] = serialize(r["received_at"]) return {"days": days, "count": len(rows), "results": rows} except Exception as e: log(traceback.format_exc()) return {"error": str(e)} @mcp.tool() def conversation_thread(conversation_id: str, limit: int = 50) -> dict: """Return all emails in one Outlook conversation thread (conversation_id from Graph). Ordered chronologically. Use to see the full back-and-forth on a topic. """ try: limit = min(max(1, limit), 200) with pg_conn() as pg, pg.cursor() as cur: cur.execute(""" SELECT mailbox, message_id, subject, sender_email, sender_name, to_addrs, received_at, folder_path, body_length, has_attachments, attachments_summary FROM emails WHERE conversation_id = %s AND ok = TRUE ORDER BY received_at ASC NULLS LAST LIMIT %s """, (conversation_id, limit)) cols = [c.name for c in cur.description] rows = [dict(zip(cols, r)) for r in cur.fetchall()] for r in rows: r["received_at"] = serialize(r["received_at"]) return {"conversation_id": conversation_id, "count": len(rows), "thread": rows} except Exception as e: log(traceback.format_exc()) return {"error": str(e)} @mcp.tool() def find_attachment( name_contains: str, mailbox: Optional[Union[str, list]] = None, since: Optional[str] = None, limit: int = 30, ) -> dict: """Find emails whose attachment filename contains the substring (case-insensitive). Use for "find emails with that protocol PDF" or "any invoice attachments". Returns emails ordered by received_at DESC. """ try: mboxes = normalize_mailbox(mailbox) since_dt = parse_since(since) limit = min(max(1, limit), 200) sql = """ SELECT mailbox, message_id, subject, sender_email, sender_name, received_at, attachment_count, attachments_summary, folder_path FROM emails WHERE ok = TRUE AND has_attachments = TRUE AND attachments_summary ILIKE %(s)s AND (%(mboxes)s::text[] IS NULL OR mailbox = ANY(%(mboxes)s::text[])) AND (%(since)s::timestamptz IS NULL OR received_at >= %(since)s::timestamptz) ORDER BY received_at DESC NULLS LAST LIMIT %(limit)s """ with pg_conn() as pg, pg.cursor() as cur: cur.execute(sql, {"s": f"%{name_contains}%", "mboxes": mboxes, "since": since_dt, "limit": limit}) cols = [c.name for c in cur.description] rows = [dict(zip(cols, r)) for r in cur.fetchall()] for r in rows: r["received_at"] = serialize(r["received_at"]) return {"name_match": name_contains, "count": len(rows), "results": rows} except Exception as e: log(traceback.format_exc()) return {"error": str(e)} @mcp.tool() def top_senders( mailbox: Optional[Union[str, list]] = None, since: Optional[str] = None, limit: int = 20, ) -> dict: """Top senders by volume (count of received emails). Optionally limit by mailbox or date window. Use for "who emails me most" or "top senders this month". """ try: mboxes = normalize_mailbox(mailbox) since_dt = parse_since(since) limit = min(max(1, limit), 100) sql = """ SELECT sender_email, count(*) AS c, max(received_at) AS last_at FROM emails WHERE ok = TRUE AND sender_email IS NOT NULL AND (%(mboxes)s::text[] IS NULL OR mailbox = ANY(%(mboxes)s::text[])) AND (%(since)s::timestamptz IS NULL OR received_at >= %(since)s::timestamptz) GROUP BY sender_email ORDER BY c DESC LIMIT %(limit)s """ with pg_conn() as pg, pg.cursor() as cur: cur.execute(sql, {"mboxes": mboxes, "since": since_dt, "limit": limit}) rows = [{"sender_email": s, "count": c, "last_at": serialize(t)} for s, c, t in cur.fetchall()] return {"count": len(rows), "results": rows} except Exception as e: log(traceback.format_exc()) return {"error": str(e)} @mcp.tool() def pipeline_status(mailbox: Optional[Union[str, list]] = None) -> dict: """End-to-end status of the email-ingest pipeline per mailbox. Reports, for each mailbox, where it stands in the 5-step pipeline: 1. parse_emails_graph -> mongo_total 2. (refetch text bodies) -> body_text_missing (legacy v1.3 emails) 3. download_attachments -> attach_done / attach_pending attach_missing (404 — marked, won't retry) attach_reference (OneDrive/SharePoint link, no content) 4. unwrap_smime -> smime_p7m_total / smime_unwrapped / smime_pending smime_p7s_count (informational; not unwrapped by design) 5. enrich_fulltext -> pg_indexed Plus: - permanently_deleted (marked by delta sync) Use this instead of running multiple Mongo count queries by hand. Returns one row per mailbox; if `mailbox` is given, returns just those rows. """ try: mbs = normalize_mailbox(mailbox) all_mb = [c for c in mongo[MONGO_DB].list_collection_names() if c not in SKIP_COLLECTIONS] targets = [m for m in all_mb if (mbs is None or m in mbs)] # PG counts in one pass pg_counts: dict[str, int] = {} with pg_conn() as pg, pg.cursor() as cur: cur.execute("SELECT mailbox, count(*) FROM emails " "WHERE ok = true GROUP BY mailbox") for mb, c in cur.fetchall(): pg_counts[mb] = c out = {} for mb in targets: col = mongo[MONGO_DB][mb] mongo_total = col.estimated_document_count() with_att = col.count_documents({"has_attachments": True}) attach_pending = col.count_documents({ "has_attachments": True, "attachments": {"$elemMatch": { "is_inline": False, "file_hash": {"$exists": False}, "attachment_missing": {"$ne": True}, "attachment_reference": {"$ne": True}, }}, }) attach_missing = col.count_documents({ "attachments.attachment_missing": True, }) attach_reference = col.count_documents({ "attachments.attachment_reference": True, }) attach_done = with_att - attach_pending - attach_missing - attach_reference smime_p7m_total = col.count_documents( {"attachments.filename": {"$regex": r"^smime\.p7m$", "$options": "i"}} ) smime_unwrapped = col.count_documents({ "attachments.filename": {"$regex": r"^smime\.p7m$", "$options": "i"}, "smime_unwrapped": True, }) smime_p7s_count = col.count_documents( {"attachments.filename": {"$regex": r"^smime\.p7s$", "$options": "i"}} ) body_text_missing = col.count_documents({ "body_html": {"$in": [None, ""]}, "body_text": {"$exists": False}, "graph_id": {"$exists": True}, }) permanently_deleted = col.count_documents({"permanently_deleted": True}) out[mb] = { "mongo_total": mongo_total, "with_attachments": with_att, "attach_done": attach_done, "attach_pending": attach_pending, "attach_missing": attach_missing, "attach_reference": attach_reference, "smime_p7m_total": smime_p7m_total, "smime_unwrapped": smime_unwrapped, "smime_pending": smime_p7m_total - smime_unwrapped, "smime_p7s_count": smime_p7s_count, "body_text_missing": body_text_missing, "pg_indexed": pg_counts.get(mb, 0), "permanently_deleted": permanently_deleted, } return {"mailboxes": out} except Exception as e: log(traceback.format_exc()) return {"error": str(e)} @mcp.tool() def sync_state_overview(mailbox: Optional[Union[str, list]] = None) -> dict: """Delta-sync state across mailboxes (collection `emaily.sync_state`). For each (mailbox, folder) pair shows: deltaLink present?, last_run_at, cumulative new/sync/removed/run_count. Use to confirm a mailbox is incrementally synced and to spot folders that haven't run in a while. """ try: sync_col = mongo[MONGO_DB]["sync_state"] q: dict = {} mbs = normalize_mailbox(mailbox) if mbs: q["mailbox"] = {"$in": mbs} cursor = sync_col.find(q, { "mailbox": 1, "folder_path": 1, "folder_id": 1, "delta_link": 1, "last_run_at": 1, "cumulative_new": 1, "cumulative_sync": 1, "cumulative_removed": 1, "run_count": 1, }).sort([("mailbox", 1), ("folder_path", 1)]) by_mailbox: dict[str, list] = {} for d in cursor: row = { "folder_path": d.get("folder_path"), "folder_id": d.get("folder_id"), "has_delta_link": bool(d.get("delta_link")), "last_run_at": serialize(d.get("last_run_at")), "cumulative_new": d.get("cumulative_new", 0), "cumulative_sync": d.get("cumulative_sync", 0), "cumulative_removed": d.get("cumulative_removed", 0), "run_count": d.get("run_count", 0), } by_mailbox.setdefault(d["mailbox"], []).append(row) # mailboxes that have collections but ZERO sync_state entries all_mb = {c for c in mongo[MONGO_DB].list_collection_names() if c not in SKIP_COLLECTIONS} not_synced = sorted(all_mb - set(by_mailbox.keys())) if mbs: not_synced = [m for m in not_synced if m in mbs] return { "mailboxes": by_mailbox, "never_delta_synced": not_synced, } except Exception as e: log(traceback.format_exc()) return {"error": str(e)} if __name__ == "__main__": log("MCP emaily server started (FastMCP)") mcp.run()