DBT Core MCP Server

Instructions

Implementation Reference

• src/dbt_core_mcp/server.py:786-952 (handler)

  The core handler function that implements the run_models tool logic. It handles state-based selection (modified models), builds dbt run commands, executes via BridgeRunner, parses results, detects schema changes, saves state for future runs, and supports progress reporting.
  async def toolImpl_run_models( self, ctx: Context | None, select: str | None = None, exclude: str | None = None, select_state_modified: bool = False, select_state_modified_plus_downstream: bool = False, full_refresh: bool = False, fail_fast: bool = False, check_schema_changes: bool = False, ) -> dict[str, Any]: """Implementation for run_models tool.""" # Prepare state-based selection (validates and returns selector) selector = await self._prepare_state_based_selection(select_state_modified, select_state_modified_plus_downstream, select) # Early return if state-based requested but no state exists if select_state_modified and not selector: return { "status": "success", "message": "No previous state found - cannot determine modifications", "results": [], "elapsed_time": 0, } # Build command args args = ["run"] # Add selector if we have one (state-based or manual) if selector: args.extend(["-s", selector, "--state", "target/state_last_run"]) elif select: args.extend(["-s", select]) if exclude: args.extend(["--exclude", exclude]) if full_refresh: args.append("--full-refresh") if fail_fast: args.append("--fail-fast") # Capture pre-run table columns for schema change detection # Also get expected count of models for progress reporting pre_run_columns: dict[str, list[str]] = {} expected_total: int | None = None if check_schema_changes or True: # Always get count for progress # Use dbt list to get models that will be run (without actually running them) list_args = ["list", "--resource-type", "model", "--output", "name"] if select_state_modified: selector = "state:modified+" if select_state_modified_plus_downstream else "state:modified" list_args.extend(["-s", selector, "--state", "target/state_last_run"]) elif select: list_args.extend(["-s", select]) if exclude: list_args.extend(["--exclude", exclude]) # Get list of models logger.info(f"Getting model list: {list_args}") list_result = await self.runner.invoke(list_args) # type: ignore if list_result.success and list_result.stdout: model_count = 0 # Parse model names from output (one per line with --output name) for line in list_result.stdout.strip().split("\n"): line = line.strip() # Skip log lines, timestamps, empty lines, and JSON output if ( not line or line.startswith("{") or ":" in line[:10] # Timestamp like "07:39:44" or "Running with dbt=" in line or "Registered adapter:" in line ): continue model_count += 1 # For schema change detection, query pre-run columns if check_schema_changes: model_name = line logger.info(f"Querying pre-run columns for {model_name}") cols = await self._get_table_columns_from_db(model_name) if cols: pre_run_columns[model_name] = cols else: # Table doesn't exist yet - mark as new pre_run_columns[model_name] = [] # Set expected total from model count if model_count > 0: expected_total = model_count logger.info(f"Expected total models to run: {expected_total}") # Execute with progress reporting logger.info(f"Running dbt models with args: {args}") logger.info(f"Expected total for progress: {expected_total}") # Define progress callback if context available async def progress_callback(current: int, total: int, message: str) -> None: if ctx: await ctx.report_progress(progress=current, total=total, message=message) result = await self.runner.invoke(args, progress_callback=progress_callback if ctx else None, expected_total=expected_total) # type: ignore if not result.success: error_msg = str(result.exception) if result.exception else "Run failed" response = { "status": "error", "message": error_msg, "command": " ".join(args), } # Include dbt output for debugging if result.stdout: response["dbt_output"] = result.stdout if result.stderr: response["stderr"] = result.stderr return response # Parse run_results.json for details run_results = self._parse_run_results() # Check for schema changes if requested schema_changes: dict[str, dict[str, list[str]]] = {} if check_schema_changes and pre_run_columns: logger.info("Detecting schema changes by comparing pre/post-run database columns") for model_name, old_columns in pre_run_columns.items(): # Query post-run columns from database new_columns = await self._get_table_columns_from_db(model_name) if not new_columns: # Model failed to build or was skipped continue # Compare columns added = [c for c in new_columns if c not in old_columns] removed = [c for c in old_columns if c not in new_columns] if old_columns else [] if added or removed: schema_changes[model_name] = {} if added: schema_changes[model_name]["added"] = added if removed: schema_changes[model_name]["removed"] = removed # Save state on success for next modified run if result.success and self.project_dir: state_dir = self.project_dir / "target" / "state_last_run" state_dir.mkdir(parents=True, exist_ok=True) manifest_path = self.runner.get_manifest_path() # type: ignore shutil.copy(manifest_path, state_dir / "manifest.json") response: dict[str, Any] = { "status": "success", "command": " ".join(args), "results": run_results.get("results", []), "elapsed_time": run_results.get("elapsed_time"), } if schema_changes: response["schema_changes"] = schema_changes response["recommendation"] = "Schema changes detected. Consider running downstream models with modified_downstream=True to propagate changes." return response
