Examples

This section provides examples of how to use pymailai in different scenarios.

OpenAI Integration

Basic Example

This example shows how to create a simple email agent that processes incoming emails using OpenAI’s GPT model and sends back AI-generated responses.

OpenAI Email Agent Example

"""Example of using PyMailAI with OpenAI's completion API."""

import asyncio
import os
from typing import Optional

from openai import OpenAI
from pymailai import EmailAgent, EmailConfig
from pymailai.message import EmailData

# Configure OpenAI
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


async def process_with_openai(message: EmailData) -> Optional[EmailData]:
    """Process email content using OpenAI's completion API.

    This example shows how to:
    1. Extract the prompt from an incoming email
    2. Send it to OpenAI's API
    3. Send the response back via email
    """
    try:
        # Get completion from OpenAI
        completion = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": message.body_text}
            ]
        )

        # Extract the response
        ai_response = completion.choices[0].message.content

        # Create email response
        return EmailData(
            message_id="",  # Will be generated by email server
            subject=f"Re: {message.subject}",
            from_address=message.to_addresses[0],  # Reply from the recipient
            to_addresses=[message.from_address],   # Reply to the sender
            cc_addresses=[],
            body_text=ai_response,
            body_html=None,
            timestamp=message.timestamp,
            in_reply_to=message.message_id,
            references=[message.message_id] if message.references is None
            else message.references + [message.message_id]
        )
    except Exception as e:
        print(f"Error processing message: {e}")
        return None


async def main():
    # Configure email settings
    config = EmailConfig(
        imap_server=os.getenv("IMAP_SERVER", "imap.gmail.com"),
        smtp_server=os.getenv("SMTP_SERVER", "smtp.gmail.com"),
        email=os.getenv("EMAIL_ADDRESS"),
        password=os.getenv("EMAIL_PASSWORD"),
        smtp_port=465  # Use implicit SSL/TLS port
    )

    # Create and run email agent
    async with EmailAgent(config, message_handler=process_with_openai) as agent:
        print(f"OpenAI Email Agent started. Monitoring {config.email}")
        print("Send an email to get an AI response!")
        print("Press Ctrl+C to stop...")

        try:
            while True:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            print("\nStopping email agent...")

if __name__ == "__main__":
    asyncio.run(main())

Advanced Example with Email Threading

This example shows how to create an email agent that maintains proper email threading, including quoted original messages and correct threading metadata.

OpenAI Email Agent with Threading Example

"""Example of using PyMailAI with OpenAI's completion API and proper email threading."""

import asyncio
import os
from typing import Optional
from dotenv import load_dotenv

from openai import OpenAI
from pymailai import EmailAgent, EmailConfig
from pymailai.message import EmailData

# Load environment variables
load_dotenv()

# Configure OpenAI
client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))


async def process_with_openai(message: EmailData) -> Optional[EmailData]:
    """Process email content using OpenAI's completion API.

    This example shows how to:
    1. Extract the prompt from an incoming email
    2. Send it to OpenAI's API
    3. Send the response back via email with proper threading
    """
    try:
        # Get completion from OpenAI
        completion = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": message.body_text}
            ]
        )

        # Extract the response
        ai_response = completion.choices[0].message.content

        # Create a properly threaded reply that includes the original message
        reply = message.create_reply(
            reply_text=ai_response,
            include_history=True  # This ensures the original message is quoted
        )

        # Set the from address since create_reply doesn't set it
        reply.from_address = message.to_addresses[0]

        return reply

    except Exception as e:
        print(f"Error processing message: {e}")
        return None


async def main():
    # Configure email settings
    config = EmailConfig(
        imap_server=os.getenv("IMAP_SERVER"),
        smtp_server=os.getenv("SMTP_SERVER"),
        email=os.getenv("EMAIL_ADDRESS"),
        password=os.getenv("EMAIL_PASSWORD"),
        smtp_port=465
    )

    # Create and run email agent
    async with EmailAgent(config, message_handler=process_with_openai) as agent:
        print(f"OpenAI Email Agent started. Monitoring {config.email}")
        print("Send an email to get an AI response with proper threading!")
        print("The response will include the original message and appear")
        print("properly threaded in your email client.")
        print("Press Ctrl+C to stop...")

        try:
            while True:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            print("\nStopping email agent...")

