Google Workspace MCP Server - Control Gmail, Calendar, Docs, Sheets, Slides, Chat, Forms & Drive

Instructions

Implementation Reference

• gmail/gmail_tools.py:1257-1306 (handler)

  The primary handler function that implements the modify_gmail_message_labels tool. It uses the Gmail API to add or remove labels from a specific message.
  @handle_http_errors("modify_gmail_message_labels", service_type="gmail") @require_google_service("gmail", GMAIL_MODIFY_SCOPE) async def modify_gmail_message_labels( service, user_google_email: str, message_id: str, add_label_ids: List[str] = Field(default=[], description="Label IDs to add to the message."), remove_label_ids: List[str] = Field(default=[], description="Label IDs to remove from the message."), ) -> str: """ Adds or removes labels from a Gmail message. To archive an email, remove the INBOX label. To delete an email, add the TRASH label. Args: user_google_email (str): The user's Google email address. Required. message_id (str): The ID of the message to modify. add_label_ids (Optional[List[str]]): List of label IDs to add to the message. remove_label_ids (Optional[List[str]]): List of label IDs to remove from the message. Returns: str: Confirmation message of the label changes applied to the message. """ logger.info( f"[modify_gmail_message_labels] Invoked. Email: '{user_google_email}', Message ID: '{message_id}'" ) if not add_label_ids and not remove_label_ids: raise Exception( "At least one of add_label_ids or remove_label_ids must be provided." ) body = {} if add_label_ids: body["addLabelIds"] = add_label_ids if remove_label_ids: body["removeLabelIds"] = remove_label_ids await asyncio.to_thread( service.users().messages().modify(userId="me", id=message_id, body=body).execute ) actions = [] if add_label_ids: actions.append(f"Added labels: {', '.join(add_label_ids)}") if remove_label_ids: actions.append(f"Removed labels: {', '.join(remove_label_ids)}") return f"Message labels updated successfully!\nMessage ID: {message_id}\n{'; '.join(actions)}"
• gmail/gmail_tools.py:1263-1265 (schema)

  Pydantic schema definition for the tool's input parameters using Field for validation and descriptions.
  add_label_ids: List[str] = Field(default=[], description="Label IDs to add to the message."), remove_label_ids: List[str] = Field(default=[], description="Label IDs to remove from the message."), ) -> str:
• gmail/gmail_tools.py:1257-1257 (registration)

  Decorators that register the function as an MCP tool and apply necessary error handling and authentication scopes.
  @handle_http_errors("modify_gmail_message_labels", service_type="gmail")
• gmail/gmail_tools.py:1130-1175 (helper)

  Helper tool to list all Gmail labels, useful for obtaining label IDs needed for modify_gmail_message_labels.
  @handle_http_errors("list_gmail_labels", is_read_only=True, service_type="gmail") @require_google_service("gmail", "gmail_read") async def list_gmail_labels(service, user_google_email: str) -> str: """ Lists all labels in the user's Gmail account. Args: user_google_email (str): The user's Google email address. Required. Returns: str: A formatted list of all labels with their IDs, names, and types. """ logger.info(f"[list_gmail_labels] Invoked. Email: '{user_google_email}'") response = await asyncio.to_thread( service.users().labels().list(userId="me").execute ) labels = response.get("labels", []) if not labels: return "No labels found." lines = [f"Found {len(labels)} labels:", ""] system_labels = [] user_labels = [] for label in labels: if label.get("type") == "system": system_labels.append(label) else: user_labels.append(label) if system_labels: lines.append("📂 SYSTEM LABELS:") for label in system_labels: lines.append(f" • {label['name']} (ID: {label['id']})") lines.append("") if user_labels: lines.append("🏷️ USER LABELS:") for label in user_labels: lines.append(f" • {label['name']} (ID: {label['id']})") return "\n".join(lines)
• gmail/gmail_tools.py:1178-1255 (helper)

  Helper tool to create, update, or delete Gmail labels, supporting label management workflow.
  @handle_http_errors("manage_gmail_label", service_type="gmail") @require_google_service("gmail", GMAIL_LABELS_SCOPE) async def manage_gmail_label( service, user_google_email: str, action: Literal["create", "update", "delete"], name: Optional[str] = None, label_id: Optional[str] = None, label_list_visibility: Literal["labelShow", "labelHide"] = "labelShow", message_list_visibility: Literal["show", "hide"] = "show", ) -> str: """ Manages Gmail labels: create, update, or delete labels. Args: user_google_email (str): The user's Google email address. Required. action (Literal["create", "update", "delete"]): Action to perform on the label. name (Optional[str]): Label name. Required for create, optional for update. label_id (Optional[str]): Label ID. Required for update and delete operations. label_list_visibility (Literal["labelShow", "labelHide"]): Whether the label is shown in the label list. message_list_visibility (Literal["show", "hide"]): Whether the label is shown in the message list. Returns: str: Confirmation message of the label operation. """ logger.info( f"[manage_gmail_label] Invoked. Email: '{user_google_email}', Action: '{action}'" ) if action == "create" and not name: raise Exception("Label name is required for create action.") if action in ["update", "delete"] and not label_id: raise Exception("Label ID is required for update and delete actions.") if action == "create": label_object = { "name": name, "labelListVisibility": label_list_visibility, "messageListVisibility": message_list_visibility, } created_label = await asyncio.to_thread( service.users().labels().create(userId="me", body=label_object).execute ) return f"Label created successfully!\nName: {created_label['name']}\nID: {created_label['id']}" elif action == "update": current_label = await asyncio.to_thread( service.users().labels().get(userId="me", id=label_id).execute ) label_object = { "id": label_id, "name": name if name is not None else current_label["name"], "labelListVisibility": label_list_visibility, "messageListVisibility": message_list_visibility, } updated_label = await asyncio.to_thread( service.users() .labels() .update(userId="me", id=label_id, body=label_object) .execute ) return f"Label updated successfully!\nName: {updated_label['name']}\nID: {updated_label['id']}" elif action == "delete": label = await asyncio.to_thread( service.users().labels().get(userId="me", id=label_id).execute ) label_name = label["name"] await asyncio.to_thread( service.users().labels().delete(userId="me", id=label_id).execute ) return f"Label '{label_name}' (ID: {label_id}) deleted successfully!"