{"id":"01KJ97484RS1VER83BD05PRMFZ","type":"skill","title":"Gmail Toolkit","slug":"camel-gmailtoolkit","short_description":"A comprehensive toolkit for Gmail operations.","description_md":"A comprehensive toolkit for Gmail operations.\n\nThis class provides methods for Gmail operations including sending emails,\nmanaging drafts, fetching messages, managing labels, and handling contacts.\nAPI keys can be accessed in google cloud console (https://console.cloud.google.com/)","ai_tags":[],"domain_tags":[],"author_user_id":"system-parser","author_name":null,"visibility":"public","pricing_type":"free","price_credits":0,"status":"published","version":"1.0.0","content_mode":"inline_text","inline_content":"### ========= Copyright 2023-2026 @ CAMEL-AI.org. All Rights Reserved. =========\n### Licensed under the Apache License, Version 2.0 (the \"License\");\n### you may not use this file except in compliance with the License.\n### You may obtain a copy of the License at\n#\n### http://www.apache.org/licenses/LICENSE-2.0\n#\n### Unless required by applicable law or agreed to in writing, software\n### distributed under the License is distributed on an \"AS IS\" BASIS,\n### WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n### See the License for the specific language governing permissions and\n### limitations under the License.\n### ========= Copyright 2023-2026 @ CAMEL-AI.org. All Rights Reserved. =========\n\nimport os\nimport re\nfrom typing import Any, Dict, List, Literal, Optional, Union\n\nfrom camel.logger import get_logger\nfrom camel.toolkits import FunctionTool\nfrom camel.toolkits.base import BaseToolkit\nfrom camel.utils import MCPServer\n\nlogger = get_logger(__name__)\n\nSCOPES = [\n 'https://www.googleapis.com/auth/gmail.readonly',\n 'https://www.googleapis.com/auth/gmail.send',\n 'https://www.googleapis.com/auth/gmail.modify',\n 'https://www.googleapis.com/auth/gmail.compose',\n 'https://www.googleapis.com/auth/gmail.labels',\n 'https://www.googleapis.com/auth/contacts.readonly',\n 'https://www.googleapis.com/auth/userinfo.profile',\n]\n\n\n@MCPServer()\nclass GmailToolkit(BaseToolkit):\n r\"\"\"A comprehensive toolkit for Gmail operations.\n\n This class provides methods for Gmail operations including sending emails,\n managing drafts, fetching messages, managing labels, and handling contacts.\n API keys can be accessed in google cloud console (https://console.cloud.google.com/)\n \"\"\"\n\n def __init__(\n self,\n timeout: Optional[float] = None,\n ):\n r\"\"\"Initializes a new instance of the GmailToolkit class.\n\n Args:\n timeout (Optional[float]): The timeout value for API requests\n in seconds. If None, no timeout is applied.\n (default: :obj:`None`)\n \"\"\"\n super().__init__(timeout=timeout)\n\n self._credentials = self._authenticate()\n\n self.gmail_service: Any = self._get_gmail_service()\n self._people_service: Any = None\n\n @property\n def people_service(self) -> Any:\n r\"\"\"Lazily initialize and return the Google People service.\"\"\"\n if self._people_service is None:\n self._people_service = self._get_people_service()\n return self._people_service\n\n @people_service.setter\n def people_service(self, service: Any) -> None:\n r\"\"\"Allow overriding/injecting the People service (e.g., in tests).\"\"\"\n self._people_service = service\n\n def gmail_send_email(\n self,\n to: Union[str, List[str]],\n subject: str,\n body: str,\n cc: Optional[Union[str, List[str]]] = None,\n bcc: Optional[Union[str, List[str]]] = None,\n attachments: Optional[List[str]] = None,\n is_html: bool = False,\n ) -> Dict[str, Any]:\n r\"\"\"Send an email through Gmail.\n\n Args:\n to (Union[str, List[str]]): Recipient email address(es).\n subject (str): Email subject.\n body (str): Email body content.\n cc (Optional[Union[str, List[str]]]): CC recipient email\n address(es).\n bcc (Optional[Union[str, List[str]]]): BCC recipient email\n address(es).\n attachments (Optional[List[str]]): List of file paths to attach.\n is_html (bool): Whether the body is HTML format. Set to True when\n sending formatted emails with HTML tags (e.g., bold,\n links, images). Use False (default) for plain text emails.\n\n Returns:\n Dict[str, Any]: A dictionary containing the result of the\n operation.\n \"\"\"\n try:\n # Normalize recipients to lists\n to_list = [to] if isinstance(to, str) else to\n cc_list = [cc] if isinstance(cc, str) else (cc or [])\n bcc_list = [bcc] if isinstance(bcc, str) else (bcc or [])\n\n # Validate email addresses\n all_recipients = to_list + cc_list + bcc_list\n for email in all_recipients:\n if not self._is_valid_email(email):\n return {\"error\": f\"Invalid email address: {email}\"}\n\n # Create message\n message = self._create_message(\n to_list, subject, body, cc_list, bcc_list, attachments, is_html\n )\n\n # Send message\n sent_message = (\n self.gmail_service.users()\n .messages()\n .send(userId='me', body=message)\n .execute()\n )\n\n return {\n \"success\": True,\n \"message_id\": sent_message.get('id'),\n \"thread_id\": sent_message.get('threadId'),\n \"message\": \"Email sent successfully\",\n }\n\n except Exception as e:\n logger.error(\"Failed to send email: %s\", e)\n return {\"error\": f\"Failed to send email: {e!s}\"}\n\n def gmail_reply_to_email(\n self,\n message_id: str,\n reply_body: str,\n reply_all: bool = False,\n is_html: bool = False,\n ) -> Dict[str, Any]:\n r\"\"\"Reply to an email message.\n\n Args:\n message_id (str): The unique identifier of the message to reply to.\n To get a message ID, first use fetch_emails() to list messages,\n or use the 'message_id' returned from send_email() or\n create_email_draft().\n reply_body (str): The reply message body.\n reply_all (bool): Whether to reply to all recipients.\n is_html (bool): Whether the body is HTML format. Set to True when\n sending formatted emails with HTML tags (e.g., bold,\n links, images). Use False (default) for plain text emails.\n\n Returns:\n Dict[str, Any]: A dictionary containing the result of the\n operation.\n \"\"\"\n try:\n # Get the original message\n original_message = (\n self.gmail_service.users()\n .messages()\n .get(userId='me', id=message_id)\n .execute()\n )\n\n # Extract headers (single pass, case-insensitive)\n headers = original_message['payload'].get('headers', [])\n subject = from_email = to_emails = cc_emails = None\n missing = {'subject', 'from', 'to', 'cc'}\n\n for header in headers:\n name = (header.get('name') or '').lower()\n if name not in missing:\n continue\n value = header.get('value')\n if name == 'subject':\n subject = value\n elif name == 'from':\n from_email = value\n elif name == 'to':\n to_emails = value\n elif name == 'cc':\n cc_emails = value\n missing.discard(name)\n if not missing:\n break\n\n # Extract identifiers for reply context\n message_id_header = self._get_header_value(\n headers, 'Message-Id'\n ) or self._get_header_value(headers, 'Message-ID')\n thread_id = original_message.get('threadId')\n\n # Prepare reply subject\n if subject and not subject.startswith('Re: '):\n subject = f\"Re: {subject}\"\n elif not subject:\n subject = \"Re: (No Subject)\"\n\n # Validate from_email\n if not from_email:\n return {\"error\": \"Original message has no sender address\"}\n\n # Prepare recipients\n if reply_all:\n recipients = [from_email]\n if to_emails:\n recipients.extend(\n [email.strip() for email in to_emails.split(',')]\n )\n if cc_emails:\n recipients.extend(\n [email.strip() for email in cc_emails.split(',')]\n )\n # Remove duplicates and None values\n recipients = [r for r in list(set(recipients)) if r]\n\n # Get current user's email and remove it from recipients\n try:\n profile_result = self.gmail_get_profile()\n if profile_result.get('success'):\n current_user_email = profile_result['profile'][\n 'email_address'\n ]\n # Remove current user from recipients (handle both\n # plain email and \"Name \" format)\n filtered_recipients = []\n for email in recipients:\n # Extract email from \"Name \" format\n match = re.search(r'<([^>]+)>$', email.strip())\n email_addr = (\n match.group(1) if match else email.strip()\n )\n if email_addr != current_user_email:\n filtered_recipients.append(email)\n recipients = filtered_recipients\n except Exception as e:\n logger.warning(\n \"Could not get current user email to filter from \"\n \"recipients: %s\",\n e,\n )\n else:\n recipients = [from_email]\n\n # Create reply message with reply headers\n message = self._create_message(\n recipients,\n subject,\n reply_body,\n is_html=is_html,\n in_reply_to=message_id_header or original_message.get('id'),\n references=[message_id_header] if message_id_header else None,\n )\n\n # Send reply in the same thread\n sent_message = (\n self.gmail_service.users()\n .messages()\n .send(userId='me', body={**message, 'threadId': thread_id})\n .execute()\n )\n\n return {\n \"success\": True,\n \"message_id\": sent_message.get('id'),\n \"thread_id\": sent_message.get('threadId'),\n \"message\": \"Reply sent successfully\",\n }\n\n except Exception as e:\n logger.error(\"Failed to reply to email: %s\", e)\n return {\"error\": f\"Failed to reply to email: {e!s}\"}\n\n def gmail_forward_email(\n self,\n message_id: str,\n to: Union[str, List[str]],\n forward_body: Optional[str] = None,\n cc: Optional[Union[str, List[str]]] = None,\n bcc: Optional[Union[str, List[str]]] = None,\n include_attachments: bool = True,\n ) -> Dict[str, Any]:\n r\"\"\"Forward an email message.\n\n Args:\n message_id (str): The unique identifier of the message to forward.\n To get a message ID, first use fetch_emails() to list messages,\n or use the 'message_id' returned from send_email() or\n create_email_draft().\n to (Union[str, List[str]]): Recipient email address(es).\n forward_body (Optional[str]): Additional message to include at\n the top of the forwarded email, before the original message\n content. If not provided, only the original message will be\n forwarded.\n cc (Optional[Union[str, List[str]]]): CC recipient email\n address(es).\n bcc (Optional[Union[str, List[str]]]): BCC recipient email\n address(es).\n include_attachments (bool): Whether to include original\n attachments. Defaults to True. Only includes real\n attachments, not inline images.\n\n Returns:\n Dict[str, Any]: A dictionary containing the result of the\n operation, including the number of attachments forwarded.\n \"\"\"\n try:\n import tempfile\n\n # Get the original message\n original_message = (\n self.gmail_service.users()\n .messages()\n .get(userId='me', id=message_id)\n .execute()\n )\n\n # Extract headers (single pass, case-insensitive)\n headers = original_message['payload'].get('headers', [])\n subject = from_email = date = None\n missing = {'subject', 'from', 'date'}\n\n for header in headers:\n name = (header.get('name') or '').lower()\n if name not in missing:\n continue\n value = header.get('value')\n if name == 'subject':\n subject = value\n elif name == 'from':\n from_email = value\n elif name == 'date':\n date = value\n missing.discard(name)\n if not missing:\n break\n\n # Prepare forward subject\n if subject and not subject.startswith('Fwd: '):\n subject = f\"Fwd: {subject}\"\n elif not subject:\n subject = \"Fwd: (No Subject)\"\n\n # Prepare forward body\n if forward_body:\n body = f\"{forward_body}\\n\\n--- Forwarded message ---\\n\"\n else:\n body = \"--- Forwarded message ---\\n\"\n\n body += f\"From: {from_email}\\n\"\n body += f\"Date: {date}\\n\"\n body += f\"Subject: {subject.replace('Fwd: ', '')}\\n\\n\"\n\n # Add original message body\n body += self._extract_message_body(original_message)\n\n # Normalize recipients\n to_list = [to] if isinstance(to, str) else to\n cc_list = [cc] if isinstance(cc, str) else (cc or [])\n bcc_list = [bcc] if isinstance(bcc, str) else (bcc or [])\n\n # Handle attachments\n attachment_paths = []\n temp_files: List[str] = []\n\n try:\n if include_attachments:\n # Extract attachment metadata\n attachments = self._extract_attachments(original_message)\n for att in attachments:\n try:\n # Create temp file\n temp_file = tempfile.NamedTemporaryFile(\n delete=False, suffix=f\"_{att['filename']}\"\n )\n temp_files.append(temp_file.name)\n\n # Download attachment\n result = self.gmail_get_attachment(\n message_id=message_id,\n attachment_id=att['attachment_id'],\n save_path=temp_file.name,\n )\n\n if result.get('success'):\n attachment_paths.append(temp_file.name)\n\n except Exception as e:\n logger.warning(\n f\"Failed to download attachment \"\n f\"{att['filename']}: {e}\"\n )\n\n # Create forward message (now with attachments!)\n message = self._create_message(\n to_list,\n subject,\n body,\n cc_list,\n bcc_list,\n attachments=attachment_paths if attachment_paths else None,\n )\n\n # Send forward\n sent_message = (\n self.gmail_service.users()\n .messages()\n .send(userId='me', body=message)\n .execute()\n )\n\n return {\n \"success\": True,\n \"message_id\": sent_message.get('id'),\n \"thread_id\": sent_message.get('threadId'),\n \"message\": \"Email forwarded successfully\",\n \"attachments_forwarded\": len(attachment_paths),\n }\n\n finally:\n # Clean up temp files\n for temp_file_path in temp_files:\n try:\n os.unlink(temp_file_path)\n except Exception as e:\n logger.warning(\n f\"Failed to delete temp file {temp_file_path}: {e}\"\n )\n\n except Exception as e:\n logger.error(\"Failed to forward email: %s\", e)\n return {\"error\": f\"Failed to forward email: {e!s}\"}\n\n def gmail_create_draft(\n self,\n to: Union[str, List[str]],\n subject: str,\n body: str,\n cc: Optional[Union[str, List[str]]] = None,\n bcc: Optional[Union[str, List[str]]] = None,\n attachments: Optional[List[str]] = None,\n is_html: bool = False,\n ) -> Dict[str, Any]:\n r\"\"\"Create an email draft.\n\n Args:\n to (Union[str, List[str]]): Recipient email address(es).\n subject (str): Email subject.\n body (str): Email body content.\n cc (Optional[Union[str, List[str]]]): CC recipient email\n address(es).\n bcc (Optional[Union[str, List[str]]]): BCC recipient email\n address(es).\n attachments (Optional[List[str]]): List of file paths to attach.\n is_html (bool): Whether the body is HTML format. Set to True when\n sending formatted emails with HTML tags (e.g., bold,\n links, images). Use False (default) for plain text emails.\n\n Returns:\n Dict[str, Any]: A dictionary containing the result of the\n operation.\n \"\"\"\n try:\n # Normalize recipients to lists\n to_list = [to] if isinstance(to, str) else to\n cc_list = [cc] if isinstance(cc, str) else (cc or [])\n bcc_list = [bcc] if isinstance(bcc, str) else (bcc or [])\n\n # Validate email addresses\n all_recipients = to_list + cc_list + bcc_list\n for email in all_recipients:\n if not self._is_valid_email(email):\n return {\"error\": f\"Invalid email address: {email}\"}\n\n # Create message\n message = self._create_message(\n to_list, subject, body, cc_list, bcc_list, attachments, is_html\n )\n\n # Create draft\n draft = (\n self.gmail_service.users()\n .drafts()\n .create(userId='me', body={'message': message})\n .execute()\n )\n\n return {\n \"success\": True,\n \"draft_id\": draft.get('id'),\n \"message_id\": draft.get('message', {}).get('id'),\n \"message\": \"Draft created successfully\",\n }\n\n except Exception as e:\n logger.error(\"Failed to create draft: %s\", e)\n return {\"error\": f\"Failed to create draft: {e!s}\"}\n\n def gmail_send_draft(self, draft_id: str) -> Dict[str, Any]:\n r\"\"\"Send a draft email.\n\n Args:\n draft_id (str): The unique identifier of the draft to send.\n To get a draft ID, first use list_drafts() to list drafts,\n or use the 'draft_id' returned from create_email_draft().\n\n Returns:\n Dict[str, Any]: A dictionary containing the result of the\n operation.\n \"\"\"\n try:\n # Send draft\n sent_message = (\n self.gmail_service.users()\n .drafts()\n .send(userId='me', body={'id': draft_id})\n .execute()\n )\n\n return {\n \"success\": True,\n \"message_id\": sent_message.get('id'),\n \"thread_id\": sent_message.get('threadId'),\n \"message\": \"Draft sent successfully\",\n }\n\n except Exception as e:\n logger.error(\"Failed to send draft: %s\", e)\n return {\"error\": f\"Failed to send draft: {e!s}\"}\n\n def gmail_fetch_emails(\n self,\n query: str = \"\",\n max_results: int = 10,\n include_spam_trash: bool = False,\n label_ids: Optional[List[str]] = None,\n page_token: Optional[str] = None,\n ) -> Dict[str, Any]:\n r\"\"\"Fetch emails with filters and pagination.\n\n Args:\n query (str): Gmail search query string. Use Gmail's search syntax:\n - 'from:example@domain.com' - emails from specific sender\n - 'subject:meeting' - emails with specific subject text\n - 'has:attachment' - emails with attachments\n - 'is:unread' - unread emails\n - 'in:sent' - emails in sent folder\n - 'after:2024/01/01 before:2024/12/31' - date range\n Examples: 'from:john@example.com subject:project', 'is:unread\n has:attachment'\n max_results (int): Maximum number of emails to fetch.\n include_spam_trash (bool): Whether to include spam and trash.\n label_ids (Optional[List[str]]): List of label IDs to filter\n emails by. Only emails with ALL of the specified\n labels will be returned.\n Label IDs can be:\n - System labels: 'INBOX', 'SENT', 'DRAFT', 'SPAM', 'TRASH',\n 'UNREAD', 'STARRED', 'IMPORTANT', 'CATEGORY_PERSONAL', etc.\n - Custom label IDs: Retrieved from list_gmail_labels() method.\n page_token (Optional[str]): Pagination token from a previous\n response. If provided, fetches the next page of results.\n\n Returns:\n Dict[str, Any]: A dictionary containing the fetched emails.\n \"\"\"\n try:\n # Build request parameters\n request_params = {\n 'userId': 'me',\n 'maxResults': max_results,\n 'includeSpamTrash': include_spam_trash,\n }\n\n if query:\n request_params['q'] = query\n if label_ids:\n request_params['labelIds'] = label_ids\n\n # List messages\n if page_token:\n request_params['pageToken'] = page_token\n\n messages_result = (\n self.gmail_service.users()\n .messages()\n .list(**request_params)\n .execute()\n )\n\n messages = messages_result.get('messages', [])\n emails = []\n\n # Fetch detailed information for each message\n for msg in messages:\n email_detail = self._get_message_details(msg['id'])\n if email_detail:\n emails.append(email_detail)\n\n return {\n \"success\": True,\n \"emails\": emails,\n \"total_count\": len(emails),\n \"next_page_token\": messages_result.get('nextPageToken'),\n }\n\n except Exception as e:\n logger.error(\"Failed to fetch emails: %s\", e)\n return {\"error\": f\"Failed to fetch emails: {e!s}\"}\n\n def gmail_fetch_thread_by_id(self, thread_id: str) -> Dict[str, Any]:\n r\"\"\"Fetch a thread by ID.\n\n Args:\n thread_id (str): The unique identifier of the thread to fetch.\n To get a thread ID, first use list_threads() to list threads,\n or use the 'thread_id' returned from send_email() or\n reply_to_email().\n\n Returns:\n Dict[str, Any]: A dictionary containing the thread details.\n \"\"\"\n try:\n thread = (\n self.gmail_service.users()\n .threads()\n .get(userId='me', id=thread_id)\n .execute()\n )\n\n messages = []\n for message in thread.get('messages', []):\n message_detail = self._get_message_details(message['id'])\n if message_detail:\n messages.append(message_detail)\n\n return {\n \"success\": True,\n \"thread_id\": thread_id,\n \"messages\": messages,\n \"message_count\": len(messages),\n }\n\n except Exception as e:\n logger.error(\"Failed to fetch thread: %s\", e)\n return {\"error\": f\"Failed to fetch thread: {e!s}\"}\n\n def gmail_modify_email_labels(\n self,\n message_id: str,\n add_labels: Optional[List[str]] = None,\n remove_labels: Optional[List[str]] = None,\n ) -> Dict[str, Any]:\n r\"\"\"Modify labels on an email message.\n\n Args:\n message_id (str): The unique identifier of the message to modify.\n To get a message ID, first use fetch_emails() to list messages,\n or use the 'message_id' returned from send_email() or\n create_email_draft().\n add_labels (Optional[List[str]]): List of label IDs to add to\n the message.\n Label IDs can be:\n - System labels: 'INBOX', 'STARRED', 'IMPORTANT',\n 'UNREAD', etc.\n - Custom label IDs: Retrieved from list_gmail_labels() method.\n Example: ['STARRED', 'IMPORTANT'] marks email as starred\n and important.\n remove_labels (Optional[List[str]]): List of label IDs to\n remove from the message. Uses the same format as add_labels.\n Example: ['UNREAD'] marks the email as read.\n\n Returns:\n Dict[str, Any]: A dictionary containing the result of the\n operation.\n \"\"\"\n try:\n body = {}\n if add_labels:\n body['addLabelIds'] = add_labels\n if remove_labels:\n body['removeLabelIds'] = remove_labels\n\n if not body:\n return {\"error\": \"No labels to add or remove\"}\n\n modified_message = (\n self.gmail_service.users()\n .messages()\n .modify(userId='me', id=message_id, body=body)\n .execute()\n )\n\n return {\n \"success\": True,\n \"message_id\": message_id,\n \"label_ids\": modified_message.get('labelIds', []),\n \"message\": \"Labels modified successfully\",\n }\n\n except Exception as e:\n logger.error(\"Failed to modify labels: %s\", e)\n return {\"error\": f\"Failed to modify labels: {e!s}\"}\n\n def gmail_move_to_trash(self, message_id: str) -> Dict[str, Any]:\n r\"\"\"Move a message to trash.\n\n Args:\n message_id (str): The unique identifier of the message to move to\n trash. To get a message ID, first use fetch_emails() to list\n messages, or use the 'message_id' returned from send_email()\n or create_email_draft().\n\n Returns:\n Dict[str, Any]: A dictionary containing the result of the\n operation.\n \"\"\"\n try:\n trashed_message = (\n self.gmail_service.users()\n .messages()\n .trash(userId='me', id=message_id)\n .execute()\n )\n\n return {\n \"success\": True,\n \"message_id\": message_id,\n \"label_ids\": trashed_message.get('labelIds', []),\n \"message\": \"Message moved to trash successfully\",\n }\n\n except Exception as e:\n logger.error(\"Failed to move message to trash: %s\", e)\n return {\"error\": f\"Failed to move message to trash: {e!s}\"}\n\n def gmail_get_attachment(\n self,\n message_id: str,\n attachment_id: str,\n save_path: Optional[str] = None,\n ) -> Dict[str, Any]:\n r\"\"\"Get an attachment from a message.\n\n Args:\n message_id (str): The unique identifier of the message containing\n the attachment. To get a message ID, first use fetch_emails()\n to list messages, or use the 'message_id' returned from\n send_email() or create_email_draft().\n attachment_id (str): The unique identifier of the attachment to\n download. To get an attachment ID, first use fetch_emails() to\n get message details, then look for 'attachment_id' in the\n 'attachments' list of each message.\n save_path (Optional[str]): Local file path where the attachment\n should be saved. If provided, the attachment will be saved to\n this location and the response will include a success message.\n If not provided, the attachment data will be returned as\n base64-encoded content in the response.\n\n Returns:\n Dict[str, Any]: A dictionary containing the attachment data or\n save result.\n \"\"\"\n try:\n import base64\n\n attachment = (\n self.gmail_service.users()\n .messages()\n .attachments()\n .get(userId='me', messageId=message_id, id=attachment_id)\n .execute()\n )\n\n # Decode the attachment data\n file_data = base64.urlsafe_b64decode(attachment['data'])\n\n if save_path:\n with open(save_path, 'wb') as f:\n f.write(file_data)\n return {\n \"success\": True,\n \"message\": f\"Attachment saved to {save_path}\",\n \"file_size\": len(file_data),\n }\n else:\n return {\n \"success\": True,\n \"data\": base64.b64encode(file_data).decode('utf-8'),\n \"file_size\": len(file_data),\n }\n\n except Exception as e:\n logger.error(\"Failed to get attachment: %s\", e)\n return {\"error\": f\"Failed to get attachment: {e!s}\"}\n\n def gmail_list_threads(\n self,\n query: str = \"\",\n max_results: int = 10,\n include_spam_trash: bool = False,\n label_ids: Optional[List[str]] = None,\n page_token: Optional[str] = None,\n ) -> Dict[str, Any]:\n r\"\"\"List email threads.\n\n Args:\n query (str): Gmail search query string. Use Gmail's search syntax:\n - 'from:example@domain.com' - threads from specific sender\n - 'subject:meeting' - threads with specific subject text\n - 'has:attachment' - threads with attachments\n - 'is:unread' - unread threads\n - 'in:sent' - threads in sent folder\n - 'after:2024/01/01 before:2024/12/31' - date range\n Examples: 'from:john@example.com subject:project', 'is:unread\n has:attachment'\n max_results (int): Maximum number of threads to fetch.\n include_spam_trash (bool): Whether to include spam and trash.\n label_ids (Optional[List[str]]): List of label IDs to filter\n threads by. Only threads with ALL of the specified labels\n will be returned.\n Label IDs can be:\n - System labels: 'INBOX', 'SENT', 'DRAFT', 'SPAM', 'TRASH',\n 'UNREAD', 'STARRED', 'IMPORTANT', 'CATEGORY_PERSONAL', etc.\n - Custom label IDs: Retrieved from list_gmail_labels() method.\n page_token (Optional[str]): Pagination token from a previous\n response. If provided, fetches the next page of results.\n\n Returns:\n Dict[str, Any]: A dictionary containing the thread list.\n \"\"\"\n try:\n # Build request parameters\n request_params = {\n 'userId': 'me',\n 'maxResults': max_results,\n 'includeSpamTrash': include_spam_trash,\n }\n\n if query:\n request_params['q'] = query\n if label_ids:\n request_params['labelIds'] = label_ids\n\n # List threads\n if page_token:\n request_params['pageToken'] = page_token\n\n threads_result = (\n self.gmail_service.users()\n .threads()\n .list(**request_params)\n .execute()\n )\n\n threads = threads_result.get('threads', [])\n thread_list = []\n\n for thread in threads:\n thread_list.append(\n {\n \"thread_id\": thread['id'],\n \"snippet\": thread.get('snippet', ''),\n \"history_id\": thread.get('historyId', ''),\n }\n )\n\n return {\n \"success\": True,\n \"threads\": thread_list,\n \"total_count\": len(thread_list),\n \"next_page_token\": threads_result.get('nextPageToken'),\n }\n\n except Exception as e:\n logger.error(\"Failed to list threads: %s\", e)\n return {\"error\": f\"Failed to list threads: {e!s}\"}\n\n def gmail_list_drafts(\n self, max_results: int = 10, page_token: Optional[str] = None\n ) -> Dict[str, Any]:\n r\"\"\"List email drafts.\n\n Args:\n max_results (int): Maximum number of drafts to fetch.\n page_token (Optional[str]): Pagination token from a previous\n response. If provided, fetches the next page of results.\n\n Returns:\n Dict[str, Any]: A dictionary containing the draft list.\n \"\"\"\n try:\n drafts_result = (\n self.gmail_service.users()\n .drafts()\n .list(\n userId='me',\n maxResults=max_results,\n **({\"pageToken\": page_token} if page_token else {}),\n )\n .execute()\n )\n\n drafts = drafts_result.get('drafts', [])\n draft_list = []\n\n for draft in drafts:\n draft_info = {\n \"draft_id\": draft['id'],\n \"message_id\": draft.get('message', {}).get('id', ''),\n \"thread_id\": draft.get('message', {}).get('threadId', ''),\n \"snippet\": draft.get('message', {}).get('snippet', ''),\n }\n draft_list.append(draft_info)\n\n return {\n \"success\": True,\n \"drafts\": draft_list,\n \"total_count\": len(draft_list),\n \"next_page_token\": drafts_result.get('nextPageToken'),\n }\n\n except Exception as e:\n logger.error(\"Failed to list drafts: %s\", e)\n return {\"error\": f\"Failed to list drafts: {e!s}\"}\n\n def gmail_list_labels(self) -> Dict[str, Any]:\n r\"\"\"List all Gmail labels.\n\n Returns:\n Dict[str, Any]: A dictionary containing the label list.\n \"\"\"\n try:\n labels_result = (\n self.gmail_service.users().labels().list(userId='me').execute()\n )\n\n labels = labels_result.get('labels', [])\n label_list = []\n\n for label in labels:\n label_info = {\n \"id\": label['id'],\n \"name\": label['name'],\n \"type\": label.get('type', 'user'),\n \"messages_total\": label.get('messagesTotal', 0),\n \"messages_unread\": label.get('messagesUnread', 0),\n \"threads_total\": label.get('threadsTotal', 0),\n \"threads_unread\": label.get('threadsUnread', 0),\n }\n label_list.append(label_info)\n\n return {\n \"success\": True,\n \"labels\": label_list,\n \"total_count\": len(label_list),\n }\n\n except Exception as e:\n logger.error(\"Failed to list labels: %s\", e)\n return {\"error\": f\"Failed to list labels: {e!s}\"}\n\n def gmail_create_label(\n self,\n name: str,\n label_list_visibility: Literal[\"labelShow\", \"labelHide\"] = \"labelShow\",\n message_list_visibility: Literal[\"show\", \"hide\"] = \"show\",\n ) -> Dict[str, Any]:\n r\"\"\"Create a new Gmail label.\n\n Args:\n name (str): The name of the label to create.\n label_list_visibility (str): How the label appears in Gmail's\n label list. - 'labelShow': Label is visible in the label list\n sidebar (default) - 'labelHide': Label is hidden from the\n label list sidebar\n message_list_visibility (str): How the label appears in message\n lists. - 'show': Label is visible on messages in inbox/lists\n (default) - 'hide': Label is hidden from message displays\n\n Returns:\n Dict[str, Any]: A dictionary containing the result of the\n operation.\n \"\"\"\n try:\n label_object = {\n 'name': name,\n 'labelListVisibility': label_list_visibility,\n 'messageListVisibility': message_list_visibility,\n }\n\n created_label = (\n self.gmail_service.users()\n .labels()\n .create(userId='me', body=label_object)\n .execute()\n )\n\n return {\n \"success\": True,\n \"label_id\": created_label['id'],\n \"label_name\": created_label['name'],\n \"message\": \"Label created successfully\",\n }\n\n except Exception as e:\n logger.error(\"Failed to create label: %s\", e)\n return {\"error\": f\"Failed to create label: {e!s}\"}\n\n def gmail_delete_label(self, label_id: str) -> Dict[str, Any]:\n r\"\"\"Delete a Gmail label.\n\n Args:\n label_id (str): The unique identifier of the user-created label to\n delete. To get a label ID, first use list_gmail_labels() to\n list all labels. Note: System labels (e.g., 'INBOX', 'SENT',\n 'DRAFT', 'SPAM', 'TRASH', 'UNREAD', 'STARRED', 'IMPORTANT',\n 'CATEGORY_PERSONAL', etc.) cannot be deleted.\n\n Returns:\n Dict[str, Any]: A dictionary containing the result of the\n operation.\n \"\"\"\n try:\n self.gmail_service.users().labels().delete(\n userId='me', id=label_id\n ).execute()\n\n return {\n \"success\": True,\n \"label_id\": label_id,\n \"message\": \"Label deleted successfully\",\n }\n\n except Exception as e:\n logger.error(\"Failed to delete label: %s\", e)\n return {\"error\": f\"Failed to delete label: {e!s}\"}\n\n def gmail_modify_thread_labels(\n self,\n thread_id: str,\n add_labels: Optional[List[str]] = None,\n remove_labels: Optional[List[str]] = None,\n ) -> Dict[str, Any]:\n r\"\"\"Modify labels on a thread.\n\n Args:\n thread_id (str): The unique identifier of the thread to modify.\n To get a thread ID, first use list_threads() to list threads,\n or use the 'thread_id' returned from send_email() or\n reply_to_email().\n add_labels (Optional[List[str]]): List of label IDs to add to all\n messages in the thread.\n Label IDs can be:\n - System labels: 'INBOX', 'STARRED', 'IMPORTANT',\n 'UNREAD', etc.\n - Custom label IDs: Retrieved from list_gmail_labels().\n Example: ['STARRED', 'IMPORTANT'] marks thread as\n starred and important.\n remove_labels (Optional[List[str]]): List of label IDs to\n remove from all messages in the thread. Uses the same\n format as add_labels.\n Example: ['UNREAD'] marks the entire thread as read.\n\n Returns:\n Dict[str, Any]: A dictionary containing the result of the\n operation.\n \"\"\"\n try:\n body = {}\n if add_labels:\n body['addLabelIds'] = add_labels\n if remove_labels:\n body['removeLabelIds'] = remove_labels\n\n if not body:\n return {\"error\": \"No labels to add or remove\"}\n\n modified_thread = (\n self.gmail_service.users()\n .threads()\n .modify(userId='me', id=thread_id, body=body)\n .execute()\n )\n\n return {\n \"success\": True,\n \"thread_id\": thread_id,\n \"label_ids\": modified_thread.get('labelIds', []),\n \"message\": \"Thread labels modified successfully\",\n }\n\n except Exception as e:\n logger.error(\"Failed to modify thread labels: %s\", e)\n return {\"error\": f\"Failed to modify thread labels: {e!s}\"}\n\n def gmail_get_profile(self) -> Dict[str, Any]:\n r\"\"\"Get Gmail profile information.\n\n Returns:\n Dict[str, Any]: A dictionary containing the profile information.\n \"\"\"\n try:\n profile = (\n self.gmail_service.users().getProfile(userId='me').execute()\n )\n\n return {\n \"success\": True,\n \"profile\": {\n \"email_address\": profile.get('emailAddress', ''),\n \"messages_total\": profile.get('messagesTotal', 0),\n \"threads_total\": profile.get('threadsTotal', 0),\n \"history_id\": profile.get('historyId', ''),\n },\n }\n\n except Exception as e:\n logger.error(\"Failed to get profile: %s\", e)\n return {\"error\": f\"Failed to get profile: {e!s}\"}\n\n def gmail_get_contacts(\n self,\n max_results: int = 100,\n page_token: Optional[str] = None,\n ) -> Dict[str, Any]:\n r\"\"\"List connections from Google People API.\n\n Args:\n max_results (int): Maximum number of contacts to fetch.\n page_token (Optional[str]): Pagination token from a previous\n response. If provided, fetches the next page of results.\n\n Returns:\n Dict[str, Any]: A dictionary containing the contacts.\n \"\"\"\n try:\n # Build request parameters\n request_params = {\n 'resourceName': 'people/me',\n 'personFields': (\n 'names,emailAddresses,phoneNumbers,organizations'\n ),\n 'pageSize': max_results,\n }\n\n # Search contacts\n if page_token:\n request_params['pageToken'] = page_token\n\n contacts_result = (\n self.people_service.people()\n .connections()\n .list(**request_params)\n .execute()\n )\n\n connections = contacts_result.get('connections', [])\n contact_list = []\n\n for person in connections:\n contact_info = {\n \"resource_name\": person.get('resourceName', ''),\n \"names\": person.get('names', []),\n \"email_addresses\": person.get('emailAddresses', []),\n \"phone_numbers\": person.get('phoneNumbers', []),\n \"organizations\": person.get('organizations', []),\n }\n contact_list.append(contact_info)\n\n return {\n \"success\": True,\n \"contacts\": contact_list,\n \"total_count\": len(contact_list),\n \"next_page_token\": contacts_result.get('nextPageToken'),\n }\n\n except Exception as e:\n logger.error(\"Failed to get contacts: %s\", e)\n return {\"error\": f\"Failed to get contacts: {e!s}\"}\n\n def gmail_search_people(\n self,\n query: str,\n max_results: int = 10,\n ) -> Dict[str, Any]:\n r\"\"\"Search for people in contacts.\n\n Args:\n query (str): Search query for people in contacts. Can search by:\n - Name: 'John Smith' or partial names like 'John'\n - Email: 'john@example.com'\n - Organization: 'Google' or 'Acme Corp'\n - Phone number: '+1234567890'\n Examples: 'John Smith', 'john@example.com', 'Google'\n max_results (int): Maximum number of results to fetch.\n\n Returns:\n Dict[str, Any]: A dictionary containing the search results.\n \"\"\"\n try:\n # Search people\n search_result = (\n self.people_service.people()\n .searchContacts(\n query=query,\n readMask='names,emailAddresses,phoneNumbers,organizations',\n pageSize=max_results,\n )\n .execute()\n )\n\n results = search_result.get('results', [])\n people_list = []\n\n for result in results:\n person = result.get('person', {})\n person_info = {\n \"resource_name\": person.get('resourceName', ''),\n \"names\": person.get('names', []),\n \"email_addresses\": person.get('emailAddresses', []),\n \"phone_numbers\": person.get('phoneNumbers', []),\n \"organizations\": person.get('organizations', []),\n }\n people_list.append(person_info)\n\n return {\n \"success\": True,\n \"people\": people_list,\n \"total_count\": len(people_list),\n }\n\n except Exception as e:\n logger.error(\"Failed to search people: %s\", e)\n return {\"error\": f\"Failed to search people: {e!s}\"}\n\n # Helper methods\n def _get_gmail_service(self):\n r\"\"\"Get Gmail service object.\"\"\"\n from googleapiclient.discovery import build\n\n try:\n # Build service with optional timeout\n if self.timeout is not None:\n import httplib2\n\n http = httplib2.Http(timeout=self.timeout)\n http = self._credentials.authorize(http)\n service = build('gmail', 'v1', http=http)\n else:\n service = build('gmail', 'v1', credentials=self._credentials)\n return service\n except Exception as e:\n raise ValueError(f\"Failed to build Gmail service: {e}\") from e\n\n def _get_people_service(self):\n r\"\"\"Get People service object.\"\"\"\n from googleapiclient.discovery import build\n\n try:\n # Build service with optional timeout\n if self.timeout is not None:\n import httplib2\n\n http = httplib2.Http(timeout=self.timeout)\n http = self._credentials.authorize(http)\n service = build('people', 'v1', http=http)\n else:\n service = build('people', 'v1', credentials=self._credentials)\n return service\n except Exception as e:\n raise ValueError(f\"Failed to build People service: {e}\") from e\n\n def _authenticate(self):\n r\"\"\"Authenticate with Google APIs using OAuth2.\n\n Automatically saves and loads credentials from\n ~/.camel/gmail_token.json to avoid repeated\n browser logins.\n \"\"\"\n import json\n from pathlib import Path\n\n from dotenv import load_dotenv\n from google.auth.transport.requests import Request\n from google.oauth2.credentials import Credentials\n from google_auth_oauthlib.flow import InstalledAppFlow\n\n # Look for .env file in the project root (camel/)\n env_file = Path(__file__).parent.parent.parent / '.env'\n load_dotenv(env_file)\n\n client_id = os.environ.get('GOOGLE_CLIENT_ID')\n client_secret = os.environ.get('GOOGLE_CLIENT_SECRET')\n\n if not client_id or not client_secret:\n missing_vars = []\n if not client_id:\n missing_vars.append('GOOGLE_CLIENT_ID')\n if not client_secret:\n missing_vars.append('GOOGLE_CLIENT_SECRET')\n raise ValueError(\n f\"Missing required environment variables: \"\n f\"{', '.join(missing_vars)}. \"\n \"Please set these in your .env file or environment variables.\"\n )\n\n token_file = Path.home() / '.camel' / 'gmail_token.json'\n creds = None\n\n # COMPONENT 1: Load saved credentials\n if token_file.exists():\n try:\n with open(token_file, 'r') as f:\n data = json.load(f)\n creds = Credentials(\n token=data.get('token'),\n refresh_token=data.get('refresh_token'),\n token_uri=data.get(\n 'token_uri', 'https://oauth2.googleapis.com/token'\n ),\n client_id=client_id,\n client_secret=client_secret,\n scopes=SCOPES,\n )\n except Exception as e:\n logger.warning(f\"Failed to load saved token: {e}\")\n creds = None\n\n # COMPONENT 2: Refresh if expired\n if creds and creds.expired and creds.refresh_token:\n try:\n creds.refresh(Request())\n logger.info(\"Access token refreshed\")\n\n # Save refreshed credentials to disk\n token_file.parent.mkdir(parents=True, exist_ok=True)\n try:\n with open(token_file, 'w') as f:\n json.dump(\n {\n 'token': creds.token,\n 'refresh_token': creds.refresh_token,\n 'token_uri': creds.token_uri,\n 'scopes': creds.scopes,\n },\n f,\n )\n os.chmod(token_file, 0o600)\n logger.info(f\"Refreshed credentials saved to {token_file}\")\n except Exception as e:\n logger.warning(\n f\"Failed to save refreshed credentials to \"\n f\"{token_file}: {e}. \"\n \"Token refreshed but not persisted.\"\n )\n\n return creds\n except Exception as e:\n logger.warning(f\"Token refresh failed: {e}\")\n creds = None\n\n # COMPONENT 3: Return if valid\n if creds and creds.valid:\n return creds\n\n # COMPONENT 4: Browser OAuth (first-time or invalid credentials)\n client_config = {\n \"installed\": {\n \"client_id\": client_id,\n \"client_secret\": client_secret,\n \"auth_uri\": \"https://accounts.google.com/o/oauth2/auth\",\n \"token_uri\": \"https://oauth2.googleapis.com/token\",\n \"redirect_uris\": [\"http://localhost\"],\n }\n }\n\n flow = InstalledAppFlow.from_client_config(client_config, SCOPES)\n creds = flow.run_local_server(port=0)\n\n # Save new credentials\n token_file.parent.mkdir(parents=True, exist_ok=True)\n try:\n with open(token_file, 'w') as f:\n json.dump(\n {\n 'token': creds.token,\n 'refresh_token': creds.refresh_token,\n 'token_uri': creds.token_uri,\n 'scopes': creds.scopes,\n },\n f,\n )\n os.chmod(token_file, 0o600)\n logger.info(f\"Credentials saved to {token_file}\")\n except Exception as e:\n logger.warning(\n f\"Failed to save credentials to {token_file}: {e}. \"\n \"You may need to re-authenticate next time.\"\n )\n\n return creds\n\n def _create_message(\n self,\n to_list: List[str],\n subject: str,\n body: str,\n cc_list: Optional[List[str]] = None,\n bcc_list: Optional[List[str]] = None,\n attachments: Optional[List[str]] = None,\n is_html: bool = False,\n in_reply_to: Optional[str] = None,\n references: Optional[List[str]] = None,\n ) -> Dict[str, str]:\n r\"\"\"Create a message object for sending.\"\"\"\n\n import base64\n from email import encoders\n from email.mime.base import MIMEBase\n from email.mime.multipart import MIMEMultipart\n from email.mime.text import MIMEText\n\n message = MIMEMultipart()\n message['to'] = ', '.join(to_list)\n message['subject'] = subject\n\n if cc_list:\n message['cc'] = ', '.join(cc_list)\n if bcc_list:\n message['bcc'] = ', '.join(bcc_list)\n\n # Set reply headers when provided\n if in_reply_to:\n message['In-Reply-To'] = in_reply_to\n if references:\n message['References'] = ' '.join(references)\n\n # Add body\n if is_html:\n message.attach(MIMEText(body, 'html'))\n else:\n message.attach(MIMEText(body, 'plain'))\n\n # Add attachments\n if attachments:\n for file_path in attachments:\n if os.path.isfile(file_path):\n with open(file_path, \"rb\") as attachment:\n part = MIMEBase('application', 'octet-stream')\n part.set_payload(attachment.read())\n encoders.encode_base64(part)\n part.add_header(\n 'Content-Disposition',\n f'attachment; filename= '\n f'{os.path.basename(file_path)}',\n )\n message.attach(part)\n\n # Encode message\n raw_message = base64.urlsafe_b64encode(message.as_bytes()).decode(\n 'utf-8'\n )\n return {'raw': raw_message}\n\n def _get_message_details(\n self, message_id: str\n ) -> Optional[Dict[str, Any]]:\n r\"\"\"Get detailed information about a message.\"\"\"\n try:\n message = (\n self.gmail_service.users()\n .messages()\n .get(userId='me', id=message_id)\n .execute()\n )\n\n headers = message['payload'].get('headers', [])\n # Build a name->value map in one pass (case-insensitive)\n header_map = {}\n for header in headers:\n name = header.get('name')\n if name:\n header_map[name.lower()] = header.get('value', '')\n\n return {\n \"message_id\": message['id'],\n \"thread_id\": message['threadId'],\n \"snippet\": message.get('snippet', ''),\n \"subject\": header_map.get('subject', ''),\n \"from\": header_map.get('from', ''),\n \"to\": header_map.get('to', ''),\n \"cc\": header_map.get('cc', ''),\n \"bcc\": header_map.get('bcc', ''),\n \"date\": header_map.get('date', ''),\n \"body\": self._extract_message_body(message),\n \"attachments\": self._extract_attachments(\n message, include_inline=True\n ),\n \"label_ids\": message.get('labelIds', []),\n \"size_estimate\": message.get('sizeEstimate', 0),\n }\n except Exception as e:\n logger.error(\"Failed to get message details: %s\", e)\n return None\n\n def _get_header_value(\n self, headers: List[Dict[str, str]], name: str\n ) -> str:\n r\"\"\"Get header value by name.\"\"\"\n for header in headers:\n if header['name'].lower() == name.lower():\n return header['value']\n return \"\"\n\n def _extract_message_body(self, message: Dict[str, Any]) -> str:\n r\"\"\"Extract message body from message payload.\n\n Recursively traverses the entire message tree and collects all text\n content from text/plain and text/html parts. Special handling for\n multipart/alternative containers: recursively searches for one format\n (preferring plain text) to avoid duplication when both formats contain\n the same content. All other text parts are collected to ensure no\n information is lost.\n\n Args:\n message (Dict[str, Any]): The Gmail message dictionary containing\n the payload to extract text from.\n\n Returns:\n str: The extracted message body text with multiple parts separated\n by double newlines, or an empty string if no text content is\n found.\n \"\"\"\n import base64\n import re\n\n text_parts = []\n\n def decode_text_data(data: str, mime_type: str) -> Optional[str]:\n \"\"\"Helper to decode base64 text data.\n\n Args:\n data: Base64 encoded text data.\n mime_type: MIME type for logging purposes.\n\n Returns:\n Decoded text string, or None if decoding fails or text\n is empty.\n \"\"\"\n if not data:\n return None\n try:\n text = base64.urlsafe_b64decode(data).decode('utf-8')\n return text if text.strip() else None\n except Exception as e:\n logger.warning(f\"Failed to decode {mime_type}: {e}\")\n return None\n\n def strip_html_tags(html_content: str) -> str:\n \"\"\"Strip HTML tags and convert to readable plain text.\n\n Uses regex to remove tags and clean up formatting while preserving\n basic document structure.\n\n Args:\n html_content: HTML content to strip.\n\n Returns:\n Plain text version of HTML content.\n \"\"\"\n if not html_content or not html_content.strip():\n return \"\"\n\n text = html_content\n\n # Remove script and style elements completely\n text = re.sub(\n r']*>.*?',\n '',\n text,\n flags=re.DOTALL | re.IGNORECASE,\n )\n text = re.sub(\n r']*>.*?',\n '',\n text,\n flags=re.DOTALL | re.IGNORECASE,\n )\n\n # Convert common HTML entities\n text = text.replace(' ', ' ')\n text = text.replace('&', '&')\n text = text.replace('<', '<')\n text = text.replace('>', '>')\n text = text.replace('"', '\"')\n text = text.replace(''', \"'\")\n text = text.replace('’', \"'\")\n text = text.replace('‘', \"'\")\n text = text.replace('”', '\"')\n text = text.replace('“', '\"')\n text = text.replace('—', '—')\n text = text.replace('–', '-')\n\n # Convert
and
to newlines\n text = re.sub(r'', '\\n', text, flags=re.IGNORECASE)\n\n # Convert block-level closing tags to newlines\n text = re.sub(\n r'', '\\n', text, flags=re.IGNORECASE\n )\n\n # Convert

to separator\n text = re.sub(r'', '\\n---\\n', text, flags=re.IGNORECASE)\n\n # Remove all remaining HTML tags\n text = re.sub(r'<[^>]+>', '', text)\n\n # Clean up whitespace\n text = re.sub(\n r'\\n\\s*\\n\\s*\\n+', '\\n\\n', text\n ) # Multiple blank lines to double newline\n text = re.sub(r' +', ' ', text) # Multiple spaces to single space\n text = re.sub(r'\\n ', '\\n', text) # Remove leading spaces on lines\n text = re.sub(\n r' \\n', '\\n', text\n ) # Remove trailing spaces on lines\n\n return text.strip()\n\n def find_text_recursive(\n part: Dict[str, Any], target_mime: str\n ) -> Optional[str]:\n \"\"\"Recursively search for text content of a specific MIME type.\n\n Args:\n part: Message part to search in.\n target_mime: Target MIME type ('text/plain' or 'text/html').\n\n Returns:\n Decoded text string if found, None otherwise.\n \"\"\"\n mime = part.get('mimeType', '')\n\n # Found the target type at this level\n if mime == target_mime:\n data = part.get('body', {}).get('data', '')\n decoded = decode_text_data(data, target_mime)\n # Strip HTML tags if this is HTML content\n if decoded and target_mime == 'text/html':\n return strip_html_tags(decoded)\n return decoded\n\n # Not found, but has nested parts? Search recursively\n if 'parts' in part:\n for nested_part in part['parts']:\n result = find_text_recursive(nested_part, target_mime)\n if result:\n return result\n\n return None\n\n def extract_from_part(part: Dict[str, Any]):\n \"\"\"Recursively collect all text from message parts.\"\"\"\n mime_type = part.get('mimeType', '')\n\n # Special handling for multipart/alternative\n if mime_type == 'multipart/alternative' and 'parts' in part:\n # Recursively search for one format (prefer plain text)\n plain_text = None\n html_text = None\n\n # Search each alternative branch recursively\n for nested_part in part['parts']:\n if not plain_text:\n plain_text = find_text_recursive(\n nested_part, 'text/plain'\n )\n if not html_text:\n html_text = find_text_recursive(\n nested_part, 'text/html'\n )\n\n # Prefer plain text, fall back to HTML\n chosen_text = plain_text if plain_text else html_text\n if chosen_text:\n text_parts.append(chosen_text)\n\n # If this part has nested parts (but not multipart/alternative)\n elif 'parts' in part:\n for nested_part in part['parts']:\n extract_from_part(nested_part)\n\n # If this is a text leaf, extract and collect it\n elif mime_type == 'text/plain':\n data = part.get('body', {}).get('data', '')\n text = decode_text_data(data, 'plain text body')\n if text:\n text_parts.append(text)\n\n elif mime_type == 'text/html':\n data = part.get('body', {}).get('data', '')\n html_text = decode_text_data(data, 'HTML body')\n if html_text:\n text = strip_html_tags(html_text)\n if text:\n text_parts.append(text)\n\n # Traverse the entire tree and collect all text parts\n payload = message.get('payload', {})\n extract_from_part(payload)\n\n if not text_parts:\n return \"\"\n\n # Return all text parts combined\n return '\\n\\n'.join(text_parts)\n\n def _extract_attachments(\n self, message: Dict[str, Any], include_inline: bool = False\n ) -> List[Dict[str, Any]]:\n r\"\"\"Extract attachment information from message payload.\n\n Recursively traverses the message tree to find all attachments\n and extracts their metadata. Distinguishes between regular attachments\n and inline images embedded in HTML content.\n\n Args:\n message (Dict[str, Any]): The Gmail message dictionary containing\n the payload to extract attachments from.\n\n Returns:\n List[Dict[str, Any]]: List of attachment dictionaries, each\n containing:\n - attachment_id: Gmail's unique identifier for the attachment\n - filename: Name of the attached file\n - mime_type: MIME type of the attachment\n - size: Size of the attachment in bytes\n - is_inline: Whether this is an inline image (embedded in HTML)\n \"\"\"\n attachments = []\n\n def is_inline_image(part: Dict[str, Any]) -> bool:\n \"\"\"Check if this part is an inline image.\"\"\"\n headers = part.get('headers', [])\n for header in headers:\n name = header.get('name', '').lower()\n value = header.get('value', '').lower()\n # Check for Content-Disposition: inline\n if name == 'content-disposition' and 'inline' in value:\n return True\n # Check for Content-ID (usually indicates inline)\n if name == 'content-id':\n return True\n return False\n\n def find_attachments(part: Dict[str, Any]):\n \"\"\"Recursively find attachments in message parts.\"\"\"\n # Check if this part has an attachmentId (indicates it's an\n # attachment)\n if 'body' in part and 'attachmentId' in part['body']:\n attachment_info = {\n 'attachment_id': part['body']['attachmentId'],\n 'filename': part.get('filename', 'unnamed'),\n 'mime_type': part.get(\n 'mimeType', 'application/octet-stream'\n ),\n 'size': part['body'].get('size', 0),\n 'is_inline': is_inline_image(part),\n }\n attachments.append(attachment_info)\n\n # Recurse into nested parts\n if 'parts' in part:\n for nested_part in part['parts']:\n find_attachments(nested_part)\n\n # Start traversal from the message payload\n payload = message.get('payload', {})\n if payload:\n find_attachments(payload)\n\n # Return based on include_inline toggle\n if include_inline:\n return attachments\n return [att for att in attachments if not att['is_inline']]\n\n def _is_valid_email(self, email: str) -> bool:\n r\"\"\"Validate email address format.\n\n Supports both formats:\n - Plain email: john@example.com\n - Named email: John Doe \n \"\"\"\n # Extract email from \"Name \" format if present\n match = re.search(r'<([^>]+)>$', email.strip())\n email_to_check = match.group(1) if match else email.strip()\n\n # Validate the email address\n pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$'\n return re.match(pattern, email_to_check) is not None\n\n def get_tools(self) -> List[FunctionTool]:\n r\"\"\"Returns a list of FunctionTool objects representing the\n functions in the toolkit.\n\n Returns:\n List[FunctionTool]: A list of FunctionTool objects\n representing the functions in the toolkit.\n \"\"\"\n return [\n FunctionTool(self.gmail_send_email),\n FunctionTool(self.gmail_reply_to_email),\n FunctionTool(self.gmail_forward_email),\n FunctionTool(self.gmail_create_draft),\n FunctionTool(self.gmail_send_draft),\n FunctionTool(self.gmail_fetch_emails),\n FunctionTool(self.gmail_fetch_thread_by_id),\n FunctionTool(self.gmail_modify_email_labels),\n FunctionTool(self.gmail_move_to_trash),\n FunctionTool(self.gmail_get_attachment),\n FunctionTool(self.gmail_list_threads),\n FunctionTool(self.gmail_list_drafts),\n FunctionTool(self.gmail_list_labels),\n FunctionTool(self.gmail_create_label),\n FunctionTool(self.gmail_delete_label),\n FunctionTool(self.gmail_modify_thread_labels),\n FunctionTool(self.gmail_get_profile),\n FunctionTool(self.gmail_get_contacts),\n FunctionTool(self.gmail_search_people),\n ]\n","metadata_json":{"methods":["people_service","people_service","gmail_send_email","gmail_reply_to_email","gmail_forward_email","gmail_create_draft","gmail_send_draft","gmail_fetch_emails","gmail_fetch_thread_by_id","gmail_modify_email_labels","gmail_move_to_trash","gmail_get_attachment","gmail_list_threads","gmail_list_drafts","gmail_list_labels","gmail_create_label","gmail_delete_label","gmail_modify_thread_labels","gmail_get_profile","gmail_get_contacts","gmail_search_people","get_tools"],"framework":"camel","class_name":"GmailToolkit"},"downloads_count":0,"rating_avg":4.0,"rating_count":1,"is_featured":false,"is_verified":false,"is_free_service":false,"external_source_name":"CAMEL-AI","external_source_url":"https://github.com/camel-ai/camel/blob/master/camel/toolkits/gmail_toolkit.py","claimed_by_user_id":null,"claimed_at":null,"files":[],"bundle_items":[],"chain_steps":[],"created_at":"2026-02-25T01:37:58.805351Z","updated_at":"2026-03-24T10:35:48.863170Z"}

Gmail Toolkit

Get this skill

Comments (0)