Hooks

Where it lives

• hooks/databricks-auth-helper.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """PostToolUse hook: suggest an auth fix when a `databricks` command fails auth.
  
  Watches shell tool results. When the command actually invoked the `databricks`
  CLI (as a segment executable, not merely "databricks" appearing in a repo
  path, URL, or argument) and the output looks like an authentication failure
  (missing credentials, expired or invalid token, OAuth refresh failure), it
  injects one line of additional context pointing at the doctor command and
  `databricks auth login`. Everything else passes through silently.
  
  No gating: this never blocks or rewrites a tool call, it only adds context
  after the fact.
  
  Contract (Claude Code PostToolUse hook, matcher: Bash; Cursor postToolUse,
  matcher: Shell, with `--platform cursor`):
    stdin : JSON with tool_name + the command and result. Claude/Codex carry
            tool_input.command and tool_response; Cursor carries tool_input /
            tool_output, either of which may itself be a JSON-encoded string.
    stdout: platform-shaped JSON carrying the hint, or "{}".
            Claude/Codex -> hookSpecificOutput.additionalContext
            Cursor       -> additional_context
    Fail-open: on ANY error print "{}" and exit 0.
  """
  import json
  import re
  import sys
  
  # `databricks` must be the executable of one of the command's shell segments,
  # not a substring anywhere in the command line: `gh pr view --repo
  # databricks/cli`, URLs, and file paths mention databricks without invoking
  # the CLI, and their output can legitimately quote auth-failure phrases (a PR
  # body, this hook's own source). Segments are split on shell connectors and
  # command-substitution openers; each segment's executable is its first token
  # after env assignments, common wrappers, and wrapper flags. Path-prefixed
  # invocations (`/usr/local/bin/databricks`) count; `databricks-test` does not.
  _SEGMENT_SPLIT_RE = re.compile(r"&&|\|\||\$\(|[;|&\n`(]")
  _ENV_ASSIGNMENT_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*=")
  _WRAPPERS = frozenset({"sudo", "env", "command", "exec", "time", "nohup", "xargs"})
  
  # Shell-running tool names across the harnesses that consume this hook:
  # Claude Code uses Bash, Cursor uses Shell, Codex documents Bash (with
  # shell/unified_exec variants in the wild), VS Code Copilot uses
  # run_in_terminal. The real filter is _invokes_databricks_cli below; this
  # gate only keeps non-shell tools (file reads, edits) from being scanned.
  _SHELL_TOOL_RE = re.compile(r"(?i)^(bash|shell|local_shell|unified_exec|run_in_terminal)$")
  
  # Per-platform names for the setup/doctor commands referenced in the hint
  # (Claude Code namespaces plugin commands as /databricks:<name>; Cursor has a
  # flat / menu, so the Cursor plugin ships them as /databricks-<name>).
  PLATFORMS = {
      "claude": {"setup_cmd": "/databricks:setup", "doctor_cmd": "/databricks:doctor"},
      "cursor": {"setup_cmd": "/databricks-setup", "doctor_cmd": "/databricks-doctor"},
  }
  
  
  def _platform_from_argv(argv):
      """Platform from `--platform <name>` / `--platform=<name>`, default claude.
  
      Unknown values fall back to claude rather than erroring: a wiring typo must
      degrade to a working hook, never a broken one.
      """
      for i, arg in enumerate(argv):
          if arg == "--platform" and i + 1 < len(argv):
              value = argv[i + 1]
          elif arg.startswith("--platform="):
              value = arg.split("=", 1)[1]
          else:
              continue
          return value if value in PLATFORMS else "claude"
      return "claude"
  
  
  def _segment_executable(tokens):
      """First token that is not an env assignment, wrapper, or wrapper flag."""
      after_wrapper = False
      for token in tokens:
          if _ENV_ASSIGNMENT_RE.match(token):
              continue
          if token in _WRAPPERS:
              after_wrapper = True
              continue
          if after_wrapper and token.startswith("-"):
              continue
          return token
      return ""
  
  
  def _invokes_databricks_cli(command):
      """True when any segment of the command runs the `databricks` executable."""
      for segment in _SEGMENT_SPLIT_RE.split(command):
          executable = _segment_executable(segment.split())
          if executable.rsplit("/", 1)[-1] == "databricks":
              return True
      return False
  
  # Phrase-shaped auth-failure signals as emitted by the CLI / Go SDK error
  # paths. Deliberately not bare status codes, so ordinary data in stdout
  # (e.g. a row containing 401) cannot trip them.
  AUTH_ERROR_PATTERNS = [
      r"cannot configure default credentials",
      r"\binvalid_grant\b",
      r"\b401 unauthorized\b",
      r"\binvalid access token\b",
      r"\btoken (?:is |has |was )?expired\b",
      r"\brefresh token (?:is |was )?(?:invalid|expired|revoked)\b",
  ]
  _AUTH_ERRORS = [re.compile(p, re.IGNORECASE) for p in AUTH_ERROR_PATTERNS]
  
  AUTH_HINT_TEMPLATE = (
      "[DATABRICKS] The `databricks` command above failed with what looks like "
      "an authentication error. Before retrying, fix auth: run "
      "`{doctor_cmd}` for a read-only diagnosis, or re-authenticate with "
      "`databricks auth login --host <workspace-url> --profile <name>` "
      "(`{setup_cmd}` walks through it). Never auto-select a profile for "
      "the user."
  )
  
  
  def _parse_maybe_json(value):
      """Return a dict from `value`, decoding it first if it's a JSON string."""
      if isinstance(value, str):
          try:
              value = json.loads(value)
          except Exception:
              return None
      return value if isinstance(value, dict) else None
  
  
  def extract_payload(data):
      """(tool_name, command, response_text) from a Claude/Codex or Cursor payload."""
      tool_name = str(data.get("tool_name", ""))
      tool_input = _parse_maybe_json(data.get("tool_input"))
      command = tool_input.get("command", "") if tool_input else ""
      if not isinstance(command, str):
          command = ""
      # Claude/Codex put the result in tool_response; Cursor in tool_output
      # (often as an already-JSON-encoded string). Serialize non-strings instead
      # of assuming their shape;

