Perf: Use IMAP SORT and batch fetch for metadata retrieval

README.md

@@ -78,6 +78,7 @@ You can also configure the email server using environment variables, which is pa
 | `MCP_EMAIL_SERVER_SMTP_SSL`                   | Enable SMTP SSL                                  | `true`        | No       |
 | `MCP_EMAIL_SERVER_SMTP_START_SSL`             | Enable STARTTLS                                  | `false`       | No       |
 | `MCP_EMAIL_SERVER_ENABLE_ATTACHMENT_DOWNLOAD` | Enable attachment download                       | `false`       | No       |
+| `MCP_EMAIL_SERVER_ENABLE_FOLDER_MANAGEMENT`   | Enable folder management tools                   | `false`       | No       |
 | `MCP_EMAIL_SERVER_SAVE_TO_SENT`               | Save sent emails to IMAP Sent folder             | `true`        | No       |
 | `MCP_EMAIL_SERVER_SENT_FOLDER_NAME`           | Custom Sent folder name (auto-detect if not set) | -             | No       |
 
@@ -114,6 +115,39 @@ enable_attachment_download = true
 
 Once enabled, you can use the `download_attachment` tool to save email attachments to a specified path.
 
+### Enabling Folder Management
+
+By default, folder management tools (list, create, delete, rename folders, and move/copy emails) are disabled for security reasons. To enable these features:
+
+**Option 1: Environment Variable**
+
+```json
+{
+  "mcpServers": {
+    "zerolib-email": {
+      "command": "uvx",
+      "args": ["mcp-email-server@latest", "stdio"],
+      "env": {
+        "MCP_EMAIL_SERVER_ENABLE_FOLDER_MANAGEMENT": "true"
+      }
+    }
+  }
+}
+```
+
+**Option 2: TOML Configuration**
+
+Add `enable_folder_management = true` to your TOML configuration file (`~/.config/zerolib/mcp_email_server/config.toml`):
+
+```toml
+enable_folder_management = true
+
+[[emails]]
+# ... your email configuration
+```
+
+Once enabled, you can use the folder management tools: `list_folders`, `create_folder`, `delete_folder`, `rename_folder`, `move_emails`, and `copy_emails`.
+
 ### Saving Sent Emails to IMAP Sent Folder
 
 By default, sent emails are automatically saved to your IMAP Sent folder. This ensures that emails sent via the MCP server appear in your email client (Thunderbird, webmail, etc.).
@@ -217,6 +251,50 @@ await send_email(
 
 The `in_reply_to` parameter sets the `In-Reply-To` header, and `references` sets the `References` header. Both are used by email clients to thread conversations properly.
 
+### Managing Folders
+
+You can list, create, delete, and rename email folders:
+
+```python
+# List all folders
+folders = await list_folders(account_name="work")
+for folder in folders.folders:
+    print(f"{folder.name} (flags: {folder.flags})")
+
+# Create a new folder
+await create_folder(account_name="work", folder_name="Projects/2024")
+
+# Rename a folder
+await rename_folder(account_name="work", old_name="Old Name", new_name="New Name")
+
+# Delete a folder (must be empty on most servers)
+await delete_folder(account_name="work", folder_name="Old Folder")
+```
+
+### Moving and Copying Emails
+
+Move or copy emails between folders:
+
+```python
+# Move emails to a different folder
+await move_emails(
+    account_name="work",
+    email_ids=["123", "456"],
+    destination_folder="Archive",
+    source_mailbox="INBOX"
+)
+
+# Copy emails (preserves original) - useful for applying labels
+await copy_emails(
+    account_name="work",
+    email_ids=["123"],
+    destination_folder="Labels/Important",
+    source_mailbox="INBOX"
+)
+```
+
+**Note for Proton Mail users:** Proton Mail Bridge exposes labels as folders under `Labels/`. Copying an email to a label folder effectively applies that label while keeping the original in place.
+
 ## Development
 
 This project is managed using [uv](https://github.com/ai-zerolab/uv).

mcp_email_server/app.py

@@ -14,7 +14,13 @@
 from mcp_email_server.emails.models import (
     AttachmentDownloadResponse,
     EmailContentBatchResponse,
+    EmailLabelsResponse,
+    EmailMarkResponse,
     EmailMetadataPageResponse,
+    EmailMoveResponse,
+    FolderListResponse,
+    FolderOperationResponse,
+    LabelListResponse,
 )
 
 mcp = FastMCP("email")
@@ -68,7 +74,21 @@ async def list_emails_metadata(
         Literal["asc", "desc"],
         Field(default=None, description="Order emails by field. `asc` or `desc`."),
     ] = "desc",
-    mailbox: Annotated[str, Field(default="INBOX", description="The mailbox to retrieve emails from.")] = "INBOX",
+    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)
 
@@ -82,6 +102,9 @@ async def list_emails_metadata(
         to_address=to_address,
         order=order,
         mailbox=mailbox,
+        seen=seen,
+        flagged=flagged,
+        answered=answered,
     )
 
 
@@ -96,7 +119,7 @@ async def get_emails_content(
             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="The mailbox to retrieve emails from.")] = "INBOX",
+    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)
@@ -170,7 +193,7 @@ async def delete_emails(
         list[str],
         Field(description="List of email_id to delete (obtained from list_emails_metadata)."),
     ],
-    mailbox: Annotated[str, Field(default="INBOX", description="The mailbox to delete emails from.")] = "INBOX",
+    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)
@@ -181,6 +204,25 @@ async def delete_emails(
     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.",
 )
@@ -193,7 +235,7 @@ async def download_attachment(
         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(description="The mailbox to search in (default: INBOX).")] = "INBOX",
+    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:
@@ -204,3 +246,177 @@ async def download_attachment(
 
     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)

mcp_email_server/config.py

@@ -223,6 +223,7 @@ class Settings(BaseSettings):
     providers: list[ProviderSettings] = []
     db_location: str = CONFIG_PATH.with_name("db.sqlite3").as_posix()
     enable_attachment_download: bool = False
+    enable_folder_management: bool = False
 
     model_config = SettingsConfigDict(toml_file=CONFIG_PATH, validate_assignment=True, revalidate_instances="always")
 
@@ -236,6 +237,12 @@ def __init__(self, **data: Any) -> None:
             self.enable_attachment_download = _parse_bool_env(env_enable_attachment, False)
             logger.info(f"Set enable_attachment_download={self.enable_attachment_download} from environment variable")
 
+        # Check for enable_folder_management from environment variable
+        env_enable_folder = os.getenv("MCP_EMAIL_SERVER_ENABLE_FOLDER_MANAGEMENT")
+        if env_enable_folder is not None:
+            self.enable_folder_management = _parse_bool_env(env_enable_folder, False)
+            logger.info(f"Set enable_folder_management={self.enable_folder_management} from environment variable")
+
         # Check for email configuration from environment variables
         env_email = EmailSettings.from_env()
         if env_email: