databricks-mcp

Instructions

Implementation Reference

• tools/metrics.py:124-157 (handler)

  Implementation of the get_cluster_events tool handler.

  def get_cluster_events(ctx: Context, cluster_id: str, limit: int = 20) -> List[Dict[str, Any]]:
      """
      Get cluster event history (start, terminate, resize, errors, etc.)
  
      Args:
          cluster_id: Cluster ID
          limit: Max number of records to return
  
      Returns:
          Event list (time in local timezone)
      """
      w = get_workspace_client()
      ctx.info(f"Querying cluster {cluster_id} events...")
  
      events_iter = w.clusters.events(cluster_id=cluster_id, limit=limit)
  
      results = []
      for event in events_iter:
          e = event.as_dict()
          ts = e.get("timestamp")
          if ts:
              local_time = datetime.utcfromtimestamp(ts / 1000) + timedelta(hours=8)
              time_str = local_time.strftime("%Y-%m-%d %H:%M:%S")
          else:
              time_str = None
  
          results.append({
              "time_local": time_str,
              "timestamp_ms": ts,
              "type": e.get("type"),
              "details": e.get("details")
          })
  
      return results

• tools/metrics.py:123-123 (registration)

  Registration of the get_cluster_events tool using the @mcp.tool decorator.

  @mcp.tool