Skip to content

Tool

EmailClient

Bases: BaseTool

Tool for working with email

Attributes:

Name Type Description
username Optional[str]

Username/email address used to send email via the SMTP protocol and retrieve email via the IMAP protocol. Example: bender@futurama.com
password Optional[str]

Password used to send email via the SMTP protocol and retrieve email via the IMAP protocol. If using gmail, this would be an App Password.
email_max_retrieve_count Optional[int]

Used to limit the number of messages retrieved during any given activities.
smtp_host Optional[str]

Hostname or url of the SMTP server. Example: smtp.gmail.com
smtp_port Optional[int]

Port of the SMTP server. Example: 465
smtp_use_ssl bool

Whether to use SSL when sending email via the SMTP protocol.
smtp_user Optional[str]

Username/email address used to send email via the SMTP protocol. Overrides username for SMTP only.
smtp_password Optional[str]

Password to send email via the SMTP protocol. Overrides password for SMTP only.
imap_url Optional[str]

Hostname or url of the IMAP server. Example: imap.gmail.com
imap_user Optional[str]

Username/email address used to retrieve email via the IMAP protocol. Overrides username for IMAP only.
imap_password Optional[str]

Password to retrieve email via the IMAP protocol. Overrides password for IMAP only.
mailboxes Optional[dict[str, str]]

Descriptions of mailboxes available for retrieving email via the IMAP protocol. Example: {'INBOX': 'default mailbox for incoming email', 'SENT': 'default mailbox for sent email'}
email_loader EmailLoader

Used to retrieve email.
Source code in griptape/griptape/tools/email_client/tool.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@define
class EmailClient(BaseTool):
    """Tool for working with email

    Attributes:
        username: Username/email address used to send email via the SMTP protocol and retrieve email via the IMAP protocol.
            Example: bender@futurama.com
        password: Password used to send email via the SMTP protocol and retrieve email via the IMAP protocol. If using gmail,
            this would be an App Password.
        email_max_retrieve_count: Used to limit the number of messages retrieved during any given activities.
        smtp_host: Hostname or url of the SMTP server. Example: smtp.gmail.com
        smtp_port: Port of the SMTP server. Example: 465
        smtp_use_ssl: Whether to use SSL when sending email via the SMTP protocol.
        smtp_user: Username/email address used to send email via the SMTP protocol. Overrides username for SMTP only.
        smtp_password: Password to send email via the SMTP protocol. Overrides password for SMTP only.
        imap_url: Hostname or url of the IMAP server. Example: imap.gmail.com
        imap_user: Username/email address used to retrieve email via the IMAP protocol. Overrides username for IMAP only.
        imap_password: Password to retrieve email via the IMAP protocol.  Overrides password for IMAP only.
        mailboxes: Descriptions of mailboxes available for retrieving email via the IMAP protocol.
            Example: {'INBOX': 'default mailbox for incoming email', 'SENT': 'default mailbox for sent email'}
        email_loader: Used to retrieve email.
    """

    username: Optional[str] = field(default=None, kw_only=True)
    password: Optional[str] = field(default=None, kw_only=True)
    email_max_retrieve_count: Optional[int] = field(default=None, kw_only=True)
    smtp_host: Optional[str] = field(default=None, kw_only=True)
    smtp_port: Optional[int] = field(default=None, kw_only=True)
    smtp_use_ssl: bool = field(default=True, kw_only=True)
    smtp_user: Optional[str] = field(default=Factory(lambda self: self.username, takes_self=True), kw_only=True)
    smtp_password: Optional[str] = field(default=Factory(lambda self: self.password, takes_self=True), kw_only=True)
    imap_url: Optional[str] = field(default=None, kw_only=True)
    imap_user: Optional[str] = field(default=Factory(lambda self: self.username, takes_self=True), kw_only=True)
    imap_password: Optional[str] = field(default=Factory(lambda self: self.password, takes_self=True), kw_only=True)
    mailboxes: Optional[dict[str, str]] = field(default=None, kw_only=True)
    email_loader: EmailLoader = field(
        default=Factory(
            lambda self: EmailLoader(imap_url=self.imap_url, username=self.imap_user, password=self.imap_password),
            takes_self=True,
        ),
        kw_only=True,
    )

    @activity(
        config={
            "description": "Can be used to retrieve emails."
            "{% if _self.mailboxes %} Available mailboxes: {{ _self.mailboxes }}{% endif %}",
            "schema": Schema(
                {
                    Literal("label", description="Label to retrieve emails from such as 'INBOX' or 'SENT'"): str,
                    schema.Optional(
                        Literal("key", description="Optional key for filtering such as 'FROM' or 'SUBJECT'")
                    ): str,
                    schema.Optional(
                        Literal("search_criteria", description="Optional search criteria to filter emails by key")
                    ): str,
                    schema.Optional(Literal("max_count", description="Optional max email count")): int,
                }
            ),
        }
    )
    def retrieve(self, params: dict) -> ListArtifact | ErrorArtifact:
        values = params["values"]
        max_count = int(values["max_count"]) if values.get("max_count") else self.email_max_retrieve_count

        return self.email_loader.load(
            EmailLoader.EmailQuery(
                label=values["label"],
                key=values.get("key"),
                search_criteria=values.get("search_criteria"),
                max_count=max_count,
            )
        )

    @activity(
        config={
            "description": "Can be used to send emails",
            "schema": Schema(
                {
                    Literal("to", description="Recipient's email address"): str,
                    Literal("subject", description="Email subject"): str,
                    Literal("body", description="Email body"): str,
                }
            ),
        }
    )
    def send(self, params: dict) -> InfoArtifact | ErrorArtifact:
        values = params["values"]

        msg = MIMEText(values["body"])
        msg["Subject"] = values["subject"]
        msg["From"] = self.smtp_user
        msg["To"] = values["to"]

        try:
            with self._create_smtp_client() as client:
                client.login(self.smtp_user, self.smtp_password)
                client.sendmail(msg["From"], [msg["To"]], msg.as_string())
                return InfoArtifact("email was successfully sent")
        except Exception as e:
            logging.error(e)
            return ErrorArtifact(f"error sending email: {e}")

    def _create_smtp_client(self) -> smtplib.SMTP | smtplib.SMTP_SSL:
        smtp_host = self.smtp_host
        smtp_port = int(self.smtp_port)

        if self.smtp_use_ssl:
            return smtplib.SMTP_SSL(smtp_host, smtp_port)
        else:
            return smtplib.SMTP(smtp_host, smtp_port)

