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
      """