if __name__ == "__main__":
    asyncio.run(main())

Key features of the threading example:

• Uses create_reply() to generate properly formatted responses

• Includes the original message as quoted text

• Sets correct threading metadata (References and In-Reply-To headers)

• Ensures emails appear properly threaded in email clients

Email Threading Best Practices

When creating email responses, it’s important to maintain proper threading for a better user experience:

1. Include the original message as quoted text using create_reply()

2. Preserve threading metadata for email client compatibility

3. Maintain conversation context with proper message quoting

4. Keep CC recipients in the loop

Example of creating a threaded reply:

# Create a properly threaded reply
reply = message.create_reply(
    reply_text="Your response here",
    include_history=True  # Include quoted original message
)
reply.from_address = "your-email@example.com"

Running the Examples

To run these examples:

1. Install the required dependencies:

   pip install pymailai[openai]
   

2. Set up the required environment variables:

   export OPENAI_API_KEY="your-openai-api-key"
   export EMAIL_ADDRESS="your-email@example.com"
   export EMAIL_PASSWORD="your-email-password"
   # Optional: Configure custom email servers
   export EMAIL_IMAP_SERVER="imap.gmail.com"
   export EMAIL_SMTP_SERVER="smtp.gmail.com"
   

3. Run the example:

   python examples/openai_completion.py
   

The agent will monitor the specified email account and:

1. Process any new incoming emails

2. Send the email content to OpenAI’s API

3. Send back the AI-generated response to the original sender

Anthropic Integration

This example shows how to create an email agent that processes incoming emails using Anthropic’s Claude model and sends back AI-generated responses.

Anthropic Email Agent Example

"""Example of using PyMailAI with Anthropic's Claude API."""

import asyncio
import os
from typing import Optional

import anthropic
from pymailai import EmailAgent, EmailConfig
from pymailai.message import EmailData

# Configure Anthropic client
client = anthropic.AsyncAnthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

async def process_with_anthropic(message: EmailData) -> Optional[EmailData]:
    """Process email content using Anthropic's Claude API.

    This example shows how to:
    1. Extract the prompt from an incoming email
    2. Send it to Anthropic's API
    3. Send the response back via email
    """
    try:
        # Get completion from Anthropic
        message_content = await client.messages.create(
            model="claude-3-opus-20240229",
            max_tokens=1024,
            messages=[
                {"role": "user", "content": message.body_text}
            ]
        )

        # Extract the response
        ai_response = message_content.content[0].text

        # Create email response
        return EmailData(
            message_id="",  # Will be generated by email server
            subject=f"Re: {message.subject}",
            from_address=message.to_addresses[0],  # Reply from the recipient
            to_addresses=[message.from_address],   # Reply to the sender
            cc_addresses=[],
            body_text=ai_response,
            body_html=None,
            timestamp=message.timestamp,
            in_reply_to=message.message_id,
            references=[message.message_id] if message.references is None
                      else message.references + [message.message_id]
        )
    except Exception as e:
        print(f"Error processing message: {e}")
        return None

async def main():
    # Configure email settings
    config = EmailConfig(
        imap_server=os.getenv("IMAP_SERVER", "imap.gmail.com"),
        smtp_server=os.getenv("SMTP_SERVER", "smtp.gmail.com"),
        email=os.getenv("EMAIL_ADDRESS"),
        password=os.getenv("EMAIL_PASSWORD")
    )

    # Create and run email agent
    async with EmailAgent(config, message_handler=process_with_anthropic) as agent:
        print(f"Anthropic Email Agent started. Monitoring {config.email}")
        print("Send an email to get an AI response!")
        print("Press Ctrl+C to stop...")

        try:
            while True:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            print("\nStopping email agent...")

if __name__ == "__main__":
    asyncio.run(main())

To run this example:

1. Install the required dependencies:

   pip install pymailai[anthropic]
   

2. Set up the required environment variables:

   export ANTHROPIC_API_KEY="your-anthropic-api-key"
   export EMAIL_ADDRESS="your-email@example.com"
   export EMAIL_PASSWORD="your-email-password"
   # Optional: Configure custom email servers
   export EMAIL_IMAP_SERVER="imap.gmail.com"
   export EMAIL_SMTP_SERVER="smtp.gmail.com"
   

