dbt-mcp

Instructions

Implementation Reference

• src/dbt_mcp/dbt_cli/tools.py:129-146 (handler)

  The handler function for the 'run' MCP tool. Executes 'dbt run' with optional selector, full_refresh, and vars.
  def run( selector: str | None = Field( default=None, description=get_prompt("dbt_cli/args/selectors") ), is_full_refresh: bool | None = Field( default=None, description=get_prompt("dbt_cli/args/full_refresh") ), vars: str | None = Field( default=None, description=get_prompt("dbt_cli/args/vars") ), ) -> str: return _run_dbt_command( ["run"], selector, is_selectable=True, is_full_refresh=is_full_refresh, vars=vars, )
• src/dbt_mcp/dbt_cli/tools.py:19-84 (helper)

  Core helper function that runs arbitrary dbt CLI commands using subprocess, applying flags like --quiet for verbose commands including 'run', selectors, vars, full-refresh, timeouts, etc.
  def _run_dbt_command( command: list[str], selector: str | None = None, resource_type: list[str] | None = None, is_selectable: bool = False, is_full_refresh: bool | None = False, vars: str | None = None, ) -> str: try: # Commands that should always be quiet to reduce output verbosity verbose_commands = [ "build", "compile", "docs", "parse", "run", "test", "list", ] if is_full_refresh is True: command.append("--full-refresh") if vars and isinstance(vars, str): command.extend(["--vars", vars]) if selector: selector_params = str(selector).split(" ") command.extend(["--select"] + selector_params) if isinstance(resource_type, Iterable): command.extend(["--resource-type"] + resource_type) full_command = command.copy() # Add --quiet flag to specific commands to reduce context window usage if len(full_command) > 0 and full_command[0] in verbose_commands: main_command = full_command[0] command_args = full_command[1:] if len(full_command) > 1 else [] full_command = [main_command, "--quiet", *command_args] # We change the path only if this is an absolute path, otherwise we can have # problems with relative paths applied multiple times as DBT_PROJECT_DIR # is applied to dbt Core and Fusion as well (but not the dbt Cloud CLI) cwd_path = config.project_dir if os.path.isabs(config.project_dir) else None # Add appropriate color disable flag based on binary type color_flag = get_color_disable_flag(config.binary_type) args = [config.dbt_path, color_flag, *full_command] process = subprocess.Popen( args=args, cwd=cwd_path, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, text=True, ) output, _ = process.communicate(timeout=config.dbt_cli_timeout) return output or "OK" except subprocess.TimeoutExpired: return "Timeout: dbt command took too long to complete." + ( " Try using a specific selector to narrow down the results." if is_selectable else "" )
• src/dbt_mcp/dbt_cli/tools.py:232-241 (registration)

  ToolDefinition instance that registers the 'run' handler function with its description and annotations.
  ToolDefinition( fn=run, description=get_prompt("dbt_cli/run"), annotations=create_tool_annotations( title="dbt run", read_only_hint=False, destructive_hint=True, idempotent_hint=False, ), ),
• src/dbt_mcp/mcp/server.py:163-170 (registration)

  Server-level registration of all dbt CLI tools, including 'run', to the FastMCP instance.
  register_dbt_cli_tools( dbt_mcp, config=config.dbt_cli_config, disabled_tools=disabled_tools, enabled_tools=enabled_tools, enabled_toolsets=enabled_toolsets, disabled_toolsets=disabled_toolsets, )
• src/dbt_mcp/tools/tool_names.py:13-13 (helper)

  Enum value defining the canonical name for the 'run' tool.
  RUN = "run"