databricks-isv-rest-api

name	databricks-isv-rest-api
description	REST API authentication for Databricks ISV integrations: PAT, OAuth M2M, U2M. Use when building or testing HTTP/REST integrations with Databricks.

REST API Authentication (ISV)

Use this skill when implementing or testing REST API authentication for Databricks partner integrations.

Golden Snippets (Copy-Paste Accurate)

Every request – headers (required):

headers = {
    "Authorization": f"Bearer {token}",
    "User-Agent": "YourCompany_YourProduct/1.0.0",
    "Content-Type": "application/json",
}
# GET: requests.get(url, headers=headers)
# POST: requests.post(url, headers=headers, json=data)

OAuth M2M – obtain token:

token_url = f"https://{host}/oidc/v1/token"
resp = requests.post(
    token_url,
    auth=(client_id, client_secret),
    data={"grant_type": "client_credentials", "scope": "all-apis"},
)
resp.raise_for_status()
token = resp.json()["access_token"]

---

Requirements

• Headers on every request: Authorization: Bearer <token>, User-Agent: <isv>_<product>/<version>, Content-Type: application/json for POST/PUT.
• Two validation tests:
  1. Unity Catalog Tables API – GET /api/2.1/unity-catalog/tables/<full_name> (no warehouse)
  2. Statement Execution API – POST /api/2.0/sql/statements with warehouse_id and SQL

---

Authentication Patterns

PAT (Personal Access Token)

import os
import requests

host = os.environ["DATABRICKS_HOST"].rstrip("/")
if not host.startswith("https://"):
    host = f"https://{host}"

token = os.environ["DATABRICKS_TOKEN"]

headers = {
    "Authorization": f"Bearer {token}",
    "User-Agent": "YourCompany_Product/1.0.0",
}

# Test: UC Tables API
table_name = "samples.nyctaxi.trips"
url = f"{host}/api/2.1/unity-catalog/tables/{table_name}"
resp = requests.get(url, headers=headers)
resp.raise_for_status()
print(f"Table: {resp.json()['name']}")

OAuth M2M (Client Credentials)

import os
import requests

host = os.environ["DATABRICKS_HOST"].rstrip("/")
if not host.startswith("https://"):
    host = f"https://{host}"

client_id = os.environ["DATABRICKS_CLIENT_ID"]
client_secret = os.environ["DATABRICKS_CLIENT_SECRET"]

# Get token
token_url = f"{host}/oidc/v1/token"
token_resp = requests.post(
    token_url,
    auth=(client_id, client_secret),
    data={"grant_type": "client_credentials", "scope": "all-apis"},
)
token_resp.raise_for_status()
token = token_resp.json()["access_token"]

headers = {
    "Authorization": f"Bearer {token}",
    "User-Agent": "YourCompany_Product/1.0.0",
}

# Test: UC Tables API
table_name = "samples.nyctaxi.trips"
url = f"{host}/api/2.1/unity-catalog/tables/{table_name}"
resp = requests.get(url, headers=headers)
resp.raise_for_status()
print(f"Table: {resp.json()['name']}")

U2M External-Browser (SDK Built-in App)

import os
from databricks.sdk.config import Config

host = os.environ["DATABRICKS_HOST"]

# Use SDK for browser-based token acquisition
# Do NOT set DATABRICKS_CLIENT_ID/CLIENT_SECRET
config = Config(host=host, auth_type="external-browser")
auth_headers = config.authenticate()
token = auth_headers["Authorization"].replace("Bearer ", "")

headers = {
    "Authorization": f"Bearer {token}",
    "User-Agent": "YourCompany_Product/1.0.0",
}

# Use token for REST calls...

U2M Custom-OAuth-App (PKCE)

Redirect URI must be consistent in three places:

1. The default in this code (http://localhost:8080/callback)
2. The DATABRICKS_REDIRECT_URI env var (if overriding)
3. Databricks Account Console → App connections → your app → Redirect URIs

A mismatch causes a silent OAuth callback failure. If you change the port, update all three locations.

import os
import requests
import secrets
import hashlib
import base64
import webbrowser
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlencode, urlparse, parse_qs

# Configuration
host = os.environ["DATABRICKS_HOST"].rstrip("/")
if not host.startswith("https://"):
    host = f"https://{host}"

client_id = os.environ["DATABRICKS_U2M_CLIENT_ID"]
client_secret = os.environ.get("DATABRICKS_U2M_CLIENT_SECRET")
redirect_uri = os.environ.get("DATABRICKS_REDIRECT_URI", "http://localhost:8080/callback")

# Generate PKCE values
verifier = secrets.token_urlsafe(64)
digest = hashlib.sha256(verifier.encode()).digest()
challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
state = secrets.token_urlsafe(16)

# Callback handler
class CallbackHandler(BaseHTTPRequestHandler):
    auth_code = None
    callback_received = threading.Event()
    
    def do_GET(self):
        if "/callback" in self.path:
            query = parse_qs(urlparse(self.path).query)
            CallbackHandler.auth_code = query.get("code", [None])[0]
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"Authentication complete. Close this tab.")
            CallbackHandler.callback_received.set()
        else:
            self.send_response(204)
            self.end_headers()
    
    def log_message(self, format, *args):
        pass

