Build a Contact Manager AI Agent with Arcade's Google Contacts Toolkit

Building AI agents that can manage contacts across different platforms requires handling complex authentication flows, OAuth tokens, and API integrations. Arcade is an AI tool-calling platform that lets large-language-model agents act on behalf of users by brokering secure, OAuth-backed access to SaaS and internal systems. This guide walks through building a production-ready contact manager AI agent using Arcade's Google Contacts toolkit.

Why Arcade for Contact Management

Traditional approaches to building contact management agents face significant challenges. Fewer than 30% of AI projects reach production because agents cannot obtain secure, user-scoped credentials to the systems they must act on. Each enterprise system demands OAuth flows, token rotation and permission scoping; building and maintaining that stack in-house is costly and error-prone, delaying time-to-value.

Create and search contacts in Google Contacts with your agents using Arcade's pre-built toolkit. The platform handles all authentication complexity while maintaining enterprise-grade security boundaries.

Setting Up Your Development Environment

Prerequisites

Before building your contact manager agent, ensure you have:

An Arcade.dev account with API key (Get an API key)
Python 3.10+ or Node.js 16+
Google Cloud Console project with OAuth 2.0 credentials
Basic understanding of async/await patterns

Installing Arcade SDK

Start by installing the necessary packages for your chosen language:

Python Installation:

pip install arcadepy
pip install arcade-mcp

JavaScript Installation:

npm install @arcadeai/arcadejs

Set up your 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"

Core Architecture Components

Authentication-First Design

Arcade.dev bridges this gap by extending MCP with enterprise-grade OAuth 2.0 authentication, transforming single-user MCP servers into production-ready, multi-user systems. The platform ensures that LLMs never see authentication tokens and authentication logic remains completely isolated from the AI model.

Google Contacts Toolkit Capabilities

The Google Contacts toolkit provides comprehensive contact management functionality:

Create contacts with full field support (name, email, phone, address, organization)
Search contacts using various filters and queries
Update existing contacts with field-level modifications
Delete contacts with proper authorization
Bulk operations for managing multiple contacts efficiently
Contact groups management and organization

Building Your Contact Manager Agent

Step 1: Initialize the Arcade Client

Arcade needs a unique identifier for your application user (this could be an email address, a UUID, etc.). This ensures proper user isolation and security.

from arcadepy import Arcade
import asyncio

# Initialize Arcade client
client = Arcade(api_key="your_arcade_api_key")

# Unique user identifier (the email you used to create your arcade account)
user_id = "user@example.com"

# List available Google Contacts tools
async def list_contact_tools():
    tools = await client.tools.list(toolkit="GoogleContacts")
    for tool in tools.items:
        print(f"Tool: {tool.name} - {tool.description}")
    return tools

Step 2: Implement User Authentication

By specifying the requires_auth parameter in the @tool decorator, you indicate that the tool needs user authorization. Arcade manages the OAuth flow, and provides the token in context.authorization.token when the tool is executed.

class ContactManagerAgent:
    def __init__(self, arcade_client, user_id):
        self.client = arcade_client
        self.user_id = user_id
        self.authorized = False

    async def authorize_google_contacts(self):
        """Handle OAuth authorization for Google Contacts"""
        # Check authorization status
        auth_response = await self.client.tools.authorize(
            tool_name="GoogleContacts.CreateContact",
            user_id=self.user_id
        )

        if auth_response.status != "completed":
            print(f"Please authorize access: {auth_response.url}")
            # Wait for user to complete authorization
            await self.client.auth.wait_for_completion(auth_response)
            self.authorized = True
            print("Authorization successful!")
        else:
            self.authorized = True
            print("Already authorized")

        return auth_response

Step 3: Create Contact Management Functions

Build core functions for managing contacts with proper error handling and user context:

