google_ads_pmax_insights

Instructions

Implementation Reference

• tools/shopping_pmax/mcp_tools_shopping_pmax.py:748-859 (handler)

  The MCP tool handler function that receives the request, delegates to the manager, and formats the response output with performance metrics, asset group details, and optimization tips.

  @mcp.tool()
  def google_ads_pmax_insights(
      customer_id: str,
      campaign_id: str,
      date_range: str = "LAST_30_DAYS"
  ) -> str:
      """Get performance insights for a Performance Max campaign.
  
      Provides comprehensive metrics including all-conversions data which
      captures conversions across all Google properties.
  
      Args:
          customer_id: Google Ads customer ID (10 digits, no hyphens)
          campaign_id: Performance Max campaign ID
          date_range: Date range (LAST_7_DAYS, LAST_30_DAYS, LAST_90_DAYS)
  
      Returns:
          Performance Max campaign insights
  
      Example:
          google_ads_pmax_insights(
              customer_id="1234567890",
              campaign_id="12345678",
              date_range="LAST_30_DAYS"
          )
      """
      with performance_logger.track_operation('pmax_insights', customer_id=customer_id):
          try:
              client = get_auth_manager().get_client()
              shopping_manager = ShoppingPMaxManager(client)
  
              result = shopping_manager.get_pmax_insights(
                  customer_id=customer_id,
                  campaign_id=campaign_id,
                  date_range=date_range
              )
  
              if 'error' in result:
                  return f"❌ {result['error']}"
  
              audit_logger.log_api_call(
                  customer_id=customer_id,
                  operation='pmax_insights',
                  campaign_id=campaign_id,
                  status='success'
              )
  
              metrics = result['metrics']
  
              output = f"# 🚀 Performance Max Insights\n\n"
              output += f"**Campaign**: {result['campaign_name']}\n"
              output += f"**Campaign ID**: {result['campaign_id']}\n"
              output += f"**Date Range**: {date_range}\n\n"
  
              output += "## Campaign Performance\n\n"
              output += f"- **Impressions**: {metrics['impressions']:,}\n"
              output += f"- **Clicks**: {metrics['clicks']:,}\n"
              output += f"- **CTR**: {metrics['ctr']:.2%}\n"
              output += f"- **Cost**: ${metrics['cost']:,.2f}\n"
              output += f"- **Conversions**: {metrics['conversions']:.1f}\n"
              output += f"- **Conversion Value**: ${metrics['conversion_value']:,.2f}\n"
              output += f"- **ROAS**: {metrics['roas']:.2f}x\n\n"
  
              output += "## All-Conversions Metrics\n\n"
              output += "*Includes conversions from all Google properties*\n\n"
              output += f"- **All Conversions**: {metrics['all_conversions']:.1f}\n"
              output += f"- **All Conversion Value**: ${metrics['all_conversions_value']:,.2f}\n\n"
  
              # Asset Groups
              output += f"## Asset Groups ({result['total_asset_groups']})\n\n"
  
              if result['total_asset_groups'] > 0:
                  output += "| Asset Group | Status | Impressions | Clicks | Conv. |\n"
                  output += "|-------------|--------|-------------|--------|-------|\n"
  
                  for ag in result['asset_groups'][:5]:  # Top 5
                      output += f"| {ag['asset_group_name'][:20]} | "
                      output += f"{ag['status']} | "
                      output += f"{ag['impressions']:,} | "
                      output += f"{ag['clicks']:,} | "
                      output += f"{ag['conversions']:.1f} |\n"
              else:
                  output += "No asset groups found. Create asset groups to start serving ads.\n"
  
              output += "\n## Performance Analysis\n\n"
  
              # ROAS analysis
              if metrics['roas'] > 4.0:
                  output += "✅ **Excellent ROAS!** Campaign is highly profitable.\n"
                  output += "💡 Consider increasing budget to scale.\n"
              elif metrics['roas'] > 2.0:
                  output += "💚 **Good ROAS.** Campaign is performing well.\n"
                  output += "💡 Monitor and optimize asset groups.\n"
              elif metrics['roas'] > 1.0:
                  output += "⚠️ **Marginal ROAS.** Room for improvement.\n"
                  output += "💡 Review assets and audience signals.\n"
              else:
                  output += "🚨 **Low ROAS.** Campaign needs optimization.\n"
                  output += "💡 Check assets, budgets, and conversion tracking.\n"
  
              output += "\n## Optimization Tips\n\n"
              output += "1. 📝 Test different headlines and descriptions\n"
              output += "2. 🖼️ Add high-quality images in multiple sizes\n"
              output += "3. 🎯 Refine audience signals based on performance\n"
              output += "4. 💰 Ensure adequate budget (at least 10x target CPA)\n"
              output += "5. ⏱️ Allow 6-8 weeks for AI learning\n"
  
              return output
  
          except Exception as e:
              error_msg = ErrorHandler.handle_error(e, context="pmax_insights")
              return f"❌ Failed to get Performance Max insights: {error_msg}"

