ArmaVita Meta Ads MCP

Instructions

Implementation Reference

• src/armavita_meta_ads_mcp/core/duplication_tools.py:368-393 (handler)

  The tool handler function `clone_ad_set` which implements the logic for duplicating an ad set by calling `_forward_duplication_request`.

  async def clone_ad_set(
      ad_set_id: str,
      meta_access_token: Optional[str] = None,
      target_campaign_id: Optional[str] = None,
      name_suffix: Optional[str] = " - Copy",
      include_ads: bool = True,
      include_creatives: bool = True,
      new_daily_budget: Optional[float] = None,
      new_targeting: Optional[Dict[str, Any]] = None,
      new_status: Optional[str] = "PAUSED",
  ) -> str:
      """Duplicate an ad set using Meta's local Graph copy edge."""
      return await _forward_duplication_request(
          "adset",
          ad_set_id,
          meta_access_token,
          {
              "target_campaign_id": target_campaign_id,
              "name_suffix": name_suffix,
              "include_ads": include_ads,
              "include_creatives": include_creatives,
              "new_daily_budget": new_daily_budget,
              "new_targeting": new_targeting,
              "new_status": new_status,
          },
      )

• src/armavita_meta_ads_mcp/core/duplication_tools.py:366-396 (registration)

  Registration of `clone_ad_set` as an MCP tool using `@mcp_server.tool()` and `@meta_api_tool`.

  @mcp_server.tool()
  @meta_api_tool
  async def clone_ad_set(
      ad_set_id: str,
      meta_access_token: Optional[str] = None,
      target_campaign_id: Optional[str] = None,
      name_suffix: Optional[str] = " - Copy",
      include_ads: bool = True,
      include_creatives: bool = True,
      new_daily_budget: Optional[float] = None,
      new_targeting: Optional[Dict[str, Any]] = None,
      new_status: Optional[str] = "PAUSED",
  ) -> str:
      """Duplicate an ad set using Meta's local Graph copy edge."""
      return await _forward_duplication_request(
          "adset",
          ad_set_id,
          meta_access_token,
          {
              "target_campaign_id": target_campaign_id,
              "name_suffix": name_suffix,
              "include_ads": include_ads,
              "include_creatives": include_creatives,
              "new_daily_budget": new_daily_budget,
              "new_targeting": new_targeting,
              "new_status": new_status,
          },
      )
  
  
  @mcp_server.tool()

• src/armavita_meta_ads_mcp/core/duplication_tools.py:450-562 (helper)

  Helper function that performs the actual API request to the Meta Graph API copy edge for all duplication tools.

  async def _forward_duplication_request(
      resource_type: str,
      resource_id: str,
      meta_access_token: Optional[str],
      options: Dict[str, Any],
  ) -> str:
      """Execute OSS-local duplication against Graph API copy edges."""
      try:
          facebook_token = meta_access_token if meta_access_token else await auth.get_current_access_token()
          if not facebook_token:
              raise DuplicationError(
                  json.dumps(
                      {
                          "success": False,
                          "error": "authentication_required",
                          "message": "Meta Ads access token not found",
                          "details": {
                              "required": "Valid Meta access token",
                              "check": "Authenticate and retry duplication request.",
                          },
                      },
                      indent=2,
                  )
              )
  
          preflight_block = await _run_v25_duplication_preflight(resource_type, resource_id, facebook_token)
          if preflight_block:
              raise DuplicationError(
                  json.dumps(
                      {
                          "success": False,
                          "error": "v25_blocked_operation",
                          "message": "Duplication is blocked for deprecated Advantage+ Shopping/App campaign flows in v25.",
                          "details": {
                              "resource_type": resource_type,
                              "resource_id": resource_id,
                              "campaign_id": preflight_block.get("campaign_id"),
                              "campaign_name": preflight_block.get("campaign_name"),
                              "campaign_objective": preflight_block.get("campaign_objective"),
                              "smart_promotion_type": preflight_block.get("smart_promotion_type"),
                              "reason": preflight_block.get("reason"),
                          },
                          "suggestion": (
                              "Use supported Advantage+ migration or rebuild the campaign with v25-compatible "
                              "flows before attempting duplication."
                          ),
                      },
                      indent=2,
                  )
              )
  
          copy_params, warnings = _build_copy_params(resource_type, options)
          endpoint = f"{resource_id}/copies"
  
          data = await make_api_request(endpoint, facebook_token, copy_params, method="POST")
          if not isinstance(data, dict):
              raise DuplicationError(
                  json.dumps(
                      {
                          "success": False,
                          "error": "duplication_failed",
                          "message": "Unexpected response from Graph API copy edge",
                          "details": {
                              "resource_type": resource_type,
                              "resource_id": resource_id,
                              "response_type": type(data).__name__,
                          },
                      },
                      indent=2,
                  )
              )
  
          graph_error = data.get("error") if isinstance(data.get("error"), dict) else None
          if graph_error:
              code = graph_error.get("code")
              if code == 4:
                  raise RateLimitError(
                      json.dumps(
                          {
                              "error": "rate_limit_exceeded",
                              "message": graph_error.get("message") or graph_error.get("primary_text", "Meta API rate limit exceeded"),
                              "details": {
                                  "code": code,
                                  "error_subcode": graph_error.get("error_subcode"),
                                  "retry_hint": "Retry with backoff.",
                              },
                          },
                          indent=2,
                      )
                  )
  
              raise DuplicationError(
                  json.dumps(
                      _build_graph_error_payload(resource_type, resource_id, graph_error),
                      indent=2,
                  )
              )
  
          new_id = _extract_new_id(resource_type, data)
          success = bool(new_id) or bool(data.get("success") is True)
  
          return json.dumps(
              {
                  "success": success,
                  "source_id": resource_id,
                  "resource_type": resource_type,
                  "new_id": new_id,
                  "warnings": warnings,
                  "meta_response": data,
              },
              indent=2,
          )