• hooks/databricks-context.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """SessionStart hook: inject a compact Databricks context banner.
  
  Local-only and fail-open by design. It never makes a network call (so it can't
  hang, hit an MCP-style timeout, or trigger an auth prompt at session start) and
  any error exits 0 with no output. It surfaces, when available:
  
    - databricks CLI presence + version
    - configured profile names, parsed straight from the config file (no network)
    - the `[__settings__].default_profile` the CLI resolves when --profile is omitted
    - env-based / in-platform auth (DATABRICKS_HOST, DATABRICKS_CONFIG_PROFILE)
  
  Token values are never printed, only their presence.
  
  Contract (Claude Code SessionStart hook; Cursor sessionStart with
  `--platform cursor`):
    stdin : JSON (drained, content unused)
    stdout: platform-shaped JSON carrying the banner string:
            Claude/Codex -> hookSpecificOutput.additionalContext
            Cursor       -> additional_context
  """
  import json
  import os
  import re
  import shutil
  import subprocess
  import sys
  from pathlib import Path
  
  VERSION_RE = re.compile(r"(\d+)\.(\d+)\.(\d+)")
  SECTION_RE = re.compile(r"^\[([^\]]+)\]", re.MULTILINE)
  # default_profile inside the [__settings__] section ([^\[]*? keeps the search
  # from crossing into the next section header).
  SETTINGS_DEFAULT_RE = re.compile(
      r"^\[__settings__\][^\[]*?^[ \t]*default_profile[ \t]*=[ \t]*(\S+)",
      re.MULTILINE,
  )
  MAX_PROFILES = 12
  
  # Per-platform differences: how the setup/doctor commands are invoked there
  # (Claude Code namespaces plugin commands as /databricks:<name>; Cursor has a
  # flat / menu, so the Cursor plugin ships them as /databricks-<name>) and the
  # hook output envelope. Codex consumes the Claude envelope.
  PLATFORMS = {
      "claude": {
          "setup_cmd": "/databricks:setup",
          "commands_blurb": "`/databricks:*` commands",
      },
      "cursor": {
          "setup_cmd": "/databricks-setup",
          "commands_blurb": "`/databricks-*` commands",
      },
  }
  
  
  def _platform_from_argv(argv):
      """Platform from `--platform <name>` / `--platform=<name>`, default claude.
  
      Unknown values fall back to claude rather than erroring: a wiring typo must
      degrade to a working hook, never a broken one.
      """
      for i, arg in enumerate(argv):
          if arg == "--platform" and i + 1 < len(argv):
              value = argv[i + 1]
          elif arg.startswith("--platform="):
              value = arg.split("=", 1)[1]
          else:
              continue
          return value if value in PLATFORMS else "claude"
      return "claude"
  
  
  def cli_version(databricks):
      """(major, minor, patch) from `databricks --version`, or None. 3s timeout."""
      try:
          out = subprocess.run(
              [databricks, "--version"],
              capture_output=True, text=True, timeout=3,
          )
      except Exception:
          return None
      m = VERSION_RE.search((out.stdout or "") + (out.stderr or ""))
      return tuple(int(x) for x in m.groups()) if m else None
  
  
  def config_profiles():
      """(config_path, [profile names], default_profile) read locally from the config file.
  
      Parsed directly rather than via `databricks auth profiles` on purpose: this
      runs at SessionStart, which must stay offline and fast (no network, no
      auth-validation round-trips). Skips CLI-internal sections like
      `[__settings__]`, which are not auth profiles, but does surface
      `[__settings__].default_profile` since the CLI resolves it when --profile
      is omitted.
      """
      cfg = os.environ.get("DATABRICKS_CONFIG_FILE") or str(Path.home() / ".databrickscfg")
      try:
          p = Path(cfg)
          # Only read a regular file under a sane size cap, so a FIFO/device or a
          # huge file pointed at by DATABRICKS_CONFIG_FILE can never hang or do
          # unbounded work at session start.
          if not p.is_file() or p.stat().st_size > 1_000_000:
              return cfg, [], None
          text = p.read_text(errors="replace")
      except Exception:
          return cfg, [], None
      names = [
          n for n in SECTION_RE.findall(text)
          if not (n.startswith("__") and n.endswith("__"))
      ]
      m = SETTINGS_DEFAULT_RE.search(text)
      return cfg, names, (m.group(1) if m else None)
  
  
  def _sanitize(value, limit=64):
      """Make a config-derived string safe to inject as one context list item.
  
      Strips control chars / newlines (so a crafted profile name or env value
      cannot inject extra bullets or instructions) and caps the length.
      """
      s = re.sub(r"[\x00-\x1f\x7f]", " ", str(value))
      s = re.sub(r"\s+", " ", s).strip()
      return s[: limit - 1].rstrip() + "…" if len(s) > limit else s
  
  
  def build_context(platform="claude"):
      """Return the context banner string, or '' to inject nothing."""
      p = PLATFORMS.get(platform, PLATFORMS["claude"])
      databricks = shutil.which("databricks")
      if not databricks:
          return (
              "Databricks CLI (`databricks`) is not on PATH. The Databricks skills "
              f"and {p['commands_blurb']} need it. Run `{p['setup_cmd']}` or see "
              "the databricks-core skill to install it."
          )
  
      lines = []
      ver = cli_version(databricks)
      if ver:
          lines.append(f"CLI v{'.'.join(map(str, ver))}.")
      else:
          lines.append("CLI present (version unknown).")
  
      cfg, profiles, default_profile = config_profiles()
      if profiles:
          shown = [_sanitize(n) for n in profiles[:MAX_PROFILES]]
          more = f" (+{len(profiles) - len(shown)} more)" if len(profiles) > len(shown) else ""
          lines.append(f"Profiles in {_sanitize(Path(cfg).name)}: {', '.join(shown)}{more}.")
          if default_profile:
              lines.append(
                  f"Default profile (from [__settings__]): `{_sanitize(default_profile)}`; "
                  "the CLI uses it when `--profile` is omitted."
              )
          lines.append("Never auto-select a profile. Pass `--profile <name>` and let the user choose.")
      else:
          # Basename only, matching the branch above: the full pa

