get_ad_accounts

Instructions

Implementation Reference

• meta_ads_mcp/core/accounts.py:33-97 (handler)

  Handler function for the get_ad_accounts tool. Decorated with @mcp.tool(). Fetches ad accounts via Meta Graph API, preferring Business Manager (/owned_ad_accounts) with fallback to user-level (/me/adaccounts). Maps numeric account status codes to human-readable names and returns accounts with rate limit info.

  @mcp.tool()
  def get_ad_accounts(limit: int = 50) -> dict:
      """
      List all ad accounts accessible to the authenticated user.
  
      Returns account IDs, names, statuses, currencies, and timezones
      for all accounts in your business portfolio.
  
      Args:
          limit: Maximum number of accounts to return (default 50).
      """
      api_client._ensure_initialized()
  
      try:
          business_id = get_business_id()
          if business_id:
              # Get accounts via business manager (preferred for agency setup)
              result = api_client.graph_get(
                  f"/{business_id}/owned_ad_accounts",
                  fields=[
                      "id", "name", "account_status", "currency",
                      "timezone_name", "amount_spent", "balance",
                      "business", "funding_source_details",
                  ],
                  params={"limit": str(limit)},
              )
          else:
              # Fallback: get accounts via user
              result = api_client.graph_get(
                  "/me/adaccounts",
                  fields=[
                      "id", "name", "account_status", "currency",
                      "timezone_name", "amount_spent", "balance",
                  ],
                  params={"limit": str(limit)},
              )
  
          accounts = result.get("data", [])
  
          # Map account_status codes to human-readable values
          status_map = {
              1: "ACTIVE",
              2: "DISABLED",
              3: "UNSETTLED",
              7: "PENDING_RISK_REVIEW",
              8: "PENDING_SETTLEMENT",
              9: "IN_GRACE_PERIOD",
              100: "PENDING_CLOSURE",
              101: "CLOSED",
              201: "ANY_ACTIVE",
              202: "ANY_CLOSED",
          }
  
          for account in accounts:
              status_code = account.get("account_status")
              account["account_status_name"] = status_map.get(status_code, f"UNKNOWN ({status_code})")
  
          return {
              "total": len(accounts),
              "accounts": accounts,
              "rate_limit_usage_pct": api_client.rate_limits.max_usage_pct,
          }
  
      except FacebookRequestError as e:
          raise api_client.handle_sdk_error(e)

• meta_ads_mcp/core/accounts.py:33-33 (registration)

  Registration of get_ad_accounts as an MCP tool via the @mcp.tool() decorator. The module is imported in meta_ads_mcp/server.py line 23 which triggers registration.

  @mcp.tool()

• meta_ads_mcp/core/auth.py:85-94 (helper)

  Helper function get_business_id() used by get_ad_accounts to determine whether to query via Business Manager or fallback to user-level endpoint.

  def get_business_id() -> Optional[str]:
      """Get the primary business ID for the authenticated user."""
      try:
          result = api_client.graph_get("/me/businesses", fields=["id", "name"])
          businesses = result.get("data", [])
          if businesses:
              return businesses[0].get("id")
          return None
      except MetaAPIError:
          return None