class ContactOperations:
    def __init__(self, arcade_client, user_id):
        self.client = arcade_client
        self.user_id = user_id

    async def create_contact(self, contact_data):
        """Create a new contact in Google Contacts"""
        try:
            response = await self.client.tools.execute(
                tool_name="GoogleContacts.CreateContact",
                input={
                    "given_name": contact_data.get("first_name"),
                    "family_name": contact_data.get("last_name"),
                    "email": contact_data.get("email"),
                },
                user_id=self.user_id
            )
            return {"success": True, "contact_id": response.output.id}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def search_contacts(self, name):
        """Search for contacts based on their name"""
        response = await self.client.tools.execute(
            tool_name="GoogleContacts.SearchContactsByName",
            input={"name": name},
            user_id=self.user_id
        )
        return response.output.contacts

    async def update_contact(self, contact_id, updates):
        """Update an existing contact"""
        response = await self.client.tools.execute(
            tool_name="GoogleContacts.UpdateContact",
            input={
                "contact_id": contact_id,
                **updates
            },
            user_id=self.user_id
        )
        return response.output

Implementing Multi-User Support

User Session Management

The class handles per-user OAuth flows, maintains session state, and executes Google actions with proper user context isolation. Apply the same pattern for Google Contacts:

from datetime import datetime
from typing import Dict, Any

class MultiUserContactManager:
    def __init__(self):
        self.client = Arcade(api_key=os.environ.get("ARCADE_API_KEY"))
        self.user_sessions: Dict[str, Any] = {}
        self.user_toolsets: Dict[str, Any] = {}

    async def authenticate_user(self, user_id: str) -> Dict[str, Any]:
        """Handle OAuth flow for a specific user"""

        # Check authorization status
        auth_response = await self.client.tools.authorize(
            tool_name="GoogleContacts.CreateContact",
            user_id=user_id
        )

        if auth_response.status != "completed":
            return {
                "authorization_required": True,
                "url": auth_response.url,
                "message": "Complete authorization to access Google Contacts"
            }

        # The user is already authenticated

        return {"authenticated": True}

Handling Concurrent Users

Implement proper isolation for concurrent user operations:

async def process_contact_request(self, user_id: str, request: Dict[str, Any]):
    """Process contact management requests with user isolation"""

    # Ensure user is authenticated
    await self.authenticate_user(user_idk)

    # Determine action type
    action = request.get("action")

    action_map = {
        "create": "GoogleContacts.CreateContact",
        "search": "GoogleContacts.SearchContacts",
        "update": "GoogleContacts.UpdateContact",
        "delete": "GoogleContacts.DeleteContact",
        "list": "GoogleContacts.ListContacts"
    }

    tool_name = action_map.get(action)

    if not tool_name:
        return {"error": "Invalid action"}

    # Execute with user context
    try:
        result = await self.client.tools.execute(
            tool_name=tool_name,
            input=request.get("parameters", {}),
            user_id=user_id
        )
        return {"success": True, "data": result.output}
    except Exception as e:
        if getattr(e, "type", "") == "authorization_required":
            return {
                "success": False,
                "authRequired": True,
                "authUrl": getattr(e, "url", "")
            }
        raise

Integration with AI Frameworks

LangChain Integration

Initialize the Arcade client const arcade = new Arcade(); // Get the Arcade tools, you can customize the toolkit (e.g. "github", "notion", "gmail", etc.). Here's how to integrate with LangChain:

from langchain_arcade import ArcadeToolManager
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

# Initialize tool manager
manager = ArcadeToolManager(api_key=arcade_api_key)

# Get Google Contacts tools
tools = manager.get_tools(toolkits=["GoogleContacts"])

# Create LLM with tools
llm = ChatOpenAI(model="gpt-4")
agent = create_react_agent(llm, tools)

# Define contact management prompt
prompt = """You are a contact management assistant.
Help users create, search, update, and organize their Google Contacts.
Always confirm actions before executing them."""

# Execute agent
response = await agent.invoke({
    "messages": [{"role": "user", "content": "Find all contacts from Acme Corp"}]
})

CrewAI Integration

Use the ArcadeToolManager to initialize, add, and get Arcade tools:

from crewai_arcade import ArcadeToolManager
from crewai import Agent, Task, Crew

