async def create_ad_creative(
ad_account_id: str,
ad_image_hash: Optional[str] = None,
meta_access_token: Optional[str] = None,
name: Optional[str] = None,
facebook_page_id: Optional[str] = None,
link_url: Optional[str] = None,
primary_text: Optional[str] = None,
primary_text_variants: Optional[List[str]] = None,
headline_text: Optional[str] = None,
headline_variants: Optional[List[str]] = None,
description_text: Optional[str] = None,
description_variants: Optional[List[str]] = None,
ad_image_hashes: Optional[List[str]] = None,
ad_video_id: Optional[str] = None,
thumbnail_url: Optional[str] = None,
optimization_type: Optional[str] = None,
dynamic_creative_spec: Optional[Dict[str, Any]] = None,
call_to_action_type: Optional[str] = None,
lead_form_id: Optional[str] = None,
instagram_actor_id: Optional[str] = None,
ad_formats: Optional[List[str]] = None,
asset_customization_rules: Optional[List[Dict[str, Any]]] = None,
) -> str:
if not ad_account_id:
return _json({"error": "No account ID provided"})
ad_image_hashes = _normalize_list_argument(ad_image_hashes)
primary_text_variants = _normalize_list_argument(primary_text_variants)
headline_variants = _normalize_list_argument(headline_variants)
description_variants = _normalize_list_argument(description_variants)
ad_formats = _normalize_list_argument(ad_formats)
asset_customization_rules = _normalize_list_argument(asset_customization_rules)
media_error = _ensure_single_media_choice(ad_image_hash, ad_image_hashes, ad_video_id)
if media_error:
return _json({"error": media_error})
if ad_image_hashes and len(ad_image_hashes) > 10:
return _json({"error": "Maximum 10 image hashes allowed for FLEX creatives"})
if thumbnail_url and not ad_video_id:
return _json({"error": "thumbnail_url can only be used with ad_video_id"})
if optimization_type and optimization_type != "DEGREES_OF_FREEDOM":
return _json({"error": f"Invalid optimization_type '{optimization_type}'. Only 'DEGREES_OF_FREEDOM' is supported."})
normalized_lists_error, normalized_assets = _normalize_flexible_asset_lists(
primary_text,
primary_text_variants,
headline_text,
headline_variants,
description_text,
description_variants,
)
if normalized_lists_error:
return _json(normalized_lists_error)
if not link_url and not lead_form_id:
return _json(
{
"error": "No link_url provided. A destination URL is required for ad creatives (unless using lead_form_id)."
}
)
if ad_video_id and not thumbnail_url:
try:
thumbnail_url = await _fetch_video_thumbnail(ad_video_id, meta_access_token)
except Exception: # noqa: BLE001
pass
normalized_account_id = _normalize_ad_account_id(ad_account_id)
final_name = name or f"Creative {int(time.time())}"
resolved_page_id, page_error = await _resolve_page_id_for_creative(normalized_account_id, meta_access_token, facebook_page_id)
if page_error:
return _json(page_error)
resolved_instagram_user_id, resolved_instagram_actor_id = _sanitize_instagram_identity(
None,
instagram_actor_id,
)
use_asset_feed = bool(primary_text_variants or headline_variants or description_variants or ad_image_hashes or optimization_type)
creative_payload: Dict[str, Any] = {"name": final_name}
if use_asset_feed:
feed, story_spec = _build_asset_feed_spec_payload(
link_url=link_url,
normalized_assets=normalized_assets,
ad_image_hash=ad_image_hash,
ad_image_hashes=ad_image_hashes,
ad_video_id=ad_video_id,
thumbnail_url=thumbnail_url,
optimization_type=optimization_type,
ad_formats=ad_formats,
call_to_action_type=call_to_action_type,
asset_customization_rules=asset_customization_rules,
)
story_spec["page_id"] = resolved_page_id
creative_payload["asset_feed_spec"] = feed
creative_payload["object_story_spec"] = story_spec
else:
if ad_video_id:
creative_payload["object_story_spec"] = _build_simple_video_story_spec(
facebook_page_id=resolved_page_id,
ad_video_id=ad_video_id,
link_url=link_url,
primary_text=primary_text,
headline_text=headline_text,
thumbnail_url=thumbnail_url,
call_to_action_type=call_to_action_type,
lead_form_id=lead_form_id,
)
else:
creative_payload["object_story_spec"] = _build_simple_image_story_spec(
facebook_page_id=resolved_page_id,
ad_image_hash=ad_image_hash,
link_url=link_url,
primary_text=primary_text,
headline_text=headline_text,
description_text=description_text,
call_to_action_type=call_to_action_type,
lead_form_id=lead_form_id,
)
if dynamic_creative_spec:
creative_payload["dynamic_creative_spec"] = dynamic_creative_spec
if resolved_instagram_user_id:
creative_payload["object_story_spec"]["instagram_user_id"] = resolved_instagram_user_id
elif resolved_instagram_actor_id:
creative_payload["object_story_spec"]["instagram_actor_id"] = resolved_instagram_actor_id
creation_result = await make_api_request(f"{normalized_account_id}/adcreatives", meta_access_token, creative_payload, method="POST")
if (resolved_instagram_user_id or resolved_instagram_actor_id) and isinstance(creation_result, dict) and creation_result.get("error"):
details = creation_result.get("error", {}).get("details", {})
inner = details.get("error", {}) if isinstance(details, dict) else {}
message_text = ""
if isinstance(inner, dict):
message_text = inner.get("message", "") or inner.get("primary_text", "")
lowered = message_text.lower()
if "valid instagram account id" in lowered or "instagram_actor_id" in lowered or "instagram_user_id" in lowered:
return _json(
{
"error": "Instagram account not authorized for advertising",
"explanation": "The Meta API rejected the Instagram identity field in object_story_spec.",
"fix": "Reconnect the Facebook account and retry with refreshed permissions.",
"instagram_user_id": resolved_instagram_user_id,
"instagram_actor_id": resolved_instagram_actor_id,
"meta_error": message_text,
}
)
if isinstance(creation_result, dict) and creation_result.get("id"):
ad_creative_id = creation_result["id"]
details = await make_api_request(ad_creative_id, meta_access_token, {"fields": _CREATIVE_FIELDS.replace(",image_urls_for_viewing", "")})
return _json({"success": True, "ad_creative_id": ad_creative_id, "details": details})
return _json(creation_result if isinstance(creation_result, dict) else {"data": creation_result})