integration-google-mail-sdd-software-design-document

---

title: Google Mail Integration - Software Design Document type: reference component_type: reference version: 1.0.0 created: '2025-12-27' updated: '2025-12-27' status: draft tags:

• ai-ml
• authentication
• deployment
• security
• testing
• api
• architecture
• automation summary: 'Google Mail Integration - Software Design Document Project: CODITECT Google Mail (Gmail) Integration Version: 1.0.0 Status: Draft Last Updated: December 17, 2025 --- Executive Summary This document defines the software design for integrating Gmail...' moe_confidence: 0.950 moe_classified: 2025-12-31

---

Google Mail Integration - Software Design Document

Project: CODITECT Google Mail (Gmail) Integration Version: 1.0.0 Status: Draft Last Updated: December 17, 2025

---

1. Executive Summary

1.1 Purpose

This document defines the software design for integrating Gmail API capabilities into the CODITECT platform as an optional component. The integration enables programmatic email sending, receiving, and processing for meeting notifications, AI-powered email analysis, and workflow automation.

1.2 Scope

• In Scope:

  • Gmail API integration (send, receive, search, labels)
  • OAuth 2.0 authentication (user and service account)
  • Email template system for notifications
  • Inbox monitoring via push notifications (Pub/Sub)
  • Thread and conversation management
  • Integration with meeting providers (Google Meet, Zoom)
  • AI pipeline integration for email analysis

• Out of Scope:

  • Gmail UI/frontend components
  • Email client features (drafts UI, signature editor)
  • Third-party email providers (Outlook handled separately)
  • Email marketing/bulk sending (use dedicated services)

1.3 Design Goals

1. Zero External Cost - Use only free-tier Gmail API quotas
2. Optional Component - Activatable only when needed
3. Provider Agnostic - Abstract interface for multi-provider email support
4. CODITECT Native - Follow framework patterns and conventions
5. Privacy First - Minimal data retention, user consent required

---

2. System Architecture

2.1 High-Level Architecture

┌─────────────────────────────────────────────────────────────────┐
│                    CODITECT Platform                            │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────────┐  ┌─────────────────┐  ┌────────────────┐  │
│  │ Google Meet     │  │ Zoom            │  │ Google Mail    │  │
│  │ Integration     │  │ Integration     │  │ Integration    │  │
│  └────────┬────────┘  └────────┬────────┘  └───────┬────────┘  │
│           │                    │                   │            │
│           └────────────────────┼───────────────────┘            │
│                                ▼                                │
│                    ┌───────────────────────┐                    │
│                    │   integration-core    │                    │
│                    │   (Orchestration)     │                    │
│                    └───────────┬───────────┘                    │
│                                │                                │
│           ┌────────────────────┼────────────────────┐           │
│           ▼                    ▼                    ▼           │
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐  │
│  │ Email Service   │  │ Template Engine │  │ AI Pipeline     │  │
│  │ (Send/Receive)  │  │ (Notifications) │  │ (Analysis)      │  │
│  └─────────────────┘  └─────────────────┘  └─────────────────┘  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘
                                 │
                                 ▼
                    ┌───────────────────────┐
                    │     Gmail API         │
                    │  (Google Workspace)   │
                    └───────────────────────┘

2.2 Component Architecture

Google-Mail/
├── src/
│   ├── __init__.py
│   ├── provider.py              # GmailProvider implementation
│   ├── auth/
│   │   ├── __init__.py
│   │   ├── oauth.py             # OAuth 2.0 manager
│   │   └── service_account.py   # Service account auth
│   ├── api/
│   │   ├── __init__.py
│   │   ├── client.py            # Gmail API client
│   │   ├── messages.py          # Message operations
│   │   ├── threads.py           # Thread operations
│   │   ├── labels.py            # Label management
│   │   └── drafts.py            # Draft operations
│   ├── send/
│   │   ├── __init__.py
│   │   ├── composer.py          # Email composition
│   │   ├── templates.py         # Template engine
│   │   └── attachments.py       # Attachment handling
│   ├── receive/
│   │   ├── __init__.py
│   │   ├── watcher.py           # Inbox watcher (Pub/Sub)
│   │   ├── parser.py            # Email parser
│   │   └── filters.py           # Message filters
│   ├── models/
│   │   ├── __init__.py
│   │   ├── message.py           # Email message model
│   │   ├── thread.py            # Thread model
│   │   ├── attachment.py        # Attachment model
│   │   └── label.py             # Label model
│   └── utils/
│       ├── __init__.py
│       ├── mime.py              # MIME utilities
│       └── rate_limiter.py      # Rate limiting
├── tests/
│   ├── __init__.py
│   ├── test_auth.py
│   ├── test_send.py
│   ├── test_receive.py
│   └── test_provider.py
├── config/
│   └── default.yaml
└── docs/
    ├── sdd-software-design-document.md
    ├── tdd-technical-design-document.md
    ├── project-plan.md
    └── adrs/

