app.py

from datetime import datetime
from typing import Annotated, Literal

from mcp.server.fastmcp import FastMCP
from pydantic import Field

from mcp_email_server.config import (
    AccountAttributes,
    EmailSettings,
    ProviderSettings,
    get_settings,
)
from mcp_email_server.emails.dispatcher import dispatch_handler
from mcp_email_server.emails.models import (
    AttachmentDownloadResponse,
    EmailContentBatchResponse,
    EmailLabelsResponse,
    EmailMarkResponse,
    EmailMetadataPageResponse,
    EmailMoveResponse,
    FolderListResponse,
    FolderOperationResponse,
    LabelListResponse,
)

mcp = FastMCP("email")


@mcp.resource("email://{account_name}")
async def get_account(account_name: str) -> EmailSettings | ProviderSettings | None:
    settings = get_settings()
    return settings.get_account(account_name, masked=True)


@mcp.tool(description="List all configured email accounts with masked credentials.")
async def list_available_accounts() -> list[AccountAttributes]:
    settings = get_settings()
    return [account.masked() for account in settings.get_accounts()]


@mcp.tool(description="Add a new email account configuration to the settings.")
async def add_email_account(email: EmailSettings) -> str:
    settings = get_settings()
    settings.add_email(email)
    settings.store()
    return f"Successfully added email account '{email.account_name}'"


@mcp.tool(
    description="List email metadata (email_id, subject, sender, recipients, date) without body content. Returns email_id for use with get_emails_content."
)
async def list_emails_metadata(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    page: Annotated[
        int,
        Field(default=1, description="The page number to retrieve (starting from 1)."),
    ] = 1,
    page_size: Annotated[int, Field(default=10, description="The number of emails to retrieve per page.")] = 10,
    before: Annotated[
        datetime | None,
        Field(default=None, description="Retrieve emails before this datetime (UTC)."),
    ] = None,
    since: Annotated[
        datetime | None,
        Field(default=None, description="Retrieve emails since this datetime (UTC)."),
    ] = None,
    subject: Annotated[str | None, Field(default=None, description="Filter emails by subject.")] = None,
    from_address: Annotated[str | None, Field(default=None, description="Filter emails by sender address.")] = None,
    to_address: Annotated[
        str | None,
        Field(default=None, description="Filter emails by recipient address."),
    ] = None,
    order: Annotated[
        Literal["asc", "desc"],
        Field(default=None, description="Order emails by field. `asc` or `desc`."),
    ] = "desc",
    mailbox: Annotated[
        str,
        Field(
            default="INBOX",
            description="IMAP folder path. Standard: INBOX, Sent, Drafts, Trash. Provider-specific: Gmail uses '[Gmail]/...' prefix (e.g., '[Gmail]/Sent Mail'); ProtonMail Bridge exposes folders as 'Folders/<name>' and labels as 'Labels/<name>'.",
        ),
    ] = "INBOX",
    seen: Annotated[
        bool | None,
        Field(default=None, description="Filter by read status: True=read, False=unread, None=all."),
    ] = None,
    flagged: Annotated[
        bool | None,
        Field(default=None, description="Filter by flagged/starred status: True=flagged, False=unflagged, None=all."),
    ] = None,
    answered: Annotated[
        bool | None,
        Field(default=None, description="Filter by replied status: True=replied, False=not replied, None=all."),
    ] = None,
) -> EmailMetadataPageResponse:
    handler = dispatch_handler(account_name)

    return await handler.get_emails_metadata(
        page=page,
        page_size=page_size,
        before=before,
        since=since,
        subject=subject,
        from_address=from_address,
        to_address=to_address,
        order=order,
        mailbox=mailbox,
        seen=seen,
        flagged=flagged,
        answered=answered,
    )


