Automated newsletter generation and email dispatch using CrewAI and Gmail API with OAuth 2.0 authentication

gmail_newsletter_dispatcher.py

import os
import base64
from dotenv import load_dotenv
from crewai import Agent, Task, Crew
from crewai_tools import BaseTool
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from email.mime.text import MIMEText

# Load environment variables
load_dotenv()

# Define OAuth 2.0 Scopes
SCOPES = ["https://www.googleapis.com/auth/gmail.send"]

class GmailEmailSenderTool(BaseTool):
    name: str = "Gmail API Email Sender Tool"
    description: str = "Sends emails using Gmail API with OAuth 2.0 authentication."

    def _run(self, recipient_list: list, subject: str, content: str) -> str:
        """Sends an email using Gmail API with OAuth 2.0 authentication."""
        try:
            creds = None
            if os.path.exists("token.json"):
                creds = Credentials.from_authorized_user_file("token.json", SCOPES)

            if not creds or not creds.valid:
                if creds and creds.expired and creds.refresh_token:
                    creds.refresh(Request())
                else:
                    flow = InstalledAppFlow.from_client_secrets_file("secret.json", SCOPES)
                    creds = flow.run_local_server(port=0)

                with open("token.json", "w") as token_file:
                    token_file.write(creds.to_json())

            service = build("gmail", "v1", credentials=creds)
            sender_email = os.getenv("GMAIL_ADDRESS")

            for recipient in recipient_list:
                message = MIMEText(content, "html")
                message["to"] = recipient
                message["from"] = sender_email
                message["subject"] = subject
                raw_message = base64.urlsafe_b64encode(message.as_bytes()).decode()

                service.users().messages().send(
                    userId="me",
                    body={"raw": raw_message}
                ).execute()

            return "Emails sent successfully."

        except Exception as e:
            return f"Failed to send emails. Error: {e}"


# Initialize the Gmail API email sender tool
email_send_tool = GmailEmailSenderTool()

# Define CrewAI Agents
content_generator = Agent(
    role="Newsletter Content Creator",
    goal="Write engaging and well-structured newsletter content based on {topic}",
    backstory="You are a professional content writer with expertise in crafting compelling newsletters.",
    allow_delegation=False,
    verbose=True
)

emailer = Agent(
    role="Email Dispatcher",
    goal="Send the generated newsletter to all specified recipients",
    backstory="You are responsible for ensuring that newsletters are delivered successfully via Gmail API.",
    allow_delegation=False,
    tools=[email_send_tool],
    verbose=True
)

# Define CrewAI Tasks
content_generation_task = Task(
    description=(
        "1. Create a concise and structured, engaging newsletter on {topic}.\n"
        "2. Ensure the newsletter has an introduction, key insights, and a compelling call to action.\n"
        "3. Format the newsletter properly for readability and impact.\n"
        "4. Generate the newsletter in HTML format."
    ),
    expected_output="A well-written newsletter in HTML format, structured for clarity and audience engagement.",
    agent=content_generator
)

email_sending_task = Task(
    description=(
        "1. Use the generated newsletter content.\n"
        "2. Send the generated newsletter to all recipients in {recipient_list}.\n"
        "3. Use the Gmail API with OAuth 2.0 authentication.\n"
        "4. Confirm successful email delivery and handle any errors."
    ),
    expected_output="A successfully delivered newsletter email, sent to all recipients with proper formatting.",
    agent=emailer,
    context=[content_generation_task]
)

# Create CrewAI Workflow
crew = Crew(
    agents=[content_generator, emailer],
    tasks=[content_generation_task, email_sending_task],
    verbose=2
)

# Run the CrewAI Workflow
if __name__ == "__main__":
    result = crew.kickoff(
        inputs={
            "topic": "The power of building in public on linkedin by Elliott Arnold",
            "recipient_list": os.getenv("DEFAULT_RECIPIENTS").split(',')
        }
    )
    print(result)