email_loader: EmailLoader = field(default=Factory(lambda : EmailLoader(imap_url=self.imap_url, username=self.imap_user, password=self.imap_password), takes_self=True), kw_only=True) class-attribute instance-attribute

email_max_retrieve_count: Optional[int] = field(default=None, kw_only=True) class-attribute instance-attribute

imap_password: Optional[str] = field(default=Factory(lambda : self.password, takes_self=True), kw_only=True) class-attribute instance-attribute

imap_url: Optional[str] = field(default=None, kw_only=True) class-attribute instance-attribute

imap_user: Optional[str] = field(default=Factory(lambda : self.username, takes_self=True), kw_only=True) class-attribute instance-attribute

mailboxes: Optional[dict[str, str]] = field(default=None, kw_only=True) class-attribute instance-attribute

password: Optional[str] = field(default=None, kw_only=True) class-attribute instance-attribute

smtp_host: Optional[str] = field(default=None, kw_only=True) class-attribute instance-attribute

smtp_password: Optional[str] = field(default=Factory(lambda : self.password, takes_self=True), kw_only=True) class-attribute instance-attribute

smtp_port: Optional[int] = field(default=None, kw_only=True) class-attribute instance-attribute

smtp_use_ssl: bool = field(default=True, kw_only=True) class-attribute instance-attribute

smtp_user: Optional[str] = field(default=Factory(lambda : self.username, takes_self=True), kw_only=True) class-attribute instance-attribute

username: Optional[str] = field(default=None, kw_only=True) class-attribute instance-attribute

retrieve(params)

Source code in griptape/griptape/tools/email_client/tool.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@activity(
    config={
        "description": "Can be used to retrieve emails."
        "{% if _self.mailboxes %} Available mailboxes: {{ _self.mailboxes }}{% endif %}",
        "schema": Schema(
            {
                Literal("label", description="Label to retrieve emails from such as 'INBOX' or 'SENT'"): str,
                schema.Optional(
                    Literal("key", description="Optional key for filtering such as 'FROM' or 'SUBJECT'")
                ): str,
                schema.Optional(
                    Literal("search_criteria", description="Optional search criteria to filter emails by key")
                ): str,
                schema.Optional(Literal("max_count", description="Optional max email count")): int,
            }
        ),
    }
)
def retrieve(self, params: dict) -> ListArtifact | ErrorArtifact:
    values = params["values"]
    max_count = int(values["max_count"]) if values.get("max_count") else self.email_max_retrieve_count

    return self.email_loader.load(
        EmailLoader.EmailQuery(
            label=values["label"],
            key=values.get("key"),
            search_criteria=values.get("search_criteria"),
            max_count=max_count,
        )
    )

send(params)

Source code in griptape/griptape/tools/email_client/tool.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@activity(
    config={
        "description": "Can be used to send emails",
        "schema": Schema(
            {
                Literal("to", description="Recipient's email address"): str,
                Literal("subject", description="Email subject"): str,
                Literal("body", description="Email body"): str,
            }
        ),
    }
)
def send(self, params: dict) -> InfoArtifact | ErrorArtifact:
    values = params["values"]

    msg = MIMEText(values["body"])
    msg["Subject"] = values["subject"]
    msg["From"] = self.smtp_user
    msg["To"] = values["to"]

    try:
        with self._create_smtp_client() as client:
            client.login(self.smtp_user, self.smtp_password)
            client.sendmail(msg["From"], [msg["To"]], msg.as_string())
            return InfoArtifact("email was successfully sent")
    except Exception as e:
        logging.error(e)
        return ErrorArtifact(f"error sending email: {e}")