""" ======================================================================= Název: mcp_vbcz_email_v1.6.py Verze: 1.6 Datum: 2026-06-09 Popis: MCP server pro schránku vladimir.buzalka@buzalka.cz přes Microsoft Graph API (application permissions). Credentials sdílí s EmailMessagingGraph.py (Knihovny/). Tooly: - list_emails — seznam posledních emailů (+ preview) - get_email — plné tělo + metadata emailu - list_attachments — přílohy emailu (název, velikost, typ) - get_attachment — stáhne přílohu na disk, vrátí cestu - create_draft_eml — vygeneruje .eml draft (X-Unsent) k ruční kontrole a odeslání v Outlooku (+ přílohy) - create_event — vytvoří událost v kalendáři (+ připomínka) - create_contact — vytvoří kontakt ve schránce Změny v1.1: + create_draft_eml (generování rozepsaných emailů) Změny v1.2: + create_draft_eml: parametr `attachments` Změny v1.3: * OPRAVA kódování: base64 (cte) + policy.SMTP (CRLF) Změny v1.4: + create_draft_eml: `original_message_id` (forward styl) Změny v1.5: + create_draft_eml: `strip_self_forward` (default True) Změny v1.6: + create_event — zapíše událost do kalendáře schránky (start/end v lokálním čase, časové pásmo, připomínka N minut předem). Vrací id + webLink. Vyžaduje Graph oprávnění Calendars.ReadWrite (application). + create_contact — založí kontakt ve schránce (jméno, e-mail, telefony, firma, funkce, poznámka). Vrací id. Vyžaduje Contacts.ReadWrite (application). + helper _graph_post (POST na Graph). POZN.: e-maily lékařům/CTA se i nadále generují jako .eml drafty (odesílají se přes JNJ server), NE přes tuto schránku. Kalendář a kontakty se píšou do osobní schránky vladimir.buzalka@buzalka.cz. Spuštění: python mcp_vbcz_email_v1.6.py ======================================================================= """ import base64 import mimetypes import sys from datetime import datetime from email.message import EmailMessage from email.policy import SMTP as SMTP_POLICY from functools import lru_cache from pathlib import Path from typing import Optional, Union import msal import requests from mcp.server.fastmcp import FastMCP # ── Konfigurace ──────────────────────────────────────────────────────── TENANT_ID = "7d269944-37a4-43a1-8140-c7517dc426e9" CLIENT_ID = "4b222bfd-78c9-4239-a53f-43006b3ed07f" CLIENT_SECRET = "Txg8Q~MjhocuopxsJyJBhPmDfMxZ2r5WpTFj1dfk" MAILBOX = "vladimir.buzalka@buzalka.cz" AUTHORITY = f"https://login.microsoftonline.com/{TENANT_ID}" SCOPE = ["https://graph.microsoft.com/.default"] GRAPH_BASE = f"https://graph.microsoft.com/v1.0/users/{MAILBOX}" # Výchozí časové pásmo pro události kalendáře (Windows TZ název pro Graph) DEFAULT_TIMEZONE = "Central European Standard Time" DOWNLOADS_DIR = Path(__file__).resolve().parent / "downloads" DOWNLOADS_DIR.mkdir(exist_ok=True) # Výchozí cíl pro generované draft .eml soubory DEFAULT_DRAFT_DIR = Path(r"u:\Dropbox\!!!Days\Downloads Z230") # Výchozí odesílatel + standardní podpis (ICON / Janssen) DEFAULT_FROM = "vbuzalka@its.jnj.com" SIGNATURE_HTML = """

S pozdravem

MUDr. Vladimír BUZALKA

ICON plc
Performing Local Trial Management Services for Janssen – Cilag s.r.o.
Global Clinical Operations
Mobile: +420 775 735 276
Fax: +420 227 012 284
E-mail: vbuzalka@its.jnj.com, vladimir.buzalka@iconplc.com

