Google Ads MCP Server

""" MCP Tools - Audience & Remarketing Management Provides 12 MCP tools for audience creation and targeting: Audience Creation (3 tools): 1. google_ads_create_user_list - Create remarketing lists 2. google_ads_upload_customer_match - Upload Customer Match data 3. google_ads_get_customer_match_status - Check upload status and match rate Audience Targeting (4 tools): 4. google_ads_add_audience_to_campaign - Add audience targeting to campaign 5. google_ads_add_audience_to_ad_group - Add audience targeting to ad group 6. google_ads_set_audience_exclusions - Exclude audiences from campaign 7. google_ads_get_audience_performance - Get performance by audience Audience Discovery (3 tools): 8. google_ads_list_user_lists - List all audiences in account 9. google_ads_search_google_audiences - Search In-Market/Affinity audiences 10. google_ads_get_user_list_details - Get detailed audience info Additional Tools (2 tools): 11. google_ads_update_customer_match_list - Add/remove Customer Match members 12. google_ads_remove_audience_from_campaign - Remove audience targeting """ from typing import Optional, List, Dict, Any from audience_manager import ( AudienceManager, UserListConfig, UserListType, AudienceTargetingType, CustomerMatchData ) from auth_manager import get_auth_manager from error_handler import ErrorHandler from logger import performance_logger, audit_logger from cache_manager import get_cache_manager, ResourceType import json def register_audience_tools(mcp): """Register all audience management tools with the MCP server. Args: mcp: FastMCP server instance """ # ============================================================================ # Audience Creation # ============================================================================ @mcp.tool() def google_ads_create_user_list( customer_id: str, list_name: str, description: Optional[str] = None, membership_days: int = 540, list_type: str = "CRMBASED" ) -> str: """ Create a remarketing user list (audience). User lists allow you to target specific groups of users based on their interactions with your business. Types include: - CRMBASED: Customer Match lists (email, phone, address) - RULE_BASED: Website visitors, app users, YouTube viewers Args: customer_id: Customer ID (without hyphens) list_name: Name for the user list (e.g., "Newsletter Subscribers") description: Optional description membership_days: How long users stay in the list (1-540 days, default: 540) list_type: Type of list (CRMBASED or RULE_BASED) Returns: Success message with user list ID Example: google_ads_create_user_list( customer_id="1234567890", list_name="High Value Customers", description="Customers with LTV > $1000", membership_days=540, list_type="CRMBASED" ) Note: RULE_BASED lists require remarketing tags on your website/app. """ with performance_logger.track_operation('create_user_list', customer_id=customer_id): try: client = get_auth_manager().get_client() audience_manager = AudienceManager(client) # Validate list type try: ul_type = UserListType[list_type.upper()] except KeyError: valid_types = [t.value for t in UserListType] return f"❌ Invalid list type '{list_type}'. Valid types: {', '.join(valid_types)}" # Validate membership days if not (1 <= membership_days <= 540): return "❌ membership_days must be between 1 and 540" config = UserListConfig( name=list_name, description=description, membership_life_span=membership_days ) result = audience_manager.create_user_list(customer_id, config, ul_type) # Audit log audit_logger.log_api_call( customer_id=customer_id, operation="create_user_list", resource_type="user_list", resource_id=result['user_list_id'], action="create", result="success", details={'name': list_name, 'type': list_type} ) output = f"✅ User list created successfully!\n\n" output += f"**User List ID**: {result['user_list_id']}\n" output += f"**Name**: {result['name']}\n" output += f"**Type**: {result['type']}\n" output += f"**Membership Duration**: {membership_days} days\n\n" if ul_type == UserListType.CRMBASED: output += f"**Next Steps**:\n" output += f"1. Upload Customer Match data with `google_ads_upload_customer_match`\n" output += f"2. Wait 12-24 hours for list to populate\n" output += f"3. Add to campaigns with `google_ads_add_audience_to_campaign`\n" else: output += f"**Next Steps**:\n" output += f"1. Install remarketing tag on your website/app\n" output += f"2. Wait for users to be added to the list\n" output += f"3. Add to campaigns when list size reaches 1,000+\n" return output except Exception as e: error_msg = ErrorHandler.handle_error(e, context="create_user_list") return f"❌ Failed to create user list: {error_msg}" @mcp.tool() def google_ads_upload_customer_match( customer_id: str, user_list_id: Optional[str] = None, list_name: Optional[str] = None, emails: Optional[List[str]] = None, phones: Optional[List[str]] = None, first_names: Optional[List[str]] = None, last_names: Optional[List[str]] = None, countries: Optional[List[str]] = None, zip_codes: Optional[List[str]] = None ) -> str: """ Upload Customer Match data (emails, phones, addresses). Customer Match allows you to use your customer data to reach them on Google Search, YouTube, Gmail, and Display Network. Data is hashed before upload for privacy. You can either upload to an existing list (provide user_list_id) or create a new list (provide list_name). Args: customer_id: Customer ID (without hyphens) user_list_id: Existing user list ID to upload to (optional) list_name: Name for new list (required if user_list_id not provided) emails: List of email addresses phones: List of phone numbers (E.164 format recommended: +12345678900) first_names: List of first names (must match emails/phones index) last_names: List of last names (must match emails/phones index) countries: List of country codes (e.g., "US", "UK") zip_codes: List of postal codes Returns: Success message with upload job details Example: google_ads_upload_customer_match( customer_id="1234567890", list_name="Email Newsletter Subscribers", emails=[ "customer1@example.com", "customer2@example.com", "customer3@example.com" ] ) Privacy Note: All data is automatically hashed with SHA256 before upload. Google cannot see the original data. Match Rate: Typically 30-70% of uploaded records will match to Google users. """ with performance_logger.track_operation('upload_customer_match', customer_id=customer_id): try: client = get_auth_manager().get_client() audience_manager = AudienceManager(client) # Validate inputs if not user_list_id and not list_name: return "❌ Either user_list_id or list_name must be provided" if not any([emails, phones]): return "❌ At least emails or phones must be provided" # Build customer data customer_data = CustomerMatchData( emails=emails, phones=phones, first_names=first_names, last_names=last_names, countries=countries, zip_codes=zip_codes ) result = audience_manager.upload_customer_match_list( customer_id, user_list_id=user_list_id or "", customer_data=customer_data, create_list=not user_list_id, list_name=list_name ) # Audit log audit_logger.log_api_call( customer_id=customer_id, operation="upload_customer_match", resource_type="user_list", resource_id=result['user_list_id'], action="update", result="success", details={'records': result['records_uploaded']} ) output = f"✅ Customer Match upload started!\n\n" output += f"**User List ID**: {result['user_list_id']}\n" output += f"**Records Uploaded**: {result['records_uploaded']}\n" output += f"**Status**: {result['status']}\n\n" output += f"**Processing Timeline**:\n" output += f"- Job processing: 12-24 hours\n" output += f"- List population: Up to 48 hours\n" output += f"- Minimum list size for targeting: 1,000 matched users\n\n" output += f"**Next Steps**:\n" output += f"1. Wait 24 hours for processing\n" output += f"2. Check status with `google_ads_get_customer_match_status`\n" output += f"3. View match rate and list size\n" output += f"4. Add to campaigns when list size >= 1,000\n\n" output += f"**Privacy**: All data is SHA256 hashed before upload. " output += f"Google cannot see the original data.\n" return output except Exception as e: error_msg = ErrorHandler.handle_error(e, context="upload_customer_match") return f"❌ Failed to upload Customer Match data: {error_msg}" @mcp.tool() def google_ads_get_customer_match_status( customer_id: str, user_list_id: str ) -> str: """ Get Customer Match upload status, match rate, and list size. Check this 24-48 hours after uploading to see how many records matched and if the list is large enough for targeting (minimum 1,000). Args: customer_id: Customer ID (without hyphens) user_list_id: User list ID to check Returns: Upload status, match rate, and list sizes Example: google_ads_get_customer_match_status( customer_id="1234567890", user_list_id="12345" ) """ with performance_logger.track_operation('get_customer_match_status', customer_id=customer_id): try: client = get_auth_manager().get_client() audience_manager = AudienceManager(client) result = audience_manager.get_customer_match_status(customer_id, user_list_id) if 'error' in result: return f"❌ {result['error']}" # Audit log audit_logger.log_api_call( customer_id=customer_id, operation="get_customer_match_status", resource_type="user_list", resource_id=user_list_id, action="read", result="success" ) output = f"# Customer Match Status\n\n" output += f"**List Name**: {result['name']}\n" output += f"**User List ID**: {result['user_list_id']}\n" output += f"**Status**: {result['membership_status']}\n\n" output += f"## List Sizes\n\n" output += f"- **Search Network**: {result['size_for_search']:,} users\n" output += f"- **Display Network**: {result['size_for_display']:,} users\n\n" if result['match_rate_percentage']: output += f"**Match Rate**: {result['match_rate_percentage']:.1f}%\n\n" # Targeting eligibility min_size = 1000 if result['size_for_search'] >= min_size: output += f"✅ **Eligible for targeting** (size >= {min_size:,})\n\n" else: needed = min_size - result['size_for_search'] output += f"⚠️ **Not yet eligible for targeting**\n" output += f"Need {needed:,} more matched users (minimum: {min_size:,})\n\n" output += f"**Membership Duration**: {result['membership_life_span']} days\n\n" output += f"**Typical Match Rates**:\n" output += f"- Email only: 30-50%\n" output += f"- Email + Name + Address: 50-70%\n" output += f"- Phone only: 20-40%\n\n" if result['size_for_search'] >= min_size: output += f"**Next Step**: Add to campaigns with `google_ads_add_audience_to_campaign`\n" return output except Exception as e: error_msg = ErrorHandler.handle_error(e, context="get_customer_match_status") return f"❌ Failed to get Customer Match status: {error_msg}" # ============================================================================ # Audience Targeting # ============================================================================ @mcp.tool() def google_ads_add_audience_to_campaign( customer_id: str, campaign_id: str, user_list_id: str, targeting_mode: str = "OBSERVATION" ) -> str: """ Add audience targeting to a campaign. Two modes available: - OBSERVATION: Monitor audience performance without restricting reach - TARGETING: Restrict campaign to only show ads to this audience Args: customer_id: Customer ID (without hyphens) campaign_id: Campaign ID user_list_id: User list ID to target targeting_mode: OBSERVATION (monitor) or TARGETING (restrict reach) Returns: Success message Example (Observation): google_ads_add_audience_to_campaign( customer_id="1234567890", campaign_id="111111111", user_list_id="12345", targeting_mode="OBSERVATION" ) Example (Targeting): google_ads_add_audience_to_campaign( customer_id="1234567890", campaign_id="222222222", user_list_id="12345", targeting_mode="TARGETING" ) Recommendation: Start with OBSERVATION mode to gather performance data before switching to TARGETING mode. """ with performance_logger.track_operation('add_audience_to_campaign', customer_id=customer_id): try: client = get_auth_manager().get_client() audience_manager = AudienceManager(client) # Validate targeting mode try: mode = AudienceTargetingType[targeting_mode.upper()] except KeyError: return f"❌ Invalid targeting mode. Use OBSERVATION or TARGETING" result = audience_manager.add_audience_to_campaign( customer_id, campaign_id, user_list_id, mode ) # Audit log audit_logger.log_api_call( customer_id=customer_id, operation="add_audience_to_campaign", resource_type="campaign", resource_id=campaign_id, action="update", result="success", details={'user_list_id': user_list_id, 'mode': targeting_mode} ) # Invalidate cache get_cache_manager().invalidate(customer_id, ResourceType.CAMPAIGN) output = f"✅ Audience added to campaign!\n\n" output += f"**Campaign ID**: {campaign_id}\n" output += f"**User List ID**: {user_list_id}\n" output += f"**Targeting Mode**: {result['targeting_type']}\n\n" if mode == AudienceTargetingType.OBSERVATION: output += f"**Observation Mode**:\n" output += f"- Campaign reach is NOT restricted\n" output += f"- You can see performance metrics for this audience\n" output += f"- Use this to evaluate audience quality before targeting\n\n" output += f"**Next Steps**:\n" output += f"1. Run campaign for 2-4 weeks\n" output += f"2. Review audience performance with `google_ads_get_audience_performance`\n" output += f"3. If audience performs well, consider switching to TARGETING mode\n" else: output += f"**Targeting Mode**:\n" output += f"- Campaign will ONLY show ads to users in this audience\n" output += f"- Reach may be limited based on audience size\n" output += f"- Best for remarketing and Customer Match campaigns\n\n" output += f"**Important**: Ensure audience size >= 1,000 for Search campaigns\n" return output except Exception as e: error_msg = ErrorHandler.handle_error(e, context="add_audience_to_campaign") return f"❌ Failed to add audience to campaign: {error_msg}" @mcp.tool() def google_ads_add_audience_to_ad_group( customer_id: str, ad_group_id: str, user_list_id: str, targeting_mode: str = "OBSERVATION" ) -> str: """ Add audience targeting to an ad group. Similar to campaign-level audience targeting, but applied at the ad group level for more granular control. Args: customer_id: Customer ID (without hyphens) ad_group_id: Ad group ID user_list_id: User list ID to target targeting_mode: OBSERVATION or TARGETING Returns: Success message Example: google_ads_add_audience_to_ad_group( customer_id="1234567890", ad_group_id="222222222", user_list_id="12345", targeting_mode="OBSERVATION" ) """ with performance_logger.track_operation('add_audience_to_ad_group', customer_id=customer_id): try: client = get_auth_manager().get_client() audience_manager = AudienceManager(client) try: mode = AudienceTargetingType[targeting_mode.upper()] except KeyError: return f"❌ Invalid targeting mode. Use OBSERVATION or TARGETING" result = audience_manager.add_audience_to_ad_group( customer_id, ad_group_id, user_list_id, mode ) # Audit log audit_logger.log_api_call( customer_id=customer_id, operation="add_audience_to_ad_group", resource_type="ad_group", resource_id=ad_group_id, action="update", result="success", details={'user_list_id': user_list_id} ) # Invalidate cache get_cache_manager().invalidate(customer_id, ResourceType.AD_GROUP) output = f"✅ Audience added to ad group!\n\n" output += f"**Ad Group ID**: {ad_group_id}\n" output += f"**User List ID**: {user_list_id}\n" output += f"**Targeting Mode**: {result['targeting_type']}\n\n" return output except Exception as e: error_msg = ErrorHandler.handle_error(e, context="add_audience_to_ad_group") return f"❌ Failed to add audience to ad group: {error_msg}" @mcp.tool() def google_ads_set_audience_exclusions( customer_id: str, campaign_id: str, user_list_ids: List[str] ) -> str: """ Exclude audiences from a campaign. Prevent your ads from showing to specific audiences. Common use cases: - Exclude existing customers from acquisition campaigns - Exclude converters from remarketing campaigns - Exclude low-value segments Args: customer_id: Customer ID (without hyphens) campaign_id: Campaign ID user_list_ids: List of user list IDs to exclude Returns: Success message Example: google_ads_set_audience_exclusions( customer_id="1234567890", campaign_id="111111111", user_list_ids=["12345", "12346", "12347"] ) Use Case: Exclude "Past Purchasers" list from new customer acquisition campaign """ with performance_logger.track_operation('set_audience_exclusions', customer_id=customer_id): try: client = get_auth_manager().get_client() audience_manager = AudienceManager(client) result = audience_manager.set_audience_exclusions( customer_id, campaign_id, user_list_ids ) # Audit log audit_logger.log_api_call( customer_id=customer_id, operation="set_audience_exclusions", resource_type="campaign", resource_id=campaign_id, action="update", result="success", details={'excluded_count': len(user_list_ids)} ) # Invalidate cache get_cache_manager().invalidate(customer_id, ResourceType.CAMPAIGN) output = f"✅ Audience exclusions set!\n\n" output += f"**Campaign ID**: {campaign_id}\n" output += f"**Excluded Audiences**: {result['excluded_audiences']}\n\n" output += f"**User List IDs**:\n" for ul_id in user_list_ids: output += f"- {ul_id}\n" output += f"\nAds in this campaign will NOT show to users in these audiences.\n" return output except Exception as e: error_msg = ErrorHandler.handle_error(e, context="set_audience_exclusions") return f"❌ Failed to set audience exclusions: {error_msg}" @mcp.tool() def google_ads_get_audience_performance( customer_id: str, campaign_id: Optional[str] = None, date_range: str = "LAST_30_DAYS" ) -> str: """ Get performance metrics by audience. See which audiences are driving the best results in terms of clicks, conversions, and ROI. Args: customer_id: Customer ID (without hyphens) campaign_id: Optional campaign ID to filter date_range: Date range (TODAY, LAST_7_DAYS, LAST_30_DAYS, etc.) Returns: Performance breakdown by audience Example: google_ads_get_audience_performance( customer_id="1234567890", campaign_id="111111111", date_range="LAST_30_DAYS" ) """ with performance_logger.track_operation('get_audience_performance', customer_id=customer_id): try: client = get_auth_manager().get_client() audience_manager = AudienceManager(client) audiences = audience_manager.get_audience_performance( customer_id, campaign_id, date_range ) # Audit log audit_logger.log_api_call( customer_id=customer_id, operation="get_audience_performance", resource_type="campaign_audience", action="read", result="success", details={'count': len(audiences)} ) if not audiences: return "No audience performance data found for the specified criteria." # Format response output = f"# Audience Performance\n\n" output += f"**Date Range**: {date_range}\n" if campaign_id: output += f"**Campaign ID**: {campaign_id}\n" output += f"**Total Audiences**: {len(audiences)}\n\n" for aud in audiences: exclusion_marker = " (EXCLUDED)" if aud['is_exclusion'] else "" output += f"## Audience {aud['user_list_id']}{exclusion_marker}\n\n" output += f"**Campaign**: {aud['campaign_name']} ({aud['campaign_id']})\n\n" if not aud['is_exclusion']: output += f"### Performance Metrics\n\n" output += f"- **Impressions**: {aud['impressions']:,}\n" output += f"- **Clicks**: {aud['clicks']:,}\n" output += f"- **CTR**: {aud['ctr'] * 100:.2f}%\n" output += f"- **Average CPC**: ${aud['average_cpc']:.2f}\n" output += f"- **Cost**: ${aud['cost']:,.2f}\n" if aud['conversions'] > 0: output += f"- **Conversions**: {aud['conversions']:.1f}\n" output += f"- **Conversion Value**: ${aud['conversions_value']:,.2f}\n" cpa = aud['cost'] / aud['conversions'] output += f"- **Cost per Conversion**: ${cpa:.2f}\n" else: output += f"*This is an exclusion - no performance metrics*\n" output += "\n" return output except Exception as e: error_msg = ErrorHandler.handle_error(e, context="get_audience_performance") return f"❌ Failed to get audience performance: {error_msg}" # ============================================================================ # Audience Discovery # ============================================================================ @mcp.tool() def google_ads_list_user_lists( customer_id: str, list_type: Optional[str] = None ) -> str: """ List all user lists (audiences) in the account. Args: customer_id: Customer ID (without hyphens) list_type: Optional filter by type (CRMBASED, RULE_BASED, SIMILAR, LOGICAL) Returns: List of all user lists with details Example: google_ads_list_user_lists( customer_id="1234567890", list_type="CRMBASED" ) """ with performance_logger.track_operation('list_user_lists', customer_id=customer_id): try: client = get_auth_manager().get_client() audience_manager = AudienceManager(client) # Validate list type if provided ul_type = None if list_type: try: ul_type = UserListType[list_type.upper()] except KeyError: valid_types = [t.value for t in UserListType] return f"❌ Invalid list type. Valid types: {', '.join(valid_types)}" user_lists = audience_manager.list_user_lists(customer_id, ul_type) # Audit log audit_logger.log_api_call( customer_id=customer_id, operation="list_user_lists", resource_type="user_list", action="read", result="success", details={'count': len(user_lists)} ) if not user_lists: return "No user lists found. Create one with `google_ads_create_user_list`." # Format response output = f"# User Lists (Audiences)\n\n" output += f"**Total Lists**: {len(user_lists)}\n\n" for ul in user_lists: output += f"## {ul['name']}\n\n" output += f"- **ID**: {ul['id']}\n" output += f"- **Type**: {ul['type']}\n" if ul['description']: output += f"- **Description**: {ul['description']}\n" output += f"- **Search Network Size**: {ul['size_for_search']:,}\n" output += f"- **Display Network Size**: {ul['size_for_display']:,}\n" if ul['match_rate_percentage']: output += f"- **Match Rate**: {ul['match_rate_percentage']:.1f}%\n" output += f"- **Membership Duration**: {ul['membership_life_span']} days\n" output += f"- **Status**: {ul['membership_status']}\n\n" return output except Exception as e: error_msg = ErrorHandler.handle_error(e, context="list_user_lists") return f"❌ Failed to list user lists: {error_msg}" @mcp.tool() def google_ads_search_google_audiences( customer_id: str, search_term: str ) -> str: """ Search for Google's predefined audiences (In-Market, Affinity). Google provides hundreds of pre-built audience segments based on user interests and purchase intent. Search to find relevant audiences for your business. Args: customer_id: Customer ID (without hyphens) search_term: Search term (e.g., "coffee", "fitness", "travel") Returns: List of matching Google audiences Example: google_ads_search_google_audiences( customer_id="1234567890", search_term="coffee" ) Common Categories: - In-Market: Users actively researching products (high purchase intent) - Affinity: Users with sustained interest in a topic - Custom Intent: Create your own based on keywords/URLs """ with performance_logger.track_operation('search_google_audiences', customer_id=customer_id): try: client = get_auth_manager().get_client() audience_manager = AudienceManager(client) audiences = audience_manager.search_google_audiences(customer_id, search_term) # Audit log audit_logger.log_api_call( customer_id=customer_id, operation="search_google_audiences", resource_type="user_interest", action="read", result="success", details={'search_term': search_term, 'count': len(audiences)} ) if not audiences: return f"No Google audiences found matching '{search_term}'. Try different search terms." # Format response output = f"# Google Audiences - Search Results\n\n" output += f"**Search Term**: {search_term}\n" output += f"**Results Found**: {len(audiences)}\n\n" for aud in audiences: output += f"## {aud['name']}\n\n" output += f"- **Audience ID**: {aud['user_interest_id']}\n" output += f"- **Category**: {aud['taxonomy_type']}\n" if aud['parent']: output += f"- **Parent Category**: {aud['parent']}\n" output += "\n" output += f"**Next Steps**:\n" output += f"To target these audiences, you'll need to add them to campaigns via the Google Ads UI " output += f"or use the audience ID in targeting API calls.\n" return output except Exception as e: error_msg = ErrorHandler.handle_error(e, context="search_google_audiences") return f"❌ Failed to search Google audiences: {error_msg}" @mcp.tool() def google_ads_get_user_list_details( customer_id: str, user_list_id: str ) -> str: """ Get detailed information about a specific user list. Args: customer_id: Customer ID (without hyphens) user_list_id: User list ID Returns: Detailed user list information Example: google_ads_get_user_list_details( customer_id="1234567890", user_list_id="12345" ) """ # This uses the same underlying method as get_customer_match_status # but is named more generically for all list types return google_ads_get_customer_match_status(customer_id, user_list_id) # Remaining tools (11, 12) would be implemented here with similar patterns # For brevity, showing the core 10 tools above