• hooks/databricks-router.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """UserPromptSubmit hook: route Databricks-related prompts into the skills.
  
  Reads the user prompt from stdin, runs a fast keyword regex (sub-50ms, no LLM,
  no network), and if the prompt is Databricks-related, injects an
  `additionalContext` instruction telling Claude to load the `databricks-core`
  skill (the parent/router) plus the matching product skill before answering.
  
  There is no second agent to delegate to: Claude itself drives the `databricks`
  CLI through the skills, so "routing" just means "make sure the Databricks skills
  are loaded." No permission gating, no cost warnings.
  
  The keyword lists and the injected instruction are generated from
  plugin.meta.json into `_routing_data.json` next to this file and loaded at
  import time; a minimal inline fallback keeps the hook working if that file is
  missing or unreadable. Regenerate with `python3 scripts/skills.py generate`.
  
  The full routing instruction is injected once per session (keyed by the
  payload's session_id via a marker file in the temp dir); later Databricks
  prompts in the same session get a one-line reminder instead, keeping repeat
  token cost low.
  
  Contract (Claude Code UserPromptSubmit hook):
    stdin : JSON, e.g. {"prompt": "...", "session_id": "..."} or {"message": "..."}
    stdout: JSON -> hookSpecificOutput.additionalContext (injected before the turn),
            or "{}" to stay out of the way.
    Fail-open: on ANY error print "{}" and exit 0, so a broken hook never blocks a
    prompt.
  """
  import json
  import re
  import sys
  import tempfile
  from pathlib import Path
  
  # Routing config (keyword lists + the injected instruction) is generated from
  # plugin.meta.json into _routing_data.json next to this file (see
  # rules/README.md and CONTRIBUTING.md) and loaded at import time. If that file
  # is missing or unreadable the hook stays fail-open via a minimal inline
  # fallback that still routes obvious Databricks prompts.
  _FALLBACK_STRONG = [r"\bdatabricks\b"]
  _FALLBACK_AMBIGUOUS = []
  _FALLBACK_SUPPRESS = []
  _FALLBACK_INSTRUCTION = (
      "[DATABRICKS] This request is Databricks-related. Handle it through the "
      "Databricks skills rather than ad hoc commands: load `databricks-core` (the "
      "parent skill) plus the matching product skill before answering."
  )
  _FALLBACK_REMINDER = (
      "[DATABRICKS] Databricks-related prompt: keep routing through the Databricks "
      "skills (databricks-core plus the matching product skill)."
  )
  
  
  def _load_routing_data(path=None):
      """Load the generated _routing_data.json next to this file; None on any problem.
  
      Returns None (so the inline fallback engages) for a missing, unreadable,
      malformed, wrong-typed, or regex-invalid file. The patterns are compiled
      here, so a corrupt data file degrades the router to "route obvious
      databricks mentions" rather than raising at import (which would otherwise
      take the whole hook down, fallback included).
      """
      try:
          if path is None:
              path = Path(__file__).resolve().parent / "_routing_data.json"
          data = json.loads(Path(path).read_text())
      except Exception:
          return None
      if not isinstance(data, dict):
          return None
      if not all(
          k in data for k in ("strong", "ambiguous", "suppress", "instruction", "reminder")
      ):
          return None
      if not all(isinstance(data[k], list) for k in ("strong", "ambiguous", "suppress")):
          return None
      if not all(isinstance(data[k], str) for k in ("instruction", "reminder")):
          return None
      try:
          for key in ("strong", "ambiguous", "suppress"):
              for pattern in data[key]:
                  re.compile(pattern)
      except (re.error, TypeError):
          return None
      return data
  
  
  _DATA = _load_routing_data()
  
  # STRONG: unambiguously Databricks -> always route, even alongside a mention of
  # an alternative platform (e.g. "migrate from redshift to databricks").
  # AMBIGUOUS: Databricks-likely but also used elsewhere -> route only when no
  # SUPPRESS signal (alternative platform / local dev) is present. STRONG matches
  # ignore SUPPRESS. All are sourced from the generated data file; the inline
  # fallback degrades gracefully (routes only clear "databricks" mentions).
  STRONG = list(_DATA["strong"]) if _DATA else list(_FALLBACK_STRONG)
  AMBIGUOUS = list(_DATA["ambiguous"]) if _DATA else list(_FALLBACK_AMBIGUOUS)
  SUPPRESS = list(_DATA["suppress"]) if _DATA else list(_FALLBACK_SUPPRESS)
  
  # The full instruction injected on the session's first Databricks prompt, plus
  # the one-line reminder for later prompts. ROUTING_INSTRUCTION stays a
  # module-level attribute (scripts/skills.py check_routing_tables reads it).
  ROUTING_INSTRUCTION = _DATA["instruction"] if _DATA else _FALLBACK_INSTRUCTION
  ROUTING_REMINDER = _DATA["reminder"] if _DATA else _FALLBACK_REMINDER
  
  _STRONG = [re.compile(p, re.IGNORECASE) for p in STRONG]
  _AMBIGUOUS = [re.compile(p, re.IGNORECASE) for p in AMBIGUOUS]
  _SUPPRESS = [re.compile(p, re.IGNORECASE) for p in SUPPRESS]
  
  # "databricks" inside a code-hosting URL (github.com/databricks/...) is an
  # org/repo name, not product intent, so URLs are blanked before matching unless
  # the hostname itself contains "databricks" (workspace and docs hosts), which
  # keeps "why is https://myco.cloud.databricks.com/jobs/123 failing?" routing.
  _URL_RE = re.compile(
      r"(?:https?://|git@)(?P<host>[\w.-]+)[/:]?\S*"
      r"|\b(?:www\.)?(?P<bare>(?:github|gitlab|bitbucket)\.(?:com|org))[/:]\S*",
      re.IGNORECASE,
  )
  
  
  def _strip_non_databricks_urls(text):
      def _keep_or_blank(match):
          host = match.group("host") or match.group("bare") or ""
          return match.group(0) if "databricks" in host.lower() else " "
  
      return _URL_RE.sub(_keep_or_blank, text)
  
  
  _SESSION_ID_SAFE_RE = re.compile(r"[^A-Za-z0-9._-]")
  
  
  def _marker_path(session_id):
      """Temp-dir marker recording that this session already got the full instruction."""
      sid = _SESSION_ID_SAFE_RE.sub("", str(session_id or ""))[:64]
      if not sid:
          

Read the script before you install anything that runs on your machine. This is the one part of a plugin that acts without being asked.