dbt CLI MCP Server

Instructions

    Returns:
        When output_format is 'json' (default):
          - With verbose=False (default): returns a simplified JSON with only name, resource_type, and depends_on.nodes
          - With verbose=True: returns a full JSON with all resource details
        When output_format is 'name', 'path', or 'selector', returns plain text with the respective format.
    

Implementation Reference

• src/tools.py:134-210 (handler)

  The core handler function for the 'dbt_ls' tool, decorated with @mcp.tool(). It constructs the 'dbt ls' command based on input parameters, executes it via execute_dbt_command, and formats the output using ls_formatter via process_command_result.

  @mcp.tool()
  async def dbt_ls(
      models: Optional[str] = Field(
          default=None,
          description="Specific models to list, using the dbt selection syntax. Note that you probably want to specify your selection here e.g. silver.fact"
      ),
      selector: Optional[str] = Field(
          default=None,
          description="Named selector to use"
      ),
      exclude: Optional[str] = Field(
          default=None,
          description="Models to exclude"
      ),
      resource_type: Optional[str] = Field(
          default=None,
          description="Type of resource to list (model, test, source, etc.)"
      ),
      project_dir: str = Field(
          default=".",
          description="ABSOLUTE PATH to the directory containing the dbt project (e.g. '/Users/username/projects/dbt_project' not '.')"
      ),
      profiles_dir: Optional[str] = Field(
          default=None,
          description="Directory containing the profiles.yml file (defaults to project_dir if not specified)"
      ),
      output_format: str = Field(
          default="json",
          description="Output format (json, name, path, or selector)"
      ),
      verbose: bool = Field(
          default=False,
          description="Return full JSON output instead of simplified version"
      )
  ) -> str:
      """List dbt resources. An AI agent should use this tool when it needs to discover available models, tests, sources, and other resources within a dbt project. This helps the agent understand the project structure, identify dependencies, and select specific resources for other operations like running or testing.
  
      Returns:
          When output_format is 'json' (default):
            - With verbose=False (default): returns a simplified JSON with only name, resource_type, and depends_on.nodes
            - With verbose=True: returns a full JSON with all resource details
          When output_format is 'name', 'path', or 'selector', returns plain text with the respective format.
      """
      # Log diagnostic information
      logger.info(f"Starting dbt_ls with project_dir={project_dir}, output_format={output_format}")
  
      command = ["ls"]
  
      if models:
          command.extend(["-s", models])
  
      if selector:
          command.extend(["--selector", selector])
  
      if exclude:
          command.extend(["--exclude", exclude])
  
      if resource_type:
          command.extend(["--resource-type", resource_type])
  
      command.extend(["--output", output_format])
  
      command.extend(["--quiet"])
  
      logger.info(f"Executing dbt command: dbt {' '.join(command)}")
      result = await execute_dbt_command(command, project_dir, profiles_dir)
      logger.info(f"dbt command result: success={result['success']}, returncode={result.get('returncode')}")
  
      # Use the centralized result processor with ls_formatter
      formatter = partial(ls_formatter, output_format=output_format, verbose=verbose)
  
      return await process_command_result(
          result,
          command_name="ls",
          output_formatter=formatter,
          include_debug_info=True  # Include extra debug info for this command
      )

• src/server.py:88-89 (registration)

  The call to register_tools(mcp) which registers the dbt_ls tool (along with others) with the FastMCP server instance.

  # Register tools
  register_tools(mcp)

