How to Build a Secure Gmail AI Agent with Arcade’s Model Context Protocol (MCP)

Building AI agents that can access and act on Gmail data represents a significant challenge in production environments. This guide demonstrates how to build a fully functional Gmail agent using OpenAI's latest models through Arcade's Model Context Protocol implementation, enabling secure OAuth-based authentication and real-world email operations.

Prerequisites

Before starting, ensure you have:

Active Arcade.dev account with API key
Python 3.10 or higher installed
OpenAI API key with access to GPT-4 or GPT-5 models
Development environment (VS Code, PyCharm, or similar)
Basic familiarity with async/await patterns in Python

Why MCP Matters for Gmail Agents

Most AI projects fail to reach production because agents cannot obtain secure, user-scoped credentials to external systems. Traditional approaches require building custom OAuth flows, managing token rotation, and handling permission scoping across multiple services.

Arcade's MCP implementation solves this by providing:

Authentication-first architecture: OAuth 2.0 flows that let AI access tools as the end user, not as a bot
Pre-built Gmail toolkit: Complete set of Gmail operations ready to use
Production infrastructure: Monitoring, logging, and evaluation capabilities built for enterprise deployment
Developer acceleration: SDKs that reduce implementation time from weeks to minutes

Installation and Setup

Install Required Packages

pip install agents-arcade arcadepy

The agents-arcade package provides integration between Arcade and the OpenAI Agents Library, while arcadepy is the core Arcade Python client.

Configure API Keys

Set up your environment variables:

export ARCADE_API_KEY="your_arcade_api_key"
export OPENAI_API_KEY="your_openai_api_key"

Get your Arcade API key from the Arcade dashboard. If you don't have one yet, visit the Get an API key page.

Building Your First Gmail Agent

Initialize the Arcade Client

Start by importing required modules and initializing the Arcade client:

from agents import Agent, Runner
from arcadepy import AsyncArcade
from agents_arcade import get_arcade_tools
from agents_arcade.errors import AuthorizationError

async def main():
    # Initialize the Arcade client
    client = AsyncArcade()

The AsyncArcade client automatically finds your ARCADE_API_KEY environment variable and handles all authentication flows behind the scenes.

Load Gmail MCP Server Tools

Retrieve the Gmail toolkit from Arcade's MCP servers:

# Get tools from the "gmail" MCP Server
    tools = await get_arcade_tools(client, toolkits=["gmail"])

This single line loads all available Gmail tools from Arcade's pre-built Gmail MCP Server, including:

Gmail.SendEmail - Send emails from user's account
Gmail.ListEmails - Read emails and extract content
Gmail.SearchThreads - Search through email threads
Gmail.WriteDraftEmail - Compose draft emails
Gmail.UpdateDraftEmail - Update existing drafts
Gmail.ListDraftEmails - List draft emails
Gmail.WhoAmI - Get user profile information

Create the Gmail Agent

Configure the agent with specific instructions and the Gmail toolkit:

# Create an agent with Gmail tools
    gmail_agent = Agent(
        name="Gmail Assistant",
        instructions="""You are a helpful Gmail assistant that can:
        - Read and summarize emails
        - Search for specific emails
        - Compose and send emails
        - Manage draft emails

        Always verify user intent before sending emails.
        Keep responses clear and concise.""",
        model="gpt-5",  # or "gpt-4o-mini" for faster responses
        tools=tools,
    )

The agent instructions guide the model's behavior when using Gmail tools. Clear instructions improve reliability and user experience.

Handling User Authorization

Implementing the Authorization Flow

When a tool requires user authorization, Arcade raises an AuthorizationError with a URL for the user to visit:

try:
        # Run the agent with a unique user_id for authorization
        result = await Runner.run(
            starting_agent=gmail_agent,
            input="What are my latest 5 emails?",
            context={"user_id": "user@example.com"},
        )

        print("Agent response:", result.final_output)

    except AuthorizationError as e:
        url = getattr(e, "url", str(e))
		    print(f"Authorization required. Please visit: {url}")
        print("After authorizing, run the agent again.")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