2.3 Integration with integration-core

# New modules in integration-core
coditect_integrations/
├── email/
│   ├── __init__.py
│   ├── interface.py         # EmailProviderInterface
│   ├── orchestrator.py      # EmailOrchestrator
│   └── exceptions.py        # Email exceptions
├── notifications/
│   ├── __init__.py
│   ├── service.py           # NotificationService
│   ├── channels.py          # Channel implementations
│   └── templates.py         # Template management
└── models/
    ├── email.py             # Email, EmailThread, Attachment
    └── notification.py      # Notification, NotificationChannel

---

3. Core Components

3.1 EmailProviderInterface

Abstract interface for email provider implementations.

from abc import ABC, abstractmethod
from typing import List, Optional
from datetime import datetime

class EmailProviderInterface(ABC):
    """Abstract interface for email providers."""

    @property
    @abstractmethod
    def provider_name(self) -> str:
        """Return provider identifier."""
        pass

    @abstractmethod
    async def authenticate(self, credentials: Dict[str, Any]) -> bool:
        """Authenticate with the email provider."""
        pass

    @abstractmethod
    async def send_email(
        self,
        to: List[str],
        subject: str,
        body: str,
        cc: List[str] = None,
        bcc: List[str] = None,
        attachments: List[Attachment] = None,
        reply_to: str = None,
        html_body: str = None,
    ) -> EmailMessage:
        """Send an email message."""
        pass

    @abstractmethod
    async def get_message(self, message_id: str) -> Optional[EmailMessage]:
        """Retrieve a specific email message."""
        pass

    @abstractmethod
    async def list_messages(
        self,
        query: str = None,
        label_ids: List[str] = None,
        max_results: int = 100,
        page_token: str = None,
    ) -> Tuple[List[EmailMessage], Optional[str]]:
        """List email messages with optional filtering."""
        pass

    @abstractmethod
    async def get_thread(self, thread_id: str) -> Optional[EmailThread]:
        """Retrieve an email thread."""
        pass

    @abstractmethod
    async def search(
        self,
        query: str,
        max_results: int = 100,
    ) -> List[EmailMessage]:
        """Search emails using provider-specific query syntax."""
        pass

    @abstractmethod
    async def watch_inbox(
        self,
        callback: Callable[[EmailMessage], Awaitable[None]],
        label_ids: List[str] = None,
    ) -> str:
        """Start watching inbox for new messages. Returns watch ID."""
        pass

    @abstractmethod
    async def stop_watch(self, watch_id: str) -> bool:
        """Stop watching inbox."""
        pass

    @abstractmethod
    async def add_label(self, message_id: str, label_id: str) -> bool:
        """Add a label to a message."""
        pass

    @abstractmethod
    async def remove_label(self, message_id: str, label_id: str) -> bool:
        """Remove a label from a message."""
        pass

3.2 GmailProvider

Concrete implementation of EmailProviderInterface for Gmail.