3. Run the example:

   python examples/anthropic_completion.py
   

The agent will monitor the specified email account and:

1. Process any new incoming emails

2. Send the email content to Anthropic’s API

3. Send back the AI-generated response to the original sender

Gmail Authentication

pymailai supports two methods for Gmail authentication:

1. OAuth2 (for personal Gmail accounts)

2. Service Account (for Google Workspace accounts)

OAuth2 Authentication

This example shows how to use Gmail OAuth2 authentication for personal Gmail accounts.

Gmail OAuth2 Example

"""Example of using PyMailAI with Gmail credentials from a file."""

import asyncio
import os
from pathlib import Path

from pymailai import EmailAgent
from pymailai.gmail import GmailCredentials
from pymailai.message import EmailData

async def echo_handler(message: EmailData) -> EmailData:
    """Simple echo handler that replies with the received message."""
    return EmailData(
        message_id="",
        subject=f"Re: {message.subject}",
        from_address=message.to_addresses[0],
        to_addresses=[message.from_address],
        cc_addresses=[],
        body_text=f"Received your message:\n\n{message.body_text}",
        body_html=None,
        timestamp=message.timestamp,
        in_reply_to=message.message_id,
        references=[message.message_id]
    )

async def main():
    # Load Gmail credentials from file
    creds_path = Path.home() / ".config" / "pymailai" / "gmail_creds.json"

    # If credentials file doesn't exist, create it from environment variables
    if not creds_path.exists():
        creds_path.parent.mkdir(parents=True, exist_ok=True)
        creds = GmailCredentials.from_oauth_credentials(
            client_id=os.getenv("GMAIL_CLIENT_ID", ""),
            client_secret=os.getenv("GMAIL_CLIENT_SECRET", ""),
            refresh_token=os.getenv("GMAIL_REFRESH_TOKEN", ""),
            save_path=creds_path
        )
    else:
        creds = GmailCredentials(creds_path)

    # Convert to EmailConfig
    email_address = os.getenv("GMAIL_ADDRESS", "")
    config = creds.to_email_config(email_address)

    # Create and run email agent
    async with EmailAgent(config, message_handler=echo_handler) as agent:
        print(f"Gmail Agent started. Monitoring {config.email}")
        print("Send an email to get an echo response!")
        print("Press Ctrl+C to stop...")

        try:
            while True:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            print("\nStopping email agent...")

if __name__ == "__main__":
    asyncio.run(main())

To use Gmail with OAuth2 credentials:

1. Set up a Google Cloud Project and enable the Gmail API:

   1. Go to the Google Cloud Console

   2. Create a new project or select an existing one

   3. Enable the Gmail API

   4. Create OAuth2 credentials: - Click “Create Credentials” and select “OAuth client ID” - Choose “Desktop application” as the application type - Download the credentials JSON file

   5. Configure the OAuth consent screen: - Add your email address as a test user - Set the necessary scopes (gmail.modify, gmail.compose, gmail.send)

2. Get a refresh token using the provided helper script:

   # First time setup - this will create the credentials file
   export GMAIL_CLIENT_ID="your-client-id"
   export GMAIL_CLIENT_SECRET="your-client-secret"
   export GMAIL_REFRESH_TOKEN="your-refresh-token"
   export GMAIL_ADDRESS="your-gmail@gmail.com"
   
   # Run the example to save credentials
   python examples/gmail_credentials.py
   

Service Account Authentication

This example shows how to use Gmail service account authentication for Google Workspace accounts.

Gmail Service Account Example

"""Example of using Gmail with service account authentication."""

import asyncio
import logging
from datetime import datetime
from typing import Optional

from pymailai.agent import EmailAgent
from pymailai.gmail import ServiceAccountCredentials
from pymailai.gmail_client import GmailClient
from pymailai.message import EmailData

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Path to the service account key file downloaded from Google Cloud Console
SERVICE_ACCOUNT_FILE = "credentials.json"

# The email address of the user to impersonate (must be set up in Google Workspace)
USER_EMAIL = "do@example.com"

# Scopes required for the Gmail API
SCOPES = ["https://www.googleapis.com/auth/gmail.modify"]

# Optional: Custom email address for replies (if different from USER_EMAIL)
CUSTOM_EMAIL = None  # e.g. "custom@example.com"