• src/formatters.py:32-90 (helper)

  The ls_formatter function used specifically by dbt_ls to format the output from 'dbt ls', handling JSON parsing, filtering, sorting, and simplification based on verbose flag.

  def ls_formatter(output: Any, output_format: str = "json", verbose: bool = False) -> str:
      """
      Formatter for dbt ls command output.
  
      Args:
          output: The command output
          output_format: The output format (json, name, path, or selector)
          verbose: Whether to return full JSON output (True) or simplified version (False)
  
      Returns:
          Formatted output string
      """
      # For name, path, or selector formats, return the raw output as string
      if output_format != "json":
          logger.info(f"Returning raw output as string for format: {output_format}")
          return str(output)
  
      # For json format, parse the output and return as JSON
      logger.info("Parsing dbt ls output as JSON")
  
      # Return raw output if it's an empty string or None
      if not output:
          logger.warning("dbt ls returned empty output")
          return "[]"
  
      # Parse the output
      parsed = parse_dbt_list_output(output)
  
      # Filter out any empty or non-model entries
      filtered_parsed = [item for item in parsed if isinstance(item, dict) and
                        item.get("resource_type") in ["model", "seed", "test", "source", "snapshot"]]
  
      # Sort the results by resource_type and name for better readability
      filtered_parsed.sort(key=lambda x: (x.get("resource_type", ""), x.get("name", "")))
  
      # Return full parsed output if filtering removed everything
      if not filtered_parsed and parsed:
          logger.warning("Filtering removed all items, returning original parsed output")
          json_output = json.dumps(parsed, indent=2)
          logger.info(f"Final JSON output length: {len(json_output)}")
          return json_output
  
      # If not verbose, simplify the output to only include name, resource_type, and depends_on.nodes
      if not verbose and filtered_parsed:
          logger.info("Simplifying output (verbose=False)")
          simplified = []
          for item in filtered_parsed:
              simplified.append({
                  "name": item.get("name"),
                  "resource_type": item.get("resource_type"),
                  "depends_on": {
                      "nodes": item.get("depends_on", {}).get("nodes", [])
                  }
              })
          filtered_parsed = simplified
  
      json_output = json.dumps(filtered_parsed, indent=2)
      logger.info(f"Final JSON output length: {len(json_output)}")
      return json_output

• src/command.py:191-316 (helper)

  Helper function parse_dbt_list_output used by ls_formatter to parse various formats of dbt ls output into a standardized list of resource dictionaries.

  def parse_dbt_list_output(output: Union[str, Dict, List]) -> List[Dict[str, Any]]:
      """
      Parse the output from dbt list command.
      
      Args:
          output: Output from dbt list command (string or parsed JSON)
          
      Returns:
          List of resources
      """
      logger.debug(f"Parsing dbt list output with type: {type(output)}")
      
      # If already parsed as JSON dictionary with nodes
      if isinstance(output, dict) and "nodes" in output:
          return [
              {"name": name, **details}
              for name, details in output["nodes"].items()
          ]
      
      # Handle dbt Cloud CLI output format - an array of objects with name property containing embedded JSON
      if isinstance(output, list) and all(isinstance(item, dict) and "name" in item for item in output):
          logger.debug(f"Found dbt Cloud CLI output format with {len(output)} items")
          extracted_models = []
          
          for item in output:
              name_value = item["name"]
              
              # Skip log messages that don't contain model data
              if any(log_msg in name_value for log_msg in [
                  "Sending project", "Created invocation", "Waiting for",
                  "Streaming", "Running dbt", "Invocation has finished"
              ]):
                  continue
              
              # Check if the name value is a JSON string
              if name_value.startswith('{') and '"name":' in name_value and '"resource_type":' in name_value:
                  try:
                      # Parse the JSON string directly
                      model_data = json.loads(name_value)
                      if isinstance(model_data, dict) and "name" in model_data and "resource_type" in model_data:
                          extracted_models.append(model_data)
                          continue
                  except json.JSONDecodeError:
                      logger.debug(f"Failed to parse JSON from: {name_value[:30]}...")
              
              # Extract model data from timestamped JSON lines (e.g., "00:59:06 {json}")
              timestamp_prefix_match = re.match(r'^(\d\d:\d\d:\d\d)\s+(.+)$', name_value)
              if timestamp_prefix_match:
                  json_string = timestamp_prefix_match.group(2)
                  try:
                      model_data = json.loads(json_string)
                      if isinstance(model_data, dict):
                          # Only add entries that have both name and resource_type
                          if "name" in model_data and "resource_type" in model_data:
                              extracted_models.append(model_data)
                  except json.JSONDecodeError:
                      # Not valid JSON, skip this line
                      logger.debug(f"Failed to parse JSON from: {json_string[:30]}...")
                      continue
          
          # If we found model data, return it
          if extracted_models:
              logger.debug(f"Successfully extracted {len(extracted_models)} models from dbt Cloud CLI output")
              return extracted_models
          
          # If no model data found, return empty list
          logger.warning("No valid model data found in dbt Cloud CLI output")
          return []
      
      # If already parsed as regular JSON list
      if isinstance(output, list):
          # For test compatibility
          if all(isinstance(item, dict) and "name" in item for item in output):
              return output
          # For empty lists or other list types, return as is
          return output
      
      # If string, try to parse as JSON
      if isinstance(output, str):
          try:
              parsed = json.loads(output)
              if isinstance(parsed, dict) and "nodes" in parsed:
                  return [
                      {"name": name, **details}
                      for name, details in parsed["nodes"].items()
                  ]
              elif isinstance(parsed, list):
                  return parsed
          except json.JSONDecodeError:
              # Not JSON, parse text format (simplified)
              models = []
              for line in output.splitlines():
                  line = line.strip()
                  if not line:
                      continue
                      
                  # Check if the line is a JSON string
                  if line.startswith('{') and '"name":' in line and '"resource_type":' in line:
                      try:
                          model_data = json.loads(line)
                          if isinstance(model_data, dict) and "name" in model_data and "resource_type" in model_data:
                              models.append(model_data)
                              continue
                      except json.JSONDecodeError:
                          pass
                  
                  # Check for dbt Cloud CLI format with timestamps (e.g., "00:59:06 {json}")
                  timestamp_match = re.match(r'^(\d\d:\d\d:\d\d)\s+(.+)$', line)
                  if timestamp_match:
                      json_part = timestamp_match.group(2)
                      try:
                          model_data = json.loads(json_part)
                          if isinstance(model_data, dict) and "name" in model_data and "resource_type" in model_data:
                              models.append(model_data)
                              continue
                      except json.JSONDecodeError:
                          pass
                  
                  # Fall back to simple name-only format
                  models.append({"name": line})
              return models
      
      # Fallback: return empty list
      logger.warning("Could not parse dbt list output in any recognized format")
      return []