class GmailProvider(EmailProviderInterface):
    """Gmail API implementation of EmailProviderInterface."""

    def __init__(
        self,
        auth_manager: GmailOAuthManager,
        rate_limiter: GmailRateLimiter = None,
    ):
        self._auth_manager = auth_manager
        self._rate_limiter = rate_limiter or GmailRateLimiter()
        self._service = None
        self._watch_id = None

    @property
    def provider_name(self) -> str:
        return "gmail"

    async def authenticate(self, credentials: Dict[str, Any]) -> bool:
        """Authenticate and build Gmail service."""
        creds = await self._auth_manager.get_credentials(credentials)
        self._service = build('gmail', 'v1', credentials=creds)
        return True

    async def send_email(
        self,
        to: List[str],
        subject: str,
        body: str,
        cc: List[str] = None,
        bcc: List[str] = None,
        attachments: List[Attachment] = None,
        reply_to: str = None,
        html_body: str = None,
    ) -> EmailMessage:
        """Send email via Gmail API."""
        await self._rate_limiter.acquire("send")

        # Build MIME message
        message = self._build_mime_message(
            to=to,
            subject=subject,
            body=body,
            cc=cc,
            bcc=bcc,
            attachments=attachments,
            reply_to=reply_to,
            html_body=html_body,
        )

        # Encode and send
        raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
        result = self._service.users().messages().send(
            userId='me',
            body={'raw': raw}
        ).execute()

        return EmailMessage(
            id=result['id'],
            thread_id=result['threadId'],
            to=to,
            subject=subject,
            body=body,
            sent_at=datetime.utcnow(),
        )

    async def watch_inbox(
        self,
        callback: Callable[[EmailMessage], Awaitable[None]],
        label_ids: List[str] = None,
    ) -> str:
        """Start Pub/Sub watch on inbox."""
        request = {
            'labelIds': label_ids or ['INBOX'],
            'topicName': f'projects/{self._project_id}/topics/gmail-push',
        }

        result = self._service.users().watch(
            userId='me',
            body=request
        ).execute()

        self._watch_id = result['historyId']
        self._watch_callback = callback

        return self._watch_id

3.3 EmailOrchestrator

Orchestrates email operations across providers and integrates with meeting systems.

class EmailOrchestrator:
    """Orchestrates email operations for CODITECT platform."""

    def __init__(
        self,
        email_provider: EmailProviderInterface,
        template_engine: TemplateEngine = None,
        ai_service: AIService = None,
    ):
        self.email_provider = email_provider
        self.template_engine = template_engine or TemplateEngine()
        self.ai_service = ai_service

    async def send_meeting_invitation(
        self,
        meeting: ScheduledMeeting,
        attendees: List[Attendee],
        template_name: str = "meeting_invitation",
    ) -> List[EmailMessage]:
        """Send meeting invitation emails to attendees."""
        sent_messages = []

        template_data = {
            "meeting_title": meeting.title,
            "start_time": meeting.start_time.isoformat(),
            "duration": meeting.duration_minutes,
            "join_url": meeting.join_url,
            "organizer": meeting.organizer.name,
        }

        html_body = self.template_engine.render(template_name, template_data)

        for attendee in attendees:
            message = await self.email_provider.send_email(
                to=[attendee.email],
                subject=f"Invitation: {meeting.title}",
                body=self._html_to_text(html_body),
                html_body=html_body,
            )
            sent_messages.append(message)

        return sent_messages

    async def send_meeting_reminder(
        self,
        meeting: ScheduledMeeting,
        attendee: Attendee,
        minutes_before: int,
    ) -> EmailMessage:
        """Send meeting reminder email."""
        template_data = {
            "meeting_title": meeting.title,
            "start_time": meeting.start_time.isoformat(),
            "minutes_until": minutes_before,
            "join_url": meeting.join_url,
        }

        html_body = self.template_engine.render("meeting_reminder", template_data)

        return await self.email_provider.send_email(
            to=[attendee.email],
            subject=f"Reminder: {meeting.title} starts in {minutes_before} minutes",
            body=self._html_to_text(html_body),
            html_body=html_body,
        )

    async def analyze_email_thread(
        self,
        thread_id: str,
    ) -> EmailAnalysis:
        """Analyze email thread with AI for action items and sentiment."""
        thread = await self.email_provider.get_thread(thread_id)

        if not self.ai_service:
            raise RuntimeError("AI service not configured")

        # Extract text from all messages in thread
        full_text = "\n\n---\n\n".join(
            f"From: {msg.sender}\nDate: {msg.sent_at}\n\n{msg.body}"
            for msg in thread.messages
        )

        analysis = await self.ai_service.analyze(
            prompt=f"Analyze this email thread and extract:\n1. Summary\n2. Action items\n3. Key decisions\n4. Sentiment\n\nThread:\n{full_text}",
        )

        return EmailAnalysis(
            thread_id=thread_id,
            summary=analysis.get("summary"),
            action_items=analysis.get("action_items", []),
            decisions=analysis.get("decisions", []),
            sentiment=analysis.get("sentiment"),
        )

