Skip to main content

GmailToolkit

class GmailToolkit(BaseToolkit):
A comprehensive toolkit for Gmail operations. This class provides methods for Gmail operations including sending emails, managing drafts, fetching messages, managing labels, and handling contacts. API keys can be accessed in google cloud console (https://console.cloud.google.com/)

init

def __init__(self, timeout: Optional[float] = None):
Initializes a new instance of the GmailToolkit class. Parameters:
  • timeout (Optional[float]): The timeout value for API requests in seconds. If None, no timeout is applied. (default: :obj:None)

people_service

def people_service(self):
Lazily initialize and return the Google People service.

people_service

def people_service(self, service: Any):
Allow overriding/injecting the People service (e.g., in tests).

gmail_send_email

def gmail_send_email(
    self,
    to: Union[str, List[str]],
    subject: str,
    body: str,
    cc: Optional[Union[str, List[str]]] = None,
    bcc: Optional[Union[str, List[str]]] = None,
    attachments: Optional[List[str]] = None,
    is_html: bool = False
):
Send an email through Gmail. Parameters:
  • to (Union[str, List[str]]): Recipient email address(es).
  • subject (str): Email subject.
  • body (str): Email body content.
  • cc (Optional[Union[str, List[str]]]): CC recipient email address(es).
  • bcc (Optional[Union[str, List[str]]]): BCC recipient email address(es).
  • attachments (Optional[List[str]]): List of file paths to attach.
  • is_html (bool): Whether the body is HTML format. Set to True when sending formatted emails with HTML tags (e.g., bold, links, images). Use False (default) for plain text emails.
Returns: Dict[str, Any]: A dictionary containing the result of the operation.

gmail_reply_to_email

def gmail_reply_to_email(
    self,
    message_id: str,
    reply_body: str,
    reply_all: bool = False,
    is_html: bool = False
):
Reply to an email message. Parameters:
  • message_id (str): The unique identifier of the message to reply to. To get a message ID, first use fetch_emails() to list messages, or use the ‘message_id’ returned from send_email() or create_email_draft().
  • reply_body (str): The reply message body.
  • reply_all (bool): Whether to reply to all recipients.
  • is_html (bool): Whether the body is HTML format. Set to True when sending formatted emails with HTML tags (e.g., bold, links, images). Use False (default) for plain text emails.
Returns: Dict[str, Any]: A dictionary containing the result of the operation.

gmail_forward_email

def gmail_forward_email(
    self,
    message_id: str,
    to: Union[str, List[str]],
    forward_body: Optional[str] = None,
    cc: Optional[Union[str, List[str]]] = None,
    bcc: Optional[Union[str, List[str]]] = None,
    include_attachments: bool = True
):
Forward an email message. Parameters:
  • message_id (str): The unique identifier of the message to forward. To get a message ID, first use fetch_emails() to list messages, or use the ‘message_id’ returned from send_email() or create_email_draft().
  • to (Union[str, List[str]]): Recipient email address(es).
  • forward_body (Optional[str]): Additional message to include at the top of the forwarded email, before the original message content. If not provided, only the original message will be forwarded.
  • cc (Optional[Union[str, List[str]]]): CC recipient email address(es).
  • bcc (Optional[Union[str, List[str]]]): BCC recipient email address(es).
  • include_attachments (bool): Whether to include original attachments. Defaults to True. Only includes real attachments, not inline images.
Returns: Dict[str, Any]: A dictionary containing the result of the operation, including the number of attachments forwarded.

gmail_create_draft

def gmail_create_draft(
    self,
    to: Union[str, List[str]],
    subject: str,
    body: str,
    cc: Optional[Union[str, List[str]]] = None,
    bcc: Optional[Union[str, List[str]]] = None,
    attachments: Optional[List[str]] = None,
    is_html: bool = False
):
Create an email draft. Parameters:
  • to (Union[str, List[str]]): Recipient email address(es).
  • subject (str): Email subject.
  • body (str): Email body content.
  • cc (Optional[Union[str, List[str]]]): CC recipient email address(es).
  • bcc (Optional[Union[str, List[str]]]): BCC recipient email address(es).
  • attachments (Optional[List[str]]): List of file paths to attach.
  • is_html (bool): Whether the body is HTML format. Set to True when sending formatted emails with HTML tags (e.g., bold, links, images). Use False (default) for plain text emails.
Returns: Dict[str, Any]: A dictionary containing the result of the operation.

gmail_send_draft

def gmail_send_draft(self, draft_id: str):
Send a draft email. Parameters:
  • draft_id (str): The unique identifier of the draft to send. To get a draft ID, first use list_drafts() to list drafts, or use the ‘draft_id’ returned from create_email_draft().
Returns: Dict[str, Any]: A dictionary containing the result of the operation.

gmail_fetch_emails

def gmail_fetch_emails(
    self,
    query: str = '',
    max_results: int = 10,
    include_spam_trash: bool = False,
    label_ids: Optional[List[str]] = None,
    page_token: Optional[str] = None
):
Fetch emails with filters and pagination. Parameters:
  • query (str): Gmail search query string. Use Gmail’s search syntax: - ‘from:[email protected]’ - emails from specific sender - ‘subject:meeting’ - emails with specific subject text - ‘has:attachment’ - emails with attachments - ‘is:unread’ - unread emails - ‘in:sent’ - emails in sent folder - ‘after:2024/01/01 before:2024/12/31’ - date range
Returns: Dict[str, Any]: A dictionary containing the fetched emails.

gmail_fetch_thread_by_id

def gmail_fetch_thread_by_id(self, thread_id: str):
Fetch a thread by ID. Parameters:
  • thread_id (str): The unique identifier of the thread to fetch. To get a thread ID, first use list_threads() to list threads, or use the ‘thread_id’ returned from send_email() or reply_to_email().
Returns: Dict[str, Any]: A dictionary containing the thread details.

gmail_modify_email_labels

def gmail_modify_email_labels(
    self,
    message_id: str,
    add_labels: Optional[List[str]] = None,
    remove_labels: Optional[List[str]] = None
):
Modify labels on an email message. Parameters:
  • message_id (str): The unique identifier of the message to modify. To get a message ID, first use fetch_emails() to list messages, or use the ‘message_id’ returned from send_email() or create_email_draft().
  • add_labels (Optional[List[str]]): List of label IDs to add to the message. Label IDs can be: - System labels: ‘INBOX’, ‘STARRED’, ‘IMPORTANT’, ‘UNREAD’, etc. - Custom label IDs: Retrieved from list_gmail_labels() method.
Returns: Dict[str, Any]: A dictionary containing the result of the operation.

gmail_move_to_trash

def gmail_move_to_trash(self, message_id: str):
Move a message to trash. Parameters:
  • message_id (str): The unique identifier of the message to move to trash. To get a message ID, first use fetch_emails() to list messages, or use the ‘message_id’ returned from send_email() or create_email_draft().
Returns: Dict[str, Any]: A dictionary containing the result of the operation.

gmail_get_attachment

def gmail_get_attachment(
    self,
    message_id: str,
    attachment_id: str,
    save_path: Optional[str] = None
):
Get an attachment from a message. Parameters:
  • message_id (str): The unique identifier of the message containing the attachment. To get a message ID, first use fetch_emails() to list messages, or use the ‘message_id’ returned from send_email() or create_email_draft().
  • attachment_id (str): The unique identifier of the attachment to download. To get an attachment ID, first use fetch_emails() to get message details, then look for ‘attachment_id’ in the ‘attachments’ list of each message.
  • save_path (Optional[str]): Local file path where the attachment should be saved. If provided, the attachment will be saved to this location and the response will include a success message. If not provided, the attachment data will be returned as base64-encoded content in the response.
Returns: Dict[str, Any]: A dictionary containing the attachment data or save result.

gmail_list_threads

def gmail_list_threads(
    self,
    query: str = '',
    max_results: int = 10,
    include_spam_trash: bool = False,
    label_ids: Optional[List[str]] = None,
    page_token: Optional[str] = None
):
List email threads. Parameters:
  • query (str): Gmail search query string. Use Gmail’s search syntax: - ‘from:[email protected]’ - threads from specific sender - ‘subject:meeting’ - threads with specific subject text - ‘has:attachment’ - threads with attachments - ‘is:unread’ - unread threads - ‘in:sent’ - threads in sent folder - ‘after:2024/01/01 before:2024/12/31’ - date range
Returns: Dict[str, Any]: A dictionary containing the thread list.

gmail_list_drafts

def gmail_list_drafts(self, max_results: int = 10, page_token: Optional[str] = None):
List email drafts. Parameters:
  • max_results (int): Maximum number of drafts to fetch.
  • page_token (Optional[str]): Pagination token from a previous response. If provided, fetches the next page of results.
Returns: Dict[str, Any]: A dictionary containing the draft list.

gmail_list_labels

def gmail_list_labels(self):
Returns: Dict[str, Any]: A dictionary containing the label list.

gmail_create_label

def gmail_create_label(
    self,
    name: str,
    label_list_visibility: Literal['labelShow', 'labelHide'] = 'labelShow',
    message_list_visibility: Literal['show', 'hide'] = 'show'
):
Create a new Gmail label. Parameters:
  • name (str): The name of the label to create.
  • label_list_visibility (str): How the label appears in Gmail’s label list. - ‘labelShow’: Label is visible in the label list sidebar (default) - ‘labelHide’: Label is hidden from the label list sidebar
  • message_list_visibility (str): How the label appears in message lists. - ‘show’: Label is visible on messages in inbox/lists (default) - ‘hide’: Label is hidden from message displays
Returns: Dict[str, Any]: A dictionary containing the result of the operation.

gmail_delete_label

def gmail_delete_label(self, label_id: str):
Delete a Gmail label. Parameters:
  • label_id (str): The unique identifier of the user-created label to delete. To get a label ID, first use list_gmail_labels() to list all labels. Note: System labels (e.g., ‘INBOX’, ‘SENT’, ‘DRAFT’, ‘SPAM’, ‘TRASH’, ‘UNREAD’, ‘STARRED’, ‘IMPORTANT’, ‘CATEGORY_PERSONAL’, etc.) cannot be deleted.
Returns: Dict[str, Any]: A dictionary containing the result of the operation.

gmail_modify_thread_labels

def gmail_modify_thread_labels(
    self,
    thread_id: str,
    add_labels: Optional[List[str]] = None,
    remove_labels: Optional[List[str]] = None
):
Modify labels on a thread. Parameters:
  • thread_id (str): The unique identifier of the thread to modify. To get a thread ID, first use list_threads() to list threads, or use the ‘thread_id’ returned from send_email() or reply_to_email().
  • add_labels (Optional[List[str]]): List of label IDs to add to all messages in the thread. Label IDs can be: - System labels: ‘INBOX’, ‘STARRED’, ‘IMPORTANT’, ‘UNREAD’, etc. - Custom label IDs: Retrieved from list_gmail_labels().
Returns: Dict[str, Any]: A dictionary containing the result of the operation.

gmail_get_profile

def gmail_get_profile(self):
Returns: Dict[str, Any]: A dictionary containing the profile information.

gmail_get_contacts

def gmail_get_contacts(self, max_results: int = 100, page_token: Optional[str] = None):
List connections from Google People API. Parameters:
  • max_results (int): Maximum number of contacts to fetch.
  • page_token (Optional[str]): Pagination token from a previous response. If provided, fetches the next page of results.
Returns: Dict[str, Any]: A dictionary containing the contacts.

gmail_search_people

def gmail_search_people(self, query: str, max_results: int = 10):
Search for people in contacts. Parameters:
  • query (str): Search query for people in contacts. Can search by: - Name: ‘John Smith’ or partial names like ‘John’ - Email: ‘[email protected]’ - Organization: ‘Google’ or ‘Acme Corp’ - Phone number: ‘+1234567890’
Returns: Dict[str, Any]: A dictionary containing the search results.

_get_gmail_service

def _get_gmail_service(self):
Get Gmail service object.

_get_people_service

def _get_people_service(self):
Get People service object.

_authenticate

def _authenticate(self):
Authenticate with Google APIs using OAuth2. Automatically saves and loads credentials from ~/.camel/gmail_token.json to avoid repeated browser logins.

_create_message

def _create_message(
    self,
    to_list: List[str],
    subject: str,
    body: str,
    cc_list: Optional[List[str]] = None,
    bcc_list: Optional[List[str]] = None,
    attachments: Optional[List[str]] = None,
    is_html: bool = False,
    in_reply_to: Optional[str] = None,
    references: Optional[List[str]] = None
):
Create a message object for sending.

_get_message_details

def _get_message_details(self, message_id: str):
Get detailed information about a message.

_get_header_value

def _get_header_value(self, headers: List[Dict[str, str]], name: str):
Get header value by name.

_extract_message_body

def _extract_message_body(self, message: Dict[str, Any]):
Extract message body from message payload. Recursively traverses the entire message tree and collects all text content from text/plain and text/html parts. Special handling for multipart/alternative containers: recursively searches for one format (preferring plain text) to avoid duplication when both formats contain the same content. All other text parts are collected to ensure no information is lost. Parameters:
  • message (Dict[str, Any]): The Gmail message dictionary containing the payload to extract text from.
Returns: str: The extracted message body text with multiple parts separated by double newlines, or an empty string if no text content is found.

_extract_attachments

def _extract_attachments(self, message: Dict[str, Any], include_inline: bool = False):
Extract attachment information from message payload. Recursively traverses the message tree to find all attachments and extracts their metadata. Distinguishes between regular attachments and inline images embedded in HTML content. Parameters:
  • message (Dict[str, Any]): The Gmail message dictionary containing the payload to extract attachments from.
Returns: List[Dict[str, Any]]: List of attachment dictionaries, each containing:
  • attachment_id: Gmail’s unique identifier for the attachment
  • filename: Name of the attached file
  • mime_type: MIME type of the attachment
  • size: Size of the attachment in bytes
  • is_inline: Whether this is an inline image (embedded in HTML)

_is_valid_email

def _is_valid_email(self, email: str):
Validate email address format. Supports both formats:

get_tools

def get_tools(self):
Returns: List[FunctionTool]: A list of FunctionTool objects representing the functions in the toolkit.