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

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

Arcade.dev Team's avatar
Arcade.dev Team
OCTOBER 21, 2025
9 MIN READ
THOUGHT LEADERSHIP
Rays decoration image
Ghost Icon

This guide shows you how to build a Python Gmail agent using Arcade's Model Context Protocol (MCP) implementation. You'll implement OAuth authentication, execute Gmail operations, and handle multi-user scenarios.

Prerequisites

Required:

  • Arcade account with API key
  • Python 3.10 or higher
  • Development environment (VS Code, PyCharm, or similar)

Optional:

  • Google Cloud Console project for custom OAuth credentials
  • Familiarity with async/await patterns in Python

What You'll Build

A production-ready Gmail agent that:

  • Authenticates users via OAuth 2.0
  • Sends and receives emails
  • Searches and filters messages
  • Manages drafts and threads
  • Handles multiple users concurrently

Setup

Install Arcade SDK

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

# Install Arcade Python client
pip install arcadepy

Configure API Key

export ARCADE_API_KEY="your_arcade_api_key"

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

Arcade Gmail Toolkit

The Gmail MCP Server provides these tools:

  • Gmail.SendEmail - Send emails
  • Gmail.SendDraftEmail - Send draft emails
  • Gmail.ListEmails - List inbox emails
  • Gmail.SearchEmails - Search emails
  • Gmail.SearchThreads - Search threads
  • Gmail.ReplyToEmail - Reply to emails
  • Gmail.WriteDraftEmail - Create drafts
  • Gmail.UpdateDraftEmail - Update drafts
  • Gmail.DeleteDraftEmail - Delete drafts

Basic Agent Implementation

Core Agent Class

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}

Send Email

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

List Emails

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

Search Emails

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

Authentication Flow

Arcade uses OAuth 2.0 for user authentication. The flow:

  1. Request authorization for a tool
  2. Receive authorization URL if user not authenticated
  3. User completes OAuth consent
  4. Arcade stores OAuth tokens securely
  5. Execute tools on behalf of user

Multi-User Support

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

Web Application Integration

FastAPI Example

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

Framework Integration

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

Full details: Using Arcade with 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
)

Full details: Using Arcade with 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
    )

Full details: Using Arcade with Google ADK

Advanced Operations

Email Search Filters

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)

Draft Management

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

Thread Operations

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

Testing

Unit Tests

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"

Integration Tests

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

Arcade CLI Testing

Use the Arcade CLI for interactive testing:

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

Production Deployment

Self-Hosted Arcade Engine

Install Arcade Engine:

# macOS
brew install arcadeai/tap/arcade-engine

# Ubuntu/Debian
sudo apt install arcade-engine

Configure with custom OAuth credentials:

# 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

Google OAuth Setup

Configure Google auth provider:

  1. Create project in Google Cloud Console
  2. Enable Gmail API
  3. Create OAuth 2.0 credentials
  4. Add authorized redirect URIs
  5. Configure in Arcade Engine

Rate Limiting

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

Error Handling

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

Monitoring

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

Complete Production Example

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

Next Steps

Resources

SHARE THIS POST

RECENT ARTICLES

Rays decoration image
THOUGHT LEADERSHIP

We Threw 4,000 Tools at Anthropic's New Tool Search. Here's What Happened.

TL;DR: Anthropic's new Tool Search is a step in the right direction-but if you're running 4,000+ tools across multiple services, it might not be ready for prime time. The promise Anthropic's Tool Search promises to let Claude "access thousands of tools without consuming its context window." Music to our ears. At Arcade, we maintain thousands of agent-optimized tools across Gmail, Slack, GitHub, HubSpot, Salesforce, and dozens more platforms. If anyone was going to stress-test this feature, it

Rays decoration image
THOUGHT LEADERSHIP

What does Anthropic's Tool Search for Claude mean for you?

I was recently in Amsterdam meeting with some of the largest enterprises, and they all raised the same challenge: how to give AI agents access to more tools without everything falling apart?  The issue is that as soon as they hit 20-30 tools, token costs became untenable and selection accuracy plummeted. The pain has been so acute that many teams have been attempting (unsuccessfully) to build their own workarounds with RAG pipelines, only to hit performance walls.  That's why I'm excited about

Rays decoration image
THOUGHT LEADERSHIP

38 Proxy Server AI Revenue Metrics: Market Growth, Data Collection ROI, and Infrastructure Performance

Comprehensive analysis of proxy server market valuations, AI-driven revenue acceleration, and performance benchmarks shaping the future of scoped, user-delegated access The convergence of proxy infrastructure and artificial intelligence represents one of the fastest-growing segments in enterprise technology, with the proxy server market valued at $1 billion in 2024. This growth reflects the need for secure, scoped access pathways as AI systems move from prototypes to real operations. Arcade.de

Blog CTA Icon

Get early access to Arcade, and start building now.

Sign up now
Docs