How to Build a LangGraph Gmail Agent with Arcade (MCP)

# Install Arcade Python client and dependencies
pip install arcadepy arcade_tdk langgraph langchain

# Set up environment variables
export ARCADE_API_KEY="your_arcade_api_key"
export GOOGLE_CLIENT_ID="your_google_client_id"
export GOOGLE_CLIENT_SECRET="your_google_client_secret"

from langgraph.graph import StateGraph, END
from arcadepy import Arcade
from typing import TypedDict, Annotated, Sequence
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[str], operator.add]
    user_id: str
    authenticated: bool
    pending_action: dict

class ArcadeGmailAgent:
    def __init__(self):
        self.arcade = Arcade(api_key=os.environ.get("ARCADE_API_KEY"))
        self.graph = self._build_graph()

    def _build_graph(self):
        workflow = StateGraph(AgentState)

        # Add nodes
        workflow.add_node("check_auth", self.check_authentication)
        workflow.add_node("handle_auth", self.handle_authorization)
        workflow.add_node("execute_gmail", self.execute_gmail_action)
        workflow.add_node("process_response", self.process_response)

        # Define edges
        workflow.add_edge("check_auth", "handle_auth")
        workflow.add_conditional_edges(
            "handle_auth",
            self.route_after_auth,
            {
                "authenticated": "execute_gmail",
                "needs_auth": "check_auth"
            }
        )
        workflow.add_edge("execute_gmail", "process_response")
        workflow.add_edge("process_response", END)

        workflow.set_entry_point("check_auth")
        return workflow.compile()

async def check_authentication(self, state: AgentState):
    """Check if user has valid Gmail authorization"""
    user_id = state["user_id"]

    # Check Gmail.SendEmail authorization status
    auth_response = await self.arcade.tools.authorize(
        tool_name="Gmail.SendEmail",
        user_id=user_id
    )

    if auth_response.status != "completed":
        state["authenticated"] = False
        state["pending_action"] = {
            "type": "authorization_required",
            "url": auth_response.url,
            "message": "Complete Gmail authorization to proceed"
        }
    else:
        state["authenticated"] = True

    return state

async def handle_authorization(self, state: AgentState):
    """Handle authorization completion"""
    if not state["authenticated"]:
        # Wait for user to complete OAuth
        return state

    # Load Gmail toolkit for authenticated user
    gmail_tools = await self.arcade.tools.list(
        toolkit="gmail",
        user_id=state["user_id"]
    )
    state["available_tools"] = gmail_tools.items
    return state

async def execute_gmail_action(self, state: AgentState):
    """Execute Gmail operations through Arcade MCP"""
    action = state["pending_action"]
    user_id = state["user_id"]

    # Map action types to Arcade Gmail tools
    tool_mapping = {
        "send_email": "Gmail.SendEmail",
        "create_draft": "Gmail.WriteDraftEmail",
        "search_emails": "Gmail.SearchThreads",
        "list_emails": "Gmail.ListEmails"
    }

    tool_name = tool_mapping.get(action["type"])

    try:
        response = await self.arcade.tools.execute(
            tool_name=tool_name,
            input=action["parameters"],
            user_id=user_id
        )
        state["messages"].append(f"Successfully executed: {response.output}")
    except Exception as e:
        if getattr(e, "type", "") == "authorization_required":
            state["authenticated"] = False
            state["pending_action"]["auth_url"] = getattr(e, "url", "")
        else:
            state["messages"].append(f"Error: {str(e)}")

    return state

class EmailCampaignAgent:
    def __init__(self):
        self.arcade = Arcade(api_key=os.environ.get("ARCADE_API_KEY"))
        self.workflow = self._build_campaign_graph()

    def _build_campaign_graph(self):
        workflow = StateGraph(CampaignState)

        # Campaign workflow nodes
        workflow.add_node("search_contacts", self.search_email_contacts)
        workflow.add_node("draft_emails", self.create_personalized_drafts)
        workflow.add_node("review_drafts", self.review_and_approve)
        workflow.add_node("send_approved", self.send_approved_emails)

        # Conditional routing based on approval
        workflow.add_conditional_edges(
            "review_drafts",
            lambda x: "send" if x["approved"] else "revise",
            {
                "send": "send_approved",
                "revise": "draft_emails"
            }
        )

        return workflow.compile()

    async def search_email_contacts(self, state):
        """Search Gmail for relevant contacts"""
        response = await self.arcade.tools.execute(
            tool_name="Gmail.SearchThreads",
            input={"query": state["search_query"]},
            user_id=state["user_id"]
        )
        state["contacts"] = self._extract_contacts(response.output)
        return state

{
  "mcp": {
    "servers": {
      "arcade-gmail": {
        "url": "https://api.arcade.dev/v1/mcps/arcade-anon/mcp"
      }
    }
  }
}