3.4 Template Engine

Manages email templates for various notification types.

class TemplateEngine:
    """Email template engine with Jinja2 support."""

    def __init__(self, template_dir: str = None):
        self.template_dir = template_dir or Path(__file__).parent / "templates"
        self.env = Environment(
            loader=FileSystemLoader(self.template_dir),
            autoescape=select_autoescape(['html', 'xml']),
        )

    def render(self, template_name: str, data: Dict[str, Any]) -> str:
        """Render a template with the given data."""
        template = self.env.get_template(f"{template_name}.html")
        return template.render(**data)

    def register_template(self, name: str, content: str) -> None:
        """Register a custom template."""
        self.env.globals[name] = content

3.5 Inbox Watcher

Monitors inbox for new messages using Pub/Sub push notifications.

class GmailInboxWatcher:
    """Watches Gmail inbox for new messages via Pub/Sub."""

    def __init__(
        self,
        gmail_provider: GmailProvider,
        pubsub_topic: str,
        pubsub_subscription: str,
    ):
        self.gmail_provider = gmail_provider
        self.pubsub_topic = pubsub_topic
        self.pubsub_subscription = pubsub_subscription
        self._handlers: Dict[str, Callable] = {}
        self._running = False

    async def start(self) -> None:
        """Start watching inbox."""
        # Set up Gmail push notifications
        await self.gmail_provider.watch_inbox(
            callback=self._handle_notification,
            label_ids=['INBOX'],
        )
        self._running = True

        # Start Pub/Sub subscriber
        await self._start_pubsub_listener()

    async def stop(self) -> None:
        """Stop watching inbox."""
        self._running = False
        await self.gmail_provider.stop_watch(self._watch_id)

    def on_message(self, filter_func: Callable[[EmailMessage], bool] = None):
        """Decorator to register message handler."""
        def decorator(handler: Callable[[EmailMessage], Awaitable[None]]):
            handler_id = str(uuid.uuid4())
            self._handlers[handler_id] = (filter_func, handler)
            return handler
        return decorator

    async def _handle_notification(self, notification: Dict) -> None:
        """Handle Pub/Sub notification for new message."""
        history_id = notification.get('historyId')
        messages = await self._get_new_messages(history_id)

        for message in messages:
            for filter_func, handler in self._handlers.values():
                if filter_func is None or filter_func(message):
                    await handler(message)

---

4. Data Models

4.1 EmailMessage

@dataclass
class EmailMessage:
    """Represents an email message."""
    id: str
    thread_id: str
    subject: str
    body: str
    html_body: Optional[str] = None
    sender: Optional[str] = None
    to: List[str] = field(default_factory=list)
    cc: List[str] = field(default_factory=list)
    bcc: List[str] = field(default_factory=list)
    reply_to: Optional[str] = None
    attachments: List[Attachment] = field(default_factory=list)
    labels: List[str] = field(default_factory=list)
    sent_at: Optional[datetime] = None
    received_at: Optional[datetime] = None
    is_read: bool = False
    is_starred: bool = False
    snippet: Optional[str] = None
    headers: Dict[str, str] = field(default_factory=dict)

4.2 EmailThread

@dataclass
class EmailThread:
    """Represents an email thread/conversation."""
    id: str
    subject: str
    messages: List[EmailMessage] = field(default_factory=list)
    participants: List[str] = field(default_factory=list)
    labels: List[str] = field(default_factory=list)
    snippet: Optional[str] = None
    last_message_at: Optional[datetime] = None
    message_count: int = 0

4.3 Attachment

