Build a LangGraph Gmail Agent with Arcade MCP

Building production-ready AI agents that can interact with Gmail requires solving authentication, tool calling, and orchestration challenges simultaneously. This guide demonstrates how to build a LangGraph Gmail agent using Arcade's Model Context Protocol (MCP) integration, enabling secure multi-user email automation with enterprise-grade authentication.

What You'll Build

This tutorial walks through creating a LangGraph-based Gmail agent that can:

Authenticate multiple users via OAuth 2.0
Send, draft, and search emails on behalf of authorized users
Maintain secure credential isolation between users
Scale to handle thousands of concurrent operations

Prerequisites and Setup

Required Components

Before starting, ensure you have:

An active Arcade.dev account with API key
Python 3.8+ installed
LangGraph installed (pip install langgraph)
A Google Cloud Console project with OAuth 2.0 credentials configured

Installation

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

Arcade MCP Architecture for Secure Gmail Automation

Arcade serves as both an MCP server and a bridge to other MCP servers using HTTP transport. This architecture provides:

Authentication Layer: OAuth 2.0 flows managed server-side
Tool Calling Interface: Unified API for Gmail operations
Security Boundaries: LLMs never see authentication tokens
Multi-User Support: Per-user credential isolation and management

The platform transforms single-user MCP servers into production-ready systems through pre-built connectors and managed authorization layers. Learn more about Arcade's architecture.

Building the LangGraph Agent Structure

Core Agent Implementation

Create a LangGraph agent that integrates with Arcade's Gmail toolkit:

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

Authentication Node Implementation

Handle OAuth flows within the LangGraph workflow:

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

Integrating Gmail Tools with LangGraph

Tool Execution Pattern

Connect LangGraph decisions to Arcade's Gmail toolkit:

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

Building Multi-Step Gmail Workflows

Create multi-step Gmail operations using LangGraph's graph structure:

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

Configuring MCP Server Connection

Desktop Client Configuration

For development with VS Code or similar IDEs:

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

Production Server Configuration

Deploy with self-hosted Arcade engine:

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

Handling Multi-User Authentication

User Session Management

Implement per-user session tracking within LangGraph:

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

OAuth Callback Handling

Process OAuth completions for multiple users:

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

Error Handling and Recovery

Implementing Robust Error Management

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

Performance Optimization

Caching Strategy for LangGraph Nodes

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

Batch Processing for Multiple Operations

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

Production Deployment

Kubernetes Configuration

Deploy your LangGraph Gmail agent at scale:

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"

Monitoring and Observability

Track agent performance and authentication metrics:

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

Testing Your LangGraph Gmail Agent

Unit Testing Authentication Flows

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

Next Steps

With your LangGraph Gmail agent built on Arcade's MCP platform, you can:

Explore additional Gmail toolkit capabilities
Integrate other Arcade connectors for Slack, GitHub, and more
Review the Arcade documentation for advanced patterns
Check out example implementations on GitHub
Consult the API reference for detailed specifications

The combination of LangGraph's orchestration capabilities with Arcade's secure authentication layer enables building production-ready Gmail agents that handle multi-user scenarios while maintaining enterprise-grade security boundaries.

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