• src/dbt_core_mcp/server.py:1527-1569 (registration)

  Registers the 'run_models' tool with FastMCP using @app.tool() decorator. Defines the tool's input parameters, comprehensive docstring describing usage, and delegates to toolImpl_run_models.
  async def run_models( ctx: Context, select: str | None = None, exclude: str | None = None, select_state_modified: bool = False, select_state_modified_plus_downstream: bool = False, full_refresh: bool = False, fail_fast: bool = False, check_schema_changes: bool = False, ) -> dict[str, Any]: """Run dbt models (compile SQL and execute against database). State-based selection modes (uses dbt state:modified selector): - select_state_modified: Run only models modified since last successful run (state:modified) - select_state_modified_plus_downstream: Run modified + downstream dependencies (state:modified+) Note: Requires select_state_modified=True Manual selection (alternative to state-based): - select: dbt selector syntax (e.g., "customers", "tag:mart", "stg_*") - exclude: Exclude specific models Args: select: Manual selector (e.g., "customers", "tag:mart", "path:marts/*") exclude: Exclude selector (e.g., "tag:deprecated") select_state_modified: Use state:modified selector (changed models only) select_state_modified_plus_downstream: Extend to state:modified+ (changed + downstream) full_refresh: Force full refresh of incremental models fail_fast: Stop execution on first failure check_schema_changes: Detect schema changes and recommend downstream runs Returns: Execution results with status, models run, timing info, and optional schema_changes Examples: - run_models(select="customers") - Run specific model - run_models(select_state_modified=True) - Run only what changed - run_models(select_state_modified=True, select_state_modified_plus_downstream=True) - Run changed + downstream - run_models(select="tag:mart", full_refresh=True) - Full refresh marts - run_models(select_state_modified=True, check_schema_changes=True) - Detect schema changes """ await self._ensure_initialized_with_context(ctx) return await self.toolImpl_run_models(ctx, select, exclude, select_state_modified, select_state_modified_plus_downstream, full_refresh, fail_fast, check_schema_changes)
• src/dbt_core_mcp/server.py:479-520 (helper)

  Helper function that validates parameters and prepares dbt state-based selectors (state:modified or state:modified+) for efficient incremental runs.
  async def _prepare_state_based_selection( self, select_state_modified: bool, select_state_modified_plus_downstream: bool, select: str | None, ) -> str | None: """Validate and prepare state-based selection. Args: select_state_modified: Use state:modified selector select_state_modified_plus_downstream: Extend to state:modified+ select: Manual selector (conflicts with state-based) Returns: The dbt selector string to use ("state:modified" or "state:modified+"), or None if: - Not using state-based selection - No previous state exists (cannot determine modifications) Raises: ValueError: If validation fails """ # Validate: hierarchical requirement if select_state_modified_plus_downstream and not select_state_modified: raise ValueError("select_state_modified_plus_downstream requires select_state_modified=True") # Validate: can't use both state-based and manual selection if select_state_modified and select: raise ValueError("Cannot use both select_state_modified* flags and select parameter") # If not using state-based selection, return None if not select_state_modified: return None # Check if state exists state_dir = self.project_dir / "target" / "state_last_run" # type: ignore if not state_dir.exists(): # No state - cannot determine modifications return None # Return selector (state exists) return "state:modified+" if select_state_modified_plus_downstream else "state:modified"
• src/dbt_core_mcp/server.py:304-341 (helper)

  Helper that parses dbt's run_results.json into a simplified dictionary format for tool responses.
  def _parse_run_results(self) -> dict[str, Any]: """Parse target/run_results.json after dbt run/test/build. Returns: Dictionary with results array and metadata """ if not self.project_dir: return {"results": [], "elapsed_time": 0} run_results_path = self.project_dir / "target" / "run_results.json" if not run_results_path.exists(): return {"results": [], "elapsed_time": 0} try: with open(run_results_path) as f: data = json.load(f) # Simplify results for output simplified_results = [] for result in data.get("results", []): simplified_results.append( { "unique_id": result.get("unique_id"), "status": result.get("status"), "message": result.get("message"), "execution_time": result.get("execution_time"), "failures": result.get("failures"), } ) return { "results": simplified_results, "elapsed_time": data.get("elapsed_time", 0), } except Exception as e: logger.warning(f"Failed to parse run_results.json: {e}") return {"results": [], "elapsed_time": 0}