@mcp.tool(
    description="Get the full content (including body) of one or more emails by their email_id. Use list_emails_metadata first to get the email_id."
)
async def get_emails_content(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    email_ids: Annotated[
        list[str],
        Field(
            description="List of email_id to retrieve (obtained from list_emails_metadata). Can be a single email_id or multiple email_ids."
        ),
    ],
    mailbox: Annotated[
        str,
        Field(
            default="INBOX",
            description="IMAP folder path. Standard: INBOX, Sent, Drafts, Trash. Provider-specific: Gmail uses '[Gmail]/...' prefix; ProtonMail Bridge uses 'Folders/<name>' and 'Labels/<name>'.",
        ),
    ] = "INBOX",
) -> EmailContentBatchResponse:
    handler = dispatch_handler(account_name)
    return await handler.get_emails_content(email_ids, mailbox)


@mcp.tool(
    description="Send an email using the specified account. Supports replying to emails with proper threading when in_reply_to is provided.",
)
async def send_email(
    account_name: Annotated[str, Field(description="The name of the email account to send from.")],
    recipients: Annotated[list[str], Field(description="A list of recipient email addresses.")],
    subject: Annotated[str, Field(description="The subject of the email.")],
    body: Annotated[str, Field(description="The body of the email.")],
    cc: Annotated[
        list[str] | None,
        Field(default=None, description="A list of CC email addresses."),
    ] = None,
    bcc: Annotated[
        list[str] | None,
        Field(default=None, description="A list of BCC email addresses."),
    ] = None,
    html: Annotated[
        bool,
        Field(default=False, description="Whether to send the email as HTML (True) or plain text (False)."),
    ] = False,
    attachments: Annotated[
        list[str] | None,
        Field(
            default=None,
            description="A list of absolute file paths to attach to the email. Supports common file types (documents, images, archives, etc.).",
        ),
    ] = None,
    in_reply_to: Annotated[
        str | None,
        Field(
            default=None,
            description="Message-ID of the email being replied to. Enables proper threading in email clients.",
        ),
    ] = None,
    references: Annotated[
        str | None,
        Field(
            default=None,
            description="Space-separated Message-IDs for the thread chain. Usually includes in_reply_to plus ancestors.",
        ),
    ] = None,
) -> str:
    handler = dispatch_handler(account_name)
    await handler.send_email(
        recipients,
        subject,
        body,
        cc,
        bcc,
        html,
        attachments,
        in_reply_to,
        references,
    )
    recipient_str = ", ".join(recipients)
    attachment_info = f" with {len(attachments)} attachment(s)" if attachments else ""
    return f"Email sent successfully to {recipient_str}{attachment_info}"


@mcp.tool(
    description="Delete one or more emails by their email_id. Use list_emails_metadata first to get the email_id."
)
async def delete_emails(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    email_ids: Annotated[
        list[str],
        Field(description="List of email_id to delete (obtained from list_emails_metadata)."),
    ],
    mailbox: Annotated[
        str,
        Field(
            default="INBOX",
            description="IMAP folder path. Standard: INBOX, Sent, Drafts, Trash. Provider-specific: Gmail uses '[Gmail]/...' prefix; ProtonMail Bridge uses 'Folders/<name>' and 'Labels/<name>'.",
        ),
    ] = "INBOX",
) -> str:
    handler = dispatch_handler(account_name)
    deleted_ids, failed_ids = await handler.delete_emails(email_ids, mailbox)

    result = f"Successfully deleted {len(deleted_ids)} email(s)"
    if failed_ids:
        result += f", failed to delete {len(failed_ids)} email(s): {', '.join(failed_ids)}"
    return result


@mcp.tool(description="Mark one or more emails as read or unread. Use list_emails_metadata first to get the email_id.")
async def mark_emails(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    email_ids: Annotated[
        list[str],
        Field(description="List of email_id to mark (obtained from list_emails_metadata)."),
    ],
    mark_as: Annotated[
        Literal["read", "unread"],
        Field(description="Mark emails as 'read' or 'unread'."),
    ],
    mailbox: Annotated[
        str,
        Field(
            default="INBOX",
            description="IMAP folder path. Standard: INBOX, Sent, Drafts, Trash. Provider-specific: Gmail uses '[Gmail]/...' prefix; ProtonMail Bridge uses 'Folders/<name>' and 'Labels/<name>'.",
        ),
    ] = "INBOX",
) -> EmailMarkResponse:
    handler = dispatch_handler(account_name)
    return await handler.mark_emails(email_ids, mark_as, mailbox)


