diff --git a/server.py b/server.py index 8ecae42..d6a8432 100644 --- a/server.py +++ b/server.py @@ -1,5 +1,5 @@ """ -MCP Email Server - OpenAPI-compatible HTTP server +MCP Email Server - MCP Streamable HTTP compatible with Open WebUI Features: - IMAP (login) + SMTP (login) @@ -9,7 +9,7 @@ Features: - Extended tools: reply, forward, drafts, move, copy, labels, attachments, schedule_send, export_conversation - Conflict check search - CalDAV / CardDAV integration (e.g., Nextcloud) for calendar and contacts -- Exposes OpenAPI 3.1 spec for integration with Open WebUI (OpenAPI tool server) +- Exposes MCP tools via Streamable HTTP (JSON-RPC 2.0) Run: - uvicorn server:app --host 0.0.0.0 --port 8000 @@ -32,8 +32,8 @@ from dataclasses import dataclass from typing import Any, Dict, List, Optional import requests -from fastapi import FastAPI, HTTPException -from pydantic import BaseModel, Field +from fastapi import FastAPI, Request, HTTPException +from fastapi.responses import JSONResponse logging.basicConfig( level=os.getenv("LOG_LEVEL", "INFO"), @@ -70,7 +70,7 @@ class EmailConfig: # Scheduling SCHEDULED_SEND_INTERVAL: int = int(os.getenv("SCHEDULED_SEND_INTERVAL", "10")) # CardDAV / CalDAV (e.g., Nextcloud) - DAV_BASE_URL: str = os.getenv("DAV_BASE_URL", "") # e.g. https://cloud.example.com/remote.php/dav + DAV_BASE_URL: str = os.getenv("DAV_BASE_URL", "") DAV_USERNAME: str = os.getenv("DAV_USERNAME", "") DAV_PASSWORD: str = os.getenv("DAV_PASSWORD", "") DAV_VERIFY_TLS: bool = os.getenv("DAV_VERIFY_TLS", "true").lower() in ("true", "1", "yes") @@ -267,16 +267,14 @@ def get_labels(imap, folder: str, uid: int): if start == -1 or end == -1: return [] flags = [f.strip() for f in line[start+1:end].split() if f.strip()] - # Consider custom labels as flags starting with "$Label_" return [f for f in flags if f.startswith("$Label_")] def list_available_labels(imap): - # Discover used labels by scanning a few folders (heuristic). labels = set() try: folders = list_all_folders(imap) - for f in folders[:20]: # limit scan + for f in folders[:20]: path = f["path"] try: ensure_selected(imap, path) @@ -284,7 +282,6 @@ def list_available_labels(imap): if status != "OK" or not data or not data[0]: continue ids = data[0].split() - # sample first 20 UIDs for uid_bytes in ids[:20]: uid = int(uid_bytes) _, flags_data = imap.uid("FETCH", str(uid), "(FLAGS)") @@ -435,11 +432,6 @@ def build_triage_hint(msg: Dict[str, Any]) -> str: # -------------------- CONFLICT CHECK SEARCH -------------------- def conflict_check_search(imap, terms: List[str], folders: Optional[List[str]] = None, max_results: int = 100): - """ - Search emails across specified folders (or INBOX if not provided) - for messages mentioning any of the given terms (names, firms, counsel). - Useful for conflict checks. - """ if not folders: folders = ["INBOX"] @@ -473,7 +465,6 @@ def conflict_check_search(imap, terms: List[str], folders: Optional[List[str]] = except Exception as e: logger.error(f"Conflict search error for term '{term}' in {folder}: {e}") - # Deduplicate by (folder, uid) seen = set() unique = [] for r in all_results: @@ -535,13 +526,12 @@ def download_attachment(imap, folder: str, uid: int, filename: str) -> Dict[str, def search_attachments(imap, folder: str, file_pattern: str, max_results: int): - # Search messages with attachments by filename pattern (case-insensitive). ensure_selected(imap, folder) status, data = imap.uid("SEARCH", None, "ALL") if status != "OK" or not data or not data[0]: return [] ids = data[0].split() - ids = ids[-max_results * 5:] # sample more to find matches + ids = ids[-max_results * 5:] pattern = file_pattern.lower() results = [] @@ -568,7 +558,7 @@ def search_attachments(imap, folder: str, file_pattern: str, max_results: int): "date": msg.get("Date", ""), "filename": part.get_filename() or "attachment", }) - break # one match per message is enough + break except Exception as e: logger.error(f"Error scanning message UID {uid}: {e}") return results @@ -648,7 +638,6 @@ scheduler_thread.start() # -------------------- EXPORT CONVERSATION -------------------- def export_conversation(imap, folder: str, message_uid: int, max_messages: int): - # Basic: fetch messages with same subject and thread-like headers in same folder. ensure_selected(imap, folder) status, msg_data = imap.uid("FETCH", str(message_uid), "(RFC822)") if status != "OK" or not msg_data or msg_data[0] is None: @@ -729,10 +718,6 @@ def dav_request(s, method, url, data=None, headers=None, timeout=20): def parse_dav_propfind(xml_text): - """ - Parse a PROPFIND response and return list of {href, props}. - props is a dict of {local-name: text}. - """ try: root = ET.fromstring(xml_text) except ET.ParseError: @@ -778,7 +763,6 @@ def list_carddav_addressbooks(): for e in entries: href = e["href"] props = e.get("props", {}) - # Check resource type for addressbook rt = props.get("resourcetype", "") if "/cards/" in href and href.endswith("/") and ("addressbook" in rt.lower() or "addressbook" in href.lower()): addressbooks.append({ @@ -899,7 +883,6 @@ def list_caldav_calendars(): def search_caldav_events(calendar_href: str, start: str, end: str): - # start/end as ISO8601, e.g. "20251201T000000Z" s = dav_session() req_body = f""" " + line for line in original.splitlines()) body = (body + "\n\n" + quoted).strip() send_email( - to=req.to, + to=to, subject=subject, body=body, - cc=req.cc, - bcc=req.bcc, - html=req.html, + cc=cc, + bcc=bcc, + html=html, ) return {"status": "forwarded"} - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) finally: imap.logout() -@app.post("/save_draft") -@run_imap_op -def save_draft_endpoint(req: SaveDraftRequest): +def save_draft_impl(to, subject, body, html=False, folder="Drafts"): imap = create_imap() try: login_imap(imap) - return save_draft_to_imap(imap, req.folder, req.to, req.subject, req.body, req.html) - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) + return save_draft_to_imap(imap, folder, to, subject, body, html) finally: imap.logout() -@app.post("/list_attachments") -@run_imap_op -def list_attachments_endpoint(req: ListAttachmentsRequest): +def list_attachments_impl(folder, uid): imap = create_imap() try: login_imap(imap) - atts = list_attachments(imap, req.folder, req.uid) + atts = list_attachments(imap, folder, uid) return {"attachments": atts} - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) finally: imap.logout() -@app.post("/download_attachment") -@run_imap_op -def download_attachment_endpoint(req: DownloadAttachmentRequest): +def download_attachment_impl(folder, uid, filename): imap = create_imap() try: login_imap(imap) - info = download_attachment(imap, req.folder, req.uid, req.filename) + info = download_attachment(imap, folder, uid, filename) return info except ValueError as e: - raise HTTPException(status_code=404, detail=str(e)) - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) + raise RuntimeError(str(e)) finally: imap.logout() -@app.post("/search_attachments") -@run_imap_op -def search_attachments_endpoint(req: SearchAttachmentsRequest): +def search_attachments_impl(folder="INBOX", file_pattern="", max_results=50): imap = create_imap() try: login_imap(imap) - results = search_attachments(imap, req.folder, req.file_pattern, req.max_results) + results = search_attachments(imap, folder, file_pattern, max_results) return {"results": results} - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) finally: imap.logout() -@app.post("/schedule_send") -def schedule_send_endpoint(req: ScheduleSendRequest): - task_id = add_scheduled_send( - to=req.to, - subject=req.subject, - body=req.body, - cc=req.cc, - bcc=req.bcc, - html=req.html, - send_at=req.send_at, - in_reply_to=req.in_reply_to, - references=req.references, - ) - return {"status": "scheduled", "task_id": task_id} - - -@app.post("/export_conversation") -@run_imap_op -def export_conversation_endpoint(req: ExportConversationRequest): +def export_conversation_impl(folder, message_uid, max_messages=50): imap = create_imap() try: login_imap(imap) - text = export_conversation(imap, req.folder, req.message_uid, req.max_messages) + text = export_conversation(imap, folder, message_uid, max_messages) return {"export": text} - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) finally: imap.logout() -@app.post("/conflict_check_search") -@run_imap_op -def conflict_check_search_endpoint(req: ConflictCheckSearchRequest): +def conflict_check_search_impl(terms, folders=None, max_results=100): imap = create_imap() try: login_imap(imap) - results = conflict_check_search(imap, req.terms, req.folders, req.max_results) + results = conflict_check_search(imap, terms, folders, max_results) return {"results": results} finally: imap.logout() -@app.post("/get_triage_config") -def get_triage_config_endpoint(_req: GetTriageConfigRequest): +def get_triage_config_impl(args): return { "business_description": config.BUSINESS_DESCRIPTION, "personnel": config.personnel, } -# CardDAV endpoints +# CardDAV tool impls + +def list_carddav_addressbooks_impl(args): + abs_list = list_carddav_addressbooks() + return {"addressbooks": abs_list} + + +def search_carddav_contacts_impl(args): + contacts = search_carddav_contacts(args["addressbook_href"], args["query"]) + return {"contacts": contacts} + + +def get_carddav_contact_impl(args): + c = get_carddav_contact(args["href"]) + return c + + +def create_carddav_contact_impl(args): + r = create_carddav_contact(args["addressbook_href"], args["vcard"]) + return r + + +def update_carddav_contact_impl(args): + r = update_carddav_contact(args["href"], args["vcard"]) + return r + + +def delete_carddav_contact_impl(args): + r = delete_carddav_contact(args["href"]) + return r + + +# CalDAV tool impls + +def list_caldav_calendars_impl(args): + cal_list = list_caldav_calendars() + return {"calendars": cal_list} + + +def search_caldav_events_impl(args): + events = search_caldav_events(args["calendar_href"], args["start"], args["end"]) + return {"events": events} + + +def create_caldav_event_impl(args): + r = create_caldav_event(args["calendar_href"], args["ical"]) + return r + + +def update_caldav_event_impl(args): + r = update_caldav_event(args["href"], args["ical"]) + return r + + +def delete_caldav_event_impl(args): + r = delete_caldav_event(args["href"]) + return r + + +# -------------------- FASTAPI / MCP ENDPOINT -------------------- + +app = FastAPI( + title="MCP Email Server", + description="Email assistant MCP server exposing IMAP/SMTP, CalDAV/CardDAV, and conflict-check operations via MCP Streamable HTTP.", + version="1.0.0", +) + + +@app.exception_handler(Exception) +async def global_exception_handler(request, exc): + logger.exception("Unhandled exception") + return JSONResponse( + status_code=500, + content={"detail": "Internal server error"} + ) + + +@app.post("/mcp") +async def mcp_endpoint(request: Request): + if request.method != "POST": + raise HTTPException(status_code=405, detail="Method Not Allowed") -@app.post("/list_carddav_addressbooks") -def list_carddav_addressbooks_endpoint(): try: - abs_list = list_carddav_addressbooks() - return {"addressbooks": abs_list} - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) + body = await request.json() + except Exception: + return JSONResponse( + status_code=400, + content=mcp_error(None, -32700, "Parse error") + ) - -@app.post("/search_carddav_contacts") -def search_carddav_contacts_endpoint(req: SearchCardDAVContactsRequest): - try: - contacts = search_carddav_contacts(req.addressbook_href, req.query) - return {"contacts": contacts} - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) - - -@app.post("/get_carddav_contact") -def get_carddav_contact_endpoint(req: GetCardDAVContactRequest): - try: - c = get_carddav_contact(req.href) - return c - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) - - -@app.post("/create_carddav_contact") -def create_carddav_contact_endpoint(req: CreateCardDAVContactRequest): - try: - r = create_carddav_contact(req.addressbook_href, req.vcard) - return r - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) - - -@app.post("/update_carddav_contact") -def update_carddav_contact_endpoint(req: UpdateCardDAVContactRequest): - try: - r = update_carddav_contact(req.href, req.vcard) - return r - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) - - -@app.post("/delete_carddav_contact") -def delete_carddav_contact_endpoint(req: DeleteCardDAVContactRequest): - try: - r = delete_carddav_contact(req.href) - return r - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) - - -# CalDAV endpoints - -@app.post("/list_caldav_calendars") -def list_caldav_calendars_endpoint(): - try: - cal_list = list_caldav_calendars() - return {"calendars": cal_list} - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) - - -@app.post("/search_caldav_events") -def search_caldav_events_endpoint(req: SearchCalDAVEventsRequest): - try: - events = search_caldav_events(req.calendar_href, req.start, req.end) - return {"events": events} - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) - - -@app.post("/create_caldav_event") -def create_caldav_event_endpoint(req: CreateCalDAVEventRequest): - try: - r = create_caldav_event(req.calendar_href, req.ical) - return r - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) - - -@app.post("/update_caldav_event") -def update_caldav_event_endpoint(req: UpdateCalDAVEventRequest): - try: - r = update_caldav_event(req.href, req.ical) - return r - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) - - -@app.post("/delete_caldav_event") -def delete_caldav_event_endpoint(req: DeleteCalDAVEventRequest): - try: - r = delete_caldav_event(req.href) - return r - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) + if isinstance(body, list): + results = [handle_rpc(msg) for msg in body] + return JSONResponse(content=results) + else: + result = handle_rpc(body) + return JSONResponse(content=result) # -------------------- ENTRYPOINT --------------------