Full Source Code
Here is the complete source code for the project.
Main Module
Source Code for run.py:"
import sys
from config import config
from logger import logger
from server import mcp
def check_configuration():
if not config.IMAP_SERVER:
logger.warning("IMAP_SERVER not available...")
if not config.SMTP_SERVER:
logger.warning("SMTP_SERVER not available...")
logger.info("Config loaded ....")
def print_banner():
logger.info("---------------------------------------")
logger.info(" IMAP MCP Server")
logger.info("---------------------------------------")
def main():
print_banner()
check_configuration()
logger.info("Start IMAP MCP-Server...")
try:
mcp.run()
except KeyboardInterrupt:
logger.info("Server stopped.")
except Exception as ex:
logger.exception(ex)
sys.exit(1)
if __name__ == "__main__":
main()
Source Code for server.py:"
from mcp.server.fastmcp import FastMCP
from tools.system import register as register_system
from tools.mail import register as register_mail
mcp = FastMCP("MCP Email - Assistant")
# just register new tools ... pretty simple!
register_system(mcp)
register_mail(mcp)
Source Code for logger.py:"
import logging
from pathlib import Path
# use current dir as BASE DIR for the logger
BASE_DIR = Path(__file__).resolve().parent
LOG_DIR = BASE_DIR / "logs"
LOG_DIR.mkdir(exist_ok=True)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
handlers=[
logging.FileHandler(LOG_DIR / "server.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger("MCPMailAssist")
Source Code for config.py:"
from dataclasses import dataclass
from pathlib import Path
import os
from dotenv import load_dotenv
# use current dir
BASE_DIR = Path(__file__).resolve().parent
# load the local .env file !!!! IMPORTANT !!!
load_dotenv(BASE_DIR / ".env")
@dataclass
class Config:
IMAP_SERVER: str = os.getenv("IMAP_SERVER", "")
IMAP_PORT: int = int(os.getenv("IMAP_PORT", "993"))
SMTP_SERVER: str = os.getenv("SMTP_SERVER", "")
SMTP_PORT: int = int(os.getenv("SMTP_PORT", "465"))
MAIL_USERNAME: str = os.getenv("MAIL_USERNAME", "")
MAIL_PASSWORD: str = os.getenv("MAIL_PASSWORD", "")
config = Config()
Source Code for .env :"
IMAP_SERVER=imap.whatever.de IMAP_PORT=993 SMTP_SERVER=smtp.whatever.de SMTP_PORT=465 MAIL_USERNAME=your@email.de MAIL_PASSWORD=yourverysecretpassword
Module Tools
Source Code for system.py:"
def register(mcp):
@mcp.tool()
def version():
return {
"server": "MCP Mail Assist by Stefan Hager",
"version": "1.0"
}
Source Code for mail.py:"
from dataclasses import asdict
from mail import mail_account
def register(mcp):
@mcp.tool()
def count_emails():
"""
Ermittelt die Anzahl der Emails in einem Postfach
"""
return mail_account.get_email_count()
@mcp.tool()
def list_recent_emails(count: int = 10):
"""
Gibt die Liste der neuesten E-Mails zurück.
Args:
count: Anzahl der zurückzugebenden E-Mails.
"""
mails = mail_account.list_recent_emails(count)
return [
asdict(mail)
for mail in mails
]
@mcp.tool()
def find_emails_by_sender(sender: str, count: int = 10):
"""
Sucht E-Mails eines bestimmten Absenders.
Args:
count: Anzahl der zurückzugebenden E-Mails.
"""
mails = mail_account.find_emails_by_sender(sender, count)
return [
asdict(mail)
for mail in mails
]
@mcp.tool()
def send_email(
recipient: str,
subject: str,
body: str
):
"""
Sendet eine E-Mail.
"""
mail_account.send_email(
recipient,
subject,
body
)
return {
"success": True
}
@mcp.tool()
def has_new_emails(self):
"""
Schaut nach, ob ungelesene Mails vorhanden sind.
"""
return len(mail_account.get_unread_uids()) > 0
Module Models
Source Code for email_summary.py:"
from dataclasses import dataclass
from datetime import datetime
@dataclass
class EmailSummary:
uid: int
subject: str
sender: str
date: datetime
unread: bool
Module Mail
Source Code for __init__.py:"
from .mail_account import MailAccount # run as Singleton mail_account = MailAccount()
Source Code for mail_account.py:"
from imapclient import IMAPClient
from mail.utils import decode_mime_header, email_address
from config import config
from logger import logger
from models.email_summary import EmailSummary
import smtplib
from email.message import EmailMessage
class MailAccount:
def __init__(self):
self.client = None
self.connected = False
def connect(self):
"""
Baut eine SSL-Verbindung zum IMAP-Server auf und meldet den Benutzer an.
"""
if self.connected:
return
logger.info("Connect to IMAP-Server...")
self.client = IMAPClient(
host=config.IMAP_SERVER,
port=config.IMAP_PORT,
ssl=True
)
self.client.login(
config.MAIL_USERNAME,
config.MAIL_PASSWORD
)
self.connected = True
logger.info("Login successful...")
def disconnect(self):
if not self.connected:
return
logger.info("Disconnect from server ....")
self.client.logout()
self.client = None
self.connected = False
def select_folder(self, folder="INBOX"):
"""
Wählt einen IMAP-Ordner aus.
"""
self.connect()
self.client.select_folder(folder)
logger.info(f"Folder '{folder}' selected...")
def get_email_count(self):
"""
Ermittelt die Anzahl der Emails in einem Postfach
"""
self.connect()
self.client.select_folder("INBOX")
return self.client.folder_status(
"INBOX",
["MESSAGES"]
)[b"MESSAGES"]
def get_email_uids(self):
"""
Liest UID einer E-Mail.
"""
self.connect()
self.client.select_folder("INBOX")
return self.client.search()
def get_email_subject(self, uid):
"""
Liest den Betreff einer E-Mail.
"""
self.connect()
data = self.client.fetch(
[uid],
["ENVELOPE"]
)
envelope = data[uid][b"ENVELOPE"]
print(type(envelope.subject))
subject = decode_mime_header(envelope.subject)
return subject
def get_email_summary(self, uid):
"""
Liest Betreff, Absender, Datum und Status einer E-Mail.
"""
data = self.client.fetch(
[uid],
["ENVELOPE", "FLAGS"]
)
envelope = data[uid][b"ENVELOPE"]
flags = data[uid][b"FLAGS"]
subject = decode_mime_header(envelope.subject)
sender = ""
if envelope.from_:
sender = email_address(envelope.from_[0])
unread = b"\\Seen" not in flags
return EmailSummary(
uid=uid,
subject=subject,
sender=sender,
date=envelope.date,
unread=unread
)
def list_recent_emails(self, count=10):
"""
Liest Betreff, Absender, Datum und Status der letzten E-Mails.
Args:
count: Anzahl der zurückzugebenden E-Mails.
"""
uids = self.get_email_uids()
emails = []
for uid in reversed(uids[-count:]):
emails.append(
self.get_email_summary(uid)
)
return emails
def search_uids(self, criteria):
"""
Sucht E-Mails anhand eines IMAP-Suchkriteriums.
Beispiel:
["FROM", "max@example.com"]
"""
self.connect()
self.select_folder("INBOX")
return self.client.search(criteria)
def find_emails_by_sender(self, sender, count=20):
"""
Liefert die neuesten E-Mails eines bestimmten Absenders.
Args:
count: Anzahl der zurückzugebenden E-Mails.
"""
uids = self.search_uids(["FROM", sender])
emails = []
for uid in reversed(uids[-count:]):
emails.append(
self.get_email_summary(uid)
)
return emails
def send_email(self,
recipient: str,
subject: str,
body: str):
"""
Versendet eine E-Mail.
"""
message = EmailMessage()
message["From"] = config.MAIL_USERNAME
message["To"] = recipient
message["Subject"] = subject
message.set_content(body)
with smtplib.SMTP_SSL(
config.SMTP_SERVER,
config.SMTP_PORT) as smtp:
smtp.login(
config.MAIL_USERNAME,
config.MAIL_PASSWORD
)
smtp.send_message(message)
logger.info(
f"E-Mail an {recipient} gesendet."
)
def get_unread_uids(self):
"""
Liefert die UIDs aller ungelesenen E-Mails.
"""
self.connect()
self.select_folder("INBOX")
return self.client.search(["UNSEEN"])
Source Code for mail_account.py:"
from email.header import decode_header
def decode_mime_header(header):
if header is None:
return ""
# Bytes -> ASCII-String
if isinstance(header, bytes):
header = header.decode("ascii", errors="replace")
parts = decode_header(header)
result = ""
for text, encoding in parts:
if isinstance(text, bytes):
result += text.decode(
encoding or "utf-8",
errors="replace"
)
else:
result += text
return result
def email_address(address):
"""
Wandelt eine IMAP-Adresse in eine E-Mail-Adresse um.
"""
if address is None:
return ""
mailbox = address.mailbox.decode()
host = address.host.decode()
return f"{mailbox}@{host}"