@mcp.tool(
    description="Download an email attachment and save it to the specified path. This feature must be explicitly enabled in settings (enable_attachment_download=true) due to security considerations.",
)
async def download_attachment(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    email_id: Annotated[
        str, Field(description="The email ID (obtained from list_emails_metadata or get_emails_content).")
    ],
    attachment_name: Annotated[
        str, Field(description="The name of the attachment to download (as shown in the attachments list).")
    ],
    save_path: Annotated[str, Field(description="The absolute path where the attachment should be saved.")],
    mailbox: Annotated[
        str,
        Field(
            default="INBOX",
            description="IMAP folder path. Standard: INBOX, Sent, Drafts, Trash. Provider-specific: Gmail uses '[Gmail]/...' prefix; ProtonMail Bridge uses 'Folders/<name>' and 'Labels/<name>'.",
        ),
    ] = "INBOX",
) -> AttachmentDownloadResponse:
    settings = get_settings()
    if not settings.enable_attachment_download:
        msg = (
            "Attachment download is disabled. Set 'enable_attachment_download=true' in settings to enable this feature."
        )
        raise PermissionError(msg)

    handler = dispatch_handler(account_name)
    return await handler.download_attachment(email_id, attachment_name, save_path, mailbox)


def _check_folder_management_enabled() -> None:
    """Check if folder management is enabled, raise PermissionError if not."""
    settings = get_settings()
    if not settings.enable_folder_management:
        msg = (
            "Folder management is disabled. Set 'enable_folder_management=true' in settings "
            "or 'MCP_EMAIL_SERVER_ENABLE_FOLDER_MANAGEMENT=true' environment variable to enable this feature."
        )
        raise PermissionError(msg)


@mcp.tool(
    description="List all folders/mailboxes for an email account. Returns folder names, hierarchy delimiters, and IMAP flags. Requires enable_folder_management=true.",
)
async def list_folders(
    account_name: Annotated[str, Field(description="The name of the email account.")],
) -> FolderListResponse:
    _check_folder_management_enabled()
    handler = dispatch_handler(account_name)
    return await handler.list_folders()


@mcp.tool(
    description="Move one or more emails to a different folder (removes from source). Use this to clear emails from INBOX. Uses IMAP MOVE command if supported, otherwise falls back to COPY + DELETE. Requires enable_folder_management=true.",
)
async def move_emails(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    email_ids: Annotated[
        list[str],
        Field(description="List of email_id to move (obtained from list_emails_metadata)."),
    ],
    destination_folder: Annotated[str, Field(description="The destination folder name.")],
    source_mailbox: Annotated[
        str, Field(default="INBOX", description="The source mailbox to move emails from.")
    ] = "INBOX",
) -> EmailMoveResponse:
    _check_folder_management_enabled()
    handler = dispatch_handler(account_name)
    return await handler.move_emails(email_ids, destination_folder, source_mailbox)


@mcp.tool(
    description="Copy one or more emails to a different folder. The original emails remain in the source folder. Useful for applying labels in providers like Proton Mail. Requires enable_folder_management=true.",
)
async def copy_emails(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    email_ids: Annotated[
        list[str],
        Field(description="List of email_id to copy (obtained from list_emails_metadata)."),
    ],
    destination_folder: Annotated[str, Field(description="The destination folder name.")],
    source_mailbox: Annotated[
        str, Field(default="INBOX", description="The source mailbox to copy emails from.")
    ] = "INBOX",
) -> EmailMoveResponse:
    _check_folder_management_enabled()
    handler = dispatch_handler(account_name)
    return await handler.copy_emails(email_ids, destination_folder, source_mailbox)


@mcp.tool(description="Create a new folder/mailbox. Requires enable_folder_management=true.")
async def create_folder(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    folder_name: Annotated[str, Field(description="The name of the folder to create.")],
) -> FolderOperationResponse:
    _check_folder_management_enabled()
    handler = dispatch_handler(account_name)
    return await handler.create_folder(folder_name)


