Meta Ads MCP

Instructions

Implementation Reference

• meta_ads_mcp/core/targeting.py:68-452 (handler)

  The primary handler function for the 'estimate_audience_size' tool. It uses Meta Ads API endpoints (reachestimate with delivery_estimate fallback) for comprehensive audience estimation, handles backwards compatibility for interest validation, includes preflight validation for locations/custom audiences, and returns formatted JSON results.
  @mcp_server.tool() @meta_api_tool async def estimate_audience_size( access_token: Optional[str] = None, account_id: Optional[str] = None, targeting: Optional[Dict[str, Any]] = None, optimization_goal: str = "REACH", # Backwards compatibility for simple interest validation interest_list: Optional[List[str]] = None, interest_fbid_list: Optional[List[str]] = None ) -> str: """ Estimate audience size for targeting specifications using Meta's delivery_estimate API. This function provides comprehensive audience estimation for complex targeting combinations including demographics, geography, interests, and behaviors. It also maintains backwards compatibility for simple interest validation. Args: access_token: Meta API access token (optional - will use cached token if not provided) account_id: Meta Ads account ID (format: act_XXXXXXXXX) - required for comprehensive estimation targeting: Complete targeting specification including demographics, geography, interests, etc. Example: { "age_min": 25, "age_max": 65, "geo_locations": {"countries": ["PL"]}, "flexible_spec": [ {"interests": [{"id": "6003371567474"}]}, {"interests": [{"id": "6003462346642"}]} ] } optimization_goal: Optimization goal for estimation (default: "REACH"). Options: "REACH", "LINK_CLICKS", "IMPRESSIONS", "CONVERSIONS", etc. interest_list: [DEPRECATED - for backwards compatibility] List of interest names to validate interest_fbid_list: [DEPRECATED - for backwards compatibility] List of interest IDs to validate Returns: JSON string with audience estimation results including estimated_audience_size, reach_estimate, and targeting validation """ # Handle backwards compatibility - simple interest validation # Check if we're in backwards compatibility mode (interest params provided OR no comprehensive params) is_backwards_compatible_call = (interest_list or interest_fbid_list) or (not account_id and not targeting) if is_backwards_compatible_call and not targeting: if not interest_list and not interest_fbid_list: return json.dumps({"error": "No interest list or FBID list provided"}, indent=2) endpoint = "search" params = { "type": "adinterestvalid" } if interest_list: params["interest_list"] = json.dumps(interest_list) if interest_fbid_list: params["interest_fbid_list"] = json.dumps(interest_fbid_list) data = await make_api_request(endpoint, access_token, params) return json.dumps(data, indent=2) # Comprehensive audience estimation using delivery_estimate API if not account_id: return json.dumps({ "error": "account_id is required for comprehensive audience estimation", "details": "For simple interest validation, use interest_list or interest_fbid_list parameters" }, indent=2) if not targeting: return json.dumps({ "error": "targeting specification is required for comprehensive audience estimation", "example": { "age_min": 25, "age_max": 65, "geo_locations": {"countries": ["US"]}, "flexible_spec": [ {"interests": [{"id": "6003371567474"}]} ] } }, indent=2) # Preflight validation: require at least one location OR a custom audience def _has_location_or_custom_audience(t: Dict[str, Any]) -> bool: if not isinstance(t, dict): return False geo = t.get("geo_locations") or {} if isinstance(geo, dict): for key in [ "countries", "regions", "cities", "zips", "geo_markets", "country_groups" ]: val = geo.get(key) if isinstance(val, list) and len(val) > 0: return True # Top-level custom audiences ca = t.get("custom_audiences") if isinstance(ca, list) and len(ca) > 0: return True # Custom audiences within flexible_spec flex = t.get("flexible_spec") if isinstance(flex, list): for spec in flex: if isinstance(spec, dict): ca_spec = spec.get("custom_audiences") if isinstance(ca_spec, list) and len(ca_spec) > 0: return True return False if not _has_location_or_custom_audience(targeting): return json.dumps({ "error": "Missing target audience location", "details": "Select at least one location in targeting.geo_locations or include a custom audience.", "action_required": "Add geo_locations with countries/regions/cities/zips or include custom_audiences.", "example": { "geo_locations": {"countries": ["US"]}, "age_min": 25, "age_max": 65 } }, indent=2) # Build reach estimate request (using correct Meta API endpoint) endpoint = f"{account_id}/reachestimate" params = { "targeting_spec": targeting } # Note: reachestimate endpoint doesn't support optimization_goal or objective parameters try: data = await make_api_request(endpoint, access_token, params, method="GET") # Surface Graph API errors directly for better diagnostics. # If reachestimate fails, optionally attempt a fallback using delivery_estimate. if isinstance(data, dict) and "error" in data: # Special handling for Missing Target Audience Location error (subcode 1885364) try: err_wrapper = data.get("error", {}) details_obj = err_wrapper.get("details", {}) raw_err = details_obj.get("error", {}) if isinstance(details_obj, dict) else {} if ( isinstance(raw_err, dict) and ( raw_err.get("error_subcode") == 1885364 or raw_err.get("error_user_title") == "Missing Target Audience Location" ) ): return json.dumps({ "error": "Missing target audience location", "details": raw_err.get("error_user_msg") or "Select at least one location, or choose a custom audience.", "endpoint_used": f"{account_id}/reachestimate", "action_required": "Add geo_locations with at least one of countries/regions/cities/zips or include custom_audiences.", "blame_field_specs": raw_err.get("error_data", {}).get("blame_field_specs") if isinstance(raw_err.get("error_data"), dict) else None }, indent=2) except Exception: pass # Allow disabling fallback via environment variable # Default: fallback disabled unless explicitly enabled by setting DISABLE flag to "0" disable_fallback = os.environ.get("META_MCP_DISABLE_DELIVERY_FALLBACK", "1") == "1" if disable_fallback: return json.dumps({ "error": "Graph API returned an error for reachestimate", "details": data.get("error"), "endpoint_used": f"{account_id}/reachestimate", "request_params": { "has_targeting_spec": bool(targeting), }, "note": "delivery_estimate fallback disabled via META_MCP_DISABLE_DELIVERY_FALLBACK" }, indent=2) # Try fallback to delivery_estimate endpoint try: fallback_endpoint = f"{account_id}/delivery_estimate" fallback_params = { "targeting_spec": json.dumps(targeting), # Some API versions accept optimization_goal here "optimization_goal": optimization_goal } fallback_data = await make_api_request(fallback_endpoint, access_token, fallback_params, method="GET") # If fallback returns usable data, format similarly if isinstance(fallback_data, dict) and "data" in fallback_data and len(fallback_data["data"]) > 0: estimate_data = fallback_data["data"][0] formatted_response = { "success": True, "account_id": account_id, "targeting": targeting, "optimization_goal": optimization_goal, "estimated_audience_size": estimate_data.get("estimate_mau", 0), "estimate_details": { "monthly_active_users": estimate_data.get("estimate_mau", 0), "daily_outcomes_curve": estimate_data.get("estimate_dau", []), "bid_estimate": estimate_data.get("bid_estimates", {}), "unsupported_targeting": estimate_data.get("unsupported_targeting", []) }, "raw_response": fallback_data, "fallback_endpoint_used": "delivery_estimate" } return json.dumps(formatted_response, indent=2) # Fallback returned but not in expected format return json.dumps({ "error": "Graph API returned an error for reachestimate; delivery_estimate fallback did not return usable data", "reachestimate_error": data.get("error"), "fallback_endpoint_used": "delivery_estimate", "fallback_raw_response": fallback_data, "endpoint_used": f"{account_id}/reachestimate", "request_params": { "has_targeting_spec": bool(targeting) } }, indent=2) except Exception as _fallback_exc: return json.dumps({ "error": "Graph API returned an error for reachestimate; delivery_estimate fallback also failed", "reachestimate_error": data.get("error"), "fallback_endpoint_used": "delivery_estimate", "fallback_exception": str(_fallback_exc), "endpoint_used": f"{account_id}/reachestimate", "request_params": { "has_targeting_spec": bool(targeting) } }, indent=2) # Format the response for easier consumption if "data" in data: response_data = data["data"] # Case 1: delivery_estimate-like list structure if isinstance(response_data, list) and len(response_data) > 0: estimate_data = response_data[0] formatted_response = { "success": True, "account_id": account_id, "targeting": targeting, "optimization_goal": optimization_goal, "estimated_audience_size": estimate_data.get("estimate_mau", 0), "estimate_details": { "monthly_active_users": estimate_data.get("estimate_mau", 0), "daily_outcomes_curve": estimate_data.get("estimate_dau", []), "bid_estimate": estimate_data.get("bid_estimates", {}), "unsupported_targeting": estimate_data.get("unsupported_targeting", []) }, "raw_response": data } return json.dumps(formatted_response, indent=2) # Case 1b: explicit handling for empty list responses if isinstance(response_data, list) and len(response_data) == 0: return json.dumps({ "error": "No estimation data returned from Meta API", "raw_response": data, "debug_info": { "response_keys": list(data.keys()) if isinstance(data, dict) else "not_a_dict", "response_type": str(type(data)), "endpoint_used": f"{account_id}/reachestimate" } }, indent=2) # Case 2: reachestimate dict structure with bounds if isinstance(response_data, dict): lower = response_data.get("users_lower_bound", response_data.get("estimate_mau_lower_bound")) upper = response_data.get("users_upper_bound", response_data.get("estimate_mau_upper_bound")) estimate_ready = response_data.get("estimate_ready") midpoint = None try: if isinstance(lower, (int, float)) and isinstance(upper, (int, float)): midpoint = int((lower + upper) / 2) except Exception: midpoint = None formatted_response = { "success": True, "account_id": account_id, "targeting": targeting, "optimization_goal": optimization_goal, "estimated_audience_size": midpoint if midpoint is not None else 0, "estimate_details": { "users_lower_bound": lower, "users_upper_bound": upper, "estimate_ready": estimate_ready }, "raw_response": data } return json.dumps(formatted_response, indent=2) else: return json.dumps({ "error": "No estimation data returned from Meta API", "raw_response": data, "debug_info": { "response_keys": list(data.keys()) if isinstance(data, dict) else "not_a_dict", "response_type": str(type(data)), "endpoint_used": f"{account_id}/reachestimate" } }, indent=2) except Exception as e: # Try fallback to delivery_estimate first when an exception occurs (unless disabled) # Default: fallback disabled unless explicitly enabled by setting DISABLE flag to "0" disable_fallback = os.environ.get("META_MCP_DISABLE_DELIVERY_FALLBACK", "1") == "1" if not disable_fallback: try: fallback_endpoint = f"{account_id}/delivery_estimate" fallback_params = { "targeting_spec": json.dumps(targeting) if isinstance(targeting, dict) else targeting, "optimization_goal": optimization_goal } fallback_data = await make_api_request(fallback_endpoint, access_token, fallback_params, method="GET") if isinstance(fallback_data, dict) and "data" in fallback_data and len(fallback_data["data"]) > 0: estimate_data = fallback_data["data"][0] formatted_response = { "success": True, "account_id": account_id, "targeting": targeting, "optimization_goal": optimization_goal, "estimated_audience_size": estimate_data.get("estimate_mau", 0), "estimate_details": { "monthly_active_users": estimate_data.get("estimate_mau", 0), "daily_outcomes_curve": estimate_data.get("estimate_dau", []), "bid_estimate": estimate_data.get("bid_estimates", {}), "unsupported_targeting": estimate_data.get("unsupported_targeting", []) }, "raw_response": fallback_data, "fallback_endpoint_used": "delivery_estimate" } return json.dumps(formatted_response, indent=2) except Exception as _fallback_exc: # If fallback also fails, proceed to detailed error handling below pass # Check if this is the specific Business Manager system user permission error error_str = str(e) if "100" in error_str and "33" in error_str: # Try to provide fallback estimation using individual interests if available interests_found = [] if targeting and "interests" in targeting: interests_found.extend([interest.get("id") for interest in targeting["interests"] if interest.get("id")]) elif targeting and "flexible_spec" in targeting: for spec in targeting["flexible_spec"]: if "interests" in spec: interests_found.extend([interest.get("id") for interest in spec["interests"] if interest.get("id")]) if interests_found: # Attempt to get individual interest data as fallback try: fallback_result = await estimate_audience_size( access_token=access_token, interest_fbid_list=interests_found ) fallback_data = json.loads(fallback_result) return json.dumps({ "comprehensive_targeting_failed": True, "error_code": "100-33", "fallback_used": True, "details": { "issue": "reachestimate endpoint returned error - possibly due to targeting parameters or account limitations", "solution": "Individual interest validation used as fallback - comprehensive targeting may have specific requirements", "endpoint_used": f"{account_id}/reachestimate" }, "individual_interest_data": fallback_data, "note": "Individual interest audience sizes provided as fallback. Comprehensive targeting via reachestimate endpoint failed." }, indent=2) except: pass return json.dumps({ "error": "reachestimate endpoint returned error (previously was incorrectly using delivery_estimate)", "error_code": "100-33", "details": { "issue": "The endpoint returned an error, possibly due to targeting parameters or account limitations", "endpoint_used": f"{account_id}/reachestimate", "previous_issue": "Code was previously using non-existent delivery_estimate endpoint - now fixed", "available_alternative": "Use interest_list or interest_fbid_list parameters for individual interest validation" }, "raw_error": error_str }, indent=2) else: return json.dumps({ "error": f"Failed to get audience estimation from reachestimate endpoint: {str(e)}", "details": "Check targeting parameters and account permissions", "error_type": "general_api_error", "endpoint_used": f"{account_id}/reachestimate" }, indent=2)
• meta_ads_mcp/core/__init__.py:14-14 (registration)

  Imports the estimate_audience_size function from targeting.py into core.__init__.py, which triggers the execution of the @mcp_server.tool() decorator for automatic MCP tool registration when the core module is imported.
  from .targeting import search_interests, get_interest_suggestions, estimate_audience_size, search_behaviors, search_demographics, search_geo_locations
• meta_ads_mcp/core/server.py:17-17 (registration)

  Creates the FastMCP server instance named 'mcp_server' that is used by the @mcp_server.tool() decorators in tool modules (like targeting.py) to register the 'estimate_audience_size' tool with the MCP server.
  mcp_server = FastMCP("meta-ads")
• meta_ads_mcp/__init__.py:38-61 (registration)

  Re-exports estimate_audience_size from the core module at the top-level package __init__.py, making it available when importing from meta_ads_mcp.
  from .core import ( get_ad_accounts, get_account_info, get_campaigns, get_campaign_details, create_campaign, get_adsets, get_adset_details, update_adset, get_ads, get_ad_details, get_ad_creatives, get_ad_image, update_ad, get_insights, login_cli, main, search_interests, get_interest_suggestions, estimate_audience_size, search_behaviors, search_demographics, search_geo_locations )
• meta_ads_mcp/core/api.py:187-329 (helper)

  The @meta_api_tool decorator applied to estimate_audience_size, providing automatic access token management, API request execution via make_api_request, comprehensive error handling, and JSON response formatting.
  def meta_api_tool(func): """Decorator for Meta API tools that handles authentication and error handling.""" @functools.wraps(func) async def wrapper(*args, **kwargs): try: # Log function call logger.debug(f"Function call: {func.__name__}") logger.debug(f"Args: {args}") # Log kwargs without sensitive info safe_kwargs = {k: ('***TOKEN***' if k == 'access_token' else v) for k, v in kwargs.items()} logger.debug(f"Kwargs: {safe_kwargs}") # Log app ID information app_id = auth_manager.app_id logger.debug(f"Current app_id: {app_id}") logger.debug(f"META_APP_ID env var: {os.environ.get('META_APP_ID')}") # If access_token is not in kwargs or not kwargs['access_token'], try to get it from auth_manager if 'access_token' not in kwargs or not kwargs['access_token']: try: access_token = await auth.get_current_access_token() if access_token: kwargs['access_token'] = access_token logger.debug("Using access token from auth_manager") else: logger.warning("No access token available from auth_manager") # Add more details about why token might be missing if (auth_manager.app_id == "YOUR_META_APP_ID" or not auth_manager.app_id) and not auth_manager.use_pipeboard: logger.error("TOKEN VALIDATION FAILED: No valid app_id configured") logger.error("Please set META_APP_ID environment variable or configure in your code") elif auth_manager.use_pipeboard: logger.error("TOKEN VALIDATION FAILED: Pipeboard authentication enabled but no valid token available") logger.error("Complete authentication via Pipeboard service or check PIPEBOARD_API_TOKEN") else: logger.error("Check logs above for detailed token validation failures") except Exception as e: logger.error(f"Error getting access token: {str(e)}") # Add stack trace for better debugging import traceback logger.error(f"Stack trace: {traceback.format_exc()}") # Final validation - if we still don't have a valid token, return authentication required if 'access_token' not in kwargs or not kwargs['access_token']: logger.warning("No access token available, authentication needed") # Add more specific troubleshooting information auth_url = auth_manager.get_auth_url() app_id = auth_manager.app_id using_pipeboard = auth_manager.use_pipeboard logger.error("TOKEN VALIDATION SUMMARY:") logger.error(f"- Current app_id: '{app_id}'") logger.error(f"- Environment META_APP_ID: '{os.environ.get('META_APP_ID', 'Not set')}'") logger.error(f"- Pipeboard API token configured: {'Yes' if os.environ.get('PIPEBOARD_API_TOKEN') else 'No'}") logger.error(f"- Using Pipeboard authentication: {'Yes' if using_pipeboard else 'No'}") # Check for common configuration issues - but only if not using Pipeboard if not using_pipeboard and (app_id == "YOUR_META_APP_ID" or not app_id): logger.error("ISSUE DETECTED: No valid Meta App ID configured") logger.error("ACTION REQUIRED: Set META_APP_ID environment variable with a valid App ID") elif using_pipeboard: logger.error("ISSUE DETECTED: Pipeboard authentication configured but no valid token available") logger.error("ACTION REQUIRED: Complete authentication via Pipeboard service") # Provide different guidance based on authentication method if using_pipeboard: return json.dumps({ "error": { "message": "Pipeboard Authentication Required", "details": { "description": "Your Pipeboard API token is invalid or has expired", "action_required": "Update your Pipeboard token", "setup_url": "https://pipeboard.co/setup", "token_url": "https://pipeboard.co/api-tokens", "configuration_status": { "app_id_configured": bool(app_id) and app_id != "YOUR_META_APP_ID", "pipeboard_enabled": True, }, "troubleshooting": "Go to https://pipeboard.co/setup to verify your account setup, then visit https://pipeboard.co/api-tokens to obtain a new API token", "setup_link": "[Verify your Pipeboard account setup](https://pipeboard.co/setup)", "token_link": "[Get a new Pipeboard API token](https://pipeboard.co/api-tokens)" } } }, indent=2) else: return json.dumps({ "error": { "message": "Authentication Required", "details": { "description": "You need to authenticate with the Meta API before using this tool", "action_required": "Please authenticate first", "auth_url": auth_url, "configuration_status": { "app_id_configured": bool(app_id) and app_id != "YOUR_META_APP_ID", "pipeboard_enabled": False, }, "troubleshooting": "Check logs for TOKEN VALIDATION FAILED messages", "markdown_link": f"[Click here to authenticate with Meta Ads API]({auth_url})" } } }, indent=2) # Call the original function result = await func(*args, **kwargs) # If the result is a string (JSON), try to parse it to check for errors if isinstance(result, str): try: result_dict = json.loads(result) if "error" in result_dict: logger.error(f"Error in API response: {result_dict['error']}") # If this is an app ID error, log more details if isinstance(result_dict.get("details", {}).get("error", {}), dict): error_obj = result_dict["details"]["error"] if error_obj.get("code") == 200 and "Provide valid app ID" in error_obj.get("message", ""): logger.error("Meta API authentication configuration issue") logger.error(f"Current app_id: {app_id}") # Replace the confusing error with a more user-friendly one return json.dumps({ "error": { "message": "Meta API Configuration Issue", "details": { "description": "Your Meta API app is not properly configured", "action_required": "Check your META_APP_ID environment variable", "current_app_id": app_id, "original_error": error_obj.get("message") } } }, indent=2) except Exception: # Not JSON or other parsing error, wrap it in a dictionary return json.dumps({"data": result}, indent=2) # If result is already a dictionary, ensure it's properly serialized if isinstance(result, dict): return json.dumps(result, indent=2) return result except Exception as e: logger.error(f"Error in {func.__name__}: {str(e)}") return json.dumps({"error": str(e)}, indent=2) return wrapper