• src/command.py:318-382 (helper)

  Centralized result processor called by dbt_ls handler to format the command result using the provided ls_formatter.

  async def process_command_result(
      result: Dict[str, Any],
      command_name: str,
      output_formatter: Optional[Callable] = None,
      include_debug_info: bool = False
  ) -> str:
      """
      Process the result of a dbt command execution.
      
      Args:
          result: The result dictionary from execute_dbt_command
          command_name: The name of the dbt command (e.g. "run", "test")
          output_formatter: Optional function to format successful output
          include_debug_info: Whether to include additional debug info in error messages
          
      Returns:
          Formatted output or error message
      """
      logger.info(f"Processing command result for {command_name}")
      logger.info(f"Result success: {result['success']}, returncode: {result.get('returncode')}")
      
      # Log the output type and a sample
      if "output" in result:
          if isinstance(result["output"], str):
              logger.info(f"Output type: str, first 100 chars: {result['output'][:100]}")
          elif isinstance(result["output"], (dict, list)):
              logger.info(f"Output type: {type(result['output'])}, sample: {json.dumps(result['output'])[:100]}")
          else:
              logger.info(f"Output type: {type(result['output'])}")
      
      # For errors, simply return the raw command output if available
      if not result["success"]:
          logger.warning(f"Command {command_name} failed with returncode {result.get('returncode')}")
          
          # If we have command output, return it directly
          if "output" in result and result["output"]:
              logger.info(f"Returning error output: {str(result['output'])[:100]}...")
              return str(result["output"])
          
          # If no command output, return the error message
          if result["error"]:
              logger.info(f"Returning error message: {str(result['error'])[:100]}...")
              return str(result["error"])
              
          # If neither output nor error is available, return a generic message
          logger.info("No output or error available, returning generic message")
          return f"Command failed with exit code {result.get('returncode', 'unknown')}"
      
      # Format successful output
      if output_formatter:
          logger.info(f"Using custom formatter for {command_name}")
          formatted_result = output_formatter(result["output"])
          logger.info(f"Formatted result type: {type(formatted_result)}, first 100 chars: {str(formatted_result)[:100]}")
          return formatted_result
      
      # Default output formatting
      logger.info(f"Using default formatting for {command_name}")
      if isinstance(result["output"], (dict, list)):
          json_result = json.dumps(result["output"])
          logger.info(f"JSON result length: {len(json_result)}, first 100 chars: {json_result[:100]}")
          return json_result
      else:
          str_result = str(result["output"])
          logger.info(f"String result length: {len(str_result)}, first 100 chars: {str_result[:100]}")
          return str_result