# Databricks Isv Python Sdk

> PWAF-compliant Databricks SDK for Python (databricks-sdk): PAT, OAuth M2M, U2M token-env, U2M custom OAuth app (PKCE); useragent.with_partner/with_product. Use when building or testing Python SDK workspace API integrations.

- Skill: `databricks-solutions/databricks-isv-python-sdk` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add databricks-solutions/databricks-isv-python-sdk`
- Raw SKILL.md: https://api.skillmd.com/api/skills/databricks-solutions/databricks-isv-python-sdk/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- Author: databricks-solutions (https://skillmd.com/u/databricks-solutions)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/databricks-solutions/databricks-isv-python-sdk

---


<!-- skill-version: 1.0.0 -->

# Databricks SDK for Python (ISV)

Use this skill when implementing or testing **Databricks SDK for Python** (`databricks-sdk`) integrations for PWAF-compliant workspace management, Unity Catalog, Jobs, and REST-style API access.

## PWAF Documentation Links

| Resource | URL |
|----------|-----|
| **PWAF Home** | https://databrickslabs.github.io/partner-architecture/ |
| **Authentication Best Practices** | https://databrickslabs.github.io/partner-architecture/isv-partners/lakehouse-patterns/access-auth/ |
| **OAuth M2M Guide** | https://databrickslabs.github.io/partner-architecture/isv-partners/lakehouse-patterns/access-auth/oauth-m2m |
| **OAuth U2M Guide** | https://databrickslabs.github.io/partner-architecture/isv-partners/lakehouse-patterns/access-auth/oauth-u2m |
| **User-Agent Telemetry** | https://databrickslabs.github.io/partner-architecture/isv-partners/telemetry-attribution/sdks |
| **Python SDK Docs** | https://docs.databricks.com/aws/en/dev-tools/sdk-python |
| **Python SDK GitHub** | https://github.com/databricks/databricks-sdk-py |

## Requirements

- **SDK:** `databricks-sdk` 0.20.0+
- **Python:** 3.8+
- **Install:** `pip install databricks-sdk`

## Authentication Decision Guide

```
Which authentication method to use?

Production / automated workloads?
  → OAuth M2M (client credentials)  ✅ RECOMMENDED

User-interactive flows?
  → U2M Custom OAuth App (PKCE)  ✅ SUPPORTED

Already have an OAuth access token?
  → U2M Token-Env  ✅ SUPPORTED

Local development/testing only?
  → PAT (Personal Access Token)  ⚠️ LIMITED
```

## Authentication Comparison

| Method | PWAF Status | Auto Token Refresh | Browser Required | Use Case |
|--------|-------------|-------------------|------------------|----------|
| PAT | ⚠️ Limited | No | No | Testing only |
| OAuth M2M | ✅ Recommended | Yes (SDK handles) | No | Production/automated |
| U2M Custom OAuth App | ✅ Supported | No | Yes | User-interactive |
| U2M Token-Env | ✅ Supported | No | No | Headless/CI |

---

## CLIENT_ID Distinction (CRITICAL)

| Variable | Purpose | Value Source | Use Case |
|----------|---------|--------------|----------|
| `DATABRICKS_CLIENT_ID` | M2M Service Principal | Workspace → Identity and access → Service principals → UUID | Automated/production |
| `DATABRICKS_U2M_CLIENT_ID` | U2M Custom OAuth App | Account Console → Settings → App connections → Client ID | User-interactive |

**Never mix these.** M2M uses service principal credentials for server-to-server auth. U2M custom OAuth uses a registered app for user login.

---

## Token Lifetime Summary

| Token Type | Typical TTL | Refresh Mechanism |
|------------|-------------|-------------------|
| PAT | 90 days (configurable) | Manual regeneration |
| M2M access_token | 1 hour | SDK auto-refreshes |
| U2M access_token | 1 hour | Manual re-auth or refresh_token |

---

## User-Agent (Required per PWAF)

Call once before creating `WorkspaceClient`:

```python
from databricks.sdk import useragent