The user_id parameter is critical. It identifies the specific user whose Gmail account the agent should access. Each user maintains separate authorization tokens, enabling true multi-user scenarios.

What Happens During Authorization

When a user runs the agent for the first time:

Arcade detects that Gmail tools require authorization
An AuthorizationError is raised with an OAuth URL
User visits the URL and grants Gmail permissions
Arcade stores the authorization token securely
Subsequent agent runs work without re-authorization

The authorization token persists until the user revokes access or the token expires. Arcade automatically handles token refresh, so your application doesn't need to manage OAuth complexity.

Complete Working Example

Here's a production-ready Gmail agent implementation:

import asyncio
from agents import Agent, Runner
from arcadepy import AsyncArcade
from agents_arcade import get_arcade_tools
from agents_arcade.errors import AuthorizationError

async def run_gmail_agent(user_id: str, query: str):
    """
    Run a Gmail agent for a specific user with a given query.

    Args:
        user_id: Unique identifier for the user
        query: The query or task for the agent
    """
    client = AsyncArcade()

    # Load Gmail tools
    tools = await get_arcade_tools(client, toolkits=["gmail"])

    # Create Gmail agent
    gmail_agent = Agent(
        name="Gmail Assistant",
        instructions="""You are a professional Gmail assistant.

        When reading emails:
        - Summarize key points clearly
        - Highlight important sender information
        - Note any action items or deadlines

        When composing emails:
        - Confirm recipient and subject with user
        - Maintain professional tone
        - Review draft before sending

        Always respect user privacy and data security.""",
        model="gpt-5",
        tools=tools,
    )

    try:
        result = await Runner.run(
            starting_agent=gmail_agent,
            input=query,
            context={"user_id": user_id},
        )

        return {
            "success": True,
            "response": result.final_output
        }

    except AuthorizationError as e:
        return {
            "success": False,
            "authorization_required": True,
            "auth_url": str(e),
            "message": "User must authorize Gmail access"
        }

    except Exception as e:
        return {
            "success": False,
            "error": str(e)
        }

async def main():
    # Example usage
    user_id = "user@example.com"

    # First request - might need authorization
    result = await run_gmail_agent(
        user_id=user_id,
        query="Read my 3 most recent emails and summarize them"
    )

    if result.get("authorization_required"):
        print(f"Please authorize at: {result['auth_url']}")
        print("Run the script again after authorizing.")
        return

    print("Agent response:", result["response"])

    # Subsequent requests work without re-authorization
    result = await run_gmail_agent(
        user_id=user_id,
        query="Search for emails from support@example.com from the last week"
    )

    print("Search results:", result["response"])

if __name__ == "__main__":
    asyncio.run(main())

This implementation includes proper error handling, clear return values, and can be integrated into web applications or services.

Advanced Operations

Sending Emails with Confirmation

Build agents that compose and send emails with user confirmation:

async def send_email_with_confirmation(user_id: str, recipient: str, subject: str, body: str):
    client = AsyncArcade()
    tools = await get_arcade_tools(client, toolkits=["gmail"])

    gmail_agent = Agent(
        name="Email Sender",
        instructions="""You are an email sending assistant.

        When asked to send an email:
        1. Show the draft to the user
        2. Ask for explicit confirmation
        3. Only send after receiving confirmation
        4. Confirm successful sending""",
        model="gpt-5",
        tools=tools,
    )

    query = f"""
    Draft an email to {recipient} with subject "{subject}" and the following content:

    {body}

    After showing me the draft, ask for confirmation before sending.
    """

    result = await Runner.run(
        starting_agent=gmail_agent,
        input=query,
        context={"user_id": user_id},
    )

    return result.final_output

Searching and Filtering Emails

Use the agent to search through emails with complex criteria:

