How to Integrate Model Context Protocol (MCP) with Arcade for Secure AI Agent Tool Access

Model Context Protocol (MCP) standardizes tool calling for AI agents. Arcade extends MCP with OAuth-backed authentication, enabling Python agents to access authenticated services like Gmail, Slack, and GitHub on behalf of users.

Prerequisites

Python 3.10 or higher
Arcade account and API key
Basic async/await knowledge in Python
Development environment (VS Code, PyCharm)

Installation

Install the Arcade Python client:

pip install arcadepy

For OpenAI Agents integration:

pip install agents-arcade

For building custom MCP servers:

pip install arcade-mcp-server

Set your API key:

export ARCADE_API_KEY="your_api_key"

Method 1: Using Pre-Built MCP Servers

Arcade hosts MCP servers for Gmail, Slack, GitHub, and other services with OAuth already configured.

Basic Python Agent Implementation

from arcadepy import Arcade

client = Arcade(api_key="your_api_key")
user_id = "user@example.com"

# List available Gmail tools
gmail_toolkit = await client.tools.list(
    toolkit="gmail",
    limit=30
)

# Execute a tool
response = await client.tools.execute(
    tool_name="Gmail.ListEmails",
    input={"max_results": 10},
    user_id=user_id
)

print(response.output)

OAuth Authorization Flow

Handle OAuth for tools requiring authentication:

from arcadepy import Arcade

async def execute_with_auth(user_id: str):
    client = Arcade()

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

    if auth_response.status != "completed":
        print(f"Authorize at: {auth_response.url}")
        await client.auth.wait_for_completion(auth_response)

    # Execute after authorization
    result = await client.tools.execute(
        tool_name="Gmail.SendEmail",
        input={
            "to": "recipient@example.com",
            "subject": "Test",
            "body": "Message from Python agent"
        },
        user_id=user_id
    )

    return result.output

OpenAI Agents Integration

Arcade provides native OpenAI Agents support:

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():
    client = AsyncArcade()

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

    # Create agent
    agent = Agent(
        name="Gmail Agent",
        instructions="Manage Gmail operations",
        model="gpt-4o-mini",
        tools=tools,
    )

    try:
        result = await Runner.run(
            starting_agent=agent,
            input="List my recent emails",
            context={"user_id": "user@example.com"},
        )
        print(result.final_output)
    except AuthorizationError as e:
        print(f"Authorization URL: {e.url}")

Method 2: Building Custom MCP Servers

Create custom MCP servers for domain-specific tools using arcade-mcp-server.

Creating a New MCP Server

Install the CLI and generate a project:

# Install CLI
uv tool install arcade-mcp

# Create new server
arcade new custom_server
cd custom_server

Project structure:

server.py - Main server file
pyproject.toml - Dependencies
.env.example - Environment variables

Defining Custom Tools

Create tools with the @app.tool decorator:

#!/usr/bin/env python3
from typing import Annotated
from arcade_mcp_server import Context, MCPApp

app = MCPApp(
    name="custom_server",
    version="1.0.0",
    instructions="Custom MCP server"
)

@app.tool
async def fetch_data(
    context: Context,
    data_id: Annotated[str, "Data identifier"],
    format: Annotated[str, "Output format"] = "json"
) -> Annotated[str, "Fetched data"]:
    """Fetch data by ID."""

    await context.log.info(f"Fetching {data_id}")

    # Your logic here
    result = f"Data for {data_id} in {format}"

    return result

if __name__ == "__main__":
    import sys
    transport = sys.argv[1] if len(sys.argv) > 1 else "stdio"
    app.run(transport=transport)

Adding OAuth Authentication

Tools requiring OAuth use auth decorators:

from arcade_mcp_server import Context, MCPApp
from arcade_mcp_server.auth import Google
from typing import Annotated
import httpx

app = MCPApp(name="google_server", version="1.0.0")

