Meta Ads MCP

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• meta_ads_mcp/core/accounts.py:9-28 (handler)

  The core handler function implementing the 'get_ad_accounts' tool logic. It constructs the Meta Graph API request for ad accounts, calls the API via make_api_request, and returns JSON-formatted results. Registered via @mcp_server.tool() decorator.
  @mcp_server.tool() @meta_api_tool async def get_ad_accounts(access_token: Optional[str] = None, user_id: str = "me", limit: int = 200) -> str: """ Get ad accounts accessible by a user. Args: access_token: Meta API access token (optional - will use cached token if not provided) user_id: Meta user ID or "me" for the current user limit: Maximum number of accounts to return (default: 200) """ endpoint = f"{user_id}/adaccounts" params = { "fields": "id,name,account_id,account_status,amount_spent,balance,currency,age,business_city,business_country_code", "limit": limit } data = await make_api_request(endpoint, access_token, params) return json.dumps(data, indent=2)
• meta_ads_mcp/core/__init__.py:4-4 (registration)

  Import that registers and exposes the get_ad_accounts tool function from the accounts module into the core package namespace.
  from .accounts import get_ad_accounts, get_account_info
• meta_ads_mcp/__init__.py:38-39 (registration)

  Package-level import that re-exports the get_ad_accounts tool from core, making it available at the top-level package.
  from .core import ( get_ad_accounts,
• meta_ads_mcp/core/api.py:41-184 (helper)

  The make_api_request helper function called by get_ad_accounts to perform the actual HTTP request to Meta Graph API, handle auth, errors, and response parsing.
  async def make_api_request( endpoint: str, access_token: str, params: Optional[Dict[str, Any]] = None, method: str = "GET" ) -> Dict[str, Any]: """ Make a request to the Meta Graph API. Args: endpoint: API endpoint path (without base URL) access_token: Meta API access token params: Additional query parameters method: HTTP method (GET, POST, DELETE) Returns: API response as a dictionary """ # Validate access token before proceeding if not access_token: logger.error("API request attempted with blank access token") return { "error": { "message": "Authentication Required", "details": "A valid access token is required to access the Meta API", "action_required": "Please authenticate first" } } url = f"{META_GRAPH_API_BASE}/{endpoint}" headers = { "User-Agent": USER_AGENT, } request_params = params or {} request_params["access_token"] = access_token # Logging the request (masking token for security) masked_params = {k: "***TOKEN***" if k == "access_token" else v for k, v in request_params.items()} logger.debug(f"API Request: {method} {url}") logger.debug(f"Request params: {masked_params}") # Check for app_id in params app_id = auth_manager.app_id logger.debug(f"Current app_id from auth_manager: {app_id}") async with httpx.AsyncClient() as client: try: if method == "GET": # For GET, JSON-encode dict/list params (e.g., targeting_spec) to proper strings encoded_params = {} for key, value in request_params.items(): if isinstance(value, (dict, list)): encoded_params[key] = json.dumps(value) else: encoded_params[key] = value response = await client.get(url, params=encoded_params, headers=headers, timeout=30.0) elif method == "POST": # For Meta API, POST requests need data, not JSON if 'targeting' in request_params and isinstance(request_params['targeting'], dict): # Convert targeting dict to string for the API request_params['targeting'] = json.dumps(request_params['targeting']) # Convert lists and dicts to JSON strings for key, value in request_params.items(): if isinstance(value, (list, dict)): request_params[key] = json.dumps(value) logger.debug(f"POST params (prepared): {masked_params}") response = await client.post(url, data=request_params, headers=headers, timeout=30.0) elif method == "DELETE": response = await client.delete(url, params=request_params, headers=headers, timeout=30.0) else: raise ValueError(f"Unsupported HTTP method: {method}") response.raise_for_status() logger.debug(f"API Response status: {response.status_code}") # Ensure the response is JSON and return it as a dictionary try: return response.json() except json.JSONDecodeError: # If not JSON, return text content in a structured format return { "text_response": response.text, "status_code": response.status_code } except httpx.HTTPStatusError as e: error_info = {} try: error_info = e.response.json() except: error_info = {"status_code": e.response.status_code, "text": e.response.text} logger.error(f"HTTP Error: {e.response.status_code} - {error_info}") # Check for authentication errors if e.response.status_code == 401 or e.response.status_code == 403: logger.warning("Detected authentication error (401/403)") auth_manager.invalidate_token() elif "error" in error_info: error_obj = error_info.get("error", {}) # Check for specific FB API errors related to auth if isinstance(error_obj, dict) and error_obj.get("code") in [190, 102, 4, 200, 10]: logger.warning(f"Detected Facebook API auth error: {error_obj.get('code')}") # Log more details about app ID related errors 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}") # Provide a clearer error message without the confusing "Provide valid app ID" message return { "error": { "message": "Meta API authentication configuration issue. Please check your app credentials.", "original_error": error_obj.get("message"), "code": error_obj.get("code") } } auth_manager.invalidate_token() # Include full details for technical users full_response = { "headers": dict(e.response.headers), "status_code": e.response.status_code, "url": str(e.response.url), "reason": getattr(e.response, "reason_phrase", "Unknown reason"), "request_method": e.request.method, "request_url": str(e.request.url) } # Return a properly structured error object return { "error": { "message": f"HTTP Error: {e.response.status_code}", "details": error_info, "full_response": full_response } } except Exception as e: logger.error(f"Request Error: {str(e)}") return {"error": {"message": str(e)}}