estimate_audience_size
Estimate audience size for Meta Ads targeting specifications using demographics, geography, interests, and behaviors to validate reach before campaign launch.
Instructions
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 validationInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| access_token | No | ||
| account_id | No | ||
| targeting | No | ||
| optimization_goal | No | REACH | |
| interest_list | No | ||
| interest_fbid_list | No |
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