google_ads_create_responsive_search_ad

Instructions

Implementation Reference

• tools/ads/mcp_tools_ads.py:32-140 (handler)

  The main tool handler function `google_ads_create_responsive_search_ad` decorated with @mcp.tool(). Validates input (3-15 headlines, 2-4 descriptions), creates a ResponsiveSearchAdConfig, delegates to AdManager.create_responsive_search_ad, performs audit logging and cache invalidation, and returns a formatted success/error message.

  def google_ads_create_responsive_search_ad(
      customer_id: str,
      ad_group_id: str,
      headlines: List[str],
      descriptions: List[str],
      final_urls: List[str],
      path1: Optional[str] = None,
      path2: Optional[str] = None,
      status: str = "PAUSED"
  ) -> str:
      """
      Create a Responsive Search Ad (RSA).
  
      Args:
          customer_id: Customer ID (without hyphens)
          ad_group_id: Ad group ID
          headlines: List of 3-15 headline texts (max 30 chars each)
          descriptions: List of 2-4 description texts (max 90 chars each)
          final_urls: List of final URLs (landing pages)
          path1: Optional display path 1 (max 15 chars)
          path2: Optional display path 2 (max 15 chars)
          status: Initial status (ENABLED or PAUSED, default: PAUSED)
  
      Returns:
          Success message with ad details
  
      Example:
          headlines = [
              "Premium Running Shoes",
              "Free Shipping Today",
              "Shop Nike & Adidas"
          ]
          descriptions = [
              "Browse our selection of top running shoes",
              "30-day money back guarantee"
          ]
          final_urls = ["https://example.com/shoes"]
      """
      with performance_logger.track_operation('create_responsive_search_ad', customer_id=customer_id):
          try:
              client = get_auth_manager().get_client()
              ad_manager = AdManager(client)
  
              # Validate headlines and descriptions
              if len(headlines) < 3 or len(headlines) > 15:
                  return "❌ Must provide 3-15 headlines"
  
              if len(descriptions) < 2 or len(descriptions) > 4:
                  return "❌ Must provide 2-4 descriptions"
  
              # Create config
              config = ResponsiveSearchAdConfig(
                  ad_group_id=ad_group_id,
                  headlines=headlines,
                  descriptions=descriptions,
                  path1=path1,
                  path2=path2,
                  final_urls=final_urls,
                  status=AdStatus[status.upper()]
              )
  
              # Create ad
              result = ad_manager.create_responsive_search_ad(customer_id, config)
  
              # Audit log
              audit_logger.log_api_call(
                  customer_id=customer_id,
                  operation="create_responsive_search_ad",
                  resource_type="ad",
                  resource_id=result['ad_id'],
                  action="create",
                  result="success",
                  details={
                      'ad_group_id': ad_group_id,
                      'headline_count': len(headlines),
                      'description_count': len(descriptions)
                  }
              )
  
              # Invalidate cache
              get_cache_manager().invalidate(customer_id, ResourceType.AD)
  
              output = f"✅ Responsive Search Ad created successfully!\n\n"
              output += f"**Ad ID**: {result['ad_id']}\n"
              output += f"**Ad Group ID**: {ad_group_id}\n"
              output += f"**Status**: {status}\n"
              output += f"**Headlines**: {result['headline_count']}\n"
              output += f"**Descriptions**: {result['description_count']}\n\n"
  
              output += "**Headlines**:\n"
              for i, h in enumerate(headlines[:5], 1):
                  output += f"{i}. {h}\n"
              if len(headlines) > 5:
                  output += f"... and {len(headlines) - 5} more\n"
  
              output += "\n**Descriptions**:\n"
              for i, d in enumerate(descriptions, 1):
                  output += f"{i}. {d}\n"
  
              output += f"\n**Final URL**: {final_urls[0]}\n"
  
              if status == "PAUSED":
                  output += "\nℹ️ Ad is paused. Enable it when ready to start serving."
  
              return output
  
          except Exception as e:
              error_msg = ErrorHandler.handle_error(e, context="create_responsive_search_ad")
              return f"❌ Failed to create ad: {error_msg}"

