Ce guide vous montre comment créer un agent Gmail Python avec l’implémentation du Model Context Protocol (MCP) d’Arcade. Vous mettrez en place l’authentification OAuth, exécuterez des opérations Gmail et gérerez des scénarios multi-utilisateurs.

Prérequis

Obligatoires :

Optionnels :

  • Projet Google Cloud Console pour des identifiants OAuth personnalisés
  • Connaissance des patterns async/await en Python

Ce que vous allez créer

Un agent Gmail prêt pour la production qui :

  • Authentifie les utilisateurs via OAuth 2.0
  • Envoie et reçoit des e-mails
  • Recherche et filtre les messages
  • Gère les brouillons et les fils de discussion
  • Gère plusieurs utilisateurs simultanément

Configuration

Installer le SDK Arcade

# Create virtual environment
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate

# Install Arcade Python client
pip install arcadepy

Configurer la clé API

export ARCADE_API_KEY="your_arcade_api_key"

Vérifier l’installation

from arcadepy import Arcade

client = Arcade()
user_id = "test@example.com"

# List available Gmail tools
response = client.tools.list(toolkit="gmail", limit=10, user_id=user_id)
print(f"Available Gmail tools: {len(response.items)}")

Boîte à outils Gmail Arcade

Le Gmail MCP Server fournit ces outils :

  • Gmail.SendEmail - Envoyer des e-mails
  • Gmail.SendDraftEmail - Envoyer des brouillons
  • Gmail.ListEmails - Lister les e-mails de la boîte de réception
  • Gmail.SearchEmails - Rechercher des e-mails
  • Gmail.SearchThreads - Rechercher des fils de discussion
  • Gmail.ReplyToEmail - Répondre à des e-mails
  • Gmail.WriteDraftEmail - Créer des brouillons
  • Gmail.UpdateDraftEmail - Modifier des brouillons
  • Gmail.DeleteDraftEmail - Supprimer des brouillons

Implémentation d’un agent basique

Classe d’agent principale

import os
from typing import Dict, Any, List
from arcadepy import Arcade

class GmailAgent:
    def __init__(self, api_key: str = None):
        self.arcade = Arcade(api_key=api_key or os.getenv("ARCADE_API_KEY"))
        self.user_sessions: Dict[str, bool] = {}

    def authenticate_user(self, user_id: str) -> Dict[str, Any]:
        """Authenticate user for Gmail access."""
        auth_response = self.arcade.tools.authorize(
            tool_name="Gmail.SendEmail",
            user_id=user_id
        )

        if auth_response.status != "completed":
            return {
                "requires_auth": True,
                "auth_url": auth_response.url,
                "message": f"Visit {auth_response.url} to authorize Gmail access"
            }

        self.arcade.auth.wait_for_completion(auth_response)
        self.user_sessions[user_id] = True

        return {"authenticated": True}

Envoyer un e-mail

def send_email(
    self,
    user_id: str,
    to: str,
    subject: str,
    body: str,
    cc: List[str] = None,
    bcc: List[str] = None
) -> Dict[str, Any]:
    """Send email via Gmail."""
    if user_id not in self.user_sessions:
        auth_result = self.authenticate_user(user_id)
        if auth_result.get("requires_auth"):
            return auth_result

    email_params = {
        "to": to,
        "subject": subject,
        "body": body
    }

    if cc:
        email_params["cc"] = cc
    if bcc:
        email_params["bcc"] = bcc

    response = self.arcade.tools.execute(
        tool_name="Gmail.SendEmail",
        input=email_params,
        user_id=user_id
    )

    return {
        "success": True,
        "message_id": response.output.get("message_id"),
        "status": "sent"
    }

Lister les e-mails

def list_emails(
    self,
    user_id: str,
    max_results: int = 10,
    query: str = None
) -> Dict[str, Any]:
    """List emails from inbox."""
    if user_id not in self.user_sessions:
        auth_result = self.authenticate_user(user_id)
        if auth_result.get("requires_auth"):
            return auth_result

    params = {"max_results": max_results}
    if query:
        params["query"] = query

    response = self.arcade.tools.execute(
        tool_name="Gmail.ListEmails",
        input=params,
        user_id=user_id
    )

    return {
        "success": True,
        "emails": response.output.get("emails", []),
        "count": len(response.output.get("emails", []))
    }

