google_ads_import_from_csv

Instructions

Campaign Name,Budget,Type,Status
My Campaign,50.00,SEARCH,PAUSED

Ad Group ID,Keyword Text,Match Type,CPC Bid
12345678,running shoes,EXACT,2.50

Implementation Reference

• tools/batch/mcp_tools_batch.py:775-844 (handler)

  MCP tool handler that imports entities from CSV data. Parses CSV, delegates to BatchOperationsManager.import_from_csv(), and formats a markdown result.

  @mcp.tool()
  def google_ads_import_from_csv(
      customer_id: str,
      entity_type: str,
      csv_data: str
  ) -> str:
      """Import entities from CSV format.
  
      Args:
          customer_id: Google Ads customer ID (10 digits, no hyphens)
          entity_type: Type to import (campaigns, keywords)
          csv_data: CSV formatted data
  
      CSV Format for Campaigns:
      ```
      Campaign Name,Budget,Type,Status
      My Campaign,50.00,SEARCH,PAUSED
      ```
  
      CSV Format for Keywords:
      ```
      Ad Group ID,Keyword Text,Match Type,CPC Bid
      12345678,running shoes,EXACT,2.50
      ```
  
      Returns:
          Import result with success/failure details
  
      Example:
          google_ads_import_from_csv(
              customer_id="1234567890",
              entity_type="campaigns",
              csv_data="Campaign Name,Budget,Type\\nTest Campaign,50.00,SEARCH"
          )
      """
      with performance_logger.track_operation('import_from_csv', customer_id=customer_id):
          try:
              client = get_auth_manager().get_client()
              batch_manager = BatchOperationsManager(client)
  
              result = batch_manager.import_from_csv(customer_id, entity_type, csv_data)
  
              audit_logger.log_api_call(
                  customer_id=customer_id,
                  operation='import_from_csv',
                  details={'entity_type': entity_type, 'total': result.total, 'succeeded': result.succeeded},
                  status='success' if result.status.value != 'FAILED' else 'failed'
              )
  
              output = f"# 📥 CSV Import ({entity_type.title()})\n\n"
              output += f"**Status**: {result.status.value}\n"
              output += f"**Total**: {result.total} {entity_type}\n"
              output += f"**Imported**: {result.succeeded} ✅\n"
              output += f"**Failed**: {result.failed} ❌\n\n"
  
              if result.succeeded > 0:
                  output += "## ✅ Successfully Imported\n\n"
                  for res in result.results:
                      output += f"- {res}\n"
  
              if result.failed > 0:
                  output += "\n## ❌ Failed\n\n"
                  for err in result.errors:
                      output += f"- {err['error']}\n"
  
              return output
  
          except Exception as e:
              error_msg = ErrorHandler.handle_error(e, context="import_from_csv")
              return f"❌ Import failed: {error_msg}"

• managers/batch_operations_manager.py:868-916 (helper)

  Core CSV import logic. Parses CSV rows, maps fields for 'campaigns' or 'keywords' entity types, then delegates to batch_create_campaigns or batch_add_keywords. Returns BatchResult.

  def import_from_csv(
      self,
      customer_id: str,
      entity_type: str,
      csv_data: str
  ) -> BatchResult:
      """Import entities from CSV format.
  
      Args:
          customer_id: Customer ID (without hyphens)
          entity_type: Type to import (campaigns, ad_groups, keywords)
          csv_data: CSV string
  
      Returns:
          BatchResult with import details
      """
      reader = csv.DictReader(io.StringIO(csv_data))
      rows = list(reader)
  
      if entity_type == 'campaigns':
          campaigns = []
          for row in rows:
              campaigns.append({
                  'name': row['Campaign Name'],
                  'budget_amount': float(row['Budget']),
                  'type': row.get('Type', 'SEARCH'),
                  'status': row.get('Status', 'PAUSED')
              })
          return self.batch_create_campaigns(customer_id, campaigns)
  
      elif entity_type == 'keywords':
          keywords = []
          for row in rows:
              keywords.append({
                  'ad_group_id': row['Ad Group ID'],
                  'text': row['Keyword Text'],
                  'match_type': row.get('Match Type', 'BROAD'),
                  'cpc_bid': float(row.get('CPC Bid', 0)) if row.get('CPC Bid') else None
              })
          return self.batch_add_keywords(customer_id, keywords)
  
      return BatchResult(
          total=0,
          succeeded=0,
          failed=0,
          status=OperationStatus.FAILED,
          results=[],
          errors=[{'error': f'Unsupported entity type: {entity_type}'}]
      )

• tools/batch/mcp_tools_batch.py:32-32 (registration)

  Registration function for all batch tools. Called via _register_all_modular_tools() in google_ads_mcp.py (line 491). The @mcp.tool() decorator registers google_ads_import_from_csv.

  def register_batch_tools(mcp):

• google_ads_mcp.py:498-514 (registration)

  Top-level registration dispatcher. The batch module (including google_ads_import_from_csv) is registered via the entry ('batch', 'tools.batch.mcp_tools_batch', 'register_batch_tools') on line 491.

  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)
              logger.info(f"  ✓ {label}")
              registered += 1
          except Exception as exc:
              logger.error(f"  ✗ {label}: {exc}")
  
      logger.info(f"Registered {registered}/{len(_TOOL_MODULES)} modular tool modules")