# Start server and open browser
port = urlparse(redirect_uri).port or 8080
auth_params = {
    "response_type": "code",
    "client_id": client_id,
    "redirect_uri": redirect_uri,
    "scope": "all-apis",
    "code_challenge": challenge,
    "code_challenge_method": "S256",
    "state": state,
}
auth_url = f"{host}/oidc/v1/authorize?{urlencode(auth_params)}"

CallbackHandler.auth_code = None
CallbackHandler.callback_received.clear()
server = HTTPServer(("localhost", port), CallbackHandler)
server.timeout = 120  # CRITICAL

webbrowser.open(auth_url)
while not CallbackHandler.callback_received.is_set():
    server.handle_request()
server.server_close()

# Exchange code for token
token_data = {
    "grant_type": "authorization_code",
    "code": CallbackHandler.auth_code,
    "redirect_uri": redirect_uri,
    "client_id": client_id,
    "code_verifier": verifier,
}
if client_secret:
    token_data["client_secret"] = client_secret

token_url = f"{host}/oidc/v1/token"
token_resp = requests.post(token_url, data=token_data)
token_resp.raise_for_status()
token = token_resp.json()["access_token"]

headers = {
    "Authorization": f"Bearer {token}",
    "User-Agent": "YourCompany_Product/1.0.0",
}
# Use token for REST calls...

U2M Token-Env (Pre-obtained Token)

import os
import requests

host = os.environ["DATABRICKS_HOST"].rstrip("/")
if not host.startswith("https://"):
    host = f"https://{host}"

# Use pre-obtained token
token = os.environ.get("DATABRICKS_ACCESS_TOKEN") or os.environ.get("DATABRICKS_TOKEN")

headers = {
    "Authorization": f"Bearer {token}",
    "User-Agent": "YourCompany_Product/1.0.0",
}

# Test: UC Tables API
table_name = "samples.nyctaxi.trips"
url = f"{host}/api/2.1/unity-catalog/tables/{table_name}"
resp = requests.get(url, headers=headers)
resp.raise_for_status()
print(f"Table: {resp.json()['name']}")

---

Multi-Auth Dispatcher Pattern

Support multiple auth types via an environment variable selector:

import os
import requests
from databricks.sdk.config import Config

auth_type = os.environ.get("APP_AUTH_TYPE", "oauth_m2m")

host = os.environ["DATABRICKS_HOST"].rstrip("/")
if not host.startswith("https://"):
    host = f"https://{host}"

def get_token():
    if auth_type == "pat":
        return os.environ["DATABRICKS_TOKEN"]
    
    elif auth_type == "oauth_m2m":
        client_id = os.environ["DATABRICKS_CLIENT_ID"]
        client_secret = os.environ["DATABRICKS_CLIENT_SECRET"]
        token_url = f"{host}/oidc/v1/token"
        resp = requests.post(
            token_url,
            auth=(client_id, client_secret),
            data={"grant_type": "client_credentials", "scope": "all-apis"},
        )
        resp.raise_for_status()
        return resp.json()["access_token"]
    
    elif auth_type == "u2m_external_browser":
        config = Config(host=host, auth_type="external-browser")
        return config.authenticate()["Authorization"].replace("Bearer ", "")
    
    elif auth_type == "u2m_custom_oauth_app":
        # Implement PKCE flow (see above)
        return run_pkce_flow()
    
    elif auth_type == "u2m_token_env":
        return os.environ.get("DATABRICKS_ACCESS_TOKEN") or os.environ.get("DATABRICKS_TOKEN")
    
    else:
        raise ValueError(f"Unknown auth type: {auth_type}")

token = get_token()
headers = {
    "Authorization": f"Bearer {token}",
    "User-Agent": "YourCompany_Product/1.0.0",
}

---

Environment Variables by Auth Type

Common Variables (All Auth Types)

Env Var	Required	Description
DATABRICKS_HOST	Yes	Workspace URL (with or without https://)
APP_AUTH_TYPE	Yes	pat | oauth_m2m | u2m_external_browser | u2m_custom_oauth_app | u2m_token_env
DATABRICKS_WAREHOUSE_ID	No	SQL Warehouse ID for Statement Execution API test

Per-Auth-Type Variables

Auth Type	Required	Optional
pat	DATABRICKS_TOKEN	—
oauth_m2m	DATABRICKS_CLIENT_ID, DATABRICKS_CLIENT_SECRET	—
u2m_external_browser	(none)	— (Unset M2M vars)