@dataclass
class Attachment:
    """Represents an email attachment."""
    id: Optional[str] = None
    filename: str = ""
    mime_type: str = ""
    size: int = 0
    data: Optional[bytes] = None
    content_id: Optional[str] = None  # For inline attachments

4.4 EmailAnalysis

@dataclass
class EmailAnalysis:
    """AI-generated analysis of an email or thread."""
    thread_id: str
    summary: Optional[str] = None
    action_items: List[str] = field(default_factory=list)
    decisions: List[str] = field(default_factory=list)
    sentiment: Optional[str] = None  # positive, negative, neutral
    key_topics: List[str] = field(default_factory=list)
    urgency: Optional[str] = None  # low, medium, high
    suggested_response: Optional[str] = None

---

5. Authentication

5.1 OAuth 2.0 Flow

class GmailOAuthManager:
    """Manages Gmail OAuth 2.0 authentication."""

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

    def __init__(
        self,
        client_id: str,
        client_secret: str,
        redirect_uri: str,
        token_storage: TokenStorage = None,
    ):
        self.client_id = client_id
        self.client_secret = client_secret
        self.redirect_uri = redirect_uri
        self.token_storage = token_storage or FileTokenStorage()

    def get_authorization_url(self, state: str = None) -> str:
        """Generate OAuth authorization URL."""
        flow = InstalledAppFlow.from_client_config(
            {
                "installed": {
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                    "redirect_uris": [self.redirect_uri],
                    "auth_uri": "https://accounts.google.com/o/oauth2/auth",
                    "token_uri": "https://oauth2.googleapis.com/token",
                }
            },
            scopes=self.SCOPES,
        )
        auth_url, _ = flow.authorization_url(
            access_type='offline',
            include_granted_scopes='true',
            state=state,
        )
        return auth_url

    async def exchange_code(self, code: str) -> Credentials:
        """Exchange authorization code for tokens."""
        flow = InstalledAppFlow.from_client_config(...)
        flow.fetch_token(code=code)
        return flow.credentials

    async def get_credentials(self, user_id: str) -> Credentials:
        """Get valid credentials, refreshing if necessary."""
        creds = await self.token_storage.load(user_id)

        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
            await self.token_storage.save(user_id, creds)

        return creds

5.2 Service Account Authentication

class GmailServiceAccountAuth:
    """Service account authentication with domain-wide delegation."""

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

    def __init__(
        self,
        service_account_file: str,
        delegated_user: str = None,
    ):
        self.service_account_file = service_account_file
        self.delegated_user = delegated_user

    async def get_credentials(self) -> Credentials:
        """Get service account credentials with delegation."""
        creds = service_account.Credentials.from_service_account_file(
            self.service_account_file,
            scopes=self.SCOPES,
        )

        if self.delegated_user:
            creds = creds.with_subject(self.delegated_user)

        return creds

---

6. Rate Limiting

6.1 Gmail API Quotas

Operation	Quota	Rate
Queries per day	1,000,000,000	Per project
Queries per 100 seconds	25,000	Per project
Queries per 100 seconds per user	250	Per user
Send per day	2,000	Per user (consumer)
Send per day	10,000	Per user (Workspace)

6.2 Rate Limiter Implementation

class GmailRateLimiter:
    """Token bucket rate limiter for Gmail API."""

    LIMITS = {
        "default": {"rate": 250, "per": 100},  # 250 per 100 seconds per user
        "send": {"rate": 100, "per": 100},     # Conservative send limit
        "batch": {"rate": 50, "per": 100},     # Batch operations
    }

    def __init__(self):
        self._buckets: Dict[str, TokenBucket] = {}

    async def acquire(self, operation: str = "default") -> None:
        """Acquire a token for the given operation type."""
        if operation not in self._buckets:
            limits = self.LIMITS.get(operation, self.LIMITS["default"])
            self._buckets[operation] = TokenBucket(
                rate=limits["rate"],
                per=limits["per"],
            )

        await self._buckets[operation].acquire()

---

7. Integration Points

7.1 Meeting Provider Integration

# In integration-core EmailOrchestrator

