Add email flags and unsubscribe features

- 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 <noreply@anthropic.com>

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

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):

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)

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: <url1>, <url2> or <url>
+    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,
+        }
+    )

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()