auth:
  providers:
    - id: gmail-provider
      description: "Gmail OAuth for LangGraph agents"
      enabled: true
      type: oauth2
      provider_id: google
      client_id: ${env:GOOGLE_CLIENT_ID}
      client_secret: ${env:GOOGLE_CLIENT_SECRET}

api:
  development: false
  host: 0.0.0.0
  port: 9099

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

workers:
  - id: "langgraph-gmail-worker"
    enabled: true
    http:
      uri: "http://localhost:8002"
      secret: ${env:WORKER_SECRET}
      timeout: 30s

class MultiUserGmailOrchestrator:
    def __init__(self):
        self.arcade = Arcade(api_key=os.environ.get("ARCADE_API_KEY"))
        self.user_graphs = {}
        self.user_sessions = {}

    async def get_user_graph(self, user_id: str):
        """Get or create user-specific LangGraph instance"""
        if user_id not in self.user_graphs:
            # Create new graph for user
            agent = ArcadeGmailAgent()
            self.user_graphs[user_id] = agent.graph

            # Initialize user session
            self.user_sessions[user_id] = {
                "authenticated": False,
                "tools_loaded": False,
                "last_activity": datetime.now()
            }

        return self.user_graphs[user_id]

    async def process_user_request(self, user_id: str, request: dict):
        """Route request through user-specific graph"""
        graph = await self.get_user_graph(user_id)

        initial_state = {
            "messages": [],
            "user_id": user_id,
            "authenticated": self.user_sessions[user_id]["authenticated"],
            "pending_action": request
        }

        result = await graph.ainvoke(initial_state)

        # Update session state
        self.user_sessions[user_id]["authenticated"] = result["authenticated"]
        self.user_sessions[user_id]["last_activity"] = datetime.now()

        return result

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

app = FastAPI()

@app.post("/api/oauth/callback")
async def oauth_callback(request: Request):
    body = await request.json()
    user_id = body.get("userId")

    # Wait for OAuth completion
    arcade = Arcade(api_key=os.environ.get("ARCADE_API_KEY"))
    auth_result = await arcade.auth.wait_for_completion(
        authorization_id=body.get("authorizationId")
    )

    if auth_result.status == "completed":
        # Update user graph state
        orchestrator = get_orchestrator_instance()
        orchestrator.user_sessions[user_id]["authenticated"] = True

        # Load Gmail tools
        gmail_tools = await arcade.tools.list(
            toolkit="gmail",
            user_id=user_id
        )

        return JSONResponse({
            "success": True,
            "tools_available": len(gmail_tools.items)
        })

    return JSONResponse({"success": False})

class ResilientGmailAgent:
    def __init__(self):
        self.arcade = Arcade(api_key=os.environ.get("ARCADE_API_KEY"))
        self.retry_config = {
            "max_attempts": 3,
            "backoff_factor": 2
        }

    async def handle_gmail_error(self, error, state):
        """Intelligent error recovery for Gmail operations"""
        error_type = getattr(error, 'type', str(type(error).__name__))

        recovery_strategies = {
            "authorization_required": self.trigger_reauth,
            "rate_limit_exceeded": self.apply_backoff,
            "insufficient_scope": self.request_additional_scopes,
            "token_expired": self.refresh_token
        }

        handler = recovery_strategies.get(error_type, self.default_error_handler)
        return await handler(error, state)

    async def trigger_reauth(self, error, state):
        """Initiate re-authentication flow"""
        state["authenticated"] = False
        state["pending_action"] = {
            "type": "reauth_required",
            "auth_url": getattr(error, "url", ""),
            "message": "Gmail authorization expired"
        }
        return state

    async def apply_backoff(self, error, state):
        """Implement exponential backoff for rate limits"""
        attempt = state.get("retry_attempt", 0) + 1
        wait_time = self.retry_config["backoff_factor"] ** attempt

        await asyncio.sleep(wait_time)
        state["retry_attempt"] = attempt

        if attempt < self.retry_config["max_attempts"]:
            state["should_retry"] = True
        else:
            state["messages"].append("Rate limit exceeded, please try later")

        return state

from collections import OrderedDict
import time

class CachedGmailAgent:
    def __init__(self):
        self.arcade = Arcade(api_key=os.environ.get("ARCADE_API_KEY"))
        self.tool_cache = OrderedDict()
        self.cache_ttl = 3600000  # 1 hour in milliseconds

    async def get_gmail_tools(self, user_id: str):
        """Retrieve Gmail tools with caching"""
        cache_key = f"gmail_tools_{user_id}"

        # Check cache
        if cache_key in self.tool_cache:
            entry = self.tool_cache[cache_key]
            if time.time() * 1000 - entry["timestamp"] < self.cache_ttl:
                self.tool_cache.move_to_end(cache_key)
                return entry["tools"]
            else:
                del self.tool_cache[cache_key]

        # Fetch fresh tools
        tools = await self.arcade.tools.list(
            toolkit="gmail",
            user_id=user_id
        )

        # Cache results
        self.tool_cache[cache_key] = {
            "tools": tools.items,
            "timestamp": time.time() * 1000
        }

        # LRU eviction
        if len(self.tool_cache) > 100:
            self.tool_cache.popitem(last=False)

        return tools.items

