databricks-mcp

Instructions

Implementation Reference

• tools/metrics.py:161-220 (handler)

  The implementation of the get_run_task_metrics tool, which retrieves and formats execution metrics for tasks within a specific job run.

  def get_run_task_metrics(ctx: Context, run_id: int) -> Dict[str, Any]:
      """
      Get job run task execution time details
  
      Args:
          run_id: Job Run ID
  
      Returns:
          Task setup/execute/cleanup times (time in local timezone)
      """
      w = get_workspace_client()
      ctx.info(f"Querying run {run_id} task metrics...")
  
      run = w.jobs.get_run(run_id=run_id)
      run_dict = run.as_dict()
  
      def ms_to_local(ms):
          if not ms:
              return None
          local_time = datetime.utcfromtimestamp(ms / 1000) + timedelta(hours=8)
          return local_time.strftime("%Y-%m-%d %H:%M:%S")
  
      tasks = []
      total_duration = 0
  
      if run_dict.get("tasks"):
          for t in run_dict["tasks"]:
              setup_ms = t.get("setup_duration", 0) or 0
              exec_ms = t.get("execution_duration", 0) or 0
              cleanup_ms = t.get("cleanup_duration", 0) or 0
              total_ms = setup_ms + exec_ms + cleanup_ms
              total_duration += total_ms
  
              tasks.append({
                  "task_key": t.get("task_key"),
                  "state": t.get("state", {}).get("result_state") or t.get("state", {}).get("life_cycle_state"),
                  "setup_sec": round(setup_ms / 1000, 1),
                  "execution_sec": round(exec_ms / 1000, 1),
                  "cleanup_sec": round(cleanup_ms / 1000, 1),
                  "total_sec": round(total_ms / 1000, 1),
                  "cluster_id": t.get("existing_cluster_id") or t.get("cluster_instance", {}).get("cluster_id")
              })
  
      tasks.sort(key=lambda x: x["execution_sec"], reverse=True)
  
      start_ms = run_dict.get("start_time")
      end_ms = run_dict.get("end_time")
      duration_min = round((end_ms - start_ms) / 1000 / 60, 2) if start_ms and end_ms else None
  
      return {
          "run_id": run_id,
          "job_id": run_dict.get("job_id"),
          "state": run_dict.get("state", {}).get("result_state"),
          "start_time_local": ms_to_local(start_ms),
          "end_time_local": ms_to_local(end_ms),
          "duration_min": duration_min,
          "total_task_duration_sec": round(total_duration / 1000, 1),
          "tasks": tasks,
          "slowest_task": tasks[0]["task_key"] if tasks else None
      }

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

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

  @mcp.tool