From ba57c5fba4c3d464db55d02126c79aab274e441d Mon Sep 17 00:00:00 2001 From: Yigit Colakoglu Date: Tue, 30 Dec 2025 17:23:49 -0800 Subject: [PATCH] Add email flags and unsubscribe features MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add set_email_flags tool for IMAP flags (standard + custom keywords) - Add unsubscribe_email tool with List-Unsubscribe header parsing - Support RFC 8058 one-click unsubscribe - Add UnsubscribeInfo model to email responses - Add data field to OperationResult for extra context - Include test.sh script for MCP server testing 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- src/models/common.py | 3 +- src/models/email_models.py | 9 ++ src/services/email_service.py | 81 +++++++++++- src/services/unsubscribe_service.py | 191 ++++++++++++++++++++++++++++ src/tools/email_tools.py | 39 ++++++ 5 files changed, 321 insertions(+), 2 deletions(-) create mode 100644 src/services/unsubscribe_service.py diff --git a/src/models/common.py b/src/models/common.py index a7383dd..c2003b2 100644 --- a/src/models/common.py +++ b/src/models/common.py @@ -1,8 +1,9 @@ from pydantic import BaseModel -from typing import Optional +from typing import Any, Optional class OperationResult(BaseModel): success: bool message: str id: Optional[str] = None + data: Optional[dict[str, Any]] = None diff --git a/src/models/email_models.py b/src/models/email_models.py index 1138a20..df3844a 100644 --- a/src/models/email_models.py +++ b/src/models/email_models.py @@ -23,6 +23,14 @@ class Attachment(BaseModel): content_id: Optional[str] = None +class UnsubscribeInfo(BaseModel): + """Information about how to unsubscribe from a mailing list.""" + available: bool = False + http_url: Optional[str] = None + mailto: Optional[str] = None + one_click: bool = False # RFC 8058 one-click unsubscribe supported + + class EmailSummary(BaseModel): id: str mailbox: str @@ -45,6 +53,7 @@ class Email(EmailSummary): headers: dict[str, str] = {} in_reply_to: Optional[str] = None references: list[str] = [] + unsubscribe: Optional[UnsubscribeInfo] = None class EmailList(BaseModel): diff --git a/src/services/email_service.py b/src/services/email_service.py index bf652e3..55b06ee 100644 --- a/src/services/email_service.py +++ b/src/services/email_service.py @@ -17,9 +17,11 @@ from models.email_models import ( EmailSummary, Email, EmailList, + UnsubscribeInfo, ) from models.common import OperationResult from config import Settings +from services.unsubscribe_service import parse_unsubscribe_header, execute_unsubscribe def decode_mime_header(header) -> str: @@ -239,11 +241,14 @@ class EmailService: # Get headers headers = {} - for key in ["Message-ID", "In-Reply-To", "References", "X-Priority"]: + for key in ["Message-ID", "In-Reply-To", "References", "X-Priority", "List-Unsubscribe", "List-Unsubscribe-Post"]: value = msg.get(key) if value: headers[key] = decode_mime_header(value) + # Parse unsubscribe info + unsubscribe_info = parse_unsubscribe_header(msg) + return Email( id=str(uid), mailbox=mailbox, @@ -262,6 +267,7 @@ class EmailService: headers=headers, in_reply_to=headers.get("In-Reply-To"), references=headers.get("References", "").split() if headers.get("References") else [], + unsubscribe=unsubscribe_info if unsubscribe_info.available else None, ) def search_emails( @@ -568,3 +574,76 @@ class EmailService: if name in trash_names or b"\\Trash" in flags: return name return None + + def set_flags( + self, + email_id: str, + mailbox: str, + add_flags: Optional[list[str]] = None, + remove_flags: Optional[list[str]] = None, + ) -> OperationResult: + """ + Set or remove IMAP flags on an email. + + Standard flags: \\Seen, \\Answered, \\Flagged, \\Deleted, \\Draft + Custom keywords are also supported (server-dependent). + + Args: + email_id: The unique ID of the email + mailbox: The mailbox containing the email + add_flags: List of flags to add + remove_flags: List of flags to remove + """ + try: + client = self._get_imap_client() + client.select_folder(mailbox) + uid = int(email_id) + + if add_flags: + client.add_flags([uid], add_flags) + if remove_flags: + client.remove_flags([uid], remove_flags) + + return OperationResult( + success=True, + message=f"Flags updated for email {email_id}", + id=email_id, + data={ + "added": add_flags or [], + "removed": remove_flags or [], + } + ) + except Exception as e: + return OperationResult(success=False, message=str(e)) + + async def unsubscribe( + self, + mailbox: str, + email_id: str, + ) -> OperationResult: + """ + Attempt to unsubscribe from a mailing list based on email headers. + + Args: + mailbox: The mailbox containing the email + email_id: The unique ID of the email + + Returns: + OperationResult with unsubscribe status + """ + # First, read the email to get unsubscribe info + email_data = self.read_email(mailbox, email_id) + if not email_data: + return OperationResult( + success=False, + message=f"Email {email_id} not found in {mailbox}" + ) + + if not email_data.unsubscribe: + return OperationResult( + success=False, + message="This email does not have unsubscribe information" + ) + + # Execute unsubscribe + return await execute_unsubscribe(email_data.unsubscribe) diff --git a/src/services/unsubscribe_service.py b/src/services/unsubscribe_service.py new file mode 100644 index 0000000..ff332bd --- /dev/null +++ b/src/services/unsubscribe_service.py @@ -0,0 +1,191 @@ +"""Service for parsing and executing email unsubscribe actions.""" + +import re +import logging +from typing import Optional +from email.message import Message + +import httpx + +from models.email_models import UnsubscribeInfo +from models.common import OperationResult + +logger = logging.getLogger(__name__) + + +def parse_unsubscribe_header(msg: Message) -> UnsubscribeInfo: + """ + Parse List-Unsubscribe and List-Unsubscribe-Post headers from an email. + + Supports: + - RFC 2369: List-Unsubscribe header with mailto: and http(s): URLs + - RFC 8058: One-Click Unsubscribe via List-Unsubscribe-Post header + + Args: + msg: The email message to parse + + Returns: + UnsubscribeInfo with available unsubscribe methods + """ + info = UnsubscribeInfo() + + list_unsub = msg.get("List-Unsubscribe", "") + list_unsub_post = msg.get("List-Unsubscribe-Post", "") + + if not list_unsub: + return info + + # Parse URLs from List-Unsubscribe header + # Format: , or + urls = re.findall(r'<([^>]+)>', list_unsub) + + for url in urls: + url = url.strip() + if url.startswith("mailto:"): + info.mailto = url + info.available = True + elif url.startswith("http://") or url.startswith("https://"): + info.http_url = url + info.available = True + + # Check for one-click unsubscribe support (RFC 8058) + # List-Unsubscribe-Post: List-Unsubscribe=One-Click + if list_unsub_post and "List-Unsubscribe=One-Click" in list_unsub_post: + info.one_click = True + + return info + + +async def execute_unsubscribe( + unsubscribe_info: UnsubscribeInfo, + prefer_http: bool = True, +) -> OperationResult: + """ + Execute an unsubscribe action. + + Args: + unsubscribe_info: The unsubscribe information from the email + prefer_http: If True, prefer HTTP unsubscribe over mailto + + Returns: + OperationResult indicating success or failure + """ + if not unsubscribe_info.available: + return OperationResult( + success=False, + message="No unsubscribe option available for this email" + ) + + # Try HTTP unsubscribe first if preferred and available + if prefer_http and unsubscribe_info.http_url: + result = await _http_unsubscribe( + unsubscribe_info.http_url, + unsubscribe_info.one_click + ) + if result.success: + return result + + # Try mailto unsubscribe + if unsubscribe_info.mailto: + return _prepare_mailto_unsubscribe(unsubscribe_info.mailto) + + # Fall back to HTTP if mailto didn't work + if unsubscribe_info.http_url and not prefer_http: + return await _http_unsubscribe( + unsubscribe_info.http_url, + unsubscribe_info.one_click + ) + + return OperationResult( + success=False, + message="Failed to unsubscribe using available methods" + ) + + +async def _http_unsubscribe(url: str, one_click: bool) -> OperationResult: + """ + Perform HTTP-based unsubscribe. + + Args: + url: The unsubscribe URL + one_click: Whether RFC 8058 one-click is supported + """ + try: + async with httpx.AsyncClient( + timeout=30.0, + follow_redirects=True + ) as client: + if one_click: + # RFC 8058: POST with List-Unsubscribe=One-Click + response = await client.post( + url, + data={"List-Unsubscribe": "One-Click"}, + headers={"Content-Type": "application/x-www-form-urlencoded"} + ) + else: + # Standard: GET request to the URL + response = await client.get(url) + + if response.status_code in (200, 201, 202, 204): + return OperationResult( + success=True, + message=f"Successfully unsubscribed via HTTP ({response.status_code})" + ) + elif response.status_code in (301, 302, 303, 307, 308): + # Follow redirect manually if needed + return OperationResult( + success=False, + message=f"Unsubscribe requires manual confirmation at: {response.headers.get('Location', url)}" + ) + else: + return OperationResult( + success=False, + message=f"HTTP unsubscribe failed with status {response.status_code}" + ) + + except httpx.TimeoutException: + return OperationResult( + success=False, + message="Unsubscribe request timed out" + ) + except Exception as e: + logger.error(f"HTTP unsubscribe error: {e}") + return OperationResult( + success=False, + message=f"HTTP unsubscribe failed: {str(e)}" + ) + + +def _prepare_mailto_unsubscribe(mailto: str) -> OperationResult: + """ + Prepare mailto unsubscribe info. + + Note: We can't send emails automatically without user consent, + so we return the mailto details for the user/agent to act on. + """ + # Parse mailto URL + # Format: mailto:unsub@example.com?subject=unsubscribe&body=... + if mailto.startswith("mailto:"): + mailto = mailto[7:] + + parts = mailto.split("?", 1) + email = parts[0] + subject = "" + body = "" + + if len(parts) > 1: + params = dict(p.split("=", 1) for p in parts[1].split("&") if "=" in p) + subject = params.get("subject", "unsubscribe") + body = params.get("body", "") + + return OperationResult( + success=True, + message=f"To unsubscribe, send an email to: {email}", + id=email, + data={ + "method": "mailto", + "email": email, + "subject": subject or "unsubscribe", + "body": body, + } + ) diff --git a/src/tools/email_tools.py b/src/tools/email_tools.py index 9dad3d8..8d509a5 100644 --- a/src/tools/email_tools.py +++ b/src/tools/email_tools.py @@ -132,3 +132,42 @@ def register_email_tools(mcp: FastMCP, service: EmailService): """ result = await service.send_email(to, subject, body, cc, bcc, reply_to, html_body) return result.model_dump() + + @mcp.tool(description="Set or remove IMAP flags on an email. Standard flags: \\Seen, \\Answered, \\Flagged, \\Deleted, \\Draft. Custom keywords are also supported.") + def set_email_flags( + email_id: str, + mailbox: str, + add_flags: Optional[list[str]] = None, + remove_flags: Optional[list[str]] = None, + ) -> dict: + """ + Set or remove flags on an email. + + Args: + email_id: The unique ID of the email + mailbox: The mailbox containing the email + add_flags: Flags to add (e.g., ["\\Flagged", "important"]) + remove_flags: Flags to remove (e.g., ["\\Seen"]) + """ + result = service.set_flags(email_id, mailbox, add_flags, remove_flags) + return result.model_dump() + + @mcp.tool(description="Unsubscribe from a mailing list. Parses List-Unsubscribe headers and attempts automatic unsubscribe via HTTP or provides mailto instructions.") + async def unsubscribe_email( + email_id: str, + mailbox: str = "INBOX", + ) -> dict: + """ + Unsubscribe from a mailing list based on email headers. + + Args: + email_id: The unique ID of the email containing unsubscribe info + mailbox: The mailbox containing the email (default: INBOX) + + Returns: + Result with unsubscribe status. For HTTP unsubscribe, it attempts + automatic unsubscription. For mailto, it returns the email details + to send. + """ + result = await service.unsubscribe(mailbox, email_id) + return result.model_dump()