dbt CLI MCP Server

Instructions

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