• meta_ads_mcp/core/api.py:138-367 (helper)

  The api_client singleton used by get_ad_accounts to make Graph API GET requests with retry logic, rate limit tracking, and error handling.

  class MetaAPIClient:
      """
      Unified client for Meta Graph API access.
  
      Provides:
      - facebook-business SDK initialization and access
      - Raw HTTP client for endpoints the SDK doesn't cover
      - Rate limit monitoring
      - appsecret_proof generation
      - Error classification
      """
  
      def __init__(self):
          self._sdk_initialized = False
          self._http_client: Optional[httpx.Client] = None
          self._access_token: Optional[str] = None
          self._app_secret: Optional[str] = None
          self._app_id: Optional[str] = None
          self.rate_limits = RateLimitStatus()
  
      def initialize(self):
          """Initialize API client from environment variables."""
          self._access_token = os.environ.get("META_ACCESS_TOKEN")
          self._app_secret = os.environ.get("META_APP_SECRET")
          self._app_id = os.environ.get("META_APP_ID")
  
          if not self._access_token:
              raise MetaAPIError("META_ACCESS_TOKEN environment variable is not set", error_code=-1)
  
          # Initialize facebook-business SDK
          FacebookAdsApi.init(
              app_id=self._app_id or "",
              app_secret=self._app_secret or "",
              access_token=self._access_token,
              api_version=GRAPH_API_VERSION,
          )
          self._sdk_initialized = True
  
          # Initialize HTTP client for raw API calls
          self._http_client = httpx.Client(
              base_url=GRAPH_API_BASE,
              timeout=60.0,
              headers={"Accept": "application/json"},
          )
  
          logger.info("Meta API client initialized (SDK + HTTP), API version %s", GRAPH_API_VERSION)
  
      @property
      def is_initialized(self) -> bool:
          return self._sdk_initialized
  
      def _ensure_initialized(self):
          if not self._sdk_initialized:
              self.initialize()
  
      def _generate_appsecret_proof(self) -> Optional[str]:
          """Generate appsecret_proof via HMAC-SHA256 for added security."""
          if self._app_secret and self._access_token:
              return hmac.new(
                  self._app_secret.encode("utf-8"),
                  self._access_token.encode("utf-8"),
                  hashlib.sha256,
              ).hexdigest()
          return None
  
      def _build_params(self, params: Optional[dict] = None) -> dict:
          """Build request parameters with token and optional appsecret_proof."""
          result = {"access_token": self._access_token}
          proof = self._generate_appsecret_proof()
          if proof:
              result["appsecret_proof"] = proof
          if params:
              result.update(params)
          return result
  
      def get_ad_account(self, account_id: str) -> AdAccount:
          """Get an AdAccount SDK object for the given account ID."""
          self._ensure_initialized()
          if not account_id.startswith("act_"):
              account_id = f"act_{account_id}"
          return AdAccount(account_id)
  
      def graph_get(self, endpoint: str, params: Optional[dict] = None, fields: Optional[list] = None) -> dict:
          """
          Make a GET request to the Graph API via raw HTTP.
  
          Use this for endpoints not covered by the facebook-business SDK.
          Retries on transient rate-limit errors with exponential backoff.
          """
          self._ensure_initialized()
          request_params = self._build_params(params)
          if fields:
              request_params["fields"] = ",".join(fields)
  
          for attempt in range(BACKOFF_MAX_RETRIES + 1):
              response = self._http_client.get(endpoint, params=request_params)
              self.rate_limits.update_from_headers(dict(response.headers))
  
              if self.rate_limits.is_warning:
                  logger.warning("Rate limit usage at %.1f%% - approaching limit", self.rate_limits.max_usage_pct)
  
              if response.status_code == 200:
                  return response.json()
  
              try:
                  self._handle_http_error(response)
              except MetaAPIError as e:
                  if e.error_code not in RETRYABLE_ERROR_CODES or attempt == BACKOFF_MAX_RETRIES:
                      raise
                  wait = self._backoff_wait(attempt)
                  logger.warning(
                      "Retryable error %d (attempt %d/%d) on GET %s - waiting %.1fs",
                      e.error_code, attempt + 1, BACKOFF_MAX_RETRIES, endpoint, wait,
                  )
                  time.sleep(wait)
  
          raise MetaAPIError("Max retries exceeded", error_code=-1)  # unreachable
  
      def graph_post(self, endpoint: str, data: Optional[dict] = None,
                     params: Optional[dict] = None, json_body: Optional[dict] = None) -> dict:
          """
          Make a POST request to the Graph API via raw HTTP.
  
          Enforces a minimum inter-request delay to stay safely under Meta's 100 QPS
          hard cap. Retries on transient rate-limit errors with exponential backoff
          (or Meta's own estimated_time_to_regain_access when set).
          """
          self._ensure_initialized()
          request_params = self._build_params(params)
  
          for attempt in range(BACKOFF_MAX_RETRIES + 1):
              # Enforce minimum inter-write delay (keeps us under 100 QPS hard cap)
              time.sleep(WRITE_THROTTLE_DELAY)
  
              if json_body:
                  response = self._http_client.post(
                      endpoint,
                      params=request_params,
                      json=json_body,
                      headers={"Content-Type": "application/json; charset=utf-8"},
                  )
              elif data:
                  response = self._http_client.post(endpoint, params=request_params, data=data)
              else:
                  response = self._http_client.post(endpoint, params=request_params)
  
              self.rate_limits.update_from_headers(dict(response.headers))
  
              if response.status_code == 200:
                  return response.json()
  
              try:
                  self._handle_http_error(response)
              except MetaAPIError as e:
                  if e.error_code not in RETRYABLE_ERROR_CODES or attempt == BACKOFF_MAX_RETRIES:
                      raise
                  wait = self._backoff_wait(attempt)
                  logger.warning(
                      "Retryable error %d (attempt %d/%d) on POST %s - waiting %.1fs",
                      e.error_code, attempt + 1, BACKOFF_MAX_RETRIES, endpoint, wait,
                  )
                  time.sleep(wait)
  
          raise MetaAPIError("Max retries exceeded", error_code=-1)  # unreachable
  
      def _backoff_wait(self, attempt: int) -> float:
          """Calculate wait time for retry attempt.
  
          Uses Meta's estimated_time_to_regain_access when available (authoritative).
          Falls back to exponential backoff with jitter.
          """
          wait_minutes = self.rate_limits.estimated_time_to_regain_access_minutes
          if wait_minutes > 0:
              logger.warning(
                  "Meta BUC throttle: estimated_time_to_regain_access=%d min. Waiting as instructed.",
                  wait_minutes,
              )
              return float(wait_minutes * 60)
          jitter = random.uniform(0, 1.0)
          return min(BACKOFF_BASE_SECONDS * (2 ** attempt) + jitter, BACKOFF_MAX_SECONDS)
  
      def _handle_http_error(self, response: httpx.Response):
          """Parse and raise structured API error."""
          try:
              body = response.json()
              error = body.get("error", {})
              raise MetaAPIError(
                  message=error.get("message", f"HTTP {response.status_code}"),
                  error_code=error.get("code", response.status_code),
                  error_subcode=error.get("error_subcode", 0),
                  error_type=error.get("type", ""),
                  fbtrace_id=error.get("fbtrace_id", ""),
                  is_transient=error.get("is_transient", False),
              )
          except (json.JSONDecodeError, KeyError):
              raise MetaAPIError(
                  message=f"HTTP {response.status_code}: {response.text[:500]}",
                  error_code=response.status_code,
              )
  
      def handle_sdk_error(self, error: FacebookRequestError) -> MetaAPIError:
          """Convert SDK error to our structured error type."""
          return MetaAPIError(
              message=error.api_error_message() or str(error),
              error_code=error.api_error_code() or 0,
              error_subcode=error.api_error_subcode() or 0,
              error_type=error.api_error_type() or "",
              fbtrace_id=error.api_transient_error() or "",
              is_transient=bool(error.api_transient_error()),
          )
  
      def check_token_health(self) -> dict:
          """Verify token validity and permissions."""
          self._ensure_initialized()
          try:
              result = self.graph_get("/me", fields=["id", "name"])
              return {
                  "status": "valid",
                  "user_id": result.get("id"),
                  "user_name": result.get("name"),
                  "rate_limit_usage_pct": self.rate_limits.max_usage_pct,
              }
          except MetaAPIError as e:
              if e.error_code in (190, 102):
                  return {"status": "expired", "error": str(e)}
              return {"status": "error", "error": str(e)}
  
  
  # Module-level singleton
  api_client = MetaAPIClient()

• meta_ads_mcp/server.py:22-24 (registration)

  The import in server.py that triggers loading of the accounts module (and thus registration of get_ad_accounts via @mcp.tool() decorator).

  # --- Accounts & Auth ---
  from meta_ads_mcp.core import accounts  # noqa: E402, F401