async def search_emails(user_id: str, search_criteria: dict):
    client = AsyncArcade()
    tools = await get_arcade_tools(client, toolkits=["gmail"])

    gmail_agent = Agent(
        name="Email Search Agent",
        instructions="""You are an email search specialist.

        Use Gmail.SearchThreads or Gmail.ListEmailsByHeader to find emails
        matching user criteria. Summarize results clearly.""",
        model="gpt-5",
        tools=tools,
    )

    query = f"""
    Search for emails matching these criteria:
    - Sender: {search_criteria.get('sender', 'any')}
    - Subject contains: {search_criteria.get('subject', 'any')}
    - Date range: {search_criteria.get('date_range', 'last_7_days')}

    Provide a summary of matching emails.
    """

    try:
        result = await Runner.run(
            starting_agent=gmail_agent,
            input=query,
            context={"user_id": user_id},
        )

        return result.final_output

    except AuthorizationError as e:
        return f"Authorization required: {e}"

Draft Management

Implement draft email workflows:

async def manage_drafts(user_id: str, action: str, draft_id: str = None, content: dict = None):
    client = AsyncArcade()
    tools = await get_arcade_tools(client, toolkits=["gmail"])

    gmail_agent = Agent(
        name="Draft Manager",
        instructions="""You manage Gmail draft emails.

        Available actions:
        - List all drafts
        - Create new draft
        - Update existing draft
        - Delete draft
        - Send draft""",
        model="gpt-5",
        tools=tools,
    )

    if action == "list":
        query = "List all my draft emails"
    elif action == "create":
        query = f"Create a draft email to {content['recipient']} with subject '{content['subject']}' and body: {content['body']}"
    elif action == "update":
        query = f"Update draft {draft_id} with new content: {content}"
    elif action == "delete":
        query = f"Delete draft {draft_id}"
    elif action == "send":
        query = f"Send draft {draft_id}"

    result = await Runner.run(
        starting_agent=gmail_agent,
        input=query,
        context={"user_id": user_id},
    )

    return result.final_output

Multi-User Scenarios

Arcade's architecture handles multiple users simultaneously without token conflicts. Each user maintains separate authentication:

async def handle_multiple_users():
    client = AsyncArcade()
    tools = await get_arcade_tools(client, toolkits=["gmail"])

    gmail_agent = Agent(
        name="Gmail Assistant",
        instructions="You are a Gmail assistant for multiple users.",
        model="gpt-5",
        tools=tools,
    )

    users = [
        {"id": "user1@example.com", "query": "List my recent emails"},
        {"id": "user2@example.com", "query": "Search for emails from boss@company.com"},
        {"id": "user3@example.com", "query": "Draft email to team@company.com"}
    ]

    results = []
    for user in users:
        try:
            result = await Runner.run(
                starting_agent=gmail_agent,
                input=user["query"],
                context={"user_id": user["id"]},
            )
            results.append({
                "user": user["id"],
                "success": True,
                "response": result.final_output
            })
        except AuthorizationError as e:
            results.append({
                "user": user["id"],
                "success": False,
                "auth_url": str(e)
            })

    return results

Each user's authorization is isolated. Tokens for user1@example.com cannot access user2@example.com's Gmail account, maintaining strict security boundaries.

Production Considerations

Error Handling

Implement comprehensive error handling for production deployments:

from arcadepy.errors import ArcadeError
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def production_gmail_agent(user_id: str, query: str):
    try:
        client = AsyncArcade()
        tools = await get_arcade_tools(client, toolkits=["gmail"])

        gmail_agent = Agent(
            name="Production Gmail Agent",
            instructions="You are a production-ready Gmail assistant.",
            model="gpt-5",
            tools=tools,
        )

        result = await Runner.run(
            starting_agent=gmail_agent,
            input=query,
            context={"user_id": user_id},
        )

        logger.info(f"Successfully processed query for user {user_id}")
        return {"success": True, "data": result.final_output}

    except AuthorizationError as e:
        logger.warning(f"Authorization required for user {user_id}")
        return {
            "success": False,
            "error_type": "authorization_required",
            "auth_url": str(e)
        }

    except ArcadeError as e:
        logger.error(f"Arcade API error for user {user_id}: {e}")
        return {
            "success": False,
            "error_type": "arcade_error",
            "message": str(e)
        }

    except Exception as e:
        logger.error(f"Unexpected error for user {user_id}: {e}")
        return {
            "success": False,
            "error_type": "unexpected_error",
            "message": "An unexpected error occurred"
        }

