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

Normalize dates to ISO 8601, sort results oldest-newest, reduce context in email tools

Browse Source
This commit is contained in:
Admin
2026-06-28 05:02:19 +00:00
parent 4095e694bf
commit dbea28b742
1 changed files with 74 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.py
+74 -17
View File
@@ -52,6 +52,48 @@ def log_context_prefix() -> str:
return ""
# -------------------- DATE UTILS --------------------
from email.utils import parsedate_to_datetime
from datetime import datetime, timezone
def parse_email_date(date_str: str) -> datetime:
"""
Parse an email Date header into a timezone-aware datetime.
Fallback: treat as UTC if ambiguous.
"""
if not date_str:
return datetime.min.replace(tzinfo=timezone.utc)
try:
dt = parsedate_to_datetime(date_str)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
except Exception:
return datetime.min.replace(tzinfo=timezone.utc)
def sort_by_date_oldest_first(items: list[dict], date_field: str = "date") -> list[dict]:
"""
Sort items (e.g. emails or thread messages) by date, oldest first.
"""
def key(item):
return parse_email_date(item.get(date_field, ""))
return sorted(items, key=key)
def normalize_date(date_str: str) -> str:
"""
Convert an email Date header to a clear ISO 8601 UTC string.
Returns original if parsing fails.
"""
dt = parse_email_date(date_str)
if dt == datetime.min:
return date_str or ""
return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
# -------------------- LOGGING (Docker-friendly) --------------------
log_level = os.getenv("LOG_LEVEL", "INFO").upper()
@@ -1212,7 +1254,7 @@ TOOLS = [
},
{
"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.",
"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. Returns results sorted oldest → newest with ISO 8601 dates.",
"inputSchema": {
"type": "object",
"properties": {
@@ -1227,7 +1269,7 @@ TOOLS = [
},
{
"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.",
"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. Date is returned in ISO 8601 format.",
"inputSchema": {
"type": "object",
"properties": {
@@ -1241,7 +1283,7 @@ TOOLS = [
},
{
"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.",
"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. Date is returned in ISO 8601 format.",
"inputSchema": {
"type": "object",
"properties": {
@@ -1255,7 +1297,7 @@ TOOLS = [
},
{
"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.",
"description": "Get a concise view of the conversation thread for a given email. Use this to understand context without loading all message bodies. Messages are returned sorted oldest → newest with ISO 8601 dates.",
"inputSchema": {
"type": "object",
"properties": {
@@ -1715,6 +1757,10 @@ def search_messages_impl(account=None, folder="INBOX", query="", since=None, max
try:
login_imap(imap, acc)
results = search_messages(imap, folder=folder, query=query, since=since, max_results=max_results)
# Normalize dates and sort oldest → newest
for m in results:
m["date"] = normalize_date(m.get("date", ""))
results = sort_by_date_oldest_first(results)
return {"messages": results}
finally:
imap.logout()
@@ -1847,9 +1893,11 @@ def get_unread_summary_impl(account=None, folder="INBOX"):
"uid": uid,
"subject": msg["subject"],
"from": msg["from"],
"date": msg["date"],
"snippet": (msg["body_plain"] or "")[:300],
"date": normalize_date(msg["date"]),
"snippet": (msg["body_plain"] or "")[:250].strip(),
})
# Sort oldest → newest
previews = sort_by_date_oldest_first(previews)
return {
"folder": folder,
"unread_count": count,
@@ -1862,6 +1910,7 @@ def get_unread_summary_impl(account=None, folder="INBOX"):
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.
Returns results sorted oldest → newest by date.
"""
acc = get_account(account)
imap = create_imap(acc)
@@ -1881,8 +1930,9 @@ def list_unread_emails_impl(account=None, folder="INBOX", max_results=50, includ
ids = data[0].split()
uids = [int(u) for u in ids]
count = len(uids)
uids = uids[-max_results:] # take most recent
total_unread = len(uids)
# take most recent, then sort chronologically
uids = uids[-max_results:]
emails = []
for uid in uids:
@@ -1891,7 +1941,7 @@ def list_unread_emails_impl(account=None, folder="INBOX", max_results=50, includ
except Exception:
continue
snippet = (msg["body_plain"] or "")[:300].strip()
snippet = (msg["body_plain"] or "")[:250].strip()
att_info = list_attachments_for_uid(imap, folder, uid)
has_attachments = len(att_info) > 0
@@ -1899,9 +1949,7 @@ def list_unread_emails_impl(account=None, folder="INBOX", max_results=50, includ
"uid": uid,
"subject": msg["subject"],
"from": msg["from"],
"to": msg["to"],
"date": msg["date"],
"flags": msg["flags"],
"date": normalize_date(msg["date"]),
"snippet": snippet,
"has_attachments": has_attachments,
}
@@ -1912,9 +1960,12 @@ def list_unread_emails_impl(account=None, folder="INBOX", max_results=50, includ
emails.append(item)
# Sort oldest → newest by date
emails = sort_by_date_oldest_first(emails)
return {
"folder": folder,
"total_unread": count,
"total_unread": total_unread,
"returned": len(emails),
"emails": emails,
}
@@ -1925,6 +1976,7 @@ def list_unread_emails_impl(account=None, folder="INBOX", max_results=50, includ
def get_email_impl(account=None, folder="INBOX", uid=None, include_html=False):
"""
Fetch the full content of a single email by UID.
Returns a normalized ISO 8601 date for reliable parsing.
"""
acc = get_account(account)
imap = create_imap(acc)
@@ -1939,7 +1991,7 @@ def get_email_impl(account=None, folder="INBOX", uid=None, include_html=False):
"from": msg["from"],
"to": msg["to"],
"cc": msg["cc"],
"date": msg["date"],
"date": normalize_date(msg["date"]),
"body_plain": msg["body_plain"],
"attachments": att_info,
}
@@ -1956,6 +2008,7 @@ 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.
Returns a normalized ISO 8601 date for reliable parsing.
"""
acc = get_account(account)
imap = create_imap(acc)
@@ -1972,7 +2025,7 @@ def summarize_email_impl(account=None, folder="INBOX", uid=None, max_length=150)
"uid": uid,
"subject": msg["subject"],
"from": msg["from"],
"date": msg["date"],
"date": normalize_date(msg["date"]),
"summary": "(No readable content)"
}
@@ -1992,7 +2045,7 @@ def summarize_email_impl(account=None, folder="INBOX", uid=None, max_length=150)
"uid": uid,
"subject": msg["subject"],
"from": msg["from"],
"date": msg["date"],
"date": normalize_date(msg["date"]),
"summary": summary + attachments_note,
}
@@ -2000,6 +2053,7 @@ def summarize_email_impl(account=None, folder="INBOX", uid=None, max_length=150)
def get_thread_impl(account=None, folder="INBOX", uid=None, max_messages=10):
"""
Return a concise thread view for a given message UID.
Messages are returned sorted oldest → newest with normalized dates.
"""
acc = get_account(account)
imap = create_imap(acc)
@@ -2067,11 +2121,14 @@ def get_thread_impl(account=None, folder="INBOX", uid=None, max_messages=10):
messages.append({
"uid": u,
"from": msg["from"],
"date": msg["date"],
"date": normalize_date(msg["date"]),
"subject": msg["subject"],
"snippet": snippet,
})
# Sort oldest → newest
messages = sort_by_date_oldest_first(messages)
return {
"subject": subject,
"participants": list(participants),