gmail_mark_as_read_by_query
Mark Gmail emails matching search queries as read. Use Gmail query syntax to target specific messages, with confirmation required for changes. Preview results before applying.
Instructions
Mark emails matching a search query as read. Use Gmail query syntax. Requires confirmation.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | Gmail search query to find emails to mark as read. Examples: 'from:newsletter@example.com', 'older_than:7d is:unread'. | |
| max_emails | No | Maximum number of emails to mark as read. Default 100, max 500. | |
| confirm | Yes | Must be true to actually mark as read. Set false to preview what would be marked. |
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_queryelif 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"] }, ),
- 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