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)
u2m_custom_oauth_app	DATABRICKS_U2M_CLIENT_ID	DATABRICKS_U2M_CLIENT_SECRET, DATABRICKS_REDIRECT_URI
u2m_token_env	DATABRICKS_ACCESS_TOKEN or DATABRICKS_TOKEN	—

CLIENT_ID Distinction

• DATABRICKS_CLIENT_ID → M2M service principal (for oauth_m2m)
• DATABRICKS_U2M_CLIENT_ID → Custom OAuth app (for u2m_custom_oauth_app)
• Using the wrong one causes "OAuth application with client_id not available"

---

Auth Isolation

Run each auth type with a clean environment. Do not set both DATABRICKS_TOKEN and DATABRICKS_CLIENT_ID/DATABRICKS_CLIENT_SECRET in the same process.

# PAT
env -i PATH="$PATH" HOME="$HOME" \
    DATABRICKS_HOST="$DATABRICKS_HOST" \
    DATABRICKS_TOKEN="$DATABRICKS_TOKEN" \
    APP_AUTH_TYPE=pat \
    python your_rest_example.py

# OAuth M2M
env -i PATH="$PATH" HOME="$HOME" \
    DATABRICKS_HOST="$DATABRICKS_HOST" \
    DATABRICKS_CLIENT_ID="$DATABRICKS_CLIENT_ID" \
    DATABRICKS_CLIENT_SECRET="$DATABRICKS_CLIENT_SECRET" \
    APP_AUTH_TYPE=oauth_m2m \
    python your_rest_example.py

See skills/testing/SKILL.md for complete runner patterns.

---

Validation Tests

Test 1: UC Tables API (No Warehouse)

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']}")

Test 2: Statement Execution API (Requires Warehouse)

warehouse_id = os.environ.get("DATABRICKS_WAREHOUSE_ID")
if warehouse_id:
    url = f"{host}/api/2.0/sql/statements"
    data = {
        "warehouse_id": warehouse_id,
        "statement": "SELECT 1 AS test",
        "wait_timeout": "30s",
    }
    resp = requests.post(url, headers=headers, json=data)
    resp.raise_for_status()
    result = resp.json()
    print(f"Statement status: {result['status']['state']}")

---

Token Endpoints

M2M Token

POST https://<host>/oidc/v1/token
Authorization: Basic base64(client_id:client_secret)
Content-Type: application/x-www-form-urlencoded

grant_type=client_credentials&scope=all-apis

PKCE Token Exchange

POST https://<host>/oidc/v1/token
Content-Type: application/x-www-form-urlencoded

grant_type=authorization_code
&code={auth_code}
&redirect_uri={redirect_uri}
&client_id={client_id}
&code_verifier={verifier}

---

HTTPServer Pitfalls (Custom OAuth App)

When implementing the local callback server:

Issue	Solution
Server blocks forever	Set server.timeout = 120 before handle_request()
shutdown() hangs	Use server.server_close() instead
Browser sends favicon requests	Loop until actual callback received
Port already in use	lsof -ti:8080 | xargs kill -9

---

Troubleshooting

Error	Cause	Fix
"more than one authorization method"	Multiple auth env vars set	Use env -i for clean environment
"OAuth application with client_id not available"	Using M2M client_id for U2M	Use DATABRICKS_U2M_CLIENT_ID
"redirect_uri not registered"	URI mismatch	Exact match required in App connections
401 Unauthorized	Token expired or invalid	Re-authenticate; tokens expire in ~1 hour
PKCE flow times out	User didn't complete sign-in	Check browser popups; increase timeout

---

Reference

• Full patterns: authentication.md
• U2M details: skills/u2m/SKILL.md
• Testing patterns: skills/testing/SKILL.md