Rechercher des e-mails

def search_emails(
    self,
    user_id: str,
    query: str,
    max_results: int = 20
) -> Dict[str, Any]:
    """Search emails with query string."""
    if user_id not in self.user_sessions:
        auth_result = self.authenticate_user(user_id)
        if auth_result.get("requires_auth"):
            return auth_result

    response = self.arcade.tools.execute(
        tool_name="Gmail.SearchEmails",
        input={"query": query, "max_results": max_results},
        user_id=user_id
    )

    return {
        "success": True,
        "results": response.output.get("emails", [])
    }

Flux d’authentification

Arcade utilise OAuth 2.0 pour l’authentification des utilisateurs. Le flux :

  1. Demander l’autorisation pour un outil
  2. Recevoir l’URL d’autorisation si l’utilisateur n’est pas authentifié
  3. L’utilisateur complète le consentement OAuth
  4. Arcade stocke les tokens OAuth de façon sécurisée
  5. Exécuter des outils au nom de l’utilisateur

Support multi-utilisateurs

from typing import Dict, Optional
from datetime import datetime
from arcadepy import Arcade

class MultiUserGmailAgent:
    def __init__(self):
        self.arcade = Arcade()
        self.user_sessions: Dict[str, Dict] = {}

    def ensure_authenticated(self, user_id: str, tool_name: str) -> Optional[str]:
        """Check authentication status. Returns auth URL if needed."""
        session = self.user_sessions.get(user_id)
        if session and session.get("authenticated"):
            return None

        auth_response = self.arcade.tools.authorize(
            tool_name=tool_name,
            user_id=user_id
        )

        if auth_response.status != "completed":
            return auth_response.url

        self.user_sessions[user_id] = {
            "authenticated": True,
            "timestamp": datetime.now()
        }

        return None

    def execute_with_auth(
        self,
        user_id: str,
        tool_name: str,
        params: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Execute tool with authentication handling."""
        auth_url = self.ensure_authenticated(user_id, tool_name)
        if auth_url:
            return {
                "success": False,
                "requires_auth": True,
                "auth_url": auth_url
            }

        try:
            response = self.arcade.tools.execute(
                tool_name=tool_name,
                input=params,
                user_id=user_id
            )

            return {
                "success": True,
                "data": response.output
            }
        except Exception as e:
            if "authorization" in str(e).lower():
                self.user_sessions.pop(user_id, None)
                auth_url = self.ensure_authenticated(user_id, tool_name)
                return {
                    "success": False,
                    "requires_auth": True,
                    "auth_url": auth_url
                }
            raise

Intégration dans une application web

Exemple FastAPI

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()
agent = MultiUserGmailAgent()

@app.post("/api/gmail/send")
async def send_email(request: Request):
    data = await request.json()
    user_id = data.get("user_id")

    result = agent.execute_with_auth(
        user_id=user_id,
        tool_name="Gmail.SendEmail",
        params={
            "to": data.get("to"),
            "subject": data.get("subject"),
            "body": data.get("body")
        }
    )

    if result.get("requires_auth"):
        return JSONResponse({
            "status": "auth_required",
            "auth_url": result["auth_url"]
        })

    return JSONResponse({
        "status": "success",
        "message_id": result["data"].get("message_id")
    })

Intégration de frameworks

LangChain

from langchain_arcade import ArcadeToolManager
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

manager = ArcadeToolManager(api_key=os.getenv("ARCADE_API_KEY"))
gmail_tools = manager.get_tools(toolkits=["gmail"])

llm = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(llm, gmail_tools)

result = agent.invoke({
    "messages": [("user", "Send email to john@example.com about 3pm meeting")]
})

Détails complets : Utiliser Arcade avec LangChain

OpenAI Agents

from arcadepy import Arcade
from openai import OpenAI

arcade = Arcade()
openai_client = OpenAI()

tools_response = arcade.tools.list(
    toolkit="gmail",
    user_id="user@example.com"
)

openai_tools = [
    {
        "type": "function",
        "function": {
            "name": tool.name,
            "description": tool.description,
            "parameters": tool.inputs
        }
    }
    for tool in tools_response.items
]

response = openai_client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "List my 5 most recent emails"}
    ],
    tools=openai_tools
)

Détails complets : Utiliser Arcade avec OpenAI Agents

Google ADK

import asyncio
from arcadepy import AsyncArcade
from google.adk import Agent, Runner
from google_adk_arcade.tools import get_arcade_tools

async def main():
    client = AsyncArcade()
    user_id = 'user@example.com'

    google_tools = await get_arcade_tools(client, tools=["Gmail.ListEmails"])

    for tool in google_tools:
        result = await client.tools.authorize(
            tool_name=tool.name,
            user_id=user_id
        )
        if result.status != "completed":
            print(f"Authorize at: {result.url}")
            await client.auth.wait_for_completion(result)

    google_agent = Agent(
        model="gemini-2.0-flash",
        name="gmail_agent",
        instruction="Manage Gmail inbox",
        tools=google_tools
    )

Détails complets : Utiliser Arcade avec Google ADK

Opérations avancées

Filtres de recherche d’e-mails

class AdvancedGmailAgent(GmailAgent):
    def search_unread_from_sender(
        self,
        user_id: str,
        sender_email: str,
        max_results: int = 10
    ) -> Dict[str, Any]:
        """Search unread emails from specific sender."""
        query = f"from:{sender_email} is:unread"
        return self.search_emails(user_id, query, max_results)

    def get_emails_with_attachments(
        self,
        user_id: str,
        days: int = 7,
        max_results: int = 20
    ) -> Dict[str, Any]:
        """Get emails with attachments from last N days."""
        query = f"has:attachment newer_than:{days}d"
        return self.search_emails(user_id, query, max_results)

    def search_important_unread(
        self,
        user_id: str,
        max_results: int = 10
    ) -> Dict[str, Any]:
        """Search important unread emails."""
        query = "is:important is:unread"
        return self.search_emails(user_id, query, max_results)

Gestion des brouillons

def create_draft(
    self,
    user_id: str,
    to: str,
    subject: str,
    body: str
) -> Dict[str, Any]:
    """Create email draft."""
    if user_id not in self.user_sessions:
        auth_result = self.authenticate_user(user_id)
        if auth_result.get("requires_auth"):
            return auth_result

    response = self.arcade.tools.execute(
        tool_name="Gmail.WriteDraftEmail",
        input={
            "to": to,
            "subject": subject,
            "body": body
        },
        user_id=user_id
    )

    return {
        "success": True,
        "draft_id": response.output.get("draft_id")
    }

def send_draft(
    self,
    user_id: str,
    draft_id: str
) -> Dict[str, Any]:
    """Send existing draft."""
    if user_id not in self.user_sessions:
        auth_result = self.authenticate_user(user_id)
        if auth_result.get("requires_auth"):
            return auth_result

    response = self.arcade.tools.execute(
        tool_name="Gmail.SendDraftEmail",
        input={"draft_id": draft_id},
        user_id=user_id
    )

    return {
        "success": True,
        "message_id": response.output.get("message_id")
    }

Opérations sur les fils de discussion

def search_threads(
    self,
    user_id: str,
    query: str,
    max_results: int = 10
) -> Dict[str, Any]:
    """Search email threads."""
    if user_id not in self.user_sessions:
        auth_result = self.authenticate_user(user_id)
        if auth_result.get("requires_auth"):
            return auth_result

    response = self.arcade.tools.execute(
        tool_name="Gmail.SearchThreads",
        input={
            "query": query,
            "max_results": max_results
        },
        user_id=user_id
    )

    return {
        "success": True,
        "threads": response.output.get("threads", [])
    }

def reply_to_email(
    self,
    user_id: str,
    message_id: str,
    body: str,
    reply_all: bool = False
) -> Dict[str, Any]:
    """Reply to email."""
    reply_type = "EVERY_RECIPIENT" if reply_all else "ONLY_THE_SENDER"

    if user_id not in self.user_sessions:
        auth_result = self.authenticate_user(user_id)
        if auth_result.get("requires_auth"):
            return auth_result

    response = self.arcade.tools.execute(
        tool_name="Gmail.ReplyToEmail",
        input={
            "message_id": message_id,
            "body": body,
            "reply_type": reply_type
        },
        user_id=user_id
    )

    return {
        "success": True,
        "message_id": response.output.get("message_id")
    }

Tests

Tests unitaires

import pytest
from unittest.mock import Mock, patch
from gmail_agent import GmailAgent

@pytest.fixture
def agent():
    return GmailAgent(api_key="test_key")

def test_authentication_required(agent):
    """Test authentication flow."""
    with patch.object(agent.arcade.tools, 'authorize') as mock_auth:
        mock_auth.return_value = Mock(
            status="pending",
            url="https://auth.arcade.dev/authorize?token=abc123"
        )

        result = agent.authenticate_user("test@example.com")

        assert result["requires_auth"] is True
        assert "auth_url" in result

def test_send_email_success(agent):
    """Test email sending."""
    agent.user_sessions["test@example.com"] = True

    with patch.object(agent.arcade.tools, 'execute') as mock_execute:
        mock_execute.return_value = Mock(
            output={"message_id": "msg_123"}
        )

        result = agent.send_email(
            user_id="test@example.com",
            to="recipient@example.com",
            subject="Test",
            body="Test message"
        )

        assert result["success"] is True
        assert result["message_id"] == "msg_123"

Tests d’intégration

def test_gmail_integration():
    """Test with actual Arcade API."""
    agent = GmailAgent()
    user_id = "test@example.com"

    auth_result = agent.authenticate_user(user_id)

    if auth_result.get("requires_auth"):
        print(f"Complete authentication at: {auth_result['auth_url']}")
        input("Press Enter after completing authentication...")

    list_result = agent.list_emails(user_id, max_results=5)
    assert list_result["success"] is True
    assert "emails" in list_result

Tests avec l’Arcade CLI

Utilisez l’Arcade CLI pour des tests interactifs :

pip install arcade-ai
arcade chat --user test@example.com

Déploiement en production

Arcade Engine auto-hébergé

Installez l’Arcade Engine :

# macOS
brew install arcadeai/tap/arcade-engine

# Ubuntu/Debian
sudo apt install arcade-engine

Configurer avec des identifiants OAuth personnalisés :

# engine.yaml
auth:
  providers:
    - id: gmail-provider
      description: 'Gmail OAuth provider'
      enabled: true
      type: oauth2
      provider_id: google
      client_id: ${GOOGLE_CLIENT_ID}
      client_secret: ${GOOGLE_CLIENT_SECRET}

api:
  host: 0.0.0.0
  port: 9099

tools:
  directors:
    - id: default
      enabled: true
      max_tools: 64

Configuration Google OAuth

Configurer le fournisseur d’auth Google :

  1. Créer un projet dans Google Cloud Console
  2. Activer l’API Gmail
  3. Créer des identifiants OAuth 2.0
  4. Ajouter les URI de redirection autorisés
  5. Configurer dans Arcade Engine

Limitation de débit

import time
from typing import Any, Callable

def retry_with_backoff(
    func: Callable,
    max_retries: int = 3,
    initial_delay: float = 1.0
) -> Any:
    """Retry with exponential backoff."""
    delay = initial_delay

    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if "rate limit" in str(e).lower() and attempt < max_retries - 1:
                time.sleep(delay)
                delay *= 2
            else:
                raise

    raise Exception(f"Failed after {max_retries} retries")

Gestion des erreurs

class GmailAgentError(Exception):
    """Base exception for Gmail agent."""
    pass

class AuthenticationError(GmailAgentError):
    """Authentication failed or required."""
    pass

class RateLimitError(GmailAgentError):
    """API rate limit exceeded."""
    pass

def safe_execute(
    self,
    user_id: str,
    tool_name: str,
    params: Dict[str, Any]
) -> Dict[str, Any]:
    """Execute tool with error handling."""
    try:
        response = self.arcade.tools.execute(
            tool_name=tool_name,
            input=params,
            user_id=user_id
        )
        return {"success": True, "data": response.output}

    except Exception as e:
        error_msg = str(e).lower()

        if "authorization" in error_msg or "auth" in error_msg:
            raise AuthenticationError(
                f"Authentication required for {tool_name}"
            )
        elif "rate limit" in error_msg or "quota" in error_msg:
            raise RateLimitError(
                "API rate limit exceeded. Retry later."
            )
        else:
            raise GmailAgentError(f"Tool execution failed: {str(e)}")

Supervision

import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)

class ProductionGmailAgent(GmailAgent):
    def send_email(self, user_id: str, to: str, subject: str, body: str) -> Dict[str, Any]:
        """Send email with logging."""
        logger.info(f"Sending email for user {user_id} to {to}")
        start_time = datetime.now()

        try:
            result = super().send_email(user_id, to, subject, body)
            duration = (datetime.now() - start_time).total_seconds()

            logger.info(
                f"Email sent in {duration:.2f}s - "
                f"message_id: {result.get('message_id')}"
            )

            return result
        except Exception as e:
            duration = (datetime.now() - start_time).total_seconds()
            logger.error(
                f"Failed to send email after {duration:.2f}s - "
                f"error: {str(e)}"
            )
            raise

Exemple de production complet

import os
import logging
from typing import Dict, Any, List, Optional
from datetime import datetime
from arcadepy import Arcade

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

class ProductionGmailAgent:
    """Production Gmail agent."""

    def __init__(self, api_key: Optional[str] = None):
        self.arcade = Arcade(api_key=api_key or os.getenv("ARCADE_API_KEY"))
        self.user_sessions: Dict[str, Dict] = {}

    def _ensure_auth(self, user_id: str, tool_name: str) -> Optional[str]:
        """Ensure user authenticated. Returns auth URL if needed."""
        session = self.user_sessions.get(user_id)
        if session and session.get("authenticated"):
            return None

        auth_response = self.arcade.tools.authorize(
            tool_name=tool_name,
            user_id=user_id
        )

        if auth_response.status != "completed":
            logger.warning(f"User {user_id} requires authentication")
            return auth_response.url

        self.user_sessions[user_id] = {
            "authenticated": True,
            "timestamp": datetime.now()
        }
        logger.info(f"User {user_id} authenticated")
        return None

    def _execute_tool(
        self,
        user_id: str,
        tool_name: str,
        params: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Execute tool with auth and error handling."""
        auth_url = self._ensure_auth(user_id, tool_name)
        if auth_url:
            return {
                "success": False,
                "requires_auth": True,
                "auth_url": auth_url
            }

        try:
            logger.info(f"Executing {tool_name} for user {user_id}")
            response = self.arcade.tools.execute(
                tool_name=tool_name,
                input=params,
                user_id=user_id
            )
            return {"success": True, "data": response.output}
        except Exception as e:
            logger.error(f"Tool execution failed: {str(e)}")
            if "authorization" in str(e).lower():
                self.user_sessions.pop(user_id, None)
                auth_url = self._ensure_auth(user_id, tool_name)
                return {
                    "success": False,
                    "requires_auth": True,
                    "auth_url": auth_url
                }
            raise

    def send_email(
        self,
        user_id: str,
        to: str,
        subject: str,
        body: str,
        cc: Optional[List[str]] = None,
        bcc: Optional[List[str]] = None
    ) -> Dict[str, Any]:
        """Send email."""
        params = {"to": to, "subject": subject, "body": body}
        if cc:
            params["cc"] = cc
        if bcc:
            params["bcc"] = bcc

        return self._execute_tool(user_id, "Gmail.SendEmail", params)

    def list_emails(
        self,
        user_id: str,
        max_results: int = 10
    ) -> Dict[str, Any]:
        """List recent emails."""
        return self._execute_tool(
            user_id,
            "Gmail.ListEmails",
            {"max_results": max_results}
        )

    def search_emails(
        self,
        user_id: str,
        query: str,
        max_results: int = 20
    ) -> Dict[str, Any]:
        """Search emails."""
        return self._execute_tool(
            user_id,
            "Gmail.SearchEmails",
            {"query": query, "max_results": max_results}
        )

    def reply_to_email(
        self,
        user_id: str,
        message_id: str,
        body: str,
        reply_all: bool = False
    ) -> Dict[str, Any]:
        """Reply to email."""
        reply_type = "EVERY_RECIPIENT" if reply_all else "ONLY_THE_SENDER"
        return self._execute_tool(
            user_id,
            "Gmail.ReplyToEmail",
            {
                "message_id": message_id,
                "body": body,
                "reply_type": reply_type
            }
        )

# Usage
if __name__ == "__main__":
    agent = ProductionGmailAgent()
    user_id = "user@example.com"

    result = agent.send_email(
        user_id=user_id,
        to="colleague@example.com",
        subject="Project Update",
        body="Project on track for delivery next week."
    )

    if result.get("requires_auth"):
        print(f"Authenticate at: {result['auth_url']}")
    elif result["success"]:
        print(f"Email sent - ID: {result['data'].get('message_id')}")

Prochaines étapes

Ressources