Databricks Sdk Foundation

Databricks SDK Foundation

Shared patterns for authentication, retry logic, error handling, cluster lifecycle, and output formatting using the databricks-sdk Python package. Other Databricks skills reference this foundation.

Prerequisites

Before using any Databricks skill, verify that the required packages are installed:

python3 -c "import databricks.sdk; import tenacity; print('OK')"

If this prints OK, you're ready. If it fails with ModuleNotFoundError, stop and tell the user:

The required Python packages databricks-sdk and tenacity are not installed. Please set up a virtual environment and install them:

Using uv (recommended):

uv venv && source .venv/bin/activate && uv pip install databricks-sdk tenacity

Using venv:

python3 -m venv .venv && source .venv/bin/activate && pip install databricks-sdk tenacity

Do NOT run pip install on behalf of the user. Wait for them to confirm the environment is ready.

dbx.py CLI Wrapper

This skill includes dbx.py, a thin CLI wrapper around the Python SDK. It provides CLI-level token efficiency with SDK-level robustness (retries, connection pooling, typed errors). All output is JSON. Errors go to stderr.

python3 dbx.py clusters list              # List all clusters
python3 dbx.py clusters get <ID>          # Get cluster details
python3 dbx.py clusters start <ID> --wait # Start and wait until RUNNING
python3 dbx.py clusters ensure <ID>       # Idempotent start
python3 dbx.py clusters find <PATTERN>    # Find by name

python3 dbx.py jobs list [--name PAT]     # List jobs
python3 dbx.py jobs submit --cluster <ID> --notebook <PATH> --wait
python3 dbx.py jobs run <JOB_ID> --wait   # Trigger existing job
python3 dbx.py jobs status <RUN_ID>       # Check run state
python3 dbx.py jobs output <RUN_ID>       # Get run output
python3 dbx.py jobs cancel <RUN_ID>

python3 dbx.py repl create <CLUSTER_ID>   # Create execution context
python3 dbx.py repl exec <CID> <CTX> <CODE>  # Execute and wait
python3 dbx.py repl destroy <CID> <CTX>   # Destroy context

python3 dbx.py workspace import <FILE> <PATH> --overwrite
python3 dbx.py workspace list <PATH>
python3 dbx.py workspace export <PATH> <FILE>

python3 dbx.py catalog list-catalogs
python3 dbx.py catalog list-schemas <CATALOG>
python3 dbx.py catalog list-tables <CATALOG> <SCHEMA>
python3 dbx.py catalog describe <FULL_NAME>
python3 dbx.py catalog search <CATALOG> --pattern <PATTERN>

Client Initialization

Auto-Detect Auth (Recommended)

The SDK follows the Databricks Unified Auth chain automatically -- picks up ~/.databrickscfg, environment variables, or managed identity:

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

Explicit Profile

Use a named profile from ~/.databrickscfg:

w = WorkspaceClient(profile="<PROFILE_NAME>")

Environment Variables

Never hardcode real tokens in code or notebooks. Prefer ~/.databrickscfg or set environment variables in your shell profile. This example is for local-only development:

import os

os.environ["DATABRICKS_HOST"] = "https://<WORKSPACE>.cloud.databricks.com"
os.environ["DATABRICKS_TOKEN"] = "<YOUR_PAT>"

w = WorkspaceClient()

Retry Pattern

Use tenacity for transient errors. The SDK raises specific exceptions for retryable conditions.

Decorator Usage

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from databricks.sdk.errors import ResourceConflict, TemporarilyUnavailable, TooManyRequests

@retry(
    retry=retry_if_exception_type((
        ResourceConflict,
        TemporarilyUnavailable,
        TooManyRequests,
        ConnectionError,
        TimeoutError,
    )),
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    reraise=True,
)
def create_cluster(w, config):
    return w.clusters.create_and_wait(**config)

Inline Usage

from tenacity import Retrying, stop_after_attempt, wait_exponential, retry_if_exception_type
from databricks.sdk.errors import ResourceConflict, TemporarilyUnavailable, TooManyRequests

for attempt in Retrying(
    retry=retry_if_exception_type((
        ResourceConflict,
        TemporarilyUnavailable,
        TooManyRequests,
        ConnectionError,
        TimeoutError,
    )),
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    reraise=True,
):
    with attempt:
        result = w.jobs.run_now(job_id=123)

Error Handling

The SDK raises typed exceptions from databricks.sdk.errors. Always handle specific errors before generic ones:

from databricks.sdk.errors import NotFound, PermissionDenied, BadRequest

try:
    cluster = w.clusters.get(cluster_id="<CLUSTER_ID>")
except NotFound:
    print(f"Cluster not found -- check the cluster ID")
except PermissionDenied:
    print(f"No access to this cluster -- verify permissions")
except BadRequest as e:
    print(f"Invalid request: {e}")

Common Exceptions

Cluster State Machine

Cluster states follow this lifecycle:

PENDING → RUNNING → RESIZING → RUNNING
                  ↘ RESTARTING → RUNNING
                  ↘ TERMINATING → TERMINATED

Start and Wait (Recommended)

The SDK provides a blocking helper that handles polling:

from databricks.sdk.errors import NotFound, ResourceConflict

cluster = w.clusters.get(cluster_id="<CLUSTER_ID>")

if cluster.state.value == "TERMINATED":
    w.clusters.start_and_wait(cluster_id="<CLUSTER_ID>")

Manual Polling (When You Need Progress Updates)

import time

def wait_for_cluster(w, cluster_id, timeout=600):
    deadline = time.time() + timeout
    while time.time() < deadline:
        info = w.clusters.get(cluster_id=cluster_id)
        state = info.state.value
        print(f"Cluster state: {state}")
        if state == "RUNNING":
            return info
        if state in ("TERMINATING", "TERMINATED", "ERROR"):
            raise RuntimeError(f"Cluster entered {state}: {info.state_message}")
        time.sleep(15)
    raise TimeoutError(f"Cluster did not reach RUNNING within {timeout}s")

Valid States

Output Formatting

SDK Objects to Dicts

Most SDK response objects support as_dict() for serialization:

cluster = w.clusters.get(cluster_id="<CLUSTER_ID>")
cluster_dict = cluster.as_dict()

JSON Formatting for Agent Output

import json

# Single object
cluster = w.clusters.get(cluster_id="<CLUSTER_ID>")
print(json.dumps(cluster.as_dict(), indent=2))

# List of objects
jobs = w.jobs.list()
print(json.dumps([j.as_dict() for j in jobs], indent=2))

Filtered Output

Extract only relevant fields to keep agent context concise:

cluster = w.clusters.get(cluster_id="<CLUSTER_ID>")
summary = {
    "cluster_id": cluster.cluster_id,
    "name": cluster.cluster_name,
    "state": cluster.state.value,
    "spark_version": cluster.spark_version,
    "num_workers": cluster.num_workers,
}
print(json.dumps(summary, indent=2))