useragent.with_partner("YourCompany")
useragent.with_product("YourCompany_YourProduct", "1.0.0")
```

**Alternative (on WorkspaceClient):**

```python
from databricks.sdk import WorkspaceClient

client = WorkspaceClient(
    host="https://myworkspace.cloud.databricks.com",
    token="dapi...",
    product="YourCompany_YourProduct",
    product_version="1.0.0"
)
```

---

## Environment Variables Reference

| Variable | Required For | Description |
|----------|-------------|-------------|
| `DATABRICKS_HOST` | All | Workspace URL (e.g., `https://myworkspace.cloud.databricks.com`) |
| `DATABRICKS_TOKEN` | PAT | Personal access token |
| `DATABRICKS_CLIENT_ID` | OAuth M2M | Service principal UUID |
| `DATABRICKS_CLIENT_SECRET` | OAuth M2M | Service principal OAuth secret |
| `DATABRICKS_U2M_CLIENT_ID` | U2M Custom OAuth | Custom OAuth app client ID from App connections |
| `DATABRICKS_U2M_CLIENT_SECRET` | U2M Custom OAuth (optional) | Custom OAuth app client secret (if confidential) |
| `DATABRICKS_REDIRECT_URI` | U2M Custom OAuth (optional) | Custom redirect URI (default: `http://localhost:8080/callback`) |
| `DATABRICKS_ACCESS_TOKEN` | U2M Token-Env | Pre-obtained OAuth access token |
| `APP_AUTH_TYPE` | Multi-auth | App-level auth selector: `pat`, `oauth_m2m`, `oauth_u2m`, `u2m_token_env` |

**Important:** 
- Use `APP_AUTH_TYPE` (not `DATABRICKS_AUTH_TYPE`) for app-level auth selection because the SDK reads `DATABRICKS_AUTH_TYPE` internally.
- Do not mix M2M and U2M environment variables. Use `env -i` for clean test environments.

---

## Config Parameters Reference

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| `host` | str | Yes | Workspace URL (include `https://`) |
| `token` | str | PAT only | Personal access token |
| `client_id` | str | OAuth M2M | Service principal UUID |
| `client_secret` | str | OAuth M2M | Service principal OAuth secret |
| `auth_type` | str | OAuth M2M | Set to `"oauth-m2m"` for M2M (avoids default credential resolution) |
| `azure_use_msi` | bool | Azure only | Enable Azure Managed Identity |
| `azure_client_id` | str | Azure MSI | Client ID for user-assigned MSI |
| `product` | str | Telemetry | Product name for User-Agent |
| `product_version` | str | Telemetry | Product version for User-Agent |

**Critical:** When using OAuth M2M, always pass `auth_type="oauth-m2m"` to Config to avoid "cannot configure default credentials" error.

---

## Complete Examples

### PAT Authentication (Testing Only)

```python
import os
from databricks.sdk import WorkspaceClient, useragent

useragent.with_partner("YourCompany")
useragent.with_product("YourProduct", "1.0.0")

def get_client_pat():
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    if not host.startswith("https://"):
        host = f"https://{host}"
    token = os.environ.get("DATABRICKS_TOKEN", "").strip()
    if not token:
        raise ValueError("DATABRICKS_TOKEN required")
    
    return WorkspaceClient(host=host, token=token)

client = get_client_pat()
me = client.current_user.me()
print(f"Authenticated as: {me.user_name}")
```

### OAuth M2M (Recommended for Production)