""" SIGNATURE_TEXT = ( "\nS pozdravem\n\n" "MUDr. Vladimír BUZALKA\n" "ICON plc\n" "Performing Local Trial Management Services for Janssen – Cilag s.r.o.\n" "Global Clinical Operations\n" "Mobile: +420 775 735 276\n" "Fax: +420 227 012 284\n" "E-mail: vbuzalka@its.jnj.com, vladimir.buzalka@iconplc.com\n" ) mcp = FastMCP("vbcz-email") def log(msg: str): print(msg, file=sys.stderr, flush=True) @lru_cache(maxsize=1) def _get_token() -> str: app = msal.ConfidentialClientApplication( CLIENT_ID, authority=AUTHORITY, client_credential=CLIENT_SECRET, ) token = app.acquire_token_for_client(scopes=SCOPE) if "access_token" not in token: raise RuntimeError(f"Graph auth failed: {token}") return token["access_token"] def _headers() -> dict: # lru_cache drží token dokud server běží; Graph tokeny platí 1 hod return { "Authorization": f"Bearer {_get_token()}", "Content-Type": "application/json", } def _graph_get(path: str, params: dict = None) -> dict: r = requests.get( f"{GRAPH_BASE}{path}", headers=_headers(), params=params or {}, timeout=30, ) if not r.ok: raise RuntimeError(f"Graph error [{r.status_code}]: {r.text[:300]}") return r.json() def _graph_post(path: str, payload: dict) -> dict: r = requests.post( f"{GRAPH_BASE}{path}", headers=_headers(), json=payload, timeout=30, ) if not r.ok: raise RuntimeError(f"Graph error [{r.status_code}]: {r.text[:400]}") return r.json() def _msg_summary(m: dict) -> dict: sender = m.get("from", {}).get("emailAddress", {}) return { "id": m.get("id"), "subject": m.get("subject"), "from": sender.get("address"), "from_name": sender.get("name"), "date": m.get("receivedDateTime"), "is_read": m.get("isRead"), "has_attachments": m.get("hasAttachments"), "preview": m.get("bodyPreview", "")[:200], "folder": m.get("parentFolderId"), } # ── MCP tooly ────────────────────────────────────────────────────────── @mcp.tool() def list_emails( folder: str = "inbox", limit: int = 10, search: Optional[str] = None, from_email: Optional[str] = None, unread_only: bool = False, ) -> list[dict]: """Vrátí seznam posledních emailů ze schránky vladimir.buzalka@buzalka.cz. - folder: 'inbox' / 'sentitems' / 'drafts' / 'deleteditems' / ID složky - limit: max počet (1–50) - search: fulltext hledání (subject nebo tělo) - from_email: filtr odesílatele (substring) - unread_only: jen nepřečtené """ limit = max(1, min(limit, 50)) filters = [] if from_email: filters.append(f"contains(from/emailAddress/address,'{from_email}')") if unread_only: filters.append("isRead eq false") params = { "$top": limit, "$orderby": "receivedDateTime desc", "$select": "id,subject,from,receivedDateTime,isRead,hasAttachments,bodyPreview,parentFolderId", } if filters: params["$filter"] = " and ".join(filters) if search: params["$search"] = f'"{search}"' params.pop("$filter", None) # $search a $filter nelze kombinovat data = _graph_get(f"/mailFolders/{folder}/messages", params) return [_msg_summary(m) for m in data.get("value", [])] @mcp.tool() def get_email(message_id: str) -> dict: """Načte plný email (tělo + metadata) podle message_id z list_emails.""" m = _graph_get(f"/messages/{message_id}") body = m.get("body", {}) sender = m.get("from", {}).get("emailAddress", {}) to_list = [ r["emailAddress"]["address"] for r in m.get("toRecipients", []) ] return { "id": m.get("id"), "subject": m.get("subject"), "from": sender.get("address"), "from_name": sender.get("name"), "to": to_list, "date": m.get("receivedDateTime"), "is_read": m.get("isRead"), "has_attachments": m.get("hasAttachments"), "body_type": body.get("contentType"), # Text / HTML "body": body.get("content", ""), "folder": m.get("parentFolderId"), } @mcp.tool() def list_attachments(message_id: str) -> list[dict]: """Vrátí seznam příloh emailu (název, velikost, contentType, attachmentId).""" data = _graph_get(f"/messages/{message_id}/attachments", {"$select": "id,name,size,contentType,isInline"}) return [ { "attachment_id": a.get("id"), "name": a.get("name"), "size_bytes": a.get("size"), "content_type": a.get("contentType"), "is_inline": a.get("isInline"), } for a in data.get("value", []) ] @mcp.tool() def get_attachment(message_id: str, attachment_id: str) -> dict: """Stáhne přílohu na disk (downloads/) a vrátí cestu + velikost. attachment_id získáš z list_attachments.""" a = _graph_get(f"/messages/{message_id}/attachments/{attachment_id}") content_b64 = a.get("contentBytes") if not content_b64: raise RuntimeError("Příloha neobsahuje contentBytes (možná referenční typ).") data = base64.b64decode(content_b64) name = a.get("name", "attachment") out_path = DOWNLOADS_DIR / name # Pokud soubor existuje, přidej suffix stem, suffix = out_path.stem, out_path.suffix counter = 1 while out_path.exists(): out_path = DOWNLOADS_DIR / f"{stem}_{counter}{suffix}" counter += 1 out_path.write_bytes(data) log(f"saved attachment: {out_path}") return { "path": str(out_path), "name": name, "size_bytes": len(data), "content_type": a.get("contentType"), } def _slug(text: str, maxlen: int = 40) -> str: """Bezpečný kousek názvu souboru z textu (jen ASCII, _ místo zbytku).""" keep = [] for ch in text: if ch.isalnum(): keep.append(ch) elif ch in " -_": keep.append("_") s = "".join(keep).strip("_") # zkrátit vícenásobné podtržítka while "__" in s: s = s.replace("__", "_") return s[:maxlen] or "draft" def _parse_attachments(attachments: Union[str, list, None]) -> list[Path]: """Z parametru attachments (str s ; nebo seznam) udělá seznam ověřených cest.""" if not attachments: return [] if isinstance(attachments, str): # akceptuj oddělovač ; nebo nový řádek raw = [p.strip() for p in attachments.replace("\n", ";").split(";")] else: raw = [str(p).strip() for p in attachments] paths = [] for p in raw: if not p: continue fp = Path(p) if not fp.exists(): raise FileNotFoundError(f"Příloha nenalezena: {p}") if not fp.is_file(): raise ValueError(f"Příloha není soubor: {p}") paths.append(fp) return paths def _strip_html_wrapper(html: str) -> str: """Vytáhne vnitřek … (nebo vrátí původní), aby šel vnořit.""" import re m = re.search(r"]*>(.*)", html, re.IGNORECASE | re.DOTALL) if m: return m.group(1) # bez body tagu — odstraň aspoň / html = re.sub(r"]*>", "", html, flags=re.IGNORECASE) html = re.sub(r"]*>.*?", "", html, flags=re.IGNORECASE | re.DOTALL) return html def _esc(text: str) -> str: """Minimální HTML escape pro hlavičkové hodnoty.""" return (text.replace("&", "&").replace("<", "<").replace(">", ">")) def _extract_inner_forward_html(inner_html: str) -> Optional[str]: """Z přeposlaného (FW) HTML těla vytáhne JEN vnitřní forwardnutý originál: najde oddělovač forward hlavičky ("From:" blok, příp. border-top div) a vrátí fragment od něj dál. Tím odřízne horní self-forward obal (prázdný forward + duplicitní podpis odesílatele). None = nenalezeno.""" import re # 1) preferuj Outlook oddělovač (border-top) bezprostředně před "From:" cut = -1 mfrom = re.search(r"(?is)(?:\s*)?From\s*:", inner_html) if mfrom: idx = mfrom.start() # najdi nejbližší otevírací blokový tag před "From:" (div má přednost) d = inner_html.rfind(" Optional[str]: """Textová varianta — vrátí vše od prvního řádku začínajícího 'From:'.""" import re m = re.search(r"(?im)^\s*From\s*:", text) if not m: return None return text[m.start():] def _build_quote_block(message_id: str, fmt: str, strip_self_forward: bool = False) -> str: """Stáhne původní email z Graphu a sestaví forward-style blok (oddělovač + From/Sent/To/Subject + původní tělo). strip_self_forward=True: pokud je `message_id` email přeposlaný k sobě (FW:), vloží JEN vnitřní skutečně odeslaný originál (i s jeho vlastní From/Sent/To/Subject hlavičkou) — odřízne horní self-forward obal. Když vnitřní forward nenajde, spadne zpět na standardní chování.""" m = _graph_get(f"/messages/{message_id}") body = m.get("body", {}) orig_type = (body.get("contentType") or "text").lower() # 'html' / 'text' orig_content = body.get("content", "") if strip_self_forward: if orig_type == "html": frag = _extract_inner_forward_html(_strip_html_wrapper(orig_content)) else: frag = _extract_inner_forward_text(orig_content) if frag: return frag # už obsahuje vlastní From/Sent/To/Subject + originál sender = m.get("from", {}).get("emailAddress", {}) from_name = sender.get("name", "") or "" from_addr = sender.get("address", "") or "" to_list = ", ".join( r["emailAddress"]["address"] for r in m.get("toRecipients", []) ) cc_list = ", ".join( r["emailAddress"]["address"] for r in m.get("ccRecipients", []) ) sent = m.get("sentDateTime") or m.get("receivedDateTime") or "" subj = m.get("subject", "") or "" if fmt == "html": hdr = ( '
' f'

From: {_esc(from_name)} ' f'<{_esc(from_addr)}>
' f'Sent: {_esc(sent)}
' f'To: {_esc(to_list)}
' ) if cc_list: hdr += f'Cc: {_esc(cc_list)}
' hdr += f'Subject: {_esc(subj)}

' if orig_type == "html": orig_html = _strip_html_wrapper(orig_content) else: orig_html = "
" \
                        + _esc(orig_content) + "
" return hdr + orig_html else: lines = [ "", "________________________________", f"From: {from_name} <{from_addr}>", f"Sent: {sent}", f"To: {to_list}", ] if cc_list: lines.append(f"Cc: {cc_list}") lines.append(f"Subject: {subj}") lines.append("") # pokud je originál HTML, dej aspoň hrubý text (bez tagů) if orig_type == "html": import re txt = re.sub(r"<[^>]+>", "", orig_content) else: txt = orig_content lines.append(txt) return "\n".join(lines) @mcp.tool() def create_draft_eml( to: str, subject: str, body: str, cc: Optional[str] = None, attachments: Optional[str] = None, original_message_id: Optional[str] = None, strip_self_forward: bool = True, from_addr: str = DEFAULT_FROM, body_format: str = "html", add_signature: bool = True, filename: Optional[str] = None, output_dir: Optional[str] = None, ) -> dict: """Vygeneruje .eml draft (X-Unsent: 1) připravený k ruční kontrole a odeslání v Outlooku. Po dvojkliku Outlook otevře rozepsaný email s předvyplněnými To/CC/Subject/tělem/přílohami — uživatel zkontroluje a pošle. - to: příjemce (jeden) nebo více oddělených čárkou - subject: předmět — libovolný text (diakritika OK), klidně s prefixem typu "[Připomínka] RE: …" - body: tělo emailu — JEN vnitřní obsah; u HTML stačí