@mcp.tool(
    description="Delete a folder/mailbox. The folder must be empty on most IMAP servers. Requires enable_folder_management=true."
)
async def delete_folder(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    folder_name: Annotated[str, Field(description="The name of the folder to delete.")],
) -> FolderOperationResponse:
    _check_folder_management_enabled()
    handler = dispatch_handler(account_name)
    return await handler.delete_folder(folder_name)


@mcp.tool(description="Rename a folder/mailbox. Requires enable_folder_management=true.")
async def rename_folder(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    old_name: Annotated[str, Field(description="The current folder name.")],
    new_name: Annotated[str, Field(description="The new folder name.")],
) -> FolderOperationResponse:
    _check_folder_management_enabled()
    handler = dispatch_handler(account_name)
    return await handler.rename_folder(old_name, new_name)


@mcp.tool(
    description="List all labels for an email account (ProtonMail: folders under Labels/ prefix). Requires enable_folder_management=true."
)
async def list_labels(
    account_name: Annotated[str, Field(description="The name of the email account.")],
) -> LabelListResponse:
    _check_folder_management_enabled()
    handler = dispatch_handler(account_name)
    return await handler.list_labels()


@mcp.tool(
    description="Apply a label to one or more emails. NOTE: This only tags emails - originals stay in INBOX. To remove from INBOX, use move_emails instead. Requires enable_folder_management=true."
)
async def apply_label(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    email_ids: Annotated[
        list[str],
        Field(description="List of email_id to label (obtained from list_emails_metadata)."),
    ],
    label_name: Annotated[str, Field(description="The label name (without Labels/ prefix).")],
    source_mailbox: Annotated[
        str, Field(default="INBOX", description="The source mailbox containing the emails.")
    ] = "INBOX",
) -> EmailMoveResponse:
    _check_folder_management_enabled()
    handler = dispatch_handler(account_name)
    return await handler.apply_label(email_ids, label_name, source_mailbox)


@mcp.tool(
    description="Remove a label from one or more emails. Deletes from label folder while preserving original emails. Requires enable_folder_management=true."
)
async def remove_label(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    email_ids: Annotated[
        list[str],
        Field(description="List of email_id to unlabel (obtained from list_emails_metadata)."),
    ],
    label_name: Annotated[str, Field(description="The label name (without Labels/ prefix).")],
) -> EmailMoveResponse:
    _check_folder_management_enabled()
    handler = dispatch_handler(account_name)
    return await handler.remove_label(email_ids, label_name)


@mcp.tool(description="Get all labels applied to a specific email. Requires enable_folder_management=true.")
async def get_email_labels(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    email_id: Annotated[str, Field(description="The email_id to check (obtained from list_emails_metadata).")],
    source_mailbox: Annotated[
        str, Field(default="INBOX", description="The source mailbox containing the email.")
    ] = "INBOX",
) -> EmailLabelsResponse:
    _check_folder_management_enabled()
    handler = dispatch_handler(account_name)
    return await handler.get_email_labels(email_id, source_mailbox)


@mcp.tool(description="Create a new label (creates Labels/name folder). Requires enable_folder_management=true.")
async def create_label(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    label_name: Annotated[str, Field(description="The label name to create (without Labels/ prefix).")],
) -> FolderOperationResponse:
    _check_folder_management_enabled()
    handler = dispatch_handler(account_name)
    return await handler.create_label(label_name)


@mcp.tool(
    description="Delete a label (deletes Labels/name folder). The label must be empty on most IMAP servers. Requires enable_folder_management=true."
)
async def delete_label(
    account_name: Annotated[str, Field(description="The name of the email account.")],
    label_name: Annotated[str, Field(description="The label name to delete (without Labels/ prefix).")],
) -> FolderOperationResponse:
    _check_folder_management_enabled()
    handler = dispatch_handler(account_name)
    return await handler.delete_label(label_name)