@app.tool(
    requires_auth=Google(
        scopes=["https://www.googleapis.com/auth/drive.readonly"]
    )
)
async def read_drive_file(
    context: Context,
    file_id: Annotated[str, "File ID"],
) -> Annotated[str, "File contents"]:
    """Read Google Drive file."""

    if not context.authorization or not context.authorization.token:
        raise ValueError("Authorization required")

    token = context.authorization.token

    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://www.googleapis.com/drive/v3/files/{file_id}",
            headers={"Authorization": f"Bearer {token}"}
        )
        response.raise_for_status()
        return response.text

Using Secrets

Access secrets through context:

@app.tool
async def call_api(
    context: Context,
    endpoint: Annotated[str, "API endpoint"],
) -> Annotated[str, "API response"]:
    """Call external API with stored credentials."""

    try:
        api_key = context.get_secret("API_KEY")
    except ValueError:
        await context.log.error("API_KEY not configured")
        return "Error: Missing API key"

    async with httpx.AsyncClient() as client:
        response = await client.get(
            endpoint,
            headers={"Authorization": f"Bearer {api_key}"}
        )
        return response.text

Running the MCP Server

For stdio transport (Claude Desktop):

python server.py stdio

For HTTP transport (Cursor, VS Code):

python server.py http --port 8000

Using Arcade CLI:

arcade serve --port 8002 --reload --mcp

Options:

-reload - Auto-reload on file changes
-mcp - Run as stdio MCP server
-no-auth - Disable authentication (development only)
-debug - Enable debug logging

Multi-User Authentication

Production agents serve multiple users with isolated authentication contexts.

Session Management

from arcadepy import Arcade
from datetime import datetime
from typing import Dict, Any

class AgentManager:
    def __init__(self):
        self.client = Arcade()
        self.sessions: Dict[str, Any] = {}

    async def auth_user(self, user_id: str) -> Dict[str, Any]:
        """Handle user OAuth flow."""

        auth_response = await self.client.tools.authorize(
            tool_name="Gmail.SendEmail",
            user_id=user_id
        )

        if auth_response.status != "completed":
            return {
                "auth_required": True,
                "url": auth_response.url
            }

        await self.client.auth.wait_for_completion(auth_response)

        self.sessions[user_id] = {
            "authenticated": True,
            "timestamp": datetime.now()
        }

        return {"authenticated": True}

    async def execute_tool(
        self,
        user_id: str,
        tool_name: str,
        params: Dict
    ):
        """Execute tool with user context."""

        if user_id not in self.sessions:
            return await self.auth_user(user_id)

        response = await self.client.tools.execute(
            tool_name=tool_name,
            input=params,
            user_id=user_id
        )

        return response.output

Toolset Caching

Optimize performance with per-user caching:

from collections import OrderedDict
import time

class ToolCache:
    def __init__(self, max_size: int = 1000, ttl: int = 3600):
        self.cache = OrderedDict()
        self.max_size = max_size
        self.ttl = ttl

    def set(self, user_id: str, tools):
        if user_id in self.cache:
            self.cache.move_to_end(user_id)

        self.cache[user_id] = {
            "tools": tools,
            "timestamp": time.time()
        }

        if len(self.cache) > self.max_size:
            self.cache.popitem(last=False)

    def get(self, user_id: str):
        entry = self.cache.get(user_id)
        if not entry:
            return None

        if time.time() - entry["timestamp"] > self.ttl:
            del self.cache[user_id]
            return None

        self.cache.move_to_end(user_id)
        return entry["tools"]

Production Deployment

Using Arcade Deploy

Arcade Deploy hosts MCP servers with OAuth support.

Create worker.toml:

[[worker]]
[worker.config]
id = "production-worker"
secret = "secure_worker_secret"

[worker.local_source]
packages = ["./toolkit"]

Deploy:

arcade deploy

Verify:

arcade workers list

Docker Deployment

Build containerized MCP server:

FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY server.py .
COPY .env .

CMD ["python", "server.py", "http", "--port", "8000"]

Run:

docker build -t mcp-server .
docker run -p 8000:8000 --env-file .env mcp-server

Kubernetes Deployment

