Gmail MCP Server

Instructions

Implementation Reference

• src/mcp_gmail/gmail_client.py:681-726 (handler)

  Core implementation: Searches for unread emails matching the Gmail query (limited to max_emails), retrieves their IDs, and marks them as read by calling mark_as_read.

  async def mark_as_read_by_query(self, query: str, max_emails: int = 100) -> dict:
      """Mark emails matching a query as read.
      
      Args:
          query: Gmail search query (e.g., "from:newsletter@example.com", "older_than:7d")
          max_emails: Maximum number of emails to mark as read (safety limit)
          
      Returns:
          Dict with success count, matched count, and any errors
      """
      try:
          # First, find matching emails
          results = (
              self.service.users()
              .messages()
              .list(userId="me", q=f"{query} is:unread", maxResults=max_emails)
              .execute()
          )
          
          messages = results.get("messages", [])
          if not messages:
              return {
                  "matched": 0,
                  "success": 0,
                  "errors": [],
                  "message": "No unread emails matched the query"
              }
          
          message_ids = [msg["id"] for msg in messages]
          
          # Mark them as read
          mark_result = await self.mark_as_read(message_ids)
          mark_result["matched"] = len(messages)
          
          # Check if there might be more
          if len(messages) == max_emails:
              mark_result["message"] = f"Marked {mark_result['success']} emails as read. There may be more matching emails (limit was {max_emails})."
          else:
              mark_result["message"] = f"Marked {mark_result['success']} emails as read."
              
          return mark_result
          
      except HttpError as e:
          logger.error(f"Failed to search and mark emails: {e}")
          return {"matched": 0, "success": 0, "errors": [str(e)], "message": f"Error: {e}"}

• src/mcp_gmail/server.py:836-869 (handler)

  MCP tool dispatch handler: Parses arguments, provides preview if confirm=false by searching and listing matching emails, otherwise delegates to GmailClient.mark_as_read_by_query

  elif name == "gmail_mark_as_read_by_query":
      query = arguments.get("query", "")
      max_emails = min(arguments.get("max_emails", 100), 500)
      confirm = arguments.get("confirm", False)
      
      if not query:
          return [TextContent(type="text", text="Error: query is required.")]
      
      if not confirm:
          # Preview mode - show what would be marked
          search_results = await client.search_emails(f"{query} is:unread", max_emails)
          if not search_results:
              return [TextContent(type="text", text=f"No unread emails match the query: {query}")]
          
          lines = [f"Preview: {len(search_results)} email(s) would be marked as read:\n"]
          for email in search_results[:20]:
              lines.append(f"- {email.subject}")
              lines.append(f"  From: {email.sender.email}")
              lines.append(f"  Date: {email.date.strftime('%Y-%m-%d %H:%M')}")
              lines.append("")
          
          if len(search_results) > 20:
              lines.append(f"... and {len(search_results) - 20} more\n")
          
          lines.append("Set confirm=true to proceed.")
          return [TextContent(type="text", text="\n".join(lines))]
      else:
          result = await client.mark_as_read_by_query(query, max_emails)
          return [TextContent(
              type="text",
              text=f"Success: {result['message']}\nMatched: {result['matched']}, Marked as read: {result['success']}"
              + (f", Errors: {result['errors']}" if result['errors'] else "")
          )]

• src/mcp_gmail/server.py:364-384 (registration)

  Tool registration in GMAIL_TOOLS list: Defines the tool name, description, and JSON schema for MCP server.

      name="gmail_mark_as_read_by_query",
      description="Mark emails matching a search query as read. Use Gmail query syntax. Requires confirmation.",
      inputSchema={
          "type": "object",
          "properties": {
              "query": {
                  "type": "string",
                  "description": "Gmail search query to find emails to mark as read. Examples: 'from:newsletter@example.com', 'older_than:7d is:unread'."
              },
              "max_emails": {
                  "type": "integer",
                  "description": "Maximum number of emails to mark as read. Default 100, max 500."
              },
              "confirm": {
                  "type": "boolean",
                  "description": "Must be true to actually mark as read. Set false to preview what would be marked."
              }
          },
          "required": ["query", "confirm"]
      },
  ),

• src/mcp_gmail/gmail_client.py:637-680 (helper)

  Supporting function: Batch removes the 'UNREAD' label from the specified message IDs using Gmail API batchModify.

  async def mark_as_read(self, message_ids: list[str]) -> dict:
      """Mark one or more emails as read by removing the UNREAD label.
      
      Args:
          message_ids: List of Gmail message IDs to mark as read
          
      Returns:
          Dict with success count and any errors
      """
      if not message_ids:
          return {"success": 0, "errors": [], "message": "No message IDs provided"}
      
      results = {"success": 0, "errors": []}
      
      try:
          # Use batchModify for efficiency (up to 1000 at a time)
          if len(message_ids) <= 1000:
              self.service.users().messages().batchModify(
                  userId="me",
                  body={
                      "ids": message_ids,
                      "removeLabelIds": ["UNREAD"]
                  }
              ).execute()
              results["success"] = len(message_ids)
          else:
              # Process in batches of 1000
              for i in range(0, len(message_ids), 1000):
                  batch = message_ids[i:i+1000]
                  self.service.users().messages().batchModify(
                      userId="me",
                      body={
                          "ids": batch,
                          "removeLabelIds": ["UNREAD"]
                      }
                  ).execute()
                  results["success"] += len(batch)
                  
      except HttpError as e:
          logger.error(f"Failed to mark emails as read: {e}")
          results["errors"].append(str(e))
          
      return results