```python
import os
from databricks.sdk import WorkspaceClient, useragent
from databricks.sdk.config import Config

useragent.with_partner("YourCompany")
useragent.with_product("YourProduct", "1.0.0")

def get_client_m2m():
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    if not host.startswith("https://"):
        host = f"https://{host}"
    
    config = Config(
        host=host,
        client_id=os.environ.get("DATABRICKS_CLIENT_ID", ""),
        client_secret=os.environ.get("DATABRICKS_CLIENT_SECRET", ""),
        auth_type="oauth-m2m",  # CRITICAL: Required for M2M
    )
    return WorkspaceClient(config=config)

client = get_client_m2m()
# SDK automatically handles token refresh
tables = list(client.tables.list("samples", "nyctaxi"))
print(f"Found {len(tables)} tables")
```

### OAuth U2M Custom OAuth App (PKCE)

```python
import os
import base64
import hashlib
import secrets
import string
import webbrowser
from http.server import HTTPServer, BaseHTTPRequestHandler
from threading import Thread
from urllib.parse import parse_qs, urlparse, urlencode
import requests
from databricks.sdk import WorkspaceClient

DEFAULT_REDIRECT_URI = "http://localhost:8080/callback"

def pkce_verifier_and_challenge():
    """Generate PKCE code_verifier and code_challenge per RFC 7636."""
    allowed = string.ascii_letters + string.digits + "-._~"
    code_verifier = "".join(secrets.choice(allowed) for _ in range(64))
    digest = hashlib.sha256(code_verifier.encode()).digest()
    code_challenge = base64.urlsafe_b64encode(digest).decode().rstrip("=")
    return code_verifier, code_challenge

def run_u2m_flow(host: str, client_id: str, redirect_uri: str = None, 
                 client_secret: str = None, scope: str = "all-apis") -> dict:
    """Run OAuth 2.0 authorization code flow with PKCE."""
    redirect_uri = redirect_uri or DEFAULT_REDIRECT_URI
    if not host.startswith("https://"):
        host = f"https://{host}"
    
    code_verifier, code_challenge = pkce_verifier_and_challenge()
    state = secrets.token_urlsafe(32)
    
    auth_url = f"{host}/oidc/v1/authorize?" + urlencode({
        "response_type": "code",
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "scope": scope,
        "code_challenge": code_challenge,
        "code_challenge_method": "S256",
        "state": state,
    })
    
    code_holder = []
    parsed = urlparse(redirect_uri)
    port = parsed.port or 8080
    
    class Handler(BaseHTTPRequestHandler):
        def do_GET(self):
            q = parse_qs(urlparse(self.path).query)
            code_holder.append(q.get("code", [None])[0])
            self.send_response(200)
            self.send_header("Content-type", "text/html; charset=utf-8")
            self.end_headers()
            self.wfile.write(b"<html><body><p>Authentication successful.</p></body></html>")
        def log_message(self, *args): pass
    
    server = HTTPServer(("localhost", port), Handler)
    Thread(target=server.handle_request, daemon=True).start()
    
    webbrowser.open(auth_url)
    print(f"Waiting for redirect at {redirect_uri}...")
    server.handle_request()
    
    if not code_holder[0]:
        raise RuntimeError("No authorization code received")
    
    token_data = {
        "grant_type": "authorization_code",
        "code": code_holder[0],
        "redirect_uri": redirect_uri,
        "client_id": client_id,
        "code_verifier": code_verifier,
    }
    if client_secret:
        token_data["client_secret"] = client_secret
    
    resp = requests.post(f"{host}/oidc/v1/token", data=token_data)
    resp.raise_for_status()
    return resp.json()

def get_client_u2m():
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    client_id = os.environ.get("DATABRICKS_U2M_CLIENT_ID", "").strip()
    redirect_uri = os.environ.get("DATABRICKS_REDIRECT_URI", "") or DEFAULT_REDIRECT_URI
    client_secret = os.environ.get("DATABRICKS_U2M_CLIENT_SECRET", "").strip() or None
    
    tokens = run_u2m_flow(host, client_id, redirect_uri, client_secret)
    access_token = tokens.get("access_token")
    
    if not host.startswith("https://"):
        host = f"https://{host}"
    
    return WorkspaceClient(
        host=host,
        token=access_token,
        product="YourCompany_YourProduct",
        product_version="1.0.0"
    )
```