• managers/ad_manager.py:47-56 (schema)

  ResponsiveSearchAdConfig dataclass defining the schema for RSA creation: ad_group_id, headlines, descriptions, optional path1/path2, optional final_urls, and AdStatus enum (default PAUSED).

  @dataclass
  class ResponsiveSearchAdConfig:
      """Configuration for creating a Responsive Search Ad."""
      ad_group_id: str
      headlines: List[str]  # 3-15 headlines
      descriptions: List[str]  # 2-4 descriptions
      path1: Optional[str] = None
      path2: Optional[str] = None
      final_urls: Optional[List[str]] = None
      status: AdStatus = AdStatus.PAUSED

• managers/ad_manager.py:79-152 (helper)

  AdManager.create_responsive_search_ad method that builds the AdGroupAd protobuf operation, sets headlines/descriptions/paths/final_urls, calls mutate_ad_group_ads API, and returns ad_id, resource_name, and counts.

  def create_responsive_search_ad(
      self,
      customer_id: str,
      config: ResponsiveSearchAdConfig
  ) -> Dict[str, Any]:
      """
      Create a Responsive Search Ad (RSA).
  
      Args:
          customer_id: Customer ID
          config: RSA configuration
  
      Returns:
          Created ad details
      """
      ad_group_ad_service = self.client.get_service("AdGroupAdService")
      ad_group_service = self.client.get_service("AdGroupService")
  
      # Create ad group ad operation
      ad_group_ad_operation = self.client.get_type("AdGroupAdOperation")
      ad_group_ad = ad_group_ad_operation.create
  
      # Set ad group
      ad_group_ad.ad_group = ad_group_service.ad_group_path(
          customer_id, config.ad_group_id
      )
  
      # Set status
      ad_group_ad.status = self.client.enums.AdGroupAdStatusEnum[config.status.value]
  
      # Create responsive search ad
      rsa = ad_group_ad.ad.responsive_search_ad
  
      # Add headlines (3-15 required)
      for headline_text in config.headlines:
          headline = self.client.get_type("AdTextAsset")
          headline.text = headline_text
          rsa.headlines.append(headline)
  
      # Add descriptions (2-4 required)
      for desc_text in config.descriptions:
          description = self.client.get_type("AdTextAsset")
          description.text = desc_text
          rsa.descriptions.append(description)
  
      # Set paths if provided
      if config.path1:
          ad_group_ad.ad.responsive_search_ad.path1 = config.path1
      if config.path2:
          ad_group_ad.ad.responsive_search_ad.path2 = config.path2
  
      # Set final URLs
      if config.final_urls:
          ad_group_ad.ad.final_urls.extend(config.final_urls)
  
      # Add ad
      response = ad_group_ad_service.mutate_ad_group_ads(
          customer_id=customer_id,
          operations=[ad_group_ad_operation]
      )
  
      ad_resource_name = response.results[0].resource_name
      ad_id = ad_resource_name.split("/")[-1]
  
      logger.info(f"Created RSA: {ad_id}")
  
      return {
          "ad_id": ad_id,
          "resource_name": ad_resource_name,
          "ad_group_id": config.ad_group_id,
          "headline_count": len(config.headlines),
          "description_count": len(config.descriptions),
          "status": config.status.value
      }

• google_ads_mcp.py:498-507 (registration)

  Registration via _register_all_modular_tools() which dynamically imports tools.ads.mcp_tools_ads and calls register_ad_tools(mcp), which decorates the handler function with @mcp.tool().

  def _register_all_modular_tools():
      """Import and register every modular tool module."""
      import importlib
  
      registered = 0
      for label, module_path, func_name in _TOOL_MODULES:
          try:
              mod = importlib.import_module(module_path)
              register_fn = getattr(mod, func_name)
              register_fn(mcp)