get_emails_content

Instructions

Implementation Reference

• mcp_email_server/emails/classic.py:1049-1081 (handler)

  ClassicEmailHandler.get_emails_content - concrete implementation that iterates over email_ids, fetches each email via IMAP, and returns EmailContentBatchResponse.

  async def get_emails_content(self, email_ids: list[str], mailbox: str = "INBOX") -> EmailContentBatchResponse:
      """Batch retrieve email body content"""
      emails = []
      failed_ids = []
  
      for email_id in email_ids:
          try:
              email_data = await self.incoming_client.get_email_body_by_id(email_id, mailbox)
              if email_data:
                  emails.append(
                      EmailBodyResponse(
                          email_id=email_data["email_id"],
                          message_id=email_data.get("message_id"),
                          subject=email_data["subject"],
                          sender=email_data["from"],
                          recipients=email_data["to"],
                          date=email_data["date"],
                          body=email_data["body"],
                          attachments=email_data["attachments"],
                      )
                  )
              else:
                  failed_ids.append(email_id)
          except Exception as e:
              logger.error(f"Failed to retrieve email {email_id}: {e}")
              failed_ids.append(email_id)
  
      return EmailContentBatchResponse(
          emails=emails,
          requested_count=len(email_ids),
          retrieved_count=len(emails),
          failed_ids=failed_ids,
      )

• mcp_email_server/app.py:103-117 (registration)

  MCP tool registration using @mcp.tool decorator. This is the entry point that dispatches to the EmailHandler implementation.

  @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="The mailbox to retrieve emails from.")] = "INBOX",
  ) -> EmailContentBatchResponse:
      handler = dispatch_handler(account_name)
      return await handler.get_emails_content(email_ids, mailbox)

• mcp_email_server/emails/models.py:49-55 (schema)

  Pydantic response model for batch email content results.

  class EmailContentBatchResponse(BaseModel):
      """Batch email content response for multiple emails"""
  
      emails: list[EmailBodyResponse]
      requested_count: int
      retrieved_count: int
      failed_ids: list[str]

• mcp_email_server/emails/models.py:43-46 (schema)

  Pydantic model for a single email's body response (extends EmailMetadata with body field).

  class EmailBodyResponse(EmailMetadata):
      """Single email body response - extends EmailMetadata with body content"""
  
      body: str

• mcp_email_server/emails/classic.py:588-624 (helper)

  EmailClient.get_email_body_by_id - helper that connects to IMAP, fetches and parses a single email's full content.

  async def get_email_body_by_id(self, email_id: str, mailbox: str = "INBOX") -> dict[str, Any] | None:
      imap = self._imap_connect()
      try:
          # Wait for the connection to be established
          await imap._client_task
          await imap.wait_hello_from_server()
  
          # Login and select inbox
          await imap.login(self.email_server.user_name, self.email_server.password.get_secret_value())
          await _send_imap_id(imap)
          await imap.select(_quote_mailbox(mailbox))
  
          # Fetch the specific email by UID
          data = await self._fetch_email_with_formats(imap, email_id)
          if not data:
              logger.error(f"Failed to fetch UID {email_id} with any format")
              return None
  
          # Extract raw email data
          raw_email = self._extract_raw_email(data)
          if not raw_email:
              logger.error(f"Could not find email data in response for email ID: {email_id}")
              return None
  
          # Parse the email
          try:
              return self._parse_email_data(raw_email, email_id)
          except Exception as e:
              logger.error(f"Error parsing email: {e!s}")
              return None
  
      finally:
          # Ensure we logout properly
          try:
              await imap.logout()
          except Exception as e:
              logger.info(f"Error during logout: {e}")