A curated collection of Agent Skills for working with dbt, to help AI agents understand and execute dbt workflows more effectively.
91
Does it follow best practices?
Validation for skill structure
"""Interactive selection widgets for CLI using Textual."""
from __future__ import annotations
import shutil
import sys
from dataclasses import dataclass
from pathlib import Path
import yaml
from textual.app import App, ComposeResult
from textual.widgets import Footer, Header, Input, OptionList
from textual.widgets.option_list import Option
@dataclass
class RunInfo:
"""Information about a run for display in selector.
Display format: '2026-01-30-120000 | 3 scenarios, 6 sets | graded'
With details: 'scenario1: set-a, set-b | scenario2: set-c'
"""
path: Path
name: str
scenario_count: int
skill_set_count: int
scenarios_detail: dict[str, list[str]] # scenario_name -> [skill_set_names]
graded: bool
@classmethod
def from_path(cls, path: Path) -> RunInfo:
"""Create RunInfo from a run directory path."""
name = path.name
scenarios_detail: dict[str, list[str]] = {}
for scenario_dir in path.iterdir():
if not scenario_dir.is_dir() or scenario_dir.name.startswith("."):
continue
skill_sets = [
d.name
for d in scenario_dir.iterdir()
if d.is_dir() and not d.name.startswith(".")
]
if skill_sets:
scenarios_detail[scenario_dir.name] = sorted(skill_sets)
scenario_count = len(scenarios_detail)
skill_set_count = sum(len(sets) for sets in scenarios_detail.values())
grades_file = path / "grades.yaml"
graded = grades_file.exists()
return cls(
path=path,
name=name,
scenario_count=scenario_count,
skill_set_count=skill_set_count,
scenarios_detail=scenarios_detail,
graded=graded,
)
def display_text(self, max_width: int = 120) -> str:
"""Format for display in selector.
Args:
max_width: Maximum width for the display line
"""
# Build header: name | counts | graded
header_parts = [self.name, f"{self.scenario_count} scenarios, {self.skill_set_count} sets"]
if self.graded:
header_parts.append("graded")
header = " | ".join(header_parts)
# Build scenario details
details_parts = []
for scenario_name in sorted(self.scenarios_detail.keys()):
sets = self.scenarios_detail[scenario_name]
details_parts.append(f"{scenario_name}: {', '.join(sets)}")
details = " | ".join(details_parts)
full_text = f"{header} | {details}" if details else header
# Truncate if needed
if len(full_text) > max_width:
return full_text[: max_width - 3] + "..."
return full_text
@dataclass
class ScenarioInfo:
"""Information about a scenario for display in selector.
Display format: 'my-scenario | 2 skill sets | Description...'
"""
path: Path
name: str
description: str
skill_set_count: int
@classmethod
def from_path(cls, path: Path) -> ScenarioInfo:
"""Create ScenarioInfo from a scenario directory path."""
name = path.name
# Count skill sets
skill_sets_file = path / "skill-sets.yaml"
skill_set_count = 0
if skill_sets_file.exists():
try:
with skill_sets_file.open() as f:
data = yaml.safe_load(f)
skill_set_count = len(data.get("sets", []))
except yaml.YAMLError:
pass
# Get description from scenario.md
description = ""
scenario_md = path / "scenario.md"
if scenario_md.exists():
try:
content = scenario_md.read_text()
# Use first non-empty line as description
for line in content.split("\n"):
line = line.strip()
if line and not line.startswith("#"):
description = line[:60] + "..." if len(line) > 60 else line
break
except (OSError, UnicodeDecodeError):
pass
return cls(
path=path,
name=name,
description=description,
skill_set_count=skill_set_count,
)
def display_text(self) -> str:
"""Format for display in selector."""
parts = [self.name, f"{self.skill_set_count} skill set(s)"]
if self.description:
parts.append(self.description)
return " | ".join(parts)
def is_interactive() -> bool:
"""Check if running in interactive mode (TTY)."""
return sys.stdin.isatty()
class RunSelectorApp(App[Path | None]):
"""Single-selection app for choosing a run."""
BINDINGS = [
("escape", "quit", "Cancel"),
("enter", "select", "Select"),
("/", "focus_search", "Search"),
]
def __init__(self, runs: list[RunInfo], title: str = "Select a run") -> None:
super().__init__()
self.runs = runs
self.title_text = title
self._selected: Path | None = None
self._term_width = shutil.get_terminal_size().columns
def compose(self) -> ComposeResult:
yield Header()
yield Input(placeholder="Type to filter...", id="search")
yield OptionList(
*[Option(run.display_text(self._term_width), id=run.name) for run in self.runs],
id="options",
)
yield Footer()
def on_mount(self) -> None:
self.theme = "tokyo-night"
self.title = "skill-eval"
self.sub_title = self.title_text
# Focus the option list by default
self.query_one("#options", OptionList).focus()
def on_input_changed(self, event: Input.Changed) -> None:
"""Filter options as user types."""
query = event.value.lower()
option_list = self.query_one("#options", OptionList)
option_list.clear_options()
for run in self.runs:
if query in run.display_text(self._term_width).lower():
option_list.add_option(Option(run.display_text(self._term_width), id=run.name))
def on_input_submitted(self, event: Input.Submitted) -> None:
"""When Enter is pressed in search, focus the list."""
self.query_one("#options", OptionList).focus()
def on_option_list_option_selected(self, event: OptionList.OptionSelected) -> None:
"""Handle selection via Enter or double-click."""
selected_name = str(event.option.id)
for run in self.runs:
if run.name == selected_name:
self._selected = run.path
break
self.exit(self._selected)
async def action_quit(self) -> None:
self.exit(None)
def action_focus_search(self) -> None:
self.query_one("#search", Input).focus()
def action_select(self) -> None:
option_list = self.query_one("#options", OptionList)
if option_list.highlighted is not None:
option = option_list.get_option_at_index(option_list.highlighted)
selected_name = str(option.id)
for run in self.runs:
if run.name == selected_name:
self._selected = run.path
break
self.exit(self._selected)
class ScenarioSelectorApp(App[list[Path]]):
"""Multi-selection app for choosing scenarios using OptionList with manual tracking."""
BINDINGS = [
("escape", "quit", "Cancel"),
("enter", "confirm", "Confirm selection"),
("space", "toggle_selection", "Toggle selection"),
("a", "select_all", "Select all"),
("/", "focus_search", "Search"),
]
def __init__(
self, scenarios: list[ScenarioInfo], title: str = "Select scenarios"
) -> None:
super().__init__()
self.scenarios = scenarios
self.title_text = title
self._selected_names: set[str] = set()
self._filtered_scenarios: list[ScenarioInfo] = list(scenarios)
def compose(self) -> ComposeResult:
yield Header()
yield Input(placeholder="Type to filter... (Space to toggle, Enter to confirm)", id="search")
yield OptionList(*self._build_options(), id="options")
yield Footer()
def _build_options(self) -> list[Option]:
"""Build option list with checkmarks for selected items."""
options = []
for scenario in self._filtered_scenarios:
prefix = "[green]✓[/] " if scenario.name in self._selected_names else " "
options.append(Option(f"{prefix}{scenario.display_text()}", id=scenario.name))
return options
def _refresh_options(self) -> None:
"""Refresh the option list to reflect current selections."""
option_list = self.query_one("#options", OptionList)
highlighted = option_list.highlighted
option_list.clear_options()
for opt in self._build_options():
option_list.add_option(opt)
if highlighted is not None and highlighted < len(self._filtered_scenarios):
option_list.highlighted = highlighted
def on_mount(self) -> None:
self.theme = "tokyo-night"
self.title = "skill-eval"
self.sub_title = self.title_text
self.query_one("#options", OptionList).focus()
def on_input_changed(self, event: Input.Changed) -> None:
"""Filter options as user types."""
query = event.value.lower()
self._filtered_scenarios = [
s for s in self.scenarios if query in s.display_text().lower()
]
self._refresh_options()
def on_input_submitted(self, event: Input.Submitted) -> None:
"""When Enter is pressed in search, focus the list."""
self.query_one("#options", OptionList).focus()
def action_toggle_selection(self) -> None:
"""Toggle selection of highlighted item."""
option_list = self.query_one("#options", OptionList)
if option_list.highlighted is not None:
idx = option_list.highlighted
if idx < len(self._filtered_scenarios):
name = self._filtered_scenarios[idx].name
if name in self._selected_names:
self._selected_names.discard(name)
else:
self._selected_names.add(name)
self._refresh_options()
def on_option_list_option_selected(self, event: OptionList.OptionSelected) -> None:
"""Confirm selection on Enter/double-click."""
self.action_confirm()
async def action_quit(self) -> None:
self.exit([])
def action_focus_search(self) -> None:
self.query_one("#search", Input).focus()
def action_confirm(self) -> None:
# If nothing selected, select the highlighted item
if not self._selected_names:
option_list = self.query_one("#options", OptionList)
if option_list.highlighted is not None:
idx = option_list.highlighted
if idx < len(self._filtered_scenarios):
self._selected_names.add(self._filtered_scenarios[idx].name)
selected_paths = [s.path for s in self.scenarios if s.name in self._selected_names]
self.exit(selected_paths)
def action_select_all(self) -> None:
"""Select all currently visible scenarios."""
for scenario in self._filtered_scenarios:
self._selected_names.add(scenario.name)
self._refresh_options()
def select_run(runs: list[Path], title: str = "Select a run") -> Path | None:
"""Show interactive selector for a single run.
Args:
runs: List of run directory paths
title: Title to display in the selector
Returns:
Selected run path, or None if cancelled
"""
if not runs:
return None
# Single run available - skip selector
if len(runs) == 1:
return runs[0]
run_infos = [RunInfo.from_path(run) for run in runs]
# Sort by name descending (most recent first)
run_infos.sort(key=lambda r: r.name, reverse=True)
app = RunSelectorApp(run_infos, title)
return app.run()
def select_scenarios(
scenarios: list[Path], title: str = "Select scenarios"
) -> list[Path]:
"""Show interactive selector for multiple scenarios.
Args:
scenarios: List of scenario directory paths
title: Title to display in the selector
Returns:
List of selected scenario paths (empty if cancelled)
"""
if not scenarios:
return []
scenario_infos = [ScenarioInfo.from_path(s) for s in scenarios]
# Sort alphabetically by name
scenario_infos.sort(key=lambda s: s.name)
app = ScenarioSelectorApp(scenario_infos, title)
result = app.run()
return result if result is not None else []Install with Tessl CLI
npx tessl i dbt-labs/dbt-agent-skills@1.1.0evals
scenarios
dbt-docs-arguments
dbt-docs-unit-test-fixtures
dbt-job-failure
dbt-unit-test-format-choice
example-yaml-error
fusion-migration-triage-basic
fusion-migration-triage-blocked
fusion-triage-cat-a-static-analysis
fusion-triage-cat-b-dict-meta-get
fusion-triage-cat-b-unexpected-config
fusion-triage-cat-b-unused-schema
fusion-triage-cat-b-yaml-syntax
fusion-triage-cat-c-hardcoded-fqn
tests
scripts
skills
dbt
skills
adding-dbt-unit-test
references
answering-natural-language-questions-with-dbt
building-dbt-semantic-layer
configuring-dbt-mcp-server
fetching-dbt-docs
scripts
running-dbt-commands
troubleshooting-dbt-job-errors
references
using-dbt-for-analytics-engineering
dbt-migration
skills
migrating-dbt-core-to-fusion
migrating-dbt-project-across-platforms