Databricks Isv REST API

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)

Per-Auth-Type Variables

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:

Troubleshooting

Reference

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