async def configure_meeting_notifications(
    self,
    meeting_provider: MeetingProviderInterface,
) -> None:
    """Configure email notifications for a meeting provider."""

    # Register notification handlers
    self._meeting_provider = meeting_provider

    # Set up templates for this provider
    provider_name = meeting_provider.provider_name
    self.template_engine.register_template(
        f"{provider_name}_invitation",
        self._load_provider_template(provider_name, "invitation"),
    )

7.2 AI Pipeline Integration

async def process_meeting_followup_emails(
    self,
    meeting_id: str,
) -> List[EmailAnalysis]:
    """Find and analyze follow-up emails for a meeting."""

    # Search for emails related to meeting
    meeting = await self._meeting_provider.get_meeting(meeting_id)
    query = f"subject:{meeting.title} after:{meeting.end_time.date()}"

    messages = await self.email_provider.search(query)

    analyses = []
    for message in messages:
        if message.thread_id:
            analysis = await self.analyze_email_thread(message.thread_id)
            analyses.append(analysis)

    return analyses

---

8. Security Considerations

8.1 Data Handling

• Minimal Retention: Email content not stored permanently
• Encryption: All stored tokens encrypted at rest (AES-256)
• Scopes: Request minimum required OAuth scopes
• Audit Logging: Log all email operations for compliance

8.2 Privacy

• User Consent: Explicit consent required before inbox access
• Data Access: Only access emails necessary for requested operation
• No Bulk Export: Prevent mass email extraction
• Revocation: Support immediate OAuth token revocation

8.3 Compliance

• GDPR: Right to erasure supported
• SOC 2: Audit logging for all operations
• Google API Terms: Comply with Gmail API terms of service

---

9. Error Handling

9.1 Exception Hierarchy

class EmailIntegrationError(Exception):
    """Base exception for email integration errors."""
    pass

class EmailAuthenticationError(EmailIntegrationError):
    """Authentication failed or token expired."""
    pass

class EmailRateLimitError(EmailIntegrationError):
    """Rate limit exceeded."""
    def __init__(self, retry_after: int = None):
        self.retry_after = retry_after

class EmailNotFoundError(EmailIntegrationError):
    """Email or thread not found."""
    pass

class EmailSendError(EmailIntegrationError):
    """Failed to send email."""
    pass

class EmailPermissionError(EmailIntegrationError):
    """Insufficient permissions for operation."""
    pass

---

10. Configuration

10.1 Default Configuration

# config/default.yaml
gmail:
  api_version: "v1"

  auth:
    type: "oauth2"  # oauth2 or service_account
    scopes:
      - "https://www.googleapis.com/auth/gmail.send"
      - "https://www.googleapis.com/auth/gmail.readonly"
      - "https://www.googleapis.com/auth/gmail.modify"

  rate_limiting:
    enabled: true
    queries_per_100_seconds: 250
    sends_per_day: 2000

  push_notifications:
    enabled: true
    pubsub_topic: "projects/{project}/topics/gmail-push"
    pubsub_subscription: "projects/{project}/subscriptions/gmail-push-sub"

  templates:
    directory: "templates/"
    cache_enabled: true

  retry:
    max_attempts: 3
    backoff_factor: 2
    max_delay: 60

---

11. Appendices

11.1 Gmail API Endpoints Used

Endpoint	Method	Purpose
/gmail/v1/users/{userId}/messages/send	POST	Send email
/gmail/v1/users/{userId}/messages	GET	List messages
/gmail/v1/users/{userId}/messages/{id}	GET	Get message
/gmail/v1/users/{userId}/threads/{id}	GET	Get thread
/gmail/v1/users/{userId}/messages/{id}/modify	POST	Modify labels
/gmail/v1/users/{userId}/watch	POST	Start push notifications
/gmail/v1/users/{userId}/stop	POST	Stop push notifications
/gmail/v1/users/{userId}/labels	GET	List labels

11.2 Required OAuth Scopes

Scope	Purpose
gmail.send	Send emails
gmail.readonly	Read emails and threads
gmail.modify	Modify labels and message state
gmail.labels	Manage labels

---

Document Control:

• Created: December 17, 2025
• Author: CODITECT Engineering Team
• Review Status: Draft
• Next Review: Upon implementation start