admin/mcp-email
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add list_unread_emails, get_email, summarize_email, get_thread tools

Browse Source
This commit is contained in:
Admin
2026-06-27 14:42:33 +00:00
parent ef6b1aa1c3
commit a43e9473b8
1 changed files with 413 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.py
+413
View File
@@ -129,6 +129,36 @@ class GlobalConfig:
LLM_ADDRESS: str = os.getenv("LLM_ADDRESS", "")
LLM_EMAIL: str = os.getenv("LLM_EMAIL", "")
# External summary model (optional)
EXTERNAL_SUMMARY_MODEL_ENDPOINT: str = os.getenv("EXTERNAL_SUMMARY_MODEL_ENDPOINT", "").strip()
EXTERNAL_SUMMARY_MODEL_API_KEY: str = os.getenv("EXTERNAL_SUMMARY_MODEL_API_KEY", "").strip()
EXTERNAL_SUMMARY_MODEL_NAME: str = os.getenv("EXTERNAL_SUMMARY_MODEL_NAME", "gpt-4o-mini").strip()
EXTERNAL_SUMMARY_MAX_TOKENS: int = int(os.getenv("EXTERNAL_SUMMARY_MAX_TOKENS", "400"))
@classmethod
def is_summary_model_configured(cls) -> bool:
return bool(cls.EXTERNAL_SUMMARY_MODEL_ENDPOINT and cls.EXTERNAL_SUMMARY_MODEL_API_KEY)
# External summary model (optional)
EXTERNAL_SUMMARY_MODEL_ENDPOINT: str = os.getenv("EXTERNAL_SUMMARY_MODEL_ENDPOINT", "").strip()
EXTERNAL_SUMMARY_MODEL_API_KEY: str = os.getenv("EXTERNAL_SUMMARY_MODEL_API_KEY", "").strip()
EXTERNAL_SUMMARY_MODEL_NAME: str = os.getenv("EXTERNAL_SUMMARY_MODEL_NAME", "gpt-4o-mini").strip()
EXTERNAL_SUMMARY_MAX_TOKENS: int = int(os.getenv("EXTERNAL_SUMMARY_MAX_TOKENS", "400"))
@classmethod
def is_summary_model_configured(cls) -> bool:
return bool(cls.EXTERNAL_SUMMARY_MODEL_ENDPOINT and cls.EXTERNAL_SUMMARY_MODEL_API_KEY)
# External summary model (optional)
EXTERNAL_SUMMARY_MODEL_ENDPOINT: str = os.getenv("EXTERNAL_SUMMARY_MODEL_ENDPOINT", "").strip()
EXTERNAL_SUMMARY_MODEL_API_KEY: str = os.getenv("EXTERNAL_SUMMARY_MODEL_API_KEY", "").strip()
EXTERNAL_SUMMARY_MODEL_NAME: str = os.getenv("EXTERNAL_SUMMARY_MODEL_NAME", "gpt-4o-mini").strip()
EXTERNAL_SUMMARY_MAX_TOKENS: int = int(os.getenv("EXTERNAL_SUMMARY_MAX_TOKENS", "400"))
@classmethod
def is_summary_model_configured(cls) -> bool:
return bool(cls.EXTERNAL_SUMMARY_MODEL_ENDPOINT and cls.EXTERNAL_SUMMARY_MODEL_API_KEY)
@classmethod
def resolve_html_signature(cls, raw: str) -> str:
if not raw:
@@ -535,6 +565,68 @@ def list_available_labels(imap):
return sorted(labels)
# -------------------- EXTERNAL SUMMARY MODEL HELPER --------------------
import urllib.request
import urllib.error
def call_summary_model(text: str, max_length: int = 200) -> str:
"""
Call an OpenAI-compatible /v1/chat/completions endpoint to summarize an email.
Returns a concise summary or the original text if it fails.
"""
if not config.is_summary_model_configured():
raise RuntimeError("External summary model not configured (EXTERNAL_SUMMARY_MODEL_ENDPOINT or API_KEY missing).")
url = config.EXTERNAL_SUMMARY_MODEL_ENDPOINT.rstrip("/") + "/v1/chat/completions"
model = config.EXTERNAL_SUMMARY_MODEL_NAME or "gpt-4o-mini"
max_tokens = min(config.EXTERNAL_SUMMARY_MAX_TOKENS or 400, 1000)
system_prompt = (
"You are assisting a lawyer. Summarize this email in bullet points: "
"sender, purpose, key facts, requested actions, deadlines, and any attachments. "
"Be concise and precise."
)
# Truncate input if extremely long
if len(text) > 20000:
text = text[:18000] + "\n[...truncated...]"
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Summarize this email in at most {max_length} words:\n\n{text}"}
],
"max_tokens": max_tokens,
"temperature": 0.2,
}
req = urllib.request.Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {config.EXTERNAL_SUMMARY_MODEL_API_KEY}",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
resp_data = json.loads(resp.read().decode("utf-8"))
content = resp_data.get("choices", [{}])[0].get("message", {}).get("content", "")
if content:
return content.strip()
raise RuntimeError("Empty response from summary model.")
except Exception as e:
logger.error(f"Error calling external summary model: {e}")
# Fallback: return first N words
words = text.split()
return " ".join(words[:max_length])
# -------------------- SMTP HELPERS --------------------
def send_email(
@@ -1118,6 +1210,63 @@ TOOLS = [
"required": []
}
},
{
"name": "list_unread_emails",
"description": "List unread emails with minimal content to save context. Use this first when triaging many emails, then use get_email or summarize_email on selected messages.",
"inputSchema": {
"type": "object",
"properties": {
"account": {"type": "string", "description": "Account ID (optional, uses default if omitted)"},
"folder": {"type": "string", "description": "IMAP folder (default: INBOX)"},
"max_results": {"type": "integer", "description": "Maximum number of emails to return (default: 50)"},
"include_body": {"type": "boolean", "description": "If true, include full body (default: false)"},
"since": {"type": "string", "description": "Optional SINCE date, e.g. '01-Jan-2024'"}
},
"required": []
}
},
{
"name": "get_email",
"description": "Fetch the full content of a single email by UID. Use this after scanning with list_unread_emails or get_unread_summary for emails you need to read in detail.",
"inputSchema": {
"type": "object",
"properties": {
"account": {"type": "string", "description": "Account ID (optional, uses default if omitted)"},
"folder": {"type": "string", "description": "IMAP folder (default: INBOX)"},
"uid": {"type": "integer", "description": "Message UID"},
"include_html": {"type": "boolean", "description": "If true, include HTML body (default: false)"}
},
"required": ["folder", "uid"]
}
},
{
"name": "summarize_email",
"description": "Summarize a single email using an external model (if configured) to reduce context usage. Use this when you need to understand an email quickly without loading its full text.",
"inputSchema": {
"type": "object",
"properties": {
"account": {"type": "string", "description": "Account ID (optional, uses default if omitted)"},
"folder": {"type": "string", "description": "IMAP folder (default: INBOX)"},
"uid": {"type": "integer", "description": "Message UID"},
"max_length": {"type": "integer", "description": "Maximum summary length in words (default: 150)"}
},
"required": ["folder", "uid"]
}
},
{
"name": "get_thread",
"description": "Get a concise view of the conversation thread for a given email. Use this to understand context without loading all message bodies.",
"inputSchema": {
"type": "object",
"properties": {
"account": {"type": "string", "description": "Account ID (optional, uses default if omitted)"},
"folder": {"type": "string", "description": "IMAP folder (default: INBOX)"},
"uid": {"type": "integer", "description": "Message UID to use as the anchor"},
"max_messages": {"type": "integer", "description": "Maximum number of thread messages to return (default: 10)"}
},
"required": ["folder", "uid"]
}
},
{
"name": "send_email",
"description": "Send an email using a specific account (or default).",
@@ -1474,6 +1623,18 @@ def call_tool(name, args):
if name == "get_unread_summary":
return imap_op(get_unread_summary_impl)
if name == "list_unread_emails":
return imap_op(list_unread_emails_impl)
if name == "get_email":
return imap_op(get_email_impl)
if name == "summarize_email":
return imap_op(summarize_email_impl)
if name == "get_thread":
return imap_op(get_thread_impl)
if name == "reply_to_message":
return imap_op(reply_to_message_impl)
@@ -1698,6 +1859,258 @@ def get_unread_summary_impl(account=None, folder="INBOX"):
imap.logout()
def list_unread_emails_impl(account=None, folder="INBOX", max_results=50, include_body=False, since=None):
"""
Metadata-only (or limited) listing of unread emails to reduce context usage.
"""
acc = get_account(account)
imap = create_imap(acc)
try:
login_imap(imap, acc)
ensure_selected(imap, folder)
criteria = ["UNSEEN"]
if since:
criteria.append(f"SINCE {since}")
search_str = " ".join(criteria)
status, data = imap.uid("SEARCH", None, search_str)
if status != "OK" or not data or not data[0]:
ids = []
else:
ids = data[0].split()
uids = [int(u) for u in ids]
count = len(uids)
uids = uids[-max_results:] # take most recent
emails = []
for uid in uids:
try:
msg = parse_message(imap, uid)
except Exception:
continue
snippet = (msg["body_plain"] or "")[:300].strip()
att_info = list_attachments_for_uid(imap, folder, uid)
has_attachments = len(att_info) > 0
item = {
"uid": uid,
"subject": msg["subject"],
"from": msg["from"],
"to": msg["to"],
"date": msg["date"],
"flags": msg["flags"],
"snippet": snippet,
"has_attachments": has_attachments,
}
if include_body:
item["body_plain"] = msg["body_plain"]
item["body_html"] = msg["body_html"]
emails.append(item)
return {
"folder": folder,
"total_unread": count,
"returned": len(emails),
"emails": emails,
}
finally:
imap.logout()
def get_email_impl(account=None, folder="INBOX", uid=None, include_html=False):
"""
Fetch the full content of a single email by UID.
"""
acc = get_account(account)
imap = create_imap(acc)
try:
login_imap(imap, acc)
msg = parse_message(imap, uid)
att_info = list_attachments_for_uid(imap, folder, uid)
result = {
"uid": uid,
"subject": msg["subject"],
"from": msg["from"],
"to": msg["to"],
"cc": msg["cc"],
"date": msg["date"],
"body_plain": msg["body_plain"],
"attachments": att_info,
}
if include_html:
result["body_html"] = msg["body_html"]
return result
finally:
imap.logout()
def summarize_email_impl(account=None, folder="INBOX", uid=None, max_length=150):
"""
Use an external model to summarize a single email.
Falls back to a short plain-text prefix if external model is unavailable.
"""
acc = get_account(account)
imap = create_imap(acc)
try:
login_imap(imap, acc)
msg = parse_message(imap, uid)
att_info = list_attachments_for_uid(imap, folder, uid)
finally:
imap.logout()
text_to_summarize = (msg["body_plain"] or msg["body_html"] or "").strip()
if not text_to_summarize:
return {
"uid": uid,
"subject": msg["subject"],
"from": msg["from"],
"date": msg["date"],
"summary": "(No readable content)"
}
try:
summary = call_summary_model(text_to_summarize, max_length=max_length)
except Exception as e:
# Fallback: first N words
words = text_to_summarize.split()
summary = " ".join(words[:max_length]) + ("..." if len(words) > max_length else "")
attachments_note = ""
if att_info:
names = [a["filename"] for a in att_info]
attachments_note = "\nAttachments: " + ", ".join(names)
return {
"uid": uid,
"subject": msg["subject"],
"from": msg["from"],
"date": msg["date"],
"summary": summary + attachments_note,
}
def get_thread_impl(account=None, folder="INBOX", uid=None, max_messages=10):
"""
Return a concise thread view for a given message UID.
"""
acc = get_account(account)
imap = create_imap(acc)
try:
login_imap(imap, acc)
ensure_selected(imap, folder)
# Fetch the base message to gather conversation identifiers
base_msg = parse_message(imap, uid)
subject = (base_msg["subject"] or "").strip()
msg_id = (base_msg.get("message_id") or "").strip()
refs = set((base_msg.get("references") or "").split())
in_reply = (base_msg.get("in_reply_to") or "").strip()
if in_reply:
refs.add(in_reply)
# Search by subject to find related messages
status, data = imap.uid("SEARCH", None, f'SUBJECT "{subject}"')
if status != "OK" or not data or not data[0]:
ids = []
else:
ids = data[0].split()
thread_uids = []
participants = set()
for uid_bytes in ids:
u = int(uid_bytes)
if len(thread_uids) >= max_messages:
break
try:
status, mdata = imap.uid("FETCH", str(u), "(RFC822)")
except Exception:
continue
if status != "OK" or not mdata or mdata[0] is None:
continue
m = email.message_from_bytes(mdata[0][1], policy=email.policy.default)
mid = (m.get("Message-ID", "") or "").strip()
mrefs = set((m.get("References", "") or "").split())
mirt = (m.get("In-Reply-To", "") or "").strip()
is_base = (u == uid or mid == msg_id)
is_related = (
refs and mrefs & refs
) or (
mirt and (mirt == msg_id or mirt in refs)
) or (
mid and (mid in refs or mid == in_reply)
)
if is_base or is_related:
thread_uids.append(u)
from_header = m.get("From", "") or ""
if from_header:
participants.add(from_header)
# Build concise thread view
messages = []
for u in thread_uids:
try:
msg = parse_message(imap, u)
except Exception:
continue
snippet = (msg["body_plain"] or "")[:200].strip()
messages.append({
"uid": u,
"from": msg["from"],
"date": msg["date"],
"subject": msg["subject"],
"snippet": snippet,
})
return {
"subject": subject,
"participants": list(participants),
"message_count": len(messages),
"messages": messages,
}
finally:
imap.logout()
def list_attachments_for_uid(imap, folder, uid):
"""
Helper: list attachments for a given UID.
"""
ensure_selected(imap, folder)
status, msg_data = imap.uid("FETCH", str(uid), "(RFC822)")
if status != "OK" or not msg_data or msg_data[0] is None:
return []
msg = email.message_from_bytes(msg_data[0][1], policy=email.policy.default)
attachments = []
if msg.is_multipart():
for part in msg.walk():
disp = (part.get("Content-Disposition") or "").lower()
if "attachment" in disp or (part.get_filename() and "inline" not in disp):
filename = part.get_filename()
if filename:
filename = email.utils.decode_param(filename)[0].decode("utf-8", errors="replace")
else:
filename = "attachment"
size = len(part.get_payload(decode=True) or b"")
attachments.append({"filename": filename, "size": size})
else:
if msg.get_filename():
filename = email.utils.decode_param(msg.get_filename())[0].decode("utf-8", errors="replace")
size = len(msg.get_payload(decode=True) or b"")
attachments.append({"filename": filename, "size": size})
return attachments
def send_email_impl(args):
account_id = args.get("account") or config.default_account_id
acc = get_account(account_id)