delete_emails
Remove specific emails from an email account by their IDs to manage mailbox storage and organization.
Instructions
Delete one or more emails by their email_id. Use list_emails_metadata first to get the email_id.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| account_name | Yes | The name of the email account. | |
| email_ids | Yes | List of email_id to delete (obtained from list_emails_metadata). | |
| mailbox | No | The mailbox to delete emails from. | INBOX |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- mcp_email_server/app.py:164-181 (handler)MCP 'delete_emails' tool handler: decorated with @mcp.tool (registration), defines input schema with Annotated[Field], dispatches to account-specific handler.delete_emails and formats response.
@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="The mailbox to delete emails from.")] = "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 - ClassicEmailHandler.delete_emails method: delegates deletion to the underlying incoming_client (EmailClient).
async def delete_emails(self, email_ids: list[str], mailbox: str = "INBOX") -> tuple[list[str], list[str]]: """Delete emails by their UIDs. Returns (deleted_ids, failed_ids).""" return await self.incoming_client.delete_emails(email_ids, mailbox) - EmailClient.delete_emails core implementation: connects to IMAP server, marks each email ID (UID) with \Deleted flag, expunges mailbox to permanently delete, handles failures.
async def delete_emails(self, email_ids: list[str], mailbox: str = "INBOX") -> tuple[list[str], list[str]]: """Delete emails by their UIDs. Returns (deleted_ids, failed_ids).""" imap = self.imap_class(self.email_server.host, self.email_server.port) deleted_ids = [] failed_ids = [] try: await imap._client_task await imap.wait_hello_from_server() await imap.login(self.email_server.user_name, self.email_server.password) await imap.select(mailbox) for email_id in email_ids: try: await imap.uid("store", email_id, "+FLAGS", r"(\Deleted)") deleted_ids.append(email_id) except Exception as e: logger.error(f"Failed to delete email {email_id}: {e}") failed_ids.append(email_id) await imap.expunge() finally: try: await imap.logout() except Exception as e: logger.info(f"Error during logout: {e}") return deleted_ids, failed_ids