AI Gmail Labeller

1
def create_trigger(
2
    user_id: str,
3
    connection_request: ConnectionRequest,
4
    trigger_config: dict | None = None,
5
):
6
    """
7
    Create a Gmail trigger for new messages.
8
9
    Args:
10
        user_id: The user ID to create the trigger for
11
        trigger_config: Optional trigger configuration, defaults to monitoring INBOX
12
13
    Returns:
14
        The trigger creation response or None if failed
15
    """
16
    connection_request.wait_for_connection(timeout=30)
17
    if trigger_config is None:
18
        trigger_config = {
19
            "interval": 1,  # Check every minute
20
            "labelids": "INBOX",  # Monitor inbox only
21
            "userId": "me",  # Current authenticated user
22
        }

1
        response = composio.triggers.create(
2
            user_id=user_id,
3
            slug="GMAIL_NEW_GMAIL_MESSAGE",
4
            trigger_config=trigger_config,
5
        )
6
7
        logger.info(f"Successfully created trigger for user {user_id}: {response}")

1
@app.post("/composio/webhook")
2
async def listen_webhooks(
3
    request: Request, 
4
    background_tasks: BackgroundTasks
5
):
6
    """
7
    Main webhook endpoint for processing Gmail 
8
    new message events. This is the core webhook 
9
    labeling functionality.
10
    """
11
    # Verify webhook signature
12
    # Check if this is a Gmail new message event
13
    if (
14
        webhook_data.get("type") == "gmail_new_message"
15
        or "gmail" in webhook_data.get("type", "").lower()
16
    ):
17
        gmail_message = GmailMessage.from_composio_payload(
18
            webhook_data
19
        )
20
        # Get the user's custom prompt or use default
21
        default_prompt = (
22
            initial_prompt or "Default email processing prompt"
23
        )
24
        user_prompt = get_user_prompt(
25
            gmail_message.user_id, default_prompt
26
        )
27
28
        # Add email processing to background tasks
29
        background_tasks.add_task(
30
            process_gmail_message, 
31
            gmail_message, 
32
            user_prompt
33
        )
34
35
        logger.info(
36
            f"Queued email {gmail_message.id} for processing"
37
        )
38
    return {"status": "received", "webhook_id": webhook_id}

1
async def process_gmail_message(message: GmailMessage, user_filter: str):
2
    """
3
    Process Gmail message with AI and apply labels.
4
5
    Args:
6
        message: The Gmail message to process
7
        user_filter: The user's custom prompt/filter for email categorization
8
9
    Returns:
10
        bool: True if processing succeeded, False otherwise
11
    """

1
        body_preview = (message.text_body or message.html_body or "")[:10000]
2
        if body_preview:
3
            logger.debug(f"Email body preview: {body_preview}...")
4
        else:

1
        tools = composio.tools.get(
2
            message.user_id,
3
            tools=[
4
                "GMAIL_ADD_LABEL_TO_EMAIL",
5
                "GMAIL_MODIFY_THREAD_LABELS",
6
                "GMAIL_PATCH_LABEL",
7
                "GMAIL_REMOVE_LABEL",
8
                "GMAIL_CREATE_LABEL",
9
                "GMAIL_LIST_LABELS",
10
            ],
11
        )

1
        reaper = Agent(
2
            name="Gmail Reaper",
3
            instructions=REAPER_SYSTEM_PROMPT,
4
            tools=tools,
5
        )
6
7
        # Prepare the prompt with user filter and email details
8
        email_content = message.text_body or message.html_body or "No content available"
9
        prompt_details = (
10
            f"{user_filter}\n\n## Email\n{email_content}\n## Message ID\n{message.id}"
11
        )
12
13
        logger.debug(
14
            f"Running agent with prompt length: {len(prompt_details)} characters"
15
        )
16
17
        # Run the agent to process and label the email
18
        result = await Runner.run(reaper, prompt_details)

1
async def verify_webhook_signature(request: Request) -> tuple[bytes, str]:
2
    """
3
    Verify the authenticity of a Composio webhook request.
4
    
5
    Args:
6
        request: The FastAPI request object
7
        
8
    Returns:
9
        tuple: (body bytes, webhook_id)
10
        
11
    Raises:
12
        HTTPException: If signature verification fails
13
    """
14
    # Get the raw body for signature verification
15
    body = await request.body()
16
    
17
    # Get the signature and timestamp from headers
18
    signature_header = request.headers.get("webhook-signature", "")
19
    timestamp = request.headers.get("webhook-timestamp", "")
20
    webhook_id = request.headers.get("webhook-id", "")
21
    
22
    # Verify webhook authenticity
23
    if COMPOSIO_WEBHOOK_SECRET and signature_header:
24
        # Extract the signature (format: "v1,signature")
25
        if "," in signature_header:
26
            version, signature = signature_header.split(",", 1)
27
        else:
28
            raise HTTPException(status_code=401, detail="Invalid signature format")
29
        
30
        # Create the signed content (webhook_id.timestamp.body)
31
        signed_content = f"{webhook_id}.{timestamp}.{body.decode('utf-8')}"
32
        
33
        # Generate expected signature
34
        expected_signature = hmac.new(
35
            COMPOSIO_WEBHOOK_SECRET.encode(), 
36
            signed_content.encode(), 
37
            hashlib.sha256
38
        ).digest()
39
        
40
        # Encode to base64
41
        expected_signature_b64 = base64.b64encode(expected_signature).decode()
42
        
43
        # Compare signatures
44
        if not hmac.compare_digest(signature, expected_signature_b64):
45
            raise HTTPException(status_code=401, detail="Invalid webhook signature")
46
    
47
    return body, webhook_id

1
@app.post("/composio/webhook")
2
async def listen_webhooks(request: Request, background_tasks: BackgroundTasks):
3
    """
4
    Main webhook endpoint for processing Gmail new message events.
5
    This is the core webhook labeling functionality.
6
    """
7
    # Verify webhook signature
8
    body, webhook_id = await verify_webhook_signature(request)