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