create_budget_schedule
Create a budget schedule for a Meta Ads campaign to increase ad spend during high-demand periods, defined by start and end times as Unix timestamps.
Instructions
Create a budget schedule for a Meta Ads campaign.
Allows scheduling budget increases based on anticipated high-demand periods.
The times should be provided as Unix timestamps.
Args:
campaign_id: Meta Ads campaign ID.
budget_value: Amount of budget increase. Interpreted based on budget_value_type.
budget_value_type: Type of budget value - "ABSOLUTE" or "MULTIPLIER".
time_start: Unix timestamp for when the high demand period should start.
time_end: Unix timestamp for when the high demand period should end.
access_token: Meta API access token (optional - will use cached token if not provided).
Returns:
A JSON string containing the ID of the created budget schedule or an error message.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| campaign_id | Yes | ||
| budget_value | Yes | ||
| budget_value_type | Yes | ||
| time_start | Yes | ||
| time_end | Yes | ||
| access_token | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- The main handler function for the create_budget_schedule tool. Decorated with @mcp_server.tool() and @meta_api_tool. It validates inputs (campaign_id, budget_value, budget_value_type, time_start, time_end), constructs a POST request to the Meta Graph API endpoint '{campaign_id}/budget_schedules', and returns the created budget schedule ID or an error message as JSON.
@mcp_server.tool() @meta_api_tool async def create_budget_schedule( campaign_id: str, budget_value: int, budget_value_type: str, time_start: int, time_end: int, access_token: Optional[str] = None ) -> str: """ Create a budget schedule for a Meta Ads campaign. Allows scheduling budget increases based on anticipated high-demand periods. The times should be provided as Unix timestamps. Args: campaign_id: Meta Ads campaign ID. budget_value: Amount of budget increase. Interpreted based on budget_value_type. budget_value_type: Type of budget value - "ABSOLUTE" or "MULTIPLIER". time_start: Unix timestamp for when the high demand period should start. time_end: Unix timestamp for when the high demand period should end. access_token: Meta API access token (optional - will use cached token if not provided). Returns: A JSON string containing the ID of the created budget schedule or an error message. """ if not campaign_id: return json.dumps({"error": "Campaign ID is required"}, indent=2) if budget_value is None: # Check for None explicitly return json.dumps({"error": "Budget value is required"}, indent=2) if not budget_value_type: return json.dumps({"error": "Budget value type is required"}, indent=2) if budget_value_type not in ["ABSOLUTE", "MULTIPLIER"]: return json.dumps({"error": "Invalid budget_value_type. Must be ABSOLUTE or MULTIPLIER"}, indent=2) if time_start is None: # Check for None explicitly to allow 0 return json.dumps({"error": "Time start is required"}, indent=2) if time_end is None: # Check for None explicitly to allow 0 return json.dumps({"error": "Time end is required"}, indent=2) endpoint = f"{campaign_id}/budget_schedules" params = { "budget_value": budget_value, "budget_value_type": budget_value_type, "time_start": time_start, "time_end": time_end, } try: data = await make_api_request(endpoint, access_token, params, method="POST") return json.dumps(data, indent=2) except Exception as e: error_msg = str(e) # Include details about the error and the parameters sent for easier debugging return json.dumps({ "error": "Failed to create budget schedule", "details": error_msg, "campaign_id": campaign_id, "params_sent": params }, indent=2) - meta_ads_mcp/core/server.py:17-17 (registration)The FastMCP server instance ('mcp_server') used to register the create_budget_schedule tool via the @mcp_server.tool() decorator. This is where the tool is registered into the MCP server.
mcp_server = FastMCP("meta-ads") - meta_ads_mcp/core/__init__.py:13-14 (registration)Re-export of create_budget_schedule from the budget_schedules module in the core package's __init__.py.
from .budget_schedules import create_budget_schedule from .targeting import search_interests, get_interest_suggestions, estimate_audience_size, search_behaviors, search_demographics, search_geo_locations - meta_ads_mcp/core/__init__.py:40-49 (registration)Export of create_budget_schedule in the __all__ list of the core package.
'create_budget_schedule', 'search_interests', 'get_interest_suggestions', 'estimate_audience_size', 'search_behaviors', 'search_demographics', 'search_geo_locations', 'search', # OpenAI MCP Deep Research search tool 'fetch', # OpenAI MCP Deep Research fetch tool ] - meta_ads_mcp/core/api.py:286-431 (helper)The meta_api_tool decorator that wraps the create_budget_schedule handler. It handles authentication (injecting access_token from auth_manager), error handling, and JSON serialization of results. The make_api_request helper function (also in api.py) performs the actual HTTP POST to Meta's Graph API.
# Generic wrapper for all Meta API tools 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 McpToolError: raise # Let FastMCP set isError: true and refund the usage credit except Exception as e: logger.error(f"Error in {func.__name__}: {str(e)}") return json.dumps({"error": str(e)}, indent=2) return wrapper