dbt CLI MCP Server

Instructions

    Returns:
        Output from the dbt compile command as text (this command does not support JSON output format)
    

Implementation Reference

• src/tools.py:212-258 (handler)

  The primary handler function for the 'dbt_compile' MCP tool. Decorated with @mcp.tool(), defines input schema via Pydantic Field descriptions, constructs the dbt compile command, executes it, and processes the result.

  @mcp.tool()
  async def dbt_compile(
      models: Optional[str] = Field(
          default=None,
          description="Specific models to compile, using the dbt selection syntax"
      ),
      selector: Optional[str] = Field(
          default=None,
          description="Named selector to use"
      ),
      exclude: Optional[str] = Field(
          default=None,
          description="Models to exclude"
      ),
      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)"
      )
  ) -> str:
      """Compile dbt models. An AI agent should use this tool when it needs to generate the SQL that will be executed without actually running it against the database. This is valuable for validating SQL syntax, previewing transformations, or investigating how dbt interprets models before committing to execution.
  
      Returns:
          Output from the dbt compile command as text (this command does not support JSON output format)
      """
      command = ["compile"]
  
      if models:
          command.extend(["-s", models])
  
      if selector:
          command.extend(["--selector", selector])
  
      if exclude:
          command.extend(["--exclude", exclude])
  
      # The --no-print flag is not supported by dbt Cloud CLI
      # We'll rely on proper parsing to handle any print macros
  
      result = await execute_dbt_command(command, project_dir, profiles_dir)
  
      # Use the centralized result processor
      return await process_command_result(result, command_name="compile")

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

  Registers all MCP tools, including dbt_compile, by calling register_tools(mcp) on the FastMCP server instance.

  # Register tools
  register_tools(mcp)

• src/command.py:53-189 (helper)

  Key helper function called by dbt_compile to execute the actual dbt CLI 'compile' command asynchronously, managing environment variables, subprocess execution, output parsing (including JSON detection), and error handling.

  async def execute_dbt_command(
      command: List[str],
      project_dir: str = ".",
      profiles_dir: Optional[str] = None
  ) -> Dict[str, Any]:
      """
      Execute a dbt command and return the result.
      
      Args:
          command: List of command arguments (without the dbt executable)
          project_dir: Directory containing the dbt project
          profiles_dir: Directory containing the profiles.yml file (defaults to project_dir if not specified)
          
      Returns:
          Dictionary containing command result:
          {
              "success": bool,
              "output": str or dict,
              "error": str or None,
              "returncode": int
          }
      """
      # Get dbt path from config
      dbt_path = get_config("dbt_path", "dbt")
      full_command = [dbt_path] + command
      
      # Load environment variables
      env_vars = load_environment(project_dir)
      
      # Explicitly set HOME environment variable in os.environ
      os.environ["HOME"] = str(Path.home())
      logger.debug(f"Explicitly setting HOME environment variable in os.environ to {os.environ['HOME']}")
      
      # Set DBT_PROFILES_DIR based on profiles_dir or project_dir
      if profiles_dir is not None:
          # Use the explicitly provided profiles_dir
          abs_profiles_dir = str(Path(profiles_dir).resolve())
          os.environ["DBT_PROFILES_DIR"] = abs_profiles_dir
          logger.debug(f"Setting DBT_PROFILES_DIR in os.environ to {abs_profiles_dir} (from profiles_dir)")
      else:
          # Check if there's a value from the .env file
          if "DBT_PROFILES_DIR" in env_vars:
              os.environ["DBT_PROFILES_DIR"] = env_vars["DBT_PROFILES_DIR"]
              logger.debug(f"Setting DBT_PROFILES_DIR from env_vars to {env_vars['DBT_PROFILES_DIR']}")
          else:
              # Default to project_dir
              abs_project_dir = str(Path(project_dir).resolve())
              os.environ["DBT_PROFILES_DIR"] = abs_project_dir
              logger.debug(f"Setting DBT_PROFILES_DIR in os.environ to {abs_project_dir} (from project_dir)")
      
      # Update env_vars with the current os.environ
      env_vars.update(os.environ)
      
      logger.debug(f"Executing command: {' '.join(full_command)} in {project_dir}")
      
      try:
          # Execute the command
          process = await asyncio.create_subprocess_exec(
              *full_command,
              cwd=project_dir,
              env=env_vars,
              stdout=asyncio.subprocess.PIPE,
              stderr=asyncio.subprocess.PIPE
          )
          
          # Communicate with the process
          stdout_bytes, stderr_bytes = await process.communicate()
          stdout = stdout_bytes.decode('utf-8') if stdout_bytes else ""
          stderr = stderr_bytes.decode('utf-8') if stderr_bytes else ""
          success = process.returncode == 0
          
          # Special case for 'show' command: detect "does not match any enabled nodes" as an error
          # Only check if --quiet is not in the command, as --quiet suppresses this output
          if success and command[0] == "show" and "--quiet" not in command and "does not match any enabled nodes" in stdout:
              success = False
              
          # For commands that failed, combine stdout and stderr for comprehensive output
          if not success and stderr:
              # If there's output from both stdout and stderr, combine them
              if stdout:
                  output = f"{stdout}\n\nSTDERR:\n{stderr}"
              else:
                  output = stderr
          else:
              # For successful commands, use stdout
              output = stdout
          
          # Check if this is dbt Cloud CLI output format with embedded JSON in log lines
          if stdout.strip().startswith('[') and '"name":' in stdout:
              try:
                  # Parse the entire output as JSON array
                  json_array = json.loads(stdout)
                  
                  # If it's an array of log objects with name field (dbt Cloud CLI format)
                  if isinstance(json_array, list) and all(isinstance(item, dict) and "name" in item for item in json_array):
                      logger.debug(f"Detected dbt Cloud CLI output format with {len(json_array)} items")
                      output = json_array
              except json.JSONDecodeError:
                  # Not valid JSON array, keep as string
                  logger.debug("Failed to parse stdout as JSON array, keeping as string")
                  pass
          else:
              # Try standard JSON parsing
              try:
                  output = json.loads(stdout)
              except json.JSONDecodeError:
                  # Not JSON, keep as string
                  logger.debug("Failed to parse stdout as standard JSON, keeping as string")
                  pass
              
          result = {
              "success": success,
              "output": output,
              "error": stderr if not success else None,
              "returncode": process.returncode
          }
          
          if not success:
              logger.warning(f"Command failed with exit code {process.returncode}: {stderr}")
              
              # Log full environment for debugging
              logger.debug(f"Full environment variables: {env_vars}")
              logger.debug(f"Current directory: {project_dir}")
              logger.debug(f"Full command: {' '.join(full_command)}")
          
          return result
      except Exception as e:
          import traceback
          stack_trace = traceback.format_exc()
          logger.error(f"Error executing command: {e}\nStack trace: {stack_trace}")
          return {
              "success": False,
              "output": None,
              "error": f"{str(e)}\nStack trace: {stack_trace}",
              "returncode": -1
          }

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

  Helper function used by dbt_compile to format and return the processed result from the dbt command execution, handling success/error cases and optional formatting.

  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