705 lines
26 KiB
Python
705 lines
26 KiB
Python
"""
|
||
=======================================================================
|
||
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 = """
|
||
<p>S pozdravem</p>
|
||
<p><strong>MUDr. Vladimír BUZALKA</strong></p>
|
||
<p style="font-size:10pt; color:#444;">
|
||
ICON plc<br>
|
||
Performing Local Trial Management Services for Janssen – Cilag s.r.o.<br>
|
||
Global Clinical Operations<br>
|
||
Mobile: +420 775 735 276<br>
|
||
Fax: +420 227 012 284<br>
|
||
E-mail: <a href="mailto:vbuzalka@its.jnj.com">vbuzalka@its.jnj.com</a>,
|
||
<a href="mailto:vladimir.buzalka@iconplc.com">vladimir.buzalka@iconplc.com</a>
|
||
</p>
|
||
"""
|
||
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 <body>…</body> (nebo vrátí původní), aby šel vnořit."""
|
||
import re
|
||
m = re.search(r"<body[^>]*>(.*)</body>", html, re.IGNORECASE | re.DOTALL)
|
||
if m:
|
||
return m.group(1)
|
||
# bez body tagu — odstraň aspoň <html>/<head>
|
||
html = re.sub(r"</?html[^>]*>", "", html, flags=re.IGNORECASE)
|
||
html = re.sub(r"<head[^>]*>.*?</head>", "", 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)(?:<b>\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("<div", 0, idx + 1)
|
||
p = inner_html.rfind("<p", 0, idx + 1)
|
||
cut = max(d, p)
|
||
if cut == -1:
|
||
cut = idx
|
||
if cut == -1:
|
||
return None
|
||
return inner_html[cut:]
|
||
|
||
|
||
def _extract_inner_forward_text(text: str) -> 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 = (
|
||
'<div style="border:none;border-top:solid #E1E1E1 1.0pt;'
|
||
'padding:6.0pt 0in 0in 0in;margin-top:18px;">'
|
||
f'<p style="font-size:11pt;"><b>From:</b> {_esc(from_name)} '
|
||
f'<{_esc(from_addr)}><br>'
|
||
f'<b>Sent:</b> {_esc(sent)}<br>'
|
||
f'<b>To:</b> {_esc(to_list)}<br>'
|
||
)
|
||
if cc_list:
|
||
hdr += f'<b>Cc:</b> {_esc(cc_list)}<br>'
|
||
hdr += f'<b>Subject:</b> {_esc(subj)}</p></div>'
|
||
if orig_type == "html":
|
||
orig_html = _strip_html_wrapper(orig_content)
|
||
else:
|
||
orig_html = "<pre style=\"font-family:inherit;white-space:pre-wrap;\">" \
|
||
+ _esc(orig_content) + "</pre>"
|
||
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čí <p>…</p>,
|
||
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_<slug>_<timestamp>.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 = (
|
||
'<html><head>'
|
||
'<meta http-equiv="Content-Type" content="text/html; charset=utf-8">'
|
||
'</head>'
|
||
'<body style="font-family: Aptos, Arial, sans-serif; font-size: 11pt;">'
|
||
f"{content}"
|
||
'</body></html>'
|
||
)
|
||
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 "<given> <surname>")
|
||
- 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()
|