# Optional: Feedback email configuration
FEEDBACK_EMAIL = "feedback@example.com"
FEEDBACK_MESSAGE = "\n\nPS: You can always send your feedback to feedback@example.com"


async def echo_handler(message: EmailData) -> Optional[EmailData]:
    """Handle incoming emails by echoing their content back."""
    logger.info(f"Processing email from: {message.from_address}, subject: {message.subject}")

    # Create reply with original content
    reply_text = f"""
I received your email with subject: {message.subject}

Here's what you sent:
-------------------
{message.body_text}
-------------------

{FEEDBACK_MESSAGE if FEEDBACK_EMAIL else ""}
"""

    # Create reply email with proper HTML formatting
    reply_html = f"""
<p>I received your email with subject: {message.subject}</p>

<p>Here's what you sent:</p>
<blockquote style="border-left: 2px solid #ccc; margin-left: 0; padding-left: 1em;">
{message.body_html if message.body_html else message.body_text}
</blockquote>

{f'<p>{FEEDBACK_MESSAGE}</p>' if FEEDBACK_EMAIL else ''}
"""

    logger.info("Creating reply email")
    reply = EmailData(
        message_id="",  # Will be generated by Gmail
        subject=f"Re: {message.subject}",
        from_address="",  # Will be set by Gmail
        to_addresses=[message.from_address],
        cc_addresses=[],
        body_text=reply_text,
        body_html=reply_html,
        timestamp=datetime.now(),
        in_reply_to=message.message_id
    )

    logger.info(f"Created reply to: {reply.to_addresses[0]}, subject: {reply.subject}")
    return reply


async def main():
    """Run the email echo service using service account authentication."""
    try:
        # Load service account credentials and set up delegation
        # The service account JSON is downloaded from Google Cloud Console
        # and contains fields like:
        # {
        #   "type": "service_account",
        #   "project_id": "your-project",
        #   "private_key_id": "...",
        #   "private_key": "-----BEGIN PRIVATE KEY-----\n...",
        #   "client_email": "service@project.iam.gserviceaccount.com",
        #   "client_id": "...",
        #   "auth_uri": "https://accounts.google.com/o/oauth2/auth",
        #   "token_uri": "https://oauth2.googleapis.com/token",
        #   ...
        # }

        # Set up Gmail service account authentication
        creds = ServiceAccountCredentials(
            credentials_path=SERVICE_ACCOUNT_FILE,
            delegated_email=USER_EMAIL,
            scopes=SCOPES
        )

        # Create Gmail client
        service = creds.get_gmail_service()
        logger.info(f"Gmail service initialized for {USER_EMAIL}")

        client = GmailClient(service)
        logger.info("Gmail client created")

        # Create and start the email agent
        agent = EmailAgent(client, message_handler=echo_handler)
        logger.info("Email agent created with echo handler")

        async with agent:
            # Keep the agent running
            while True:
                await asyncio.sleep(1)

    except KeyboardInterrupt:
        logger.info("Shutting down email echo service")
    except Exception as e:
        logger.error(f"Service error: {e}")


if __name__ == "__main__":
    asyncio.run(main())

To use Gmail with service account authentication:

1. Set up Google Cloud Project:

   1. Go to the Google Cloud Console

   2. Create a new project or select an existing one

   3. Enable the Gmail API

   4. Create a service account: - Click “Create Service Account” - Download the JSON key file

   5. Configure domain-wide delegation: - Go to Google Workspace Admin Console - Security -> API Controls -> Domain-wide Delegation - Add the service account’s client ID - Add scope: https://www.googleapis.com/auth/gmail.modify

2. Run the example:

   # Update SERVICE_ACCOUNT_FILE and USER_EMAIL in the example
   python examples/gmail_service_account.py
   

The service account provides: - Better security through domain-wide delegation - No need for user interaction - Support for multiple user impersonation - Direct Gmail API access

Logging Configuration

pymailai includes comprehensive logging for debugging and monitoring. You can configure the logging level and output format:

import logging

