We provide all the information about MCP servers via our MCP API.
curl -X GET 'https://glama.ai/api/mcp/v1/servers/pipeboard-co/meta-ads-mcp'
If you have feedback or need assistance with the MCP directory API, please join our Discord server
"""Duplication functionality for Meta Ads API."""
import json
import os
import httpx
from typing import Optional, Dict, Any, List, Union
from .server import mcp_server
from .api import meta_api_tool
from . import auth
from .http_auth_integration import FastMCPAuthIntegration
# Only register the duplication functions if the environment variable is set
ENABLE_DUPLICATION = bool(os.environ.get("META_ADS_ENABLE_DUPLICATION", ""))
if ENABLE_DUPLICATION:
@mcp_server.tool()
@meta_api_tool
async def duplicate_campaign(
campaign_id: str,
access_token: Optional[str] = None,
name_suffix: Optional[str] = " - Copy",
include_ad_sets: bool = True,
include_ads: bool = True,
include_creatives: bool = True,
copy_schedule: bool = False,
new_daily_budget: Optional[float] = None,
new_status: Optional[str] = "PAUSED"
) -> str:
"""
Duplicate a Meta Ads campaign with all its ad sets and ads.
Recommended: Use this to run robust experiments.
Args:
campaign_id: Meta Ads campaign ID to duplicate
name_suffix: Suffix to add to the duplicated campaign name
include_ad_sets: Whether to duplicate ad sets within the campaign
include_ads: Whether to duplicate ads within ad sets
include_creatives: Whether to duplicate ad creatives
copy_schedule: Whether to copy the campaign schedule
new_daily_budget: Override the daily budget for the new campaign
new_status: Status for the new campaign (ACTIVE or PAUSED)
"""
return await _forward_duplication_request(
"campaign",
campaign_id,
access_token,
{
"name_suffix": name_suffix,
"include_ad_sets": include_ad_sets,
"include_ads": include_ads,
"include_creatives": include_creatives,
"copy_schedule": copy_schedule,
"new_daily_budget": new_daily_budget,
"new_status": new_status
}
)
@mcp_server.tool()
@meta_api_tool
async def duplicate_adset(
adset_id: str,
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 a Meta Ads ad set with its ads.
Recommended: Use this to run robust experiments.
Args:
adset_id: Meta Ads ad set ID to duplicate
target_campaign_id: Campaign ID to move the duplicated ad set to (optional)
name_suffix: Suffix to add to the duplicated ad set name
include_ads: Whether to duplicate ads within the ad set
include_creatives: Whether to duplicate ad creatives
new_daily_budget: Override the daily budget for the new ad set
new_targeting: Override targeting settings for the new ad set
new_status: Status for the new ad set (ACTIVE or PAUSED)
"""
return await _forward_duplication_request(
"adset",
adset_id,
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()
@meta_api_tool
async def duplicate_ad(
ad_id: str,
access_token: Optional[str] = None,
target_adset_id: Optional[str] = None,
name_suffix: Optional[str] = " - Copy",
duplicate_creative: bool = True,
new_creative_name: Optional[str] = None,
new_status: Optional[str] = "PAUSED"
) -> str:
"""
Duplicate a Meta Ads ad.
Recommended: Use this to run robust experiments.
Args:
ad_id: Meta Ads ad ID to duplicate
target_adset_id: Ad set ID to move the duplicated ad to (optional)
name_suffix: Suffix to add to the duplicated ad name
duplicate_creative: Whether to duplicate the ad creative
new_creative_name: Override name for the duplicated creative
new_status: Status for the new ad (ACTIVE or PAUSED)
"""
return await _forward_duplication_request(
"ad",
ad_id,
access_token,
{
"target_adset_id": target_adset_id,
"name_suffix": name_suffix,
"duplicate_creative": duplicate_creative,
"new_creative_name": new_creative_name,
"new_status": new_status
}
)
@mcp_server.tool()
@meta_api_tool
async def duplicate_creative(
creative_id: str,
access_token: Optional[str] = None,
name_suffix: Optional[str] = " - Copy",
new_primary_text: Optional[str] = None,
new_headline: Optional[str] = None,
new_description: Optional[str] = None,
new_cta_type: Optional[str] = None,
new_destination_url: Optional[str] = None
) -> str:
"""
Duplicate a Meta Ads creative.
Recommended: Use this to run robust experiments.
Args:
creative_id: Meta Ads creative ID to duplicate
name_suffix: Suffix to add to the duplicated creative name
new_primary_text: Override the primary text for the new creative
new_headline: Override the headline for the new creative
new_description: Override the description for the new creative
new_cta_type: Override the call-to-action type for the new creative
new_destination_url: Override the destination URL for the new creative
"""
return await _forward_duplication_request(
"creative",
creative_id,
access_token,
{
"name_suffix": name_suffix,
"new_primary_text": new_primary_text,
"new_headline": new_headline,
"new_description": new_description,
"new_cta_type": new_cta_type,
"new_destination_url": new_destination_url
}
)
async def _forward_duplication_request(resource_type: str, resource_id: str, access_token: str, options: Dict[str, Any]) -> str:
"""
Forward duplication request to the cloud-hosted MCP API using dual-header authentication.
This implements the dual-header authentication pattern for MCP server callbacks:
- Authorization: Bearer <facebook_token> - Facebook access token for Meta API calls
- X-Pipeboard-Token: <pipeboard_token> - Pipeboard API token for authentication
Args:
resource_type: Type of resource to duplicate (campaign, adset, ad, creative)
resource_id: ID of the resource to duplicate
access_token: Meta API access token (optional, will use context if not provided)
options: Duplication options
"""
try:
# Get tokens from the request context that were set by the HTTP auth middleware
# In the dual-header authentication pattern:
# - Pipeboard token comes from X-Pipeboard-Token header (for authentication)
# - Facebook token comes from Authorization header (for Meta API calls)
# Get tokens from context set by AuthInjectionMiddleware
pipeboard_token = FastMCPAuthIntegration.get_pipeboard_token()
facebook_token = FastMCPAuthIntegration.get_auth_token()
# Use provided access_token parameter if no Facebook token found in context
if not facebook_token:
facebook_token = access_token if access_token else await auth.get_current_access_token()
# Validate we have both required tokens
if not pipeboard_token:
return json.dumps({
"error": "authentication_required",
"message": "Pipeboard API token not found",
"details": {
"required": "Valid Pipeboard token via X-Pipeboard-Token header",
"received_headers": "Check that the MCP server is forwarding the X-Pipeboard-Token header"
}
}, indent=2)
if not facebook_token:
return json.dumps({
"error": "authentication_required",
"message": "Meta Ads access token not found",
"details": {
"required": "Valid Meta access token from authenticated session",
"check": "Ensure Facebook account is connected and token is valid"
}
}, indent=2)
# Construct the API endpoint
base_url = "https://mcp.pipeboard.co"
endpoint = f"{base_url}/api/meta/duplicate/{resource_type}/{resource_id}"
# Prepare the dual-header authentication as per API documentation
headers = {
"Authorization": f"Bearer {facebook_token}", # Facebook token for Meta API
"X-Pipeboard-Token": pipeboard_token, # Pipeboard token for auth
"Content-Type": "application/json",
"User-Agent": "meta-ads-mcp/1.0"
}
# Remove None values from options
clean_options = {k: v for k, v in options.items() if v is not None}
# Make the request to the cloud service
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
endpoint,
headers=headers,
json=clean_options
)
if response.status_code == 200:
result = response.json()
return json.dumps(result, indent=2)
elif response.status_code == 400:
# Validation failed
try:
error_data = response.json()
return json.dumps({
"success": False,
"error": "validation_failed",
"errors": error_data.get("errors", [response.text]),
"warnings": error_data.get("warnings", [])
}, indent=2)
except:
return json.dumps({
"success": False,
"error": "validation_failed",
"errors": [response.text],
"warnings": []
}, indent=2)
elif response.status_code == 401:
return json.dumps({
"success": False,
"error": "authentication_error",
"message": "Invalid or expired API token"
}, indent=2)
elif response.status_code == 402:
try:
error_data = response.json()
return json.dumps({
"success": False,
"error": "subscription_required",
"message": error_data.get("message", "This feature is not available in your current plan"),
"upgrade_url": error_data.get("upgrade_url", "https://pipeboard.co/upgrade"),
"suggestion": error_data.get("suggestion", "Please upgrade your account to access this feature")
}, indent=2)
except:
return json.dumps({
"success": False,
"error": "subscription_required",
"message": "This feature is not available in your current plan",
"upgrade_url": "https://pipeboard.co/upgrade",
"suggestion": "Please upgrade your account to access this feature"
}, indent=2)
elif response.status_code == 403:
try:
error_data = response.json()
# Check if this is a premium feature error
if error_data.get("error") == "premium_feature":
return json.dumps({
"success": False,
"error": "premium_feature_required",
"message": error_data.get("message", "This is a premium feature that requires subscription"),
"details": error_data.get("details", {
"upgrade_url": "https://pipeboard.co/upgrade",
"suggestion": "Please upgrade your account to access this feature"
})
}, indent=2)
else:
# Default to facebook connection required
return json.dumps({
"success": False,
"error": "facebook_connection_required",
"message": error_data.get("message", "You need to connect your Facebook account first"),
"details": error_data.get("details", {
"login_flow_url": "/connections",
"auth_flow_url": "/api/meta/auth"
})
}, indent=2)
except:
return json.dumps({
"success": False,
"error": "facebook_connection_required",
"message": "You need to connect your Facebook account first",
"details": {
"login_flow_url": "/connections",
"auth_flow_url": "/api/meta/auth"
}
}, indent=2)
elif response.status_code == 404:
return json.dumps({
"success": False,
"error": "resource_not_found",
"message": f"{resource_type.title()} not found or access denied",
"suggestion": f"Verify the {resource_type} ID and your Facebook account permissions"
}, indent=2)
elif response.status_code == 429:
return json.dumps({
"error": "rate_limit_exceeded",
"message": "Meta API rate limit exceeded",
"details": {
"suggestion": "Please wait before retrying",
"retry_after": response.headers.get("Retry-After", "60")
}
}, indent=2)
elif response.status_code == 502:
try:
error_data = response.json()
return json.dumps({
"success": False,
"error": "meta_api_error",
"message": error_data.get("message", "Facebook API error"),
"recoverable": True,
"suggestion": "Please wait 5 minutes before retrying"
}, indent=2)
except:
return json.dumps({
"success": False,
"error": "meta_api_error",
"message": "Facebook API error",
"recoverable": True,
"suggestion": "Please wait 5 minutes before retrying"
}, indent=2)
else:
error_detail = response.text
try:
error_json = response.json()
error_detail = error_json.get("message", error_detail)
except:
pass
return json.dumps({
"error": "duplication_failed",
"message": f"Failed to duplicate {resource_type}",
"details": {
"status_code": response.status_code,
"error_detail": error_detail,
"resource_type": resource_type,
"resource_id": resource_id
}
}, indent=2)
except httpx.TimeoutException:
return json.dumps({
"error": "request_timeout",
"message": "Request to duplication service timed out",
"details": {
"suggestion": "Please try again later",
"timeout": "30 seconds"
}
}, indent=2)
except httpx.RequestError as e:
return json.dumps({
"error": "network_error",
"message": "Failed to connect to duplication service",
"details": {
"error": str(e),
"suggestion": "Check your internet connection and try again"
}
}, indent=2)
except Exception as e:
return json.dumps({
"error": "unexpected_error",
"message": f"Unexpected error during {resource_type} duplication",
"details": {
"error": str(e),
"resource_type": resource_type,
"resource_id": resource_id
}
}, indent=2)
def _get_estimated_components(resource_type: str, options: Dict[str, Any]) -> Dict[str, Any]:
"""Get estimated components that would be duplicated."""
if resource_type == "campaign":
components = {"campaigns": 1}
if options.get("include_ad_sets", True):
components["ad_sets"] = "3-5 (estimated)"
if options.get("include_ads", True):
components["ads"] = "5-15 (estimated)"
if options.get("include_creatives", True):
components["creatives"] = "5-15 (estimated)"
return components
elif resource_type == "adset":
components = {"ad_sets": 1}
if options.get("include_ads", True):
components["ads"] = "2-5 (estimated)"
if options.get("include_creatives", True):
components["creatives"] = "2-5 (estimated)"
return components
elif resource_type == "ad":
components = {"ads": 1}
if options.get("duplicate_creative", True):
components["creatives"] = 1
return components
elif resource_type == "creative":
return {"creatives": 1}
return {}