### OAuth U2M Token-Env (Pre-obtained Token)

```python
import os
from databricks.sdk import WorkspaceClient, useragent

useragent.with_partner("YourCompany")
useragent.with_product("YourProduct", "1.0.0")

def get_client_token_env():
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    if not host.startswith("https://"):
        host = f"https://{host}"
    
    # Use pre-obtained OAuth token
    access_token = os.environ.get("DATABRICKS_ACCESS_TOKEN", "").strip()
    if not access_token:
        raise ValueError("DATABRICKS_ACCESS_TOKEN required")
    
    return WorkspaceClient(host=host, token=access_token)
```

---

## Error Handling Patterns

```python
from databricks.sdk.core import DatabricksError

def classify_error(e: Exception) -> tuple[str, bool, str]:
    """Classify error and determine if retryable.
    
    Returns: (category, retryable, message)
    """
    if isinstance(e, DatabricksError):
        error_code = getattr(e, 'error_code', '')
        status_code = getattr(e, 'status_code', 0)
        
        if status_code == 401:
            return ("AUTHENTICATION_ERROR", False, "Check credentials")
        if status_code == 403:
            return ("PERMISSION_DENIED", False, "Check user/SP permissions")
        if status_code == 404:
            return ("NOT_FOUND", False, "Resource doesn't exist")
        if status_code == 429:
            return ("RATE_LIMITED", True, "Retry with exponential backoff")
        if status_code >= 500:
            return ("SERVER_ERROR", True, "Retry")
    
    msg = str(e).lower()
    if "more than one authorization method" in msg:
        return ("AUTH_CONFLICT", False, "Use env -i for clean environment")
    if "cannot configure default credentials" in msg:
        return ("CONFIG_ERROR", False, "Pass auth_type='oauth-m2m' to Config")
    if "timeout" in msg or "connection" in msg:
        return ("NETWORK_ERROR", True, "Check connectivity")
    
    return ("UNKNOWN_ERROR", False, str(e))
```

---

## Retry with Exponential Backoff

```python
import time
import random

def retry_with_backoff(operation, max_retries: int = 3):
    """Execute operation with exponential backoff on retryable errors."""
    for attempt in range(max_retries + 1):
        try:
            return operation()
        except Exception as e:
            category, retryable, _ = classify_error(e)
            if not retryable or attempt >= max_retries:
                raise
            backoff = (2 ** attempt) + random.uniform(0, 1)
            print(f"Retry {attempt + 1}/{max_retries} in {backoff:.1f}s ({category})")
            time.sleep(backoff)
```

---

## Multi-Auth Pattern

```python
import os
from databricks.sdk import WorkspaceClient, useragent
from databricks.sdk.config import Config

useragent.with_partner("YourCompany")
useragent.with_product("YourProduct", "1.0.0")

def create_client(auth_type: str = None) -> WorkspaceClient:
    """Factory for creating WorkspaceClient with different auth methods."""
    auth_type = auth_type or os.environ.get("APP_AUTH_TYPE", "oauth_m2m")
    host = os.environ.get("DATABRICKS_HOST", "").strip()
    if not host.startswith("https://"):
        host = f"https://{host}"
    
    if auth_type == "pat":
        return WorkspaceClient(
            host=host,
            token=os.environ.get("DATABRICKS_TOKEN", "")
        )
    
    if auth_type == "oauth_m2m":
        config = Config(
            host=host,
            client_id=os.environ.get("DATABRICKS_CLIENT_ID", ""),
            client_secret=os.environ.get("DATABRICKS_CLIENT_SECRET", ""),
            auth_type="oauth-m2m",
        )
        return WorkspaceClient(config=config)
    
    if auth_type == "u2m_token_env":
        return WorkspaceClient(
            host=host,
            token=os.environ.get("DATABRICKS_ACCESS_TOKEN", "")
        )
    
    raise ValueError(f"Unknown auth_type: {auth_type}")
```