# Configure logging for the entire package
logging.basicConfig(
    level=logging.INFO,  # or logging.DEBUG for more detailed output
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Or configure logging for specific components
logger = logging.getLogger('pymailai.client')
logger.setLevel(logging.DEBUG)

The following components have dedicated loggers:

• pymailai.client: Email client operations (SMTP/IMAP connections)

• pymailai.agent: AI agent activities (message processing)

• pymailai.gmail: Gmail-specific operations (OAuth2, service accounts)

• pymailai.gmail_client: Gmail API operations

Example log output at INFO level:

2024-03-20 10:15:30,123 - pymailai.gmail - INFO - Gmail service initialized for user@domain.com
2024-03-20 10:15:30,456 - pymailai.gmail_client - INFO - Found 1 unread messages
2024-03-20 10:15:30,789 - pymailai.agent - INFO - Processing message from: sender@example.com
2024-03-20 10:15:31,012 - pymailai.gmail_client - INFO - Message sent successfully with ID: msg-123

Ollama Integration

This example shows how to create an email agent that processes incoming emails using Ollama’s local LLM models and sends back AI-generated responses.

Ollama Email Agent Example

"""Example of using PyMailAI with Ollama's chat API."""

import asyncio
import os
from typing import Optional

from ollama import chat, ChatResponse
from pymailai import EmailAgent, EmailConfig
from pymailai.message import EmailData


async def process_with_ollama(message: EmailData) -> Optional[EmailData]:
    """Process email content using Ollama's chat API.

    This example shows how to:
    1. Extract the prompt from an incoming email
    2. Send it to Ollama's API
    3. Send the response back via email
    """
    try:
        # Get completion from Ollama
        response: ChatResponse = chat(
            model="llama3.2",  # Latest Llama version
            messages=[
                {
                    "role": "system",
                    "content": "You are a helpful assistant.",
                },
                {
                    "role": "user",
                    "content": message.body_text,
                },
            ],
        )

        # Extract the response
        ai_response = response.message.content

        # Create email response
        return EmailData(
            message_id="",  # Will be generated by email server
            subject=f"Re: {message.subject}",
            from_address=message.to_addresses[0],  # Reply from the recipient
            to_addresses=[message.from_address],  # Reply to the sender
            cc_addresses=[],
            body_text=ai_response,
            body_html=None,
            timestamp=message.timestamp,
            in_reply_to=message.message_id,
            references=[message.message_id]
            if message.references is None
            else message.references + [message.message_id],
        )
    except Exception as e:
        print(f"Error processing message: {e}")
        return None


async def main():
    # Configure email settings - requires explicit server configuration
    config = EmailConfig(
        imap_server=os.getenv("IMAP_SERVER"),
        smtp_server=os.getenv("SMTP_SERVER"),
        email=os.getenv("EMAIL_ADDRESS"),
        password=os.getenv("EMAIL_PASSWORD"),
        smtp_port=int(os.getenv("SMTP_PORT", "465")),  # SSL/TLS port by default
        imap_port=int(os.getenv("IMAP_PORT", "993")),  # SSL/TLS port by default
    )

    # Create and run email agent
    async with EmailAgent(config, message_handler=process_with_ollama) as agent:
        print(f"Ollama Email Agent started. Monitoring {config.email}")
        print("Send an email to get an AI response!")
        print("Press Ctrl+C to stop...")

        try:
            while True:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            print("\nStopping email agent...")


if __name__ == "__main__":
    asyncio.run(main())

To run this example:

1. Install the required dependencies:

   pip install pymailai[ollama]
   

2. Install and start Ollama:

   Follow the instructions at https://ollama.ai to install Ollama for your platform. Then pull your desired model:

   ollama pull llama3.2  # Latest Llama version
   

3. Set up the required environment variables:

   export EMAIL_ADDRESS="your-email@example.com"
   export EMAIL_PASSWORD="your-email-password"
   export EMAIL_IMAP_SERVER="your-imap-server"
   export EMAIL_SMTP_SERVER="your-smtp-server"
   export EMAIL_IMAP_PORT="993"  # Default SSL/TLS port for IMAP
   export EMAIL_SMTP_PORT="465"  # Default SSL/TLS port for SMTP
   

4. Run the example:

   python examples/ollama_completion.py
   

The agent will monitor the specified email account and:

1. Process any new incoming emails

2. Send the email content to the local Ollama instance

3. Send back the AI-generated response to the original sender

Simple AI Agent

For a simpler example without external AI providers, check out the examples/simple_ai_agent.py file in the repository. This example shows the basic structure of creating a custom message handler.