ArmaVita Meta Ads MCP

Instructions

Implementation Reference

• src/armavita_meta_ads_mcp/core/adset_tools.py:215-337 (handler)

  The `create_ad_set` handler function performs validation and makes a POST request to the Meta Ads API to create a new ad set.

  async def create_ad_set(
      ad_account_id: str,
      campaign_id: str,
      name: str,
      optimization_goal: str,
      billing_event: str,
      status: str = "PAUSED",
      daily_budget: Optional[int] = None,
      lifetime_budget: Optional[int] = None,
      targeting: Optional[Dict[str, Any]] = None,
      bid_amount: Optional[int] = None,
      bid_strategy: Optional[str] = None,
      bid_constraints: Optional[Dict[str, Any]] = None,
      start_time: Optional[str] = None,
      end_time: Optional[str] = None,
      dsa_beneficiary: Optional[str] = None,
      promoted_object: Optional[Dict[str, Any]] = None,
      destination_type: Optional[str] = None,
      is_dynamic_creative: Optional[bool] = None,
      meta_access_token: Optional[str] = None,
  ) -> str:
      """Create an ad set under a campaign."""
      if not ad_account_id:
          return _json({"error": "No account ID provided"})
      if not campaign_id:
          return _json({"error": "No campaign ID provided"})
      if not name:
          return _json({"error": "No ad set name provided"})
      if not optimization_goal:
          return _json({"error": "No optimization goal provided"})
      if not billing_event:
          return _json({"error": "No billing event provided"})
  
      app_error = _validate_promoted_object_for_app_installs(optimization_goal, promoted_object)
      if app_error:
          return _json(app_error)
  
      bid_error = _validate_bid_controls(bid_strategy, bid_amount, bid_constraints)
      if bid_error:
          return _json(bid_error)
  
      normalized_bid_strategy = _normalize_bid_strategy(bid_strategy)
  
      if bid_amount is None:
          try:
              parent_strategy = await _parent_campaign_bid_strategy(campaign_id, meta_access_token)
          except Exception:  # noqa: BLE001
              parent_strategy = None
  
          if parent_strategy in (_BID_STRATEGIES_REQUIRING_BID_AMOUNT | {"TARGET_COST"}):
              return _json(
                  {
                      "error": (
                          f"bid_amount is required because the parent campaign uses bid_strategy "
                          f"'{parent_strategy}'"
                      ),
                      "details": "Provide bid_amount in cents or update parent campaign strategy.",
                      "example_with_bid_amount": {"bid_amount": 500},
                  }
              )
  
      payload: Dict[str, Any] = {
          "name": name,
          "campaign_id": campaign_id,
          "status": status,
          "optimization_goal": optimization_goal,
          "billing_event": billing_event,
          "targeting": json.dumps(_normalize_targeting(targeting)),
      }
  
      if daily_budget is not None:
          payload["daily_budget"] = str(daily_budget)
      if lifetime_budget is not None:
          payload["lifetime_budget"] = str(lifetime_budget)
      if bid_amount is not None:
          payload["bid_amount"] = str(bid_amount)
      if normalized_bid_strategy is not None:
          payload["bid_strategy"] = normalized_bid_strategy
      if bid_constraints is not None:
          payload["bid_constraints"] = json.dumps(bid_constraints)
      if start_time:
          payload["start_time"] = start_time
      if end_time:
          payload["end_time"] = end_time
      if dsa_beneficiary:
          payload["dsa_beneficiary"] = dsa_beneficiary
      if promoted_object is not None:
          payload["promoted_object"] = json.dumps(promoted_object)
      if destination_type is not None:
          payload["destination_type"] = destination_type
      if is_dynamic_creative is not None:
          payload["is_dynamic_creative"] = "true" if bool(is_dynamic_creative) else "false"
  
      result = await make_api_request(f"{ad_account_id}/adsets", meta_access_token, payload, method="POST")
  
      if isinstance(result, dict) and result.get("error"):
          rendered_error = json.dumps(result.get("error", {}), default=str).lower()
          if "permission" in rendered_error or "insufficient" in rendered_error:
              return _json(
                  {
                      "error": "Insufficient permissions to set DSA beneficiary. Please ensure business_management permissions.",
                      "details": result,
                      "permission_required": True,
                  }
              )
          if "dsa_beneficiary" in rendered_error and ("not supported" in rendered_error or "parameter" in rendered_error):
              return _json(
                  {
                      "error": "DSA beneficiary parameter not supported in this API version.",
                      "details": result,
                      "manual_setup_required": True,
                  }
              )
          if "benefits from ads" in rendered_error or "dsa beneficiary" in rendered_error:
              return _json(
                  {
                      "error": "DSA beneficiary required for European compliance.",
                      "details": result,
                      "dsa_required": True,
                  }
              )
  
      return _json(result)