Rate Limiting and Quotas

Monitor tool usage to stay within Gmail API quotas:

class GmailAgentManager:
    def __init__(self):
        self.client = AsyncArcade()
        self.request_count = {}

    async def execute_with_rate_limit(self, user_id: str, query: str, max_requests: int = 100):
        # Track requests per user
        count = self.request_count.get(user_id, 0)

        if count >= max_requests:
            return {
                "success": False,
                "error": "Rate limit exceeded"
            }

        tools = await self.client.tools.list(toolkit="gmail", user_id=user_id)

        gmail_agent = Agent(
            name="Rate-Limited Agent",
            instructions="You are a Gmail assistant with rate limiting.",
            model="gpt-5",
            tools=tools.items,
        )

        result = await Runner.run(
            starting_agent=gmail_agent,
            input=query,
            context={"user_id": user_id},
        )

        self.request_count[user_id] = count + 1

        return {
            "success": True,
            "data": result.final_output,
            "requests_remaining": max_requests - (count + 1)
        }

Logging and Monitoring

Implement monitoring for production visibility:

import time
from datetime import datetime

class MonitoredGmailAgent:
    def __init__(self):
        self.metrics = {
            "total_requests": 0,
            "successful_requests": 0,
            "authorization_errors": 0,
            "api_errors": 0
        }

    async def execute_with_monitoring(self, user_id: str, query: str):
        start_time = time.time()
        self.metrics["total_requests"] += 1

        try:
            client = AsyncArcade()
            tools = await get_arcade_tools(client, toolkits=["gmail"])

            gmail_agent = Agent(
                name="Monitored Agent",
                instructions="You are a monitored Gmail assistant.",
                model="gpt-5",
                tools=tools,
            )

            result = await Runner.run(
                starting_agent=gmail_agent,
                input=query,
                context={"user_id": user_id},
            )

            self.metrics["successful_requests"] += 1
            execution_time = time.time() - start_time

            logger.info(f"""
            Request completed:
            - User: {user_id}
            - Execution time: {execution_time:.2f}s
            - Timestamp: {datetime.now().isoformat()}
            """)

            return {"success": True, "data": result.final_output}

        except AuthorizationError:
            self.metrics["authorization_errors"] += 1
            raise

        except Exception:
            self.metrics["api_errors"] += 1
            raise

    def get_metrics(self):
        return {
            **self.metrics,
            "success_rate": self.metrics["successful_requests"] / max(self.metrics["total_requests"], 1)
        }

Integrating with Web Applications

FastAPI Integration

Build a production API endpoint:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class EmailQuery(BaseModel):
    user_id: str
    query: str

class EmailSend(BaseModel):
    user_id: str
    recipient: str
    subject: str
    body: str

@app.post("/api/gmail/query")
async def query_gmail(request: EmailQuery):
    """
    Execute a Gmail query for a specific user.
    """
    result = await run_gmail_agent(request.user_id, request.query)

    if result.get("authorization_required"):
        raise HTTPException(
            status_code=401,
            detail={
                "message": "Authorization required",
                "auth_url": result["auth_url"]
            }
        )

    if not result["success"]:
        raise HTTPException(
            status_code=500,
            detail={"message": result.get("error")}
        )

    return {"response": result["response"]}

@app.post("/api/gmail/send")
async def send_email(request: EmailSend):
    """
    Send an email on behalf of a user.
    """
    query = f"Send an email to {request.recipient} with subject '{request.subject}' and body: {request.body}"
    result = await run_gmail_agent(request.user_id, query)

    if result.get("authorization_required"):
        raise HTTPException(
            status_code=401,
            detail={
                "message": "Authorization required",
                "auth_url": result["auth_url"]
            }
        )

    return {"message": "Email sent successfully"}

