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

How to Query Postgres from Python Agent via Arcade (MCP)

Python agents require direct database access to analyze data and generate insights. This guide shows how to build Python agents that query PostgreSQL databases through Arcade's Model Context Protocol implementation, enabling secure database operations without exposing credentials to language models. Prerequisites Install required components: * Python 3.8+ * PostgreSQL database with connection details * Arcade.dev account with API key * Basic SQL and Python knowledge Architecture Ov

Rays decoration image
THOUGHT LEADERSHIP

How to Connect Python Agent to Slack with Arcade (MCP)

AI agents need direct access to Slack workspaces to read messages, send notifications, and manage conversations. Arcade provides OAuth-backed authentication and a complete Slack toolkit through the Model Context Protocol (MCP), eliminating the need to build authentication infrastructure from scratch. This guide covers the technical implementation of Python agents that interact with Slack through Arcade's authentication layer and pre-built tools. Prerequisites * Python 3.10+ * Arcade acc

Rays decoration image
THOUGHT LEADERSHIP

How to Call Custom Tools from Open Agents SDK via Arcade

LangChain's Open Agent Platform connects to tools through the Model Context Protocol (MCP). Arcade provides the infrastructure to build, deploy, and serve custom tools as MCP servers that OAP agents can call. This guide covers building custom tools with Arcade's SDK, deploying them as MCP servers, and integrating them into LangChain Open Agent Platform. Architecture Overview The integration flow works as follows: LangGraph Agent → MCP Protocol → Arcade MCP Server → Custom Tool Execution

Blog CTA Icon

Get early access to Arcade, and start building now.

Sign up now
Docs