async def batch_gmail_operations(self, operations: list, user_id: str):
    """Process multiple Gmail operations efficiently"""
    results = []

    # Group operations by type for optimization
    grouped_ops = {}
    for op in operations:
        op_type = op["type"]
        if op_type not in grouped_ops:
            grouped_ops[op_type] = []
        grouped_ops[op_type].append(op)

    # Execute batched operations
    for op_type, ops in grouped_ops.items():
        if op_type == "send_email":
            # Send multiple emails in sequence
            for op in ops:
                result = await self.arcade.tools.execute(
                    tool_name="Gmail.SendEmail",
                    input=op["parameters"],
                    user_id=user_id
                )
                results.append(result)

        elif op_type == "search":
            # Combine search queries when possible
            combined_query = " OR ".join([op["query"] for op in ops])
            result = await self.arcade.tools.execute(
                tool_name="Gmail.SearchThreads",
                input={"query": combined_query},
                user_id=user_id
            )
            results.append(result)

    return results

apiVersion: apps/v1
kind: Deployment
metadata:
  name: langgraph-gmail-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: langgraph-gmail
  template:
    metadata:
      labels:
        app: langgraph-gmail
    spec:
      containers:
        - name: arcade-engine
          image: ghcr.io/arcadeai/engine:latest
          env:
            - name: ARCADE_API_KEY
              valueFrom:
                secretKeyRef:
                  name: arcade-secrets
                  key: api-key
            - name: GOOGLE_CLIENT_ID
              valueFrom:
                secretKeyRef:
                  name: oauth-secrets
                  key: google-client-id
            - name: GOOGLE_CLIENT_SECRET
              valueFrom:
                secretKeyRef:
                  name: oauth-secrets
                  key: google-client-secret
          resources:
            requests:
              memory: "512Mi"
              cpu: "500m"
            limits:
              memory: "1Gi"
              cpu: "1000m"

class AgentMonitor:
    def __init__(self):
        self.metrics = {
            "graph_executions": 0,
            "auth_successes": 0,
            "gmail_operations": 0,
            "error_recoveries": 0
        }

    async def track_graph_execution(self, user_id: str, execution_time: float):
        """Monitor LangGraph execution metrics"""
        self.metrics["graph_executions"] += 1

        # Log to monitoring system
        await self.send_metrics({
            "timestamp": datetime.now().isoformat(),
            "user_id": self.hash_user_id(user_id),
            "execution_time_ms": execution_time * 1000,
            "service": "langgraph-gmail",
            "protocol": "mcp"
        })

    def generate_health_report(self):
        """Generate system health metrics"""
        return {
            "status": "healthy",
            "graph_execution_rate": self.calculate_execution_rate(),
            "auth_success_rate": self.calculate_auth_rate(),
            "active_graphs": len(self.get_active_graphs()),
            "gmail_quota_usage": self.check_gmail_quota()
        }

import pytest
from unittest.mock import AsyncMock, MagicMock

@pytest.mark.asyncio
async def test_gmail_authentication():
    """Test Gmail authentication through LangGraph"""
    agent = ArcadeGmailAgent()

    # Mock Arcade client
    agent.arcade = AsyncMock()
    agent.arcade.tools.authorize.return_value = MagicMock(
        status="completed"
    )

    state = {
        "user_id": "test_user",
        "messages": [],
        "authenticated": False,
        "pending_action": {"type": "send_email"}
    }

    result = await agent.check_authentication(state)

    assert result["authenticated"] == True
    agent.arcade.tools.authorize.assert_called_once_with(
        tool_name="Gmail.SendEmail",
        user_id="test_user"
    )

@pytest.mark.asyncio
async def test_gmail_operation_execution():
    """Test Gmail operation execution"""
    agent = ArcadeGmailAgent()

    # Mock successful execution
    agent.arcade = AsyncMock()
    agent.arcade.tools.execute.return_value = MagicMock(
        output={"message_id": "123", "status": "sent"}
    )

    state = {
        "user_id": "test_user",
        "messages": [],
        "authenticated": True,
        "pending_action": {
            "type": "send_email",
            "parameters": {
                "to": "test@example.com",
                "subject": "Test",
                "body": "Test message"
            }
        }
    }

    result = await agent.execute_gmail_action(state)

    assert "Successfully executed" in result["messages"][0]
    agent.arcade.tools.execute.assert_called_once()