dbt-mcp

Instructions

Implementation Reference

• src/dbt_mcp/dbt_cli/tools.py:110-124 (handler)

  The `ls` function is the direct handler for the MCP tool named 'list'. It runs the 'dbt list' CLI command using the shared _run_dbt_command helper, supporting selector and resource_type parameters for filtering.
  def ls( selector: str | None = Field( default=None, description=get_prompt("dbt_cli/args/selectors") ), resource_type: list[str] | None = Field( default=None, description=get_prompt("dbt_cli/args/resource_type"), ), ) -> str: return _run_dbt_command( ["list"], selector, resource_type=resource_type, is_selectable=True, )
• src/dbt_mcp/dbt_cli/tools.py:211-221 (registration)

  Registration of the 'list' tool via ToolDefinition, specifying the name, handler function `ls`, description from prompts, and annotations indicating it's read-only and idempotent.
  ToolDefinition( name="list", fn=ls, description=get_prompt("dbt_cli/list"), annotations=create_tool_annotations( title="dbt list", read_only_hint=True, destructive_hint=False, idempotent_hint=True, ), ),
• src/dbt_mcp/dbt_cli/tools.py:19-84 (helper)

  Core helper function `_run_dbt_command` that executes the dbt CLI subprocess for all dbt_cli tools, including 'list'. Handles parameters, adds --quiet for verbose commands like 'list', manages working directory, colors, and timeouts.
  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:265-281 (registration)

  The `register_dbt_cli_tools` function registers all dbt CLI tools, including 'list', to the FastMCP server by calling `register_tools` with the list of ToolDefinitions from `create_dbt_cli_tool_definitions`.
  def register_dbt_cli_tools( dbt_mcp: FastMCP, config: DbtCliConfig, *, disabled_tools: set[ToolName], enabled_tools: set[ToolName], enabled_toolsets: set[Toolset], disabled_toolsets: set[Toolset], ) -> None: register_tools( dbt_mcp, tool_definitions=create_dbt_cli_tool_definitions(config), disabled_tools=disabled_tools, enabled_tools=enabled_tools, enabled_toolsets=enabled_toolsets, disabled_toolsets=disabled_toolsets, )