Gmail MCP Server

Instructions

Implementation Reference

• src/mcp_gmail/server.py:899-932 (handler)

  MCP tool handler for gmail_mark_as_unread_by_query. Handles input arguments, confirmation preview (searches read emails with query and shows list), and executes GmailClient.mark_as_unread_by_query on confirm.

  elif name == "gmail_mark_as_unread_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 read emails match the query: {query}")]
          
          lines = [f"Preview: {len(search_results)} email(s) would be marked as unread:\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_unread_by_query(query, max_emails)
          return [TextContent(
              type="text",
              text=f"Success: {result['message']}\nMatched: {result['matched']}, Marked as unread: {result['success']}"
              + (f", Errors: {result['errors']}" if result['errors'] else "")
          )]

• src/mcp_gmail/server.py:406-427 (schema)

  JSON Schema definition for the tool input parameters: query (required), max_emails (optional), confirm (required). Included in GMAIL_TOOLS list.

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

• src/mcp_gmail/gmail_client.py:771-816 (handler)

  GmailClient.mark_as_unread_by_query: Searches Gmail API for read emails (query -is:unread), gets message IDs, calls mark_as_unread to batch-add UNREAD label.

  async def mark_as_unread_by_query(self, query: str, max_emails: int = 100) -> dict:
      """Mark emails matching a query as unread.
      
      Args:
          query: Gmail search query (e.g., "from:newsletter@example.com", "older_than:7d")
          max_emails: Maximum number of emails to mark as unread (safety limit)
          
      Returns:
          Dict with success count, matched count, and any errors
      """
      try:
          # First, find matching emails (that are currently read)
          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 read emails matched the query"
              }
          
          message_ids = [msg["id"] for msg in messages]
          
          # Mark them as unread
          mark_result = await self.mark_as_unread(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 unread. There may be more matching emails (limit was {max_emails})."
          else:
              mark_result["message"] = f"Marked {mark_result['success']} emails as unread."
              
          return mark_result
          
      except HttpError as e:
          logger.error(f"Failed to search and mark emails as unread: {e}")
          return {"matched": 0, "success": 0, "errors": [str(e)], "message": f"Error: {e}"}

• src/mcp_gmail/gmail_client.py:727-769 (helper)

  Low-level GmailClient.mark_as_unread: Uses Gmail API messages.batchModify to add 'UNREAD' label to given message IDs (batched for large lists).

  async def mark_as_unread(self, message_ids: list[str]) -> dict:
      """Mark one or more emails as unread by adding the UNREAD label.
      
      Args:
          message_ids: List of Gmail message IDs to mark as unread
          
      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,
                      "addLabelIds": ["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,
                          "addLabelIds": ["UNREAD"]
                      }
                  ).execute()
                  results["success"] += len(batch)
                  
      except HttpError as e:
          logger.error(f"Failed to mark emails as unread: {e}")
          results["errors"].append(str(e))
          
      return results

• src/mcp_gmail/server.py:1128-1131 (registration)

  MCP Server registration: Decorator registers list_tools handler that returns GMAIL_TOOLS containing the gmail_mark_as_unread_by_query tool schema.

  @server.list_tools()
  async def list_tools() -> list[Tool]:
      return GMAIL_TOOLS