MCP Email Server

Instructions

Implementation Reference

• mcp_email_server/emails/classic.py:745-778 (handler)

  The primary handler function in ClassicEmailHandler that orchestrates batch retrieval of email contents by ID. Fetches individual emails, handles errors, constructs response objects, and reports failures.
  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:88-102 (registration)

  MCP tool registration using FastMCP @mcp.tool decorator. Defines input parameters with descriptions and types (schema), and delegates execution to the dispatched handler.
  @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/classic.py:360-393 (helper)

  EmailClient method called by handler to fetch a single email's full content via IMAP UID fetch (trying multiple formats), extract raw bytes, and parse into structured data.
  async def get_email_body_by_id(self, email_id: str, mailbox: str = "INBOX") -> dict[str, Any] | None: imap = self.imap_class(self.email_server.host, self.email_server.port) 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) 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(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
• mcp_email_server/emails/dispatcher.py:12-20 (helper)

  Dispatches to the appropriate EmailHandler implementation (currently only ClassicEmailHandler) based on account settings.
  def dispatch_handler(account_name: str) -> EmailHandler: settings = get_settings() account = settings.get_account(account_name) if isinstance(account, ProviderSettings): raise NotImplementedError if isinstance(account, EmailSettings): return ClassicEmailHandler(account) raise ValueError(f"Account {account_name} not found, available accounts: {settings.get_accounts()}")