• managers/shopping_pmax_manager.py:754-844 (helper)

  The manager helper that executes GAQL queries via GoogleAdsService to fetch campaign-level metrics (impressions, clicks, cost, conversions, ROAS) and asset-group-level performance data, then returns structured results.

  def get_pmax_insights(
      self,
      customer_id: str,
      campaign_id: str,
      date_range: str = "LAST_30_DAYS"
  ) -> Dict[str, Any]:
      """Get Performance Max campaign insights.
  
      Args:
          customer_id: Customer ID (without hyphens)
          campaign_id: Performance Max campaign ID
          date_range: Date range for metrics
  
      Returns:
          Performance Max insights
      """
      ga_service = self.client.get_service("GoogleAdsService")
  
      # Get campaign performance
      campaign_query = f"""
          SELECT
              campaign.id,
              campaign.name,
              metrics.impressions,
              metrics.clicks,
              metrics.ctr,
              metrics.cost_micros,
              metrics.conversions,
              metrics.conversions_value,
              metrics.all_conversions,
              metrics.all_conversions_value
          FROM campaign
          WHERE campaign.id = {campaign_id}
            AND segments.date DURING {date_range}
      """
  
      campaign_response = ga_service.search(customer_id=customer_id, query=campaign_query)
      campaign_results = list(campaign_response)
  
      if not campaign_results:
          return {'error': 'Campaign not found or no data available'}
  
      row = campaign_results[0]
  
      # Get asset group performance
      asset_group_query = f"""
          SELECT
              asset_group.id,
              asset_group.name,
              asset_group.status,
              metrics.impressions,
              metrics.clicks,
              metrics.conversions
          FROM asset_group
          WHERE campaign.id = {campaign_id}
            AND segments.date DURING {date_range}
      """
  
      asset_group_response = ga_service.search(customer_id=customer_id, query=asset_group_query)
  
      asset_groups = []
      for ag_row in asset_group_response:
          asset_groups.append({
              'asset_group_id': str(ag_row.asset_group.id),
              'asset_group_name': ag_row.asset_group.name,
              'status': ag_row.asset_group.status.name,
              'impressions': ag_row.metrics.impressions,
              'clicks': ag_row.metrics.clicks,
              'conversions': ag_row.metrics.conversions
          })
  
      cost = row.metrics.cost_micros / 1_000_000
      roas = (row.metrics.conversions_value / cost) if cost > 0 else 0
  
      return {
          'campaign_id': str(row.campaign.id),
          'campaign_name': row.campaign.name,
          'metrics': {
              'impressions': row.metrics.impressions,
              'clicks': row.metrics.clicks,
              'ctr': row.metrics.ctr,
              'cost': cost,
              'conversions': row.metrics.conversions,
              'conversion_value': row.metrics.conversions_value,
              'all_conversions': row.metrics.all_conversions,
              'all_conversions_value': row.metrics.all_conversions_value,
              'roas': roas
          },
          'asset_groups': asset_groups,
          'total_asset_groups': len(asset_groups)
      }

• tools/shopping_pmax/mcp_tools_shopping_pmax.py:37-37 (registration)

  The registration function that uses the @mcp.tool() decorator to register google_ads_pmax_insights (and other tools) as MCP tools.

  def register_shopping_pmax_tools(mcp):

• google_ads_mcp.py:492-492 (registration)

  The module registration entry in the _TOOL_MODULES list in the main MCP server file, which triggers dynamic import and registration.

  ("shopping_pmax", "tools.shopping_pmax.mcp_tools_shopping_pmax", "register_shopping_pmax_tools"),