MCP-Ambari-API

Instructions

Implementation Reference

• src/mcp_ambari_api/functions.py:711-713 (handler)

  This is the core handler function that implements listing available Ambari Metrics System (AMS) application IDs (apps). It fetches the dynamic metric catalog from AMS metadata and returns the sorted list of available appIds, which directly corresponds to the 'list_ambari_metric_apps' tool functionality.

  async def get_available_app_ids(use_cache: bool = True) -> List[str]:
      catalog, _ = await ensure_metric_catalog(use_cache=use_cache)
      return sorted(catalog.keys())

• src/mcp_ambari_api/functions.py:601-668 (helper)

  Supporting function that builds and caches the metric catalog (appId -> list of metrics) from Ambari Metrics metadata API, including fallback probing of curated apps and synonym resolution. Called by the handler.

  async def ensure_metric_catalog(use_cache: bool = True) -> Tuple[Dict[str, List[str]], Dict[str, str]]:
      """Return (catalog, lookup) built from AMS metadata, refreshing when cache expires."""
      global _DYNAMIC_CATALOG_CACHE
  
      now = time.monotonic()
      cached_catalog = _DYNAMIC_CATALOG_CACHE.get("catalog") or {}
      cached_lookup = _DYNAMIC_CATALOG_CACHE.get("lookup") or {}
      cached_timestamp = float(_DYNAMIC_CATALOG_CACHE.get("timestamp", 0.0))
  
      if use_cache and cached_catalog and now - cached_timestamp < AMBARI_METRICS_METADATA_TTL:
          return cached_catalog, cached_lookup
  
      entries = await get_metrics_metadata(None, use_cache=use_cache)
  
      # Fallback: probe common apps individually if the bulk metadata query returns nothing.
      if not entries:
          aggregated: List[Dict[str, Any]] = []
          for fallback_app in CURATED_METRIC_APP_IDS:
              if _is_excluded_app(fallback_app):
                  continue
              fallback_entries = await get_metrics_metadata(fallback_app, use_cache=use_cache)
              if fallback_entries:
                  aggregated.extend(fallback_entries)
          entries = aggregated
  
      metrics_by_app: Dict[str, Set[str]] = {}
      lookup: Dict[str, str] = {}
  
      for entry in entries:
          app_hint = entry.get("appid") or entry.get("appId") or entry.get("application")
          metric_hint = entry.get("metricname") or entry.get("metricName") or entry.get("metric_name")
          if not app_hint or not metric_hint:
              continue
  
          app_name = str(app_hint).strip()
          metric_name = str(metric_hint).strip()
          if not app_name or not metric_name:
              continue
  
          if _is_excluded_app(app_name):
              continue
  
          metrics_by_app.setdefault(app_name, set()).add(metric_name)
          lookup.setdefault(_normalize_app_key(app_name), app_name)
  
      catalog = {app: sorted(values) for app, values in metrics_by_app.items()}
  
      # Ensure canonical entries exist even when AMS metadata is sparse.
      for canonical in APP_SYNONYMS.keys():
          if _is_excluded_app(canonical):
              continue
          lower = _normalize_app_key(canonical)
          lookup.setdefault(lower, canonical)
          catalog.setdefault(canonical, [])
  
      for excluded in list(catalog.keys()):
          if _is_excluded_app(excluded):
              catalog.pop(excluded, None)
              lookup.pop(_normalize_app_key(excluded), None)
  
      _DYNAMIC_CATALOG_CACHE = {
          "timestamp": now,
          "catalog": catalog,
          "lookup": lookup,
      }
  
      return catalog, lookup

• src/mcp_ambari_api/functions.py:86-92 (helper)

  Curated list of prioritized Ambari metric appIds used as fallback when bulk metadata query returns empty.

  CURATED_METRIC_APP_IDS = [
      "ambari_server",
      "namenode",
      "datanode",
      "nodemanager",
      "resourcemanager",
  ]