list_emails_metadata
Retrieve email metadata including subject, sender, recipients, and date for filtering and pagination, returning email IDs to access full content.
Instructions
List email metadata (email_id, subject, sender, recipients, date) without body content. Returns email_id for use with get_emails_content.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| account_name | Yes | The name of the email account. | |
| page | No | The page number to retrieve (starting from 1). | |
| page_size | No | The number of emails to retrieve per page. | |
| before | No | Retrieve emails before this datetime (UTC). | |
| since | No | Retrieve emails since this datetime (UTC). | |
| subject | No | Filter emails by subject. | |
| from_address | No | Filter emails by sender address. | |
| to_address | No | Filter emails by recipient address. | |
| order | No | Order emails by field. `asc` or `desc`. | desc |
| mailbox | No | The mailbox to retrieve emails from. | INBOX |
Implementation Reference
- mcp_email_server/app.py:43-85 (handler)The main MCP tool handler and registration for 'list_emails_metadata'. Dispatches to account handler's get_emails_metadata method.
@mcp.tool( description="List email metadata (email_id, subject, sender, recipients, date) without body content. Returns email_id for use with get_emails_content." ) async def list_emails_metadata( account_name: Annotated[str, Field(description="The name of the email account.")], page: Annotated[ int, Field(default=1, description="The page number to retrieve (starting from 1)."), ] = 1, page_size: Annotated[int, Field(default=10, description="The number of emails to retrieve per page.")] = 10, before: Annotated[ datetime | None, Field(default=None, description="Retrieve emails before this datetime (UTC)."), ] = None, since: Annotated[ datetime | None, Field(default=None, description="Retrieve emails since this datetime (UTC)."), ] = None, subject: Annotated[str | None, Field(default=None, description="Filter emails by subject.")] = None, from_address: Annotated[str | None, Field(default=None, description="Filter emails by sender address.")] = None, to_address: Annotated[ str | None, Field(default=None, description="Filter emails by recipient address."), ] = None, order: Annotated[ Literal["asc", "desc"], Field(default=None, description="Order emails by field. `asc` or `desc`."), ] = "desc", mailbox: Annotated[str, Field(default="INBOX", description="The mailbox to retrieve emails from.")] = "INBOX", ) -> EmailMetadataPageResponse: handler = dispatch_handler(account_name) return await handler.get_emails_metadata( page=page, page_size=page_size, before=before, since=since, subject=subject, from_address=from_address, to_address=to_address, order=order, mailbox=mailbox, ) - Pydantic models EmailMetadata and EmailMetadataPageResponse defining the input parameters and output structure for list_emails_metadata.
class EmailMetadata(BaseModel): """Email metadata""" email_id: str message_id: str | None = None # RFC 5322 Message-ID header for reply threading subject: str sender: str recipients: list[str] # Recipient list date: datetime attachments: list[str] @classmethod def from_email(cls, email: dict[str, Any]): return cls( email_id=email["email_id"], message_id=email.get("message_id"), subject=email["subject"], sender=email["from"], recipients=email.get("to", []), date=email["date"], attachments=email["attachments"], ) class EmailMetadataPageResponse(BaseModel): """Paged email metadata response""" page: int page_size: int before: datetime | None since: datetime | None subject: str | None emails: list[EmailMetadata] total: int - dispatch_handler selects and returns the appropriate EmailHandler (ClassicEmailHandler) for the given account_name.
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()}") - ClassicEmailHandler.get_emails_metadata collects metadata from the stream, computes total count, and constructs the paged response.
async def get_emails_metadata( self, page: int = 1, page_size: int = 10, before: datetime | None = None, since: datetime | None = None, subject: str | None = None, from_address: str | None = None, to_address: str | None = None, order: str = "desc", mailbox: str = "INBOX", ) -> EmailMetadataPageResponse: emails = [] async for email_data in self.incoming_client.get_emails_metadata_stream( page, page_size, before, since, subject, from_address, to_address, order, mailbox ): emails.append(EmailMetadata.from_email(email_data)) total = await self.incoming_client.get_email_count( before, since, subject, from_address=from_address, to_address=to_address, mailbox=mailbox ) return EmailMetadataPageResponse( page=page, page_size=page_size, before=before, since=since, subject=subject, emails=emails, total=total, ) - EmailClient.get_emails_metadata_stream performs the actual IMAP operations: search, paginate, fetch headers, parse and yield individual email metadata.
async def get_emails_metadata_stream( # noqa: C901 self, page: int = 1, page_size: int = 10, before: datetime | None = None, since: datetime | None = None, subject: str | None = None, from_address: str | None = None, to_address: str | None = None, order: str = "desc", mailbox: str = "INBOX", ) -> AsyncGenerator[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) search_criteria = self._build_search_criteria( before, since, subject, from_address=from_address, to_address=to_address ) logger.info(f"Get metadata: Search criteria: {search_criteria}") # Search for messages - use UID SEARCH for better compatibility _, messages = await imap.uid_search(*search_criteria) # Handle empty or None responses if not messages or not messages[0]: logger.warning("No messages returned from search") email_ids = [] else: email_ids = messages[0].split() logger.info(f"Found {len(email_ids)} email IDs") start = (page - 1) * page_size end = start + page_size if order == "desc": email_ids.reverse() # Fetch each message's metadata only for _, email_id in enumerate(email_ids[start:end]): try: # Convert email_id from bytes to string email_id_str = email_id.decode("utf-8") # Fetch only headers to get metadata without body _, data = await imap.uid("fetch", email_id_str, "BODY.PEEK[HEADER]") if not data: logger.error(f"Failed to fetch headers for UID {email_id_str}") continue # Find the email headers in the response raw_headers = None if len(data) > 1 and isinstance(data[1], bytearray): raw_headers = bytes(data[1]) else: # Search through all items for header content for item in data: if isinstance(item, bytes | bytearray) and len(item) > 10: # Skip IMAP protocol responses if isinstance(item, bytes) and b"FETCH" in item: continue # This is likely the header content raw_headers = bytes(item) if isinstance(item, bytearray) else item break if raw_headers: try: # Parse headers only parser = BytesParser(policy=default) email_message = parser.parsebytes(raw_headers) # Extract metadata subject = email_message.get("Subject", "") sender = email_message.get("From", "") date_str = email_message.get("Date", "") # Extract recipients to_addresses = [] to_header = email_message.get("To", "") if to_header: to_addresses = [addr.strip() for addr in to_header.split(",")] cc_header = email_message.get("Cc", "") if cc_header: to_addresses.extend([addr.strip() for addr in cc_header.split(",")]) # Parse date try: date_tuple = email.utils.parsedate_tz(date_str) date = ( datetime.fromtimestamp(email.utils.mktime_tz(date_tuple), tz=timezone.utc) if date_tuple else datetime.now(timezone.utc) ) except Exception: date = datetime.now(timezone.utc) # For metadata, we don't fetch attachments to save bandwidth # We'll mark it as unknown for now metadata = { "email_id": email_id_str, "subject": subject, "from": sender, "to": to_addresses, "date": date, "attachments": [], # We don't fetch attachment info for metadata } yield metadata except Exception as e: # Log error but continue with other emails logger.error(f"Error parsing email metadata: {e!s}") else: logger.error(f"Could not find header data in response for email ID: {email_id_str}") except Exception as e: logger.error(f"Error fetching email metadata {email_id}: {e!s}") finally: # Ensure we logout properly try: await imap.logout() except Exception as e: logger.info(f"Error during logout: {e}")