Scale with Kubernetes:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: mcp-server
  template:
    metadata:
      labels:
        app: mcp-server
    spec:
      containers:
        - name: server
          image: registry/mcp-server:latest
          env:
            - name: ARCADE_API_KEY
              valueFrom:
                secretKeyRef:
                  name: arcade-secrets
                  key: api-key
          ports:
            - containerPort: 8000
          resources:
            requests:
              memory: "512Mi"
              cpu: "500m"
            limits:
              memory: "1Gi"
              cpu: "1000m"

Error Handling

Authorization Errors

Handle authorization failures:

from arcadepy import Arcade
import asyncio

async def handle_execution(user_id: str, tool_name: str, params: dict):
    """Execute tool with error handling."""

    client = Arcade()

    try:
        result = await client.tools.execute(
            tool_name=tool_name,
            input=params,
            user_id=user_id
        )
        return {"success": True, "data": result.output}

    except Exception as e:
        error_type = getattr(e, 'type', type(e).__name__)

        if error_type == "authorization_required":
            return {
                "success": False,
                "auth_required": True,
                "url": getattr(e, 'url', '')
            }

        elif error_type == "token_expired":
            refresh = await client.auth.refresh(
                user_id=user_id,
                provider='google'
            )

            if refresh.success:
                return await handle_execution(user_id, tool_name, params)
            else:
                return {"success": False, "reauth_required": True}

        elif error_type == "rate_limit_exceeded":
            retry_count = getattr(e, 'retry_count', 1)
            await asyncio.sleep(2 ** retry_count)
            return {"success": False, "retry": True}

        else:
            return {"success": False, "error": str(e)}

Monitoring

Track metrics for production systems:

import logging
from datetime import datetime
from typing import Dict

class Monitor:
    def __init__(self):
        self.logger = logging.getLogger("mcp.monitor")
        self.metrics = {
            "auth_attempts": 0,
            "auth_success": 0,
            "tool_calls": 0,
            "errors": 0
        }

    async def track_auth(self, user_id: str, success: bool):
        """Track authentication metrics."""
        self.metrics["auth_attempts"] += 1
        if success:
            self.metrics["auth_success"] += 1

        await self.log_event({
            "timestamp": datetime.now().isoformat(),
            "event": "auth",
            "user": self.hash_id(user_id),
            "success": success
        })

    async def track_tool(self, tool: str, success: bool):
        """Track tool execution."""
        self.metrics["tool_calls"] += 1
        if not success:
            self.metrics["errors"] += 1

    def health_status(self) -> Dict:
        """Generate health report."""
        auth_rate = (
            self.metrics["auth_success"] / self.metrics["auth_attempts"]
            if self.metrics["auth_attempts"] > 0 else 0
        )

        error_rate = (
            self.metrics["errors"] / self.metrics["tool_calls"]
            if self.metrics["tool_calls"] > 0 else 0
        )

        return {
            "status": "healthy" if error_rate < 0.05 else "degraded",
            "auth_rate": auth_rate,
            "error_rate": error_rate,
            "total_calls": self.metrics["tool_calls"]
        }

    def hash_id(self, user_id: str) -> str:
        """Hash user ID for privacy."""
        import hashlib
        return hashlib.sha256(user_id.encode()).hexdigest()[:16]

Common Issues

Authentication Failures

If users cannot authenticate:

Verify API key is set correctly
Check OAuth credentials in Arcade dashboard
Confirm redirect URLs match configuration
Verify scopes in tool definitions

Tools Not Loading

If tools don't appear:

Confirm toolkit name matches documented server name
Check MCP server registration
Verify user completed authorization
Review server startup logs

Rate Limiting

Handle rate limits:

Implement exponential backoff
Cache tool results when appropriate
Batch operations
Contact support for quota increases

Resources

MCP provides the protocol foundation for tool calling. Arcade extends it with OAuth authentication, enabling Python agents to interact with services securely. Use pre-built servers for rapid development or build custom MCP servers for specific requirements. The platform handles authentication, letting you focus on agent functionality.

How to Use MCP with Python Agent through Arcade