Cómo construir un agente Python para Gmail con Arcade.dev (MCP)

Esta guía muestra cómo construir un agente Python para Gmail usando la implementación del Model Context Protocol (MCP) de Arcade. Implementarás autenticación OAuth, ejecutarás operaciones en Gmail y manejarás escenarios con múltiples usuarios.

Prerequisitos

Requeridos:

• Cuenta en Arcade.dev con API key
• Python 3.10 o superior
• Entorno de desarrollo (VS Code, PyCharm o similar)

Opcionales:

• Proyecto en Google Cloud Console para credenciales OAuth personalizadas
• Familiaridad con patrones async/await en Python

Lo que vas a construir

Un agente Gmail listo para producción que:

• Autentica usuarios con OAuth 2.0
• Envía y recibe correos
• Busca y filtra mensajes
• Gestiona borradores e hilos
• Maneja múltiples usuarios de forma concurrente

Configuración

Instalar el SDK de Arcade

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

# Install Arcade Python client
pip install arcadepy

Configurar la API Key

export ARCADE_API_KEY="your_arcade_api_key"

Verificar la instalación

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)}")

Toolkit de Gmail en Arcade

El Gmail MCP Server ofrece estas herramientas:

• Gmail.SendEmail - Enviar correos
• Gmail.SendDraftEmail - Enviar borradores
• Gmail.ListEmails - Listar correos de bandeja de entrada
• Gmail.SearchEmails - Buscar correos
• Gmail.SearchThreads - Buscar hilos
• Gmail.ReplyToEmail - Responder correos
• Gmail.WriteDraftEmail - Crear borradores
• Gmail.UpdateDraftEmail - Actualizar borradores
• Gmail.DeleteDraftEmail - Eliminar borradores

Implementación básica del agente

Clase principal del agente

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}

Enviar correo

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"
    }

Listar correos

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", []))
    }

Buscar correos

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", [])
    }

Flujo de autenticación

Arcade usa OAuth 2.0 para autenticar usuarios. El flujo:

1. Solicitar autorización para una herramienta
2. Recibir URL de autorización si el usuario no está autenticado
3. El usuario completa el consentimiento OAuth
4. Arcade almacena los tokens OAuth de forma segura
5. Ejecutar herramientas en nombre del usuario

Soporte multiusuario

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

Integración con aplicaciones web

Ejemplo con 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")
    })

Integración con 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")]
})

Más detalles: Uso de Arcade con 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
)

Más detalles: Uso de Arcade con 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
    )

Más detalles: Uso de Arcade con Google ADK

Operaciones avanzadas

Filtros de búsqueda de correos

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)

Gestión de borradores

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")
    }

Operaciones con hilos

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")
    }

Pruebas

Pruebas unitarias

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"

Pruebas de integración

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

Pruebas con Arcade CLI

Usa el Arcade CLI para pruebas interactivas:

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

Despliegue en producción

Arcade Engine autoalojado

Instalar Arcade Engine:

# macOS
brew install arcadeai/tap/arcade-engine

# Ubuntu/Debian
sudo apt install arcade-engine

Configurar con credenciales OAuth personalizadas:

# 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

Configuración de OAuth con Google

Configurar proveedor de autenticación de Google:

1. Crear proyecto en Google Cloud Console
2. Habilitar la API de Gmail
3. Crear credenciales OAuth 2.0
4. Agregar URIs de redirección autorizadas
5. Configurar en Arcade Engine

Límite de solicitudes

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")

Manejo de errores

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)}")

Monitoreo

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

Ejemplo completo en producción

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')}")

Siguientes pasos

• Amplía con otros toolkits de Arcade (Calendar, Drive, Slack)
• Crea herramientas personalizadas con Arcade Tool SDK
• Despliega Arcade Engine autoalojado
• Integra con LangChain, CrewAI o OpenAI Agents
• Prueba Google ADK o Vercel AI SDK

Recursos

• Documentación de Arcade
• Arcade en GitHub
• Referencia del toolkit de Gmail
• Referencia de la API
• Blog de Arcade
• Ejemplo de agente Gmail