> ## Documentation Index
> Fetch the complete documentation index at: https://docs.timbal.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Mail Assistant

> AI-powered Gmail integration that triages emails and autonomously drafts intelligent responses

This guide shows how to:

* Connect to the Gmail API to retrieve incoming messages
* Use agents to generate draft replies

## Setup Requirements

<AccordionGroup>
  <Accordion title="Installing Packages">
    Install the required Google API client libraries:

    ```python theme={"dark"}
    google-api-core 
    google-api-python-client
    google-auth
    google-auth-oauthlib
    ```
  </Accordion>

  <Accordion title="Enabling Gmail API to your account">
    <Steps>
      <Step title="Create a Google Cloud Project">
        Go to the [Google Cloud Console](https://console.cloud.google.com/) and create a new project or select an existing one.
      </Step>

      <Step title="Enable Gmail API">
        1. Navigate to **APIs & Services** > **Library**
        2. Search for "Gmail API"
        3. Click on **Gmail API** and then **Enable**
      </Step>

      <Step title="Choose Authentication Method">
        There are two ways to authenticate with Gmail API. Choose the one that fits your use case:

        ## Method 1: OAuth 2.0 (Personal/Desktop Apps)

        **Use this if:**

        * You're building a personal application
        * You want users to authenticate with their own Google account
        * You're using a personal Gmail account (not Google Workspace)

        ### Setup Steps:

        **1. Create OAuth 2.0 Credentials**

        1. Go to **APIs & Services** > **Credentials**
        2. Click **Create Credentials** > **OAuth client ID**
        3. Choose **Desktop app** as the application type
        4. Download the credentials JSON file and save it as `credentials.json`

        **2. Configure OAuth Scopes**
        The following scopes are required:

        ```python theme={"dark"}
        SCOPES = [
          "https://www.googleapis.com/auth/gmail.readonly",
          "https://www.googleapis.com/auth/gmail.compose"
        ]
        ```

        **3. Generate Access Token**
        Run this script once to generate `token.json`:

        ```python theme={"dark"}
        from google_auth_oauthlib.flow import InstalledAppFlow

        SCOPES = [
            'https://www.googleapis.com/auth/gmail.readonly',
            'https://www.googleapis.com/auth/gmail.compose'
        ]

        flow = InstalledAppFlow.from_client_secrets_file('credentials.json', SCOPES)
        creds = flow.run_local_server(port=0)

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

        print("token.json generated successfully!")
        ```

        This will open a browser window for Google sign-in.

        **4. Initialize in Code**

        ```python theme={"dark"}
        from google.oauth2.credentials import Credentials
        from google.auth.transport.requests import Request
        from googleapiclient.discovery import build
        import os

        SCOPES = [
            'https://www.googleapis.com/auth/gmail.readonly',
            'https://www.googleapis.com/auth/gmail.compose'
        ]

        def initialize_gmail():
            """Initialize Gmail API connection with OAuth"""
            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:
                        print("No valid credentials found.")
                        return None
                
                gmail_service = build('gmail', 'v1', credentials=creds)

                # Test connection and get profile
                profile = gmail_service.users().getProfile(userId='me').execute()
                print(f"Connected to Gmail: {profile['emailAddress']}")
                
                return gmail_service
                
            except Exception as e:
                print(f"Error initializing Gmail API: {e}")
                return None
        ```

        ***

        ## Method 2: Service Account (Google Workspace)

        **Use this if:**

        * You have a Google Workspace account
        * You need to access Gmail on behalf of multiple users
        * You're building a server-side application

        <Warning>
          **Requirements:**

          * Google Workspace account (not personal Gmail)
          * Google Workspace Admin access for domain-wide delegation
        </Warning>

        ### Setup Steps:

        **1. Create a Service Account**

        1. Go to **IAM & Admin** > **Service Accounts**
        2. Click **CREATE SERVICE ACCOUNT**
        3. Enter name: `gmail-service-account`
        4. Add description: `Service account for Gmail API access`
        5. Click **CREATE AND CONTINUE**, then **DONE**
        6. Note the service account email: `gmail-service-account@your-project-id.iam.gserviceaccount.com`

        **2. Create and Download Service Account Key**

        1. Click on your service account
        2. Go to **KEYS** tab
        3. Click **ADD KEY** > **Create new key**
        4. Select **JSON** and click **CREATE**
        5. Save the downloaded file as `credentials.json`
        6. **Keep this file secure!**

        **3. Configure Domain-Wide Delegation**

        <Warning>
          Requires Google Workspace Admin privileges
        </Warning>

        1. Open `credentials.json` and copy the `client_id` value (e.g., `103635629912027933995`)
        2. Go to [Google Workspace Admin Console](https://admin.google.com/)
        3. Navigate to **Security** > **Access and data control** > **API Controls**
        4. Click **Manage Domain Wide Delegation** > **Add new**
        5. Paste the **Client ID** and add scopes:

        ```
        https://www.googleapis.com/auth/gmail.readonly,https://www.googleapis.com/auth/gmail.compose
        ```

        6. Click **Authorize**

        **4. Initialize Service Account in Code**

        ```python theme={"dark"}
        from google.oauth2 import service_account
        from googleapiclient.discovery import build

        SCOPES = [
            'https://www.googleapis.com/auth/gmail.readonly',
            'https://www.googleapis.com/auth/gmail.compose'
        ]

        CREDENTIALS_FILE = 'credentials.json'
        DELEGATED_USER = 'your-email@yourdomain.com'

        def initialize_gmail():
            """Initialize Gmail API connection with service account"""
            try:
                credentials = service_account.Credentials.from_service_account_file(
                    CREDENTIALS_FILE, scopes=SCOPES)

                delegated_credentials = credentials.with_subject(DELEGATED_USER)

                gmail_service = build('gmail', 'v1', credentials=delegated_credentials)

                # Test connection and get profile
                profile = gmail_service.users().getProfile(userId='me').execute()
                print(f"Connected to Gmail: {profile['emailAddress']}")
                
                return gmail_service
                
            except Exception as e:
                print(f"Error initializing Gmail API: {e}")
                return None
        ```
      </Step>
    </Steps>
  </Accordion>
</AccordionGroup>

## Implementation

This section walks through the essential components needed to implement Gmail integration with Timbal.
For the complete example, visit our [GitHub repository](https://github.com/timbal-ai/timbal/tree/main/examples/gmail).

### Gmail Initialization

Establish a connection to the Gmail API using your credentials. Choose the method that matches your setup:

<Tabs>
  <Tab title="OAuth 2.0">
    ```python theme={"dark"}
    def initialize_gmail(self):
        """Initialize Gmail API connection with OAuth"""
        try:
            creds = None
            token_file = 'token.json'
            
            if os.path.exists(token_file):
                creds = Credentials.from_authorized_user_file(token_file, self.scopes)
            
            if not creds or not creds.valid:
                if creds and creds.expired and creds.refresh_token:
                    print("Refreshing expired credentials...")
                    creds.refresh(Request())
                else:
                    print("No valid credentials found.")
                    return False
            
            self.gmail_service = build('gmail', 'v1', credentials=creds)

            # Test connection and get initial history ID
            profile = self.gmail_service.users().getProfile(userId='me').execute()
            print(f"Connected to Gmail: {profile['emailAddress']}")
            
            # Get initial history ID from profile
            self.last_history_id = profile.get('historyId')
            if self.last_history_id:
                print(f"Starting from history ID: {self.last_history_id}")
            else:
                print("No history ID found, will monitor from now")
            
            return True
            
        except Exception as e:
            print(f"Error initializing Gmail API: {e}")
            return False
    ```
  </Tab>

  <Tab title="Service Account">
    ```python theme={"dark"}
    def initialize_gmail(self):
        """Initialize Gmail API connection with Service Account"""
        try:
            CREDENTIALS_FILE = 'credentials.json'
            DELEGATED_USER = 'your-email@yourdomain.com'
            
            credentials = service_account.Credentials.from_service_account_file(
                CREDENTIALS_FILE, scopes=self.scopes)

            delegated_credentials = credentials.with_subject(DELEGATED_USER)

            self.gmail_service = build('gmail', 'v1', credentials=delegated_credentials)

            # Test connection and get initial history ID
            profile = self.gmail_service.users().getProfile(userId='me').execute()
            print(f"Connected to Gmail: {profile['emailAddress']}")
            
            # Get initial history ID from profile
            self.last_history_id = profile.get('historyId')
            if self.last_history_id:
                print(f"Starting from history ID: {self.last_history_id}")
            else:
                print("No history ID found, will monitor from now")
            
            return True
            
        except Exception as e:
            print(f"Error initializing Gmail API: {e}")
            return False
    ```
  </Tab>
</Tabs>

### Email monitoring

Monitor the Gmail inbox for new messages using polling:

```python theme={"dark"}
async def check_for_new_messages(self):
    """Check for new messages since last check"""
    try:           
        # Check for new messages since last history ID
        history = self.gmail_service.users().history().list(
            userId='me',
            startHistoryId=self.last_history_id,
            historyTypes=['messageAdded']
        ).execute()
        
        new_messages = []
        for history_record in history.get('history', []):
            for message_added in history_record.get('messagesAdded', []):
                message_id = message_added['message']['id']
                message_details = self.get_message_details(message_id)
                if message_details:
                    # Filter out draft messages
                    if not self.is_draft_message(message_details):
                        new_messages.append(message_details)
                    else:
                        print(f"📝 Skipping draft message: {message_details.get('subject', 'No Subject')}")
        
        if new_messages:
            print(f"\n🔔 Found {len(new_messages)} new message(s) at {datetime.now().strftime('%H:%M:%S')}")
            for message in new_messages:
                await self.generate_draft(message)
        
        # Update history ID
        if history.get('history'):
            self.last_history_id = history['history'][-1]['id']
        
    except HttpError as error:
        print(f"Error checking for new messages: {error}")


async def start_monitoring(self):
    """Start monitoring Gmail for new messages using polling"""   
    while True:
        await self.check_for_new_messages()
        await asyncio.sleep(10)  # Check new emails every 10 seconds
```

### Create a Timbal Agent for intelligent email responses

Create an Agent to generate emails responses:

```python theme={"dark"}
agent = Agent(
    name="email_response_generator", 
    model="openai/gpt-4o-mini",
    system_prompt="You are an assistant that helps draft professional email responses.",
    post_hook=self.save_draft,
)  
```

### Call a Timbal Agent to generate the draft

Process incoming emails by creating the prompt and invoking the Agent to generate a response:

```python theme={"dark"}
def _create_email_prompt(self, email):
    # Extract email information
    subject = email.get('subject', 'No Subject')
    sender = email.get('from', 'Unknown Sender')
    body = email.get('body', '')
    snippet = email.get('snippet', '')
    content = body if body else snippet
    
    prompt = f"""
You are an AI assistant that helps draft professional email responses. 

Please write a helpful and appropriate response to this email:

**Received Email:**
From: {sender}
Subject: {subject}
Content: {content}

**Instructions:**
- Write a professional response
- Do not assume anything that is not explicitly stated in the email
- Address any questions or requests appropriately
- Keep the tone friendly but business-appropriate
- If you need more information, ask clarifying questions
- End with a professional closing
- Keep the response concise but complete

**Your Response:**
"""
    return prompt.strip()


async def generate_draft(self, message):
    prompt = self._create_email_prompt(message)
    
    # Generate and create draft
    try:
        response_text = await self.agent(prompt=prompt, input_email=message).collect()
        if response_text:
            print("Generated Response:")
            print("-" * 40)
            print(response_text)
            print("-" * 40)
            
            print("Draft created successfully!")
            print("="*60)
        else:
            print("Failed to generate response")
    except Exception as e:
        print(f"Error generating/creating draft: {e}")
    
    print("="*60)
```

### Save the draft to the emails thread

Define the post-hook function that automatically saves the AI-generated response as a draft reply in the original email thread:

```python theme={"dark"}
def save_draft(self):
    """Create a draft response to the original message"""
    span = get_run_context().current_span()
    original_message = span.input['input_email']
    response_text = span.output.collect_text()

    try:
        # Extract sender email from the original message
        from_header = original_message.get('from', '')
        if '<' in from_header and '>' in from_header:
            # Extract email from "Name <email@domain.com>" format
            sender_email = from_header.split('<')[1].split('>')[0].strip()
        else:
            # Assume the entire from header is the email
            sender_email = from_header.strip()
        
        # Create response subject
        original_subject = original_message.get('subject', '')
        if not original_subject.startswith('Re:'):
            response_subject = f"Re: {original_subject}"
        else:
            response_subject = original_subject
        
        # Create draft message
        draft_message = self._create_message(sender_email, response_subject, response_text)
        
        # Create the draft with thread ID to link it to the original conversation
        draft_body = {
            'message': {
                'raw': draft_message
            }
        }
        
        # Include thread ID if available to link the draft to the original conversation
        if original_message.get('thread_id'):
            draft_body['message']['threadId'] = original_message['thread_id']
        
        draft = self.gmail_service.users().drafts().create(
            userId='me',
            body=draft_body
        ).execute()
        
        print(f"Draft ID: {draft['id']}")
        print(f"To: {sender_email}")
        print(f"Subject: {response_subject}")
        return draft
        
    except Exception as e:
        print(f"Error creating draft: {e}")
        return None
```

## Key Features

* **Real-time Processing**: Constantly polls new emails
* **Intelligent Responses**: Uses AI to generate contextually appropriate draft replies
* **History Tracking**: Maintains state to avoid processing duplicate emails or generated drafts

## Troubleshooting

<AccordionGroup>
  <Accordion title="Authentication Issues">
    * **Missing scopes**: Verify all required Gmail API scopes are included
    * **OAuth issues**: Check that `credentials.json` and `token.json` files exist and are valid
    * **Service Account issues**: Verify domain-wide delegation is configured and `DELEGATED_USER` has correct email
  </Accordion>

  <Accordion title="API Connection Problems">
    * **Rate limits**: Gmail API has daily quotas. Check your usage.
    * **Permission denied**: For Service Accounts, ensure domain-wide delegation is authorized with correct Client ID
  </Accordion>

  <Accordion title="Code Issues">
    * **Import errors**: Ensure all required packages are installed
    * **File errors**: Check that your credentials file exists and is properly formatted
  </Accordion>
</AccordionGroup>