# Initialize manager
manager = ArcadeToolManager(default_user_id=user_id)

# Get contact management tools
tools = manager.get_tools(toolkits=["GoogleContacts"])

# Create specialized agent
contact_agent = Agent(
    role='Contact Manager',
    goal='Efficiently manage and organize Google Contacts',
    backstory='Expert at maintaining clean, organized contact databases',
    tools=tools,
    verbose=True
)

# Define contact management task
organize_task = Task(
    description='Review and organize all contacts, removing duplicates',
    agent=contact_agent
)

# Create crew
crew = Crew(
    agents=[contact_agent],
    tasks=[organize_task]
)

Advanced Features

Custom Contact Enrichment

Build custom tools that extend Google Contacts functionality:

from arcade_mcp_server import Context, tool
from arcade_mcp_server.auth import Google

# The decorator will ensure the tool is executed ONLY if the
# OAuth requirements are met
@tool(
    requires_auth=Google(
        scopes=[
            "https://www.googleapis.com/auth/contacts",
            "https://www.googleapis.com/auth/contacts.readonly"
        ]
    )
)
async def enrich_contact_with_company_data(
    context: ToolContext,
    contact_id: str,
    company_data: dict
):
    """Enrich contact with additional company information"""

    # Use the authorized token to make API calls
    oauth_token = context.get_auth_token_or_empty()
    headers = {
        "Authorization": f"Bearer {oauth_token}",
        "Content-Type": "application/json"
    }

    # Add custom fields to contact
    enriched_data = {
        "company_size": company_data.get("size"),
        "industry": company_data.get("industry"),
        "linkedin_url": company_data.get("linkedin"),
        "last_interaction": datetime.now().isoformat()
    }

    # Update contact with enriched data
    return await update_contact_custom_fields(
        contact_id,
        enriched_data,
        headers
    )

Bulk Operations Handler

Implement efficient bulk operations for managing multiple contacts:

class BulkContactOperations:
    def __init__(self, arcade_client, user_id):
        self.client = arcade_client
        self.user_id = user_id

    async def bulk_import_contacts(self, csv_data):
        """Import multiple contacts from CSV"""
        results = []

        for row in csv_data:
            try:
                response = await self.client.tools.execute(
                    tool_name="GoogleContacts.CreateContact",
                    input={
                        "given_name": row.get("first_name"),
                        "family_name": row.get("last_name"),
                        "email": row.get("email"),
                        "phone": row.get("phone"),
                        "organization": row.get("company")
                    },
                    user_id=self.user_id
                )
                results.append({
                    "email": row.get("email"),
                    "status": "created",
                    "id": response.output.id
                })
            except Exception as e:
                results.append({
                    "email": row.get("email"),
                    "status": "failed",
                    "error": str(e)
                })

        return results

    async def deduplicate_contacts(self):
        """Find and merge duplicate contacts"""
        # Search all contacts
        all_contacts = await self.client.tools.execute(
            tool_name="GoogleContacts.ListContacts",
            input={"limit": 1000},
            user_id=self.user_id
        )

        # Group by email for deduplication
        email_groups = {}
        for contact in all_contacts.output.contacts:
            email = contact.get("email")
            if email:
                if email not in email_groups:
                    email_groups[email] = []
                email_groups[email].append(contact)

        # Identify duplicates
        duplicates = {
            email: contacts
            for email, contacts in email_groups.items()
            if len(contacts) > 1
        }

        return duplicates

Production Deployment

Cloud Deployment with Arcade Deploy

Use Arcade Deploy for managed cloud deployment:

cd <path to your project>

# Deploy your contact manager
arcade deploy --name contact-manager

# View deployment status
arcade server list

Security Best Practices

Token Management

Arcade manages access/refresh token rotation server-side; never log or persist tokens in application code. Implement secure token handling:

