MCP Email Server

Instructions

Implementation Reference

• mcp_email_server/app.py:184-206 (registration)

  MCP tool registration for 'download_attachment' including input schema via Annotated Fields, permission check, and dispatch to account-specific handler.
  @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.")], ) -> 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)
• mcp_email_server/emails/classic.py:401-474 (handler)

  Core implementation in EmailClient (or similar) that performs IMAP fetch, MIME parsing, attachment extraction, file saving, and returns result dict.
  async def download_attachment( self, email_id: str, attachment_name: str, save_path: str, ) -> dict[str, Any]: """Download a specific attachment from an email and save it to disk.""" imap = self.imap_class(self.email_server.host, self.email_server.port) try: await imap._client_task await imap.wait_hello_from_server() await imap.login(self.email_server.user_name, self.email_server.password) try: await imap.id(name="mcp-email-server", version="1.0.0") except Exception as e: logger.warning(f"IMAP ID command failed: {e!s}") await imap.select("INBOX") data = await self._fetch_email_with_formats(imap, email_id) if not data: msg = f"Failed to fetch email with UID {email_id}" logger.error(msg) raise ValueError(msg) raw_email = self._extract_raw_email(data) if not raw_email: msg = f"Could not find email data for email ID: {email_id}" logger.error(msg) raise ValueError(msg) parser = BytesParser(policy=default) email_message = parser.parsebytes(raw_email) # Find the attachment attachment_data = None mime_type = None if email_message.is_multipart(): for part in email_message.walk(): content_disposition = str(part.get("Content-Disposition", "")) if "attachment" in content_disposition: filename = part.get_filename() if filename == attachment_name: attachment_data = part.get_payload(decode=True) mime_type = part.get_content_type() break if attachment_data is None: msg = f"Attachment '{attachment_name}' not found in email {email_id}" logger.error(msg) raise ValueError(msg) # Save to disk save_file = Path(save_path) save_file.parent.mkdir(parents=True, exist_ok=True) save_file.write_bytes(attachment_data) logger.info(f"Attachment '{attachment_name}' saved to {save_path}") return { "email_id": email_id, "attachment_name": attachment_name, "mime_type": mime_type or "application/octet-stream", "size": len(attachment_data), "saved_path": str(save_file.resolve()), } finally: try: await imap.logout() except Exception as e: logger.info(f"Error during logout: {e}")
• mcp_email_server/emails/models.py:65-72 (schema)

  Pydantic model defining the output schema for attachment download response.
  class AttachmentDownloadResponse(BaseModel): """Attachment download response""" email_id: str attachment_name: str mime_type: str size: int saved_path: str
• mcp_email_server/emails/classic.py:807-821 (helper)

  ClassicEmailHandler implementation that delegates to incoming_client and converts result to typed response.
  async def download_attachment( self, email_id: str, attachment_name: str, save_path: str, ) -> AttachmentDownloadResponse: """Download an email attachment and save it to the specified path.""" result = await self.incoming_client.download_attachment(email_id, attachment_name, save_path) return AttachmentDownloadResponse( email_id=result["email_id"], attachment_name=result["attachment_name"], mime_type=result["mime_type"], size=result["size"], saved_path=result["saved_path"], )
• mcp_email_server/emails/__init__.py:58-67 (schema)

  Abstract base class method defining the EmailHandler interface for download_attachment.
  @abc.abstractmethod async def download_attachment( self, email_id: str, attachment_name: str, save_path: str, ) -> "AttachmentDownloadResponse": """ Download an email attachment and save it to the specified path """