318 lines
12 KiB
Python
318 lines
12 KiB
Python
"""
|
|
=======================================================================
|
|
Název: mcp_owa_v1.0.py
|
|
Verze: 1.0
|
|
Datum: 2026-06-04
|
|
Popis: MCP server pro práci s otevřeným OWA oknem (Playwright).
|
|
Drží persistent session a vystavuje tooly:
|
|
- vyhledání emailu v MongoDB OperativniEmailyJNJ.messages
|
|
- otevření emailu v OWA UI přes search
|
|
- Forward → vepsání úvodního textu na začátek body
|
|
- zavření původního čtecího panelu
|
|
Odeslání forwardu dělá uživatel sám.
|
|
Profil sdílí s import_emails_to_mongo_v1.0.py (outlook_profile/).
|
|
Spuštění: python mcp_owa_v1.0.py (registrace přes .mcp.json)
|
|
=======================================================================
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
from playwright.async_api import async_playwright, Page, BrowserContext
|
|
from pymongo import MongoClient, DESCENDING
|
|
|
|
# ── Konfigurace ────────────────────────────────────────────────────────
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
PROFILE_DIR = BASE_DIR / "outlook_profile"
|
|
START_URL = "https://outlook.cloud.microsoft/mail/"
|
|
|
|
MONGO_URI = "mongodb://192.168.1.76:27017"
|
|
DB_NAME = "OperativniEmailyJNJ"
|
|
COL_NAME = "messages"
|
|
|
|
SEARCH_READY = (
|
|
'[placeholder*="Search"], [aria-label*="Search"], '
|
|
'[placeholder*="Hledat"], [aria-label*="Hledat"]'
|
|
)
|
|
|
|
|
|
def log(msg: str):
|
|
print(msg, file=sys.stderr, flush=True)
|
|
|
|
|
|
# ── Globální stav (persistent session) ─────────────────────────────────
|
|
class State:
|
|
pw = None # playwright instance
|
|
context: Optional[BrowserContext] = None
|
|
main_page: Optional[Page] = None
|
|
draft_page: Optional[Page] = None # forward composer (může být totéž okno)
|
|
lock = asyncio.Lock()
|
|
|
|
|
|
_mongo = MongoClient(MONGO_URI)
|
|
_col = _mongo[DB_NAME][COL_NAME]
|
|
|
|
mcp = FastMCP("owa")
|
|
|
|
|
|
# ── Helpers ────────────────────────────────────────────────────────────
|
|
|
|
def _doc_summary(doc: dict) -> dict:
|
|
return {
|
|
"message_id": doc.get("message_id"),
|
|
"subject": doc.get("subject"),
|
|
"from": doc.get("from"),
|
|
"to": doc.get("to", [])[:5],
|
|
"date": doc.get("date").isoformat() if doc.get("date") else None,
|
|
"folder": doc.get("folder"),
|
|
"has_attachments": doc.get("has_attachments", False),
|
|
}
|
|
|
|
|
|
async def _ensure_started():
|
|
if State.context is None:
|
|
raise RuntimeError("OWA není spuštěné. Zavolej nejprve start_owa.")
|
|
|
|
|
|
async def _wait_ready(page: Page):
|
|
await page.wait_for_load_state("domcontentloaded")
|
|
await page.wait_for_selector(SEARCH_READY, timeout=30_000)
|
|
|
|
|
|
# ── Tools: lifecycle ───────────────────────────────────────────────────
|
|
|
|
@mcp.tool()
|
|
async def start_owa() -> dict:
|
|
"""Spustí Playwright s persistent profilem a otevře OWA. Okno zůstane otevřené."""
|
|
async with State.lock:
|
|
if State.context is not None:
|
|
return {"status": "already_running", "url": State.main_page.url if State.main_page else None}
|
|
|
|
if not PROFILE_DIR.exists():
|
|
return {"status": "error", "error": f"Profil nenalezen: {PROFILE_DIR}. Spusť outlook_login_v1.0.py."}
|
|
|
|
State.pw = await async_playwright().start()
|
|
State.context = await State.pw.chromium.launch_persistent_context(
|
|
user_data_dir=str(PROFILE_DIR),
|
|
headless=False,
|
|
no_viewport=True,
|
|
accept_downloads=True,
|
|
args=[
|
|
"--disable-blink-features=AutomationControlled",
|
|
"--start-maximized",
|
|
],
|
|
)
|
|
State.main_page = State.context.pages[0] if State.context.pages else await State.context.new_page()
|
|
await State.main_page.goto(START_URL)
|
|
await _wait_ready(State.main_page)
|
|
return {"status": "started", "url": State.main_page.url}
|
|
|
|
|
|
@mcp.tool()
|
|
async def stop_owa() -> dict:
|
|
"""Zavře Playwright context (a tím i okno OWA)."""
|
|
async with State.lock:
|
|
if State.context is None:
|
|
return {"status": "not_running"}
|
|
try:
|
|
await State.context.close()
|
|
finally:
|
|
if State.pw:
|
|
await State.pw.stop()
|
|
State.context = None
|
|
State.main_page = None
|
|
State.draft_page = None
|
|
State.pw = None
|
|
return {"status": "stopped"}
|
|
|
|
|
|
@mcp.tool()
|
|
async def status() -> dict:
|
|
"""Vrátí stav session: běží/neběží, URL, počet tabů, draft otevřen?"""
|
|
if State.context is None:
|
|
return {"running": False}
|
|
return {
|
|
"running": True,
|
|
"url": State.main_page.url if State.main_page else None,
|
|
"tabs": len(State.context.pages),
|
|
"draft_open": State.draft_page is not None and not State.draft_page.is_closed(),
|
|
}
|
|
|
|
|
|
# ── Tools: MongoDB lookup ──────────────────────────────────────────────
|
|
|
|
@mcp.tool()
|
|
async def find_emails(
|
|
query: Optional[str] = None,
|
|
from_email: Optional[str] = None,
|
|
folder: Optional[str] = None,
|
|
since_iso: Optional[str] = None,
|
|
limit: int = 10,
|
|
) -> list[dict]:
|
|
"""Hledá v MongoDB. `query` = substring v subjectu (case-insensitive).
|
|
`since_iso` = ISO datum, vrátí jen emaily od něj. Seřazeno od nejnovějšího."""
|
|
flt: dict = {}
|
|
if query:
|
|
flt["subject"] = {"$regex": query, "$options": "i"}
|
|
if from_email:
|
|
flt["from.email"] = {"$regex": from_email, "$options": "i"}
|
|
if folder:
|
|
flt["folder"] = folder
|
|
if since_iso:
|
|
try:
|
|
dt = datetime.fromisoformat(since_iso.replace("Z", "+00:00"))
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
flt["date"] = {"$gte": dt}
|
|
except Exception:
|
|
pass
|
|
|
|
cur = _col.find(flt).sort("date", DESCENDING).limit(max(1, min(limit, 50)))
|
|
return [_doc_summary(d) for d in cur]
|
|
|
|
|
|
@mcp.tool()
|
|
async def find_last_email(from_email: Optional[str] = None, folder: Optional[str] = None) -> Optional[dict]:
|
|
"""Vrátí nejnovější email (volitelně filtr podle odesílatele / složky)."""
|
|
flt: dict = {}
|
|
if from_email:
|
|
flt["from.email"] = {"$regex": from_email, "$options": "i"}
|
|
if folder:
|
|
flt["folder"] = folder
|
|
doc = _col.find(flt).sort("date", DESCENDING).limit(1)
|
|
docs = list(doc)
|
|
return _doc_summary(docs[0]) if docs else None
|
|
|
|
|
|
# ── Tools: OWA UI ──────────────────────────────────────────────────────
|
|
|
|
@mcp.tool()
|
|
async def open_email_by_subject(subject: str) -> dict:
|
|
"""Vyhledá v OWA podle subjectu a otevře první výsledek v reading pane."""
|
|
await _ensure_started()
|
|
page = State.main_page
|
|
|
|
search = page.locator(SEARCH_READY).first
|
|
await search.click()
|
|
await search.fill("")
|
|
await search.type(subject, delay=20)
|
|
await page.keyboard.press("Enter")
|
|
await page.wait_for_timeout(2_000)
|
|
|
|
# první výsledek v listu
|
|
msgs = page.locator('div[role="option"]')
|
|
try:
|
|
await msgs.first.wait_for(state="visible", timeout=10_000)
|
|
except Exception:
|
|
return {"status": "no_results"}
|
|
await msgs.first.click()
|
|
await page.wait_for_timeout(800)
|
|
return {"status": "opened", "count_visible": await msgs.count()}
|
|
|
|
|
|
@mcp.tool()
|
|
async def forward_current(body_prefix: str = "", subject_prefix: str = "") -> dict:
|
|
"""Klikne Forward na otevřeném emailu. Pokud je `body_prefix`, vepíše ho na začátek body
|
|
(Home → text → Enter). Pokud `subject_prefix`, předřadí ho do předmětu draftu."""
|
|
await _ensure_started()
|
|
page = State.main_page
|
|
|
|
# Forward přes klávesovou zkratku (funguje, když je reading pane focus)
|
|
await page.keyboard.press("Control+Shift+F")
|
|
await page.wait_for_timeout(1_500)
|
|
|
|
# Body editor
|
|
body = page.locator('div[aria-label*="Message body" i][contenteditable="true"], '
|
|
'div[aria-label*="Tělo zprávy" i][contenteditable="true"]').first
|
|
try:
|
|
await body.wait_for(state="visible", timeout=10_000)
|
|
except Exception:
|
|
return {"status": "forward_failed", "hint": "Body editor forwardu nenalezen."}
|
|
|
|
State.draft_page = page # OWA forward bývá v tom samém tabu (inline composer)
|
|
|
|
if subject_prefix:
|
|
subj = page.locator('input[aria-label*="subject" i], input[aria-label*="předmět" i]').first
|
|
try:
|
|
await subj.wait_for(state="visible", timeout=5_000)
|
|
current = await subj.input_value()
|
|
await subj.fill(f"{subject_prefix}{current}")
|
|
except Exception:
|
|
pass
|
|
|
|
if body_prefix:
|
|
await body.click()
|
|
await page.keyboard.press("Control+Home")
|
|
await page.keyboard.type(body_prefix, delay=10)
|
|
await page.keyboard.press("Enter")
|
|
|
|
return {"status": "forward_ready"}
|
|
|
|
|
|
@mcp.tool()
|
|
async def write_at_top(text: str) -> dict:
|
|
"""Vepíše text na začátek body otevřeného draftu (před existující obsah / podpis)."""
|
|
await _ensure_started()
|
|
if State.draft_page is None or State.draft_page.is_closed():
|
|
return {"status": "no_draft"}
|
|
page = State.draft_page
|
|
body = page.locator('div[aria-label*="Message body" i][contenteditable="true"], '
|
|
'div[aria-label*="Tělo zprávy" i][contenteditable="true"]').first
|
|
await body.click()
|
|
await page.keyboard.press("Control+Home")
|
|
await page.keyboard.type(text, delay=10)
|
|
return {"status": "written", "chars": len(text)}
|
|
|
|
|
|
@mcp.tool()
|
|
async def set_recipients(to: list[str], cc: Optional[list[str]] = None) -> dict:
|
|
"""Vyplní To / Cc v otevřeném draftu. Adresy oddělené středníkem, OWA si je sám zvalidčuje."""
|
|
await _ensure_started()
|
|
if State.draft_page is None or State.draft_page.is_closed():
|
|
return {"status": "no_draft"}
|
|
page = State.draft_page
|
|
|
|
to_field = page.locator('div[aria-label*="To" i][contenteditable="true"], '
|
|
'div[aria-label*="Komu" i][contenteditable="true"]').first
|
|
await to_field.click()
|
|
await page.keyboard.type("; ".join(to) + ";", delay=15)
|
|
await page.wait_for_timeout(500)
|
|
|
|
if cc:
|
|
cc_field = page.locator('div[aria-label*="Cc" i][contenteditable="true"], '
|
|
'div[aria-label*="Kopie" i][contenteditable="true"]').first
|
|
if await cc_field.count():
|
|
await cc_field.click()
|
|
await page.keyboard.type("; ".join(cc) + ";", delay=15)
|
|
|
|
return {"status": "filled", "to": to, "cc": cc or []}
|
|
|
|
|
|
@mcp.tool()
|
|
async def close_reading_pane() -> dict:
|
|
"""Zavře otevřený email v reading pane (Escape). Forward draft tím nezavře."""
|
|
await _ensure_started()
|
|
await State.main_page.keyboard.press("Escape")
|
|
await State.main_page.wait_for_timeout(300)
|
|
return {"status": "closed"}
|
|
|
|
|
|
@mcp.tool()
|
|
async def screenshot(path: str = "owa_screenshot.png") -> dict:
|
|
"""Uloží screenshot aktivního okna pro orientaci."""
|
|
await _ensure_started()
|
|
out = (BASE_DIR / path).resolve()
|
|
await State.main_page.screenshot(path=str(out), full_page=False)
|
|
return {"status": "ok", "path": str(out)}
|
|
|
|
|
|
# ── Entry ──────────────────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
log("mcp_owa v1.0 starting (stdio)…")
|
|
mcp.run()
|