@app.get("/api/gmail/status/{user_id}")
async def check_authorization_status(user_id: str):
    """
    Check if a user has authorized Gmail access.
    """
    client = AsyncArcade()

    try:
        auth_response = await client.tools.authorize(
            tool_name="Gmail.ListEmails",
            user_id=user_id
        )

        return {
            "authorized": auth_response.status == "completed",
            "status": auth_response.status
        }

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

WebSocket Support for Real-Time Updates

Implement streaming responses:

from fastapi import WebSocket
import json

@app.websocket("/ws/gmail/{user_id}")
async def gmail_websocket(websocket: WebSocket, user_id: str):
    await websocket.accept()

    try:
        while True:
            # Receive query from client
            data = await websocket.receive_text()
            query_data = json.loads(data)

            # Execute agent
            result = await run_gmail_agent(user_id, query_data["query"])

            # Send response
            await websocket.send_json(result)

    except Exception as e:
        await websocket.send_json({
            "error": str(e)
        })
    finally:
        await websocket.close()

Troubleshooting Common Issues

Authorization Failures

If users cannot complete authorization:

Verify Google OAuth credentials are properly configured in Arcade dashboard
Check that redirect URLs match exactly
Ensure proper OAuth scopes are requested
Verify the user is clicking the authorization link from the same browser session

Token Expiration

Arcade automatically handles token refresh, but if issues persist:

async def force_reauthorization(user_id: str):
    """
    Force a user to re-authorize Gmail access.
    """
    client = AsyncArcade()

    # This will trigger a new authorization flow
    auth_response = await client.tools.authorize(
        tool_name="Gmail.ListEmails",
        user_id=user_id,
        force=True
    )

    return auth_response.url

Agent Not Using Tools Correctly

Improve agent instructions to guide tool usage:

gmail_agent = Agent(
    name="Gmail Assistant",
    instructions="""You are a Gmail assistant. Follow these guidelines:

    READING EMAILS:
    - Use Gmail.ListEmails to get recent emails
    - Use Gmail.SearchThreads for specific searches
    - Always specify how many emails to retrieve

    SENDING EMAILS:
    - Use Gmail.SendEmail with recipient, subject, and body
    - Confirm details before sending
    - Check for successful sending

    DRAFTS:
    - Use Gmail.WriteDraftEmail to create drafts
    - Use Gmail.UpdateDraftEmail to modify drafts
    - Use Gmail.SendDraftEmail to send existing drafts

    Always provide clear, actionable responses.""",
    model="gpt-5",
    tools=tools,
)

Next Steps

Now that you have a working Gmail agent, explore these advanced capabilities:

Combine Multiple Toolkits: Add Slack, Calendar, or Drive tools for comprehensive workflow automation
Build Custom Tools: Create specialized tools using Arcade's Tool SDK for your specific use cases
Deploy to Production: Follow Arcade's deployment guide for self-hosted or hybrid deployments
Implement Evaluations: Use Arcade's evaluation suite to test tool reliability
Explore Other Frameworks: Try Arcade with LangChain, CrewAI, or Mastra

Check the Arcade GitHub repository for example implementations and the latest updates.

Summary

Building production-ready Gmail agents with Arcade and MCP eliminates the authentication complexity that typically prevents AI projects from reaching production. By handling OAuth flows, token management, and permission scoping, Arcade lets you focus on building agent functionality rather than infrastructure.

The combination of OpenAI's powerful models with Arcade's secure, user-scoped authentication creates agents that can take real actions in Gmail accounts while maintaining enterprise-grade security. This approach scales from single-user prototypes to multi-tenant production applications without architectural changes.

Visit the Arcade documentation to explore additional MCP servers and build agents that connect to the full ecosystem of tools and services your users rely on.

How to Build a GPT-5 Gmail Agent with Arcade (MCP)