---

## Auth Isolation

Use `env -i` to ensure each authentication method is tested in isolation:

### PAT Test

```bash
env -i PATH=$PATH HOME=$HOME \
  DATABRICKS_HOST=$DATABRICKS_HOST \
  DATABRICKS_TOKEN=$DATABRICKS_TOKEN \
  python your_script.py
```

### OAuth M2M Test

```bash
env -i PATH=$PATH HOME=$HOME \
  DATABRICKS_HOST=$DATABRICKS_HOST \
  DATABRICKS_CLIENT_ID=$DATABRICKS_CLIENT_ID \
  DATABRICKS_CLIENT_SECRET=$DATABRICKS_CLIENT_SECRET \
  python your_script.py
```

### U2M Custom OAuth App Test

```bash
env -i PATH=$PATH HOME=$HOME USER=$USER DISPLAY=$DISPLAY \
  DATABRICKS_HOST=$DATABRICKS_HOST \
  DATABRICKS_U2M_CLIENT_ID=$DATABRICKS_U2M_CLIENT_ID \
  python your_script.py
```

**Note:** U2M requires `DISPLAY`, `HOME`, `USER` for browser launching.

### U2M Token-Env Test

```bash
env -i PATH=$PATH HOME=$HOME \
  DATABRICKS_HOST=$DATABRICKS_HOST \
  DATABRICKS_ACCESS_TOKEN=$DATABRICKS_ACCESS_TOKEN \
  python your_script.py
```

---

## Redirect URI Reference

| Scenario | Default URI | Notes |
|----------|-------------|-------|
| Custom OAuth App (Python) | `http://localhost:8080/callback` | Default for PKCE helper |
| SDK external-browser | `http://localhost:8020` | SDK default |
| Port conflict | `http://localhost:8040/callback` | Use alternative port |

**Registering Custom Redirect URIs:**
1. Databricks Account Console → Settings → App connections
2. Select your OAuth app → Edit
3. Add redirect URI under "Redirect URIs"
4. Set `DATABRICKS_REDIRECT_URI` environment variable to match

---

## Validation Query

Use this query to verify authentication and User-Agent:

```sql
SELECT event_time, user_agent, action_name, request_params
FROM system.access.audit
WHERE event_time > current_timestamp() - INTERVAL 1 HOUR
  AND lower(user_agent) LIKE '%yourcompany%'
ORDER BY event_time DESC
LIMIT 10;
```

### SDK Validation

```python
def validate_connection(client: WorkspaceClient) -> bool:
    """Validate SDK connection by calling current_user API."""
    try:
        me = client.current_user.me()
        print(f"Connected as: {me.user_name}")
        return True
    except Exception as e:
        print(f"Validation failed: {e}")
        return False
```

---

## U2M Token Refresh Pattern

```python
import time
import threading

class TokenRefreshManager:
    """Manages automatic U2M token refresh."""
    
    def __init__(self, authenticator, refresh_interval_sec: int = 45 * 60):
        self.authenticator = authenticator
        self.refresh_interval = refresh_interval_sec
        self.current_token = None
        self._stop = False
        self._thread = None
    
    def start(self):
        self._refresh()
        self._thread = threading.Thread(target=self._refresh_loop, daemon=True)
        self._thread.start()
    
    def _refresh_loop(self):
        while not self._stop:
            time.sleep(self.refresh_interval)
            if not self._stop:
                self._refresh()
    
    def _refresh(self):
        try:
            self.current_token = self.authenticator()
            print("Token refreshed successfully")
        except Exception as e:
            print(f"Token refresh failed: {e}")
    
    def get_token(self) -> str:
        return self.current_token
    
    def stop(self):
        self._stop = True
```

---

## Troubleshooting