class SecureContactManager:
    def __init__(self):
        self.arcade = Arcade()
        # Never store tokens directly
        self.token_store = {}  # Use encrypted storage in production

    def check_user_status(self, tool_name: str, user_id: str) -> Dict:
        """Retrieve user context without exposing tokens"""
        # Arcade handles token management internally
        # Your code never sees the actual tokens
        requirements_met = True
        tool = client.tools.get(tool_name=TOOL_NAME, user_id=USER_ID)
        if tool.requirements:
            requirements_met = tool.requirements.mer
        return {
            'authenticated': requirements_met,
            'user_id': user_id
        }

Rate Limiting

Implement proper rate limiting and error handling:

import asyncio
from typing import Optional

class RateLimitedContactManager:
    def __init__(self):
        self.request_counts = {}
        self.rate_limit = 100  # requests per minute

    async def execute_with_rate_limit(
        self,
        tool_name: str,
        input: dict,
        user_id: str
    ):
        """Execute tool with rate limiting"""

        # Check rate limit
        if self.is_rate_limited(user_id):
            await self.handle_rate_limit()

        try:
            response = await self.client.tools.execute(
                tool_name=tool_name,
                input=input,
                user_id=user_id
            )
            self.increment_request_count(user_id)
            return response
        except Exception as e:
            if "rate_limit" in str(e):
                await asyncio.sleep(60)  # Wait before retry
                return await self.execute_with_rate_limit(
                    tool_name, input, user_id
                )
            raise

Monitoring and Observability

Track authentication metrics and system health:

class ContactManagerMonitor:
    def __init__(self):
        self.metrics = {
            "auth_attempts": 0,
            "auth_successes": 0,
            "contacts_created": 0,
            "search_queries": 0,
            "api_errors": 0
        }

    async def track_operation(self, operation_type: str, user_id: str):
        """Track contact management operations"""
        self.metrics[operation_type] += 1

        # Log to monitoring system
        await self.send_to_monitoring({
            'timestamp': datetime.now().isoformat(),
            'user_id': self.hash_user_id(user_id),
            'operation': operation_type,
            'service': 'google_contacts'
        })

    def generate_health_report(self) -> Dict:
        """Generate system health metrics"""
        return {
            'status': 'healthy',
            'total_operations': sum(self.metrics.values()),
            'error_rate': self.metrics['api_errors'] / max(sum(self.metrics.values()), 1),
            'auth_success_rate': self.metrics['auth_successes'] / max(self.metrics['auth_attempts'], 1)
        }

Testing Your Contact Manager

Create comprehensive tests for your agent:

import pytest
from unittest.mock import AsyncMock

@pytest.fixture
async def contact_manager():
    """Create test contact manager instance"""
    client = AsyncMock()
    manager = ContactManagerAgent(client, "test_user@example.com")
    return manager

@pytest.mark.asyncio
async def test_create_contact(contact_manager):
    """Test contact creation"""
    contact_data = {
        "first_name": "John",
        "last_name": "Doe",
        "email": "john@example.com",
        "phone": "+1234567890"
    }

    result = await contact_manager.create_contact(contact_data)
    assert result["success"] == True
    assert "contact_id" in result

@pytest.mark.asyncio
async def test_search_contacts(contact_manager):
    """Test contact search functionality"""
    results = await contact_manager.search_contacts("example.com")
    assert isinstance(results, list)

Conclusion

Building a contact manager AI agent with Arcade's Google Contacts toolkit provides a production-ready solution that handles authentication, multi-user support, and secure API access without the complexity of managing OAuth flows manually. Cut integration build time from weeks to minutes with pre-built connectors while maintaining enterprise-grade security and scalability.

The platform's authentication-first approach ensures your AI agents can safely act on behalf of users while maintaining proper security boundaries. With support for multiple AI frameworks and deployment options, you can build and scale contact management solutions that integrate seamlessly with existing workflows.

Next Steps

Explore additional toolkits in the Arcade Toolkits documentation
Build custom tools using the Tool Development Kit
Join the Arcade community on GitHub
Check the API Reference for detailed implementation guidance

Start building your contact manager agent today with Arcade.dev and transform how your applications interact with Google Contacts.

How to Build a Contact Manager AI Agent Using Arcade's Google Contacts Toolkit