Source code for langchain_google_community.gmail.create_draft

import base64
from email.message import EmailMessage
from typing import List, Optional, Type

from langchain_core.callbacks import CallbackManagerForToolRun
from pydantic import BaseModel, Field

from langchain_google_community.gmail.base import GmailBaseTool


[docs] class CreateDraftSchema(BaseModel): """Input for CreateDraftTool.""" message: str = Field( ..., description="The message to include in the draft.", ) to: List[str] = Field( ..., description="The list of recipients.", ) subject: str = Field( ..., description="The subject of the message.", ) cc: Optional[List[str]] = Field( default=None, description="The list of CC recipients.", ) bcc: Optional[List[str]] = Field( default=None, description="The list of BCC recipients.", )
[docs] class GmailCreateDraft(GmailBaseTool): """Tool that creates a draft email for Gmail.""" name: str = "create_gmail_draft" description: str = ( "Use this tool to create a draft email with the provided message fields." ) args_schema: Type[CreateDraftSchema] = CreateDraftSchema def _prepare_draft_message( self, message: str, to: List[str], subject: str, cc: Optional[List[str]] = None, bcc: Optional[List[str]] = None, ) -> dict: draft_message = EmailMessage() draft_message.set_content(message) draft_message["To"] = ", ".join(to) draft_message["Subject"] = subject if cc is not None: draft_message["Cc"] = ", ".join(cc) if bcc is not None: draft_message["Bcc"] = ", ".join(bcc) encoded_message = base64.urlsafe_b64encode(draft_message.as_bytes()).decode() return {"message": {"raw": encoded_message}} def _run( self, message: str, to: List[str], subject: str, cc: Optional[List[str]] = None, bcc: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForToolRun] = None, ) -> str: try: create_message = self._prepare_draft_message(message, to, subject, cc, bcc) draft = ( self.api_resource.users() .drafts() .create(userId="me", body=create_message) .execute() ) output = f'Draft created. Draft Id: {draft["id"]}' return output except Exception as e: raise Exception(f"An error occurred: {e}")