, plain text se zalomí dle řádků - cc: kopie, oddělené čárkou (volitelné) - attachments: cesta k souboru nebo více cest oddělených ';' (např. "C:\\a.pdf;C:\\b.docx"). MIME typ se detekuje sám. - original_message_id: message_id původního emailu (z list_emails). Když je zadán, původní email se vloží POD nový text ve forward stylu (oddělovač + From/Sent/To/Subject + tělo) — tj. připomínka nahoře, zachovaný originál pod ní. - strip_self_forward: (default True) když je `original_message_id` email přeposlaný k sobě (FW:), odřízne horní self-forward obal (prázdný forward + duplicitní podpis + FW hlavičku k sobě) a vloží JEN vnitřní skutečně odeslaný originál. Když vnitřní forward nenajde, ocituje celý email (v1.4). - from_addr: odesílatel (default vbuzalka@its.jnj.com) - body_format: 'html' (default) nebo 'text' - add_signature:připojí standardní ICON/Janssen podpis (default True) - filename: vlastní název .eml (default draft__.eml) - output_dir: cílová složka (default Dropbox/Downloads Z230) Vrací: {path, to, cc, subject, attachments: [{name, size_bytes}], quoted_original: bool} """ fmt = (body_format or "html").lower().strip() if fmt not in ("html", "text"): raise ValueError("body_format musí být 'html' nebo 'text'") att_paths = _parse_attachments(attachments) # Forward-style blok s původním emailem (pod nový text) quote_block = "" if original_message_id: quote_block = _build_quote_block( original_message_id, fmt, strip_self_forward=strip_self_forward ) out_dir = Path(output_dir) if output_dir else DEFAULT_DRAFT_DIR out_dir.mkdir(parents=True, exist_ok=True) # Sestavení těla if fmt == "html": content = body if add_signature: content = f"{content}\n{SIGNATURE_HTML}" if quote_block: content = f"{content}\n{quote_block}" html_doc = ( '' '' '' '' f"{content}" '' ) msg = EmailMessage() # base64 (cte) – vyhne se quoted-printable soft-breakům, které # Outlook mršil (zbytky =C4/=C3, rozbitá diakritika, "<=a>") msg.set_content("Tento e-mail vyžaduje HTML zobrazení.", charset="utf-8", cte="base64") # plain fallback msg.add_alternative(html_doc, subtype="html", charset="utf-8", cte="base64") else: content = body if add_signature: content = f"{content}\n{SIGNATURE_TEXT}" if quote_block: content = f"{content}\n{quote_block}" msg = EmailMessage() msg.set_content(content, charset="utf-8", cte="base64") # Hlavičky msg["From"] = from_addr msg["To"] = to if cc: msg["Cc"] = cc msg["Subject"] = subject msg["X-Unsent"] = "1" # Outlook otevře jako rozepsaný (draft) # Přílohy attached = [] for fp in att_paths: ctype, encoding = mimetypes.guess_type(str(fp)) if ctype is None or encoding is not None: ctype = "application/octet-stream" maintype, subtype = ctype.split("/", 1) data = fp.read_bytes() msg.add_attachment(data, maintype=maintype, subtype=subtype, filename=fp.name) attached.append({"name": fp.name, "size_bytes": len(data)}) # Název souboru if filename: name = filename if filename.lower().endswith(".eml") else f"{filename}.eml" else: ts = datetime.now().strftime("%Y%m%d_%H%M") name = f"draft_{_slug(subject)}_{ts}.eml" out_path = out_dir / name stem, suffix = out_path.stem, out_path.suffix counter = 1 while out_path.exists(): out_path = out_dir / f"{stem}_{counter}{suffix}" counter += 1 # SMTP policy → CRLF konce řádků (MIME standard); Outlook tak korektně # dekóduje base64 i hlavičky out_path.write_bytes(msg.as_bytes(policy=SMTP_POLICY)) log(f"draft .eml created: {out_path} ({len(attached)} příloh)") return { "path": str(out_path), "to": to, "cc": cc or "", "subject": subject, "attachments": attached, "quoted_original": bool(quote_block), } @mcp.tool() def create_event( subject: str, start: str, end: Optional[str] = None, duration_minutes: int = 30, body: Optional[str] = None, location: Optional[str] = None, reminder_minutes_before: int = 0, timezone: str = DEFAULT_TIMEZONE, is_all_day: bool = False, body_is_html: bool = False, ) -> dict: """Vytvoří událost v kalendáři schránky vladimir.buzalka@buzalka.cz (Graph POST /events). Vyžaduje Calendars.ReadWrite (application). - subject: název události - start: začátek v LOKÁLNÍM čase, ISO bez offsetu, např. "2026-06-11T09:00:00" - end: konec (ISO bez offsetu). Když je None, dopočítá se z `duration_minutes`. - duration_minutes: délka v minutách, když není zadán `end` (def 30) - body: poznámka k události (plain text nebo HTML) - location: místo (např. "tel. 0911 926 046") - reminder_minutes_before: připomínka N minut před začátkem (0 = v čase začátku). Připomínka je vždy zapnutá. - timezone: Windows TZ název (def "Central European Standard Time") - is_all_day: celodenní událost (start/end musí být půlnoci) - body_is_html: True = `body` je HTML, jinak plain text Vrací: {id, web_link, subject, start, end, reminder_minutes_before} """ from datetime import datetime as _dt, timedelta as _td if end: end_val = end else: try: st = _dt.fromisoformat(start) except ValueError: raise ValueError("start musí být ISO formát, např. 2026-06-11T09:00:00") end_val = (st + _td(minutes=max(1, duration_minutes))).isoformat() payload = { "subject": subject, "start": {"dateTime": start, "timeZone": timezone}, "end": {"dateTime": end_val, "timeZone": timezone}, "isAllDay": bool(is_all_day), "isReminderOn": True, "reminderMinutesBeforeStart": max(0, int(reminder_minutes_before)), } if body is not None: payload["body"] = { "contentType": "HTML" if body_is_html else "Text", "content": body, } if location: payload["location"] = {"displayName": location} ev = _graph_post("/events", payload) log(f"event created: {ev.get('id')} — {subject}") return { "id": ev.get("id"), "web_link": ev.get("webLink"), "subject": ev.get("subject"), "start": ev.get("start"), "end": ev.get("end"), "reminder_minutes_before": ev.get("reminderMinutesBeforeStart"), } @mcp.tool() def create_contact( given_name: str, surname: str, display_name: Optional[str] = None, email: Optional[str] = None, mobile_phone: Optional[str] = None, business_phone: Optional[str] = None, company: Optional[str] = None, job_title: Optional[str] = None, note: Optional[str] = None, ) -> dict: """Založí kontakt ve schránce vladimir.buzalka@buzalka.cz (Graph POST /contacts). Vyžaduje Contacts.ReadWrite (application). - given_name / surname: jméno / příjmení (povinné) - display_name: zobrazované jméno (def " ") - email: e-mailová adresa - mobile_phone: mobil - business_phone: pracovní telefon - company: firma / pracoviště - job_title: funkce - note: poznámka (personalNotes) Vrací: {id, display_name, email} """ disp = display_name or f"{given_name} {surname}".strip() payload = { "givenName": given_name, "surname": surname, "displayName": disp, } if email: payload["emailAddresses"] = [{"address": email, "name": disp}] if mobile_phone: payload["mobilePhone"] = mobile_phone if business_phone: payload["businessPhones"] = [business_phone] if company: payload["companyName"] = company if job_title: payload["jobTitle"] = job_title if note: payload["personalNotes"] = note c = _graph_post("/contacts", payload) log(f"contact created: {c.get('id')} — {disp}") return { "id": c.get("id"), "display_name": c.get("displayName"), "email": email or "", } if __name__ == "__main__": log("mcp_vbcz_email v1.6 starting (stdio)…") mcp.run()