| Error | Cause | Solution |
|-------|-------|----------|
| `more than one authorization method configured` | Both PAT and M2M env vars set | Use `env -i` for clean environment |
| `cannot configure default credentials` | M2M Config without auth_type | Pass `auth_type="oauth-m2m"` to Config |
| `OAuth application with client_id not available` | Using M2M client_id for U2M | Use `DATABRICKS_U2M_CLIENT_ID` for U2M |
| `redirect_uri not registered` | URI mismatch in App connections | Verify exact match including port and path |
| Connection timeout | Network issues or firewall | Check connectivity to workspace |
| `401 Unauthorized` | Invalid/expired credentials | Regenerate token or re-authenticate |

---

## Host Normalization

```python
def normalize_host(host: str) -> str:
    """Ensure host has https:// prefix."""
    host = host.strip()
    if not host.startswith("https://") and not host.startswith("http://"):
        host = f"https://{host}"
    return host.rstrip("/")
```

---

## Implementation Learnings

1. **Config import matters**: Use `from databricks.sdk.config import Config` (not `from databricks.sdk import Config`)
2. **auth_type is critical for M2M**: Always pass `auth_type="oauth-m2m"` when using client_id/client_secret
3. **UserAgent is global**: Call `useragent.with_partner()` and `useragent.with_product()` once before creating any client
4. **M2M auto-refreshes**: SDK handles token lifecycle automatically for M2M
5. **U2M requires manual refresh**: Implement refresh timer for long-running U2M applications
6. **Host normalization**: SDK accepts both `https://host` and `host`, but be explicit

### PKCE Flow Implementation Details

When implementing a custom OAuth U2M PKCE flow, follow these patterns for reliable browser callback handling:

**Use Event for callback signaling:**
```python
from threading import Thread, Event

callback_received = Event()

def serve_until_callback():
    while not callback_received.is_set():
        server.handle_request()

thread = Thread(target=serve_until_callback, daemon=True)
thread.start()
```

**Handle multiple requests (ignore favicon):**
```python
class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        req_path = urlparse(self.path).path
        if req_path == callback_path:  # Only process OAuth callback
            # Extract code and signal completion
            callback_received.set()
        else:
            self.send_response(404)  # Ignore favicon, etc.
            self.end_headers()
```

**Use output flushing for immediate feedback:**
```python
print("Waiting for browser auth...", flush=True)
```

**Set server timeout for responsiveness:**
```python
server = HTTPServer((host, port), Handler)
server.timeout = 1  # 1 second timeout per handle_request call
```

**Wait with timeout:**
```python
CALLBACK_TIMEOUT = 120  # seconds
if not callback_received.wait(timeout=CALLBACK_TIMEOUT):
    raise RuntimeError(f"Timed out after {CALLBACK_TIMEOUT}s")
```

**Key insight:** Using `server.handle_request()` in a loop (instead of just once) ensures that browser prefetch requests (like favicon) don't consume the single handler and block the OAuth callback.

---

## SDK Workarounds Summary

| Issue | Workaround |
|-------|------------|
| Default credential resolution error | Pass `auth_type="oauth-m2m"` explicitly |
| Multiple auth methods conflict | Use `env -i` for clean environment per auth type |
| U2M token expiry | Implement manual refresh timer (tokens ~1hr TTL) |
| Config import error | Use `from databricks.sdk.config import Config` |

---

## Key Differences from SQL Connectors

| Aspect | Python SDK | databricks-sql-connector |
|--------|------------|--------------------------|
| **Use case** | REST APIs (UC, Jobs, Clusters) | SQL queries via warehouse |
| **SQL Warehouse** | Not required | Required |
| **Import** | `from databricks.sdk import WorkspaceClient` | `from databricks import sql` |
| **User-Agent** | `useragent.with_partner()` | Connection parameter |
| **M2M auth** | `Config(auth_type="oauth-m2m")` | `OAuthM2MAuthProvider` |