dbt CLI MCP Server

Instructions

Implementation Reference

• src/tools.py:476-529 (handler)

  The core handler function for the 'dbt_build' MCP tool. It defines the input schema using Pydantic Field descriptions, constructs the dbt 'build' command with optional parameters, executes it via execute_dbt_command from src.command, processes the result, and returns the output.
  @mcp.tool() async def dbt_build( models: Optional[str] = Field( default=None, description="Specific models to build, 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)" ), full_refresh: bool = Field( default=False, description="Whether to perform a full refresh" ) ) -> str: """Run build command (seeds, tests, snapshots, and models). An AI agent should use this tool when it needs to execute a comprehensive build process that runs seeds, snapshots, models, and tests in the correct order. This is ideal for complete project deployment or ensuring all components work together. Returns: Output from the dbt build command as text (this command does not support JSON output format) """ command = ["build"] if models: command.extend(["-s", models]) if selector: command.extend(["--selector", selector]) if exclude: command.extend(["--exclude", exclude]) if full_refresh: command.append("--full-refresh") # 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="build")
• src/server.py:88-90 (registration)

  The registration of all MCP tools, including 'dbt_build', by invoking register_tools(mcp) imported from src.tools. This is called during FastMCP server initialization.
  # Register tools register_tools(mcp)
• src/tools.py:25-532 (registration)

  The register_tools function that defines all nested MCP tool functions (including dbt_build) and registers them using the @mcp.tool() decorator.
  def register_tools(mcp: FastMCP) -> None: """ Register all tools with the MCP server. Args: mcp: The FastMCP server instance """ @mcp.tool() async def dbt_run( models: Optional[str] = Field( default=None, description="Specific models to run, using the dbt selection syntax (e.g., \"model_name+\")" ), 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)" ), full_refresh: bool = Field( default=False, description="Whether to perform a full refresh" ) ) -> str: """Run dbt models. An AI agent should use this tool when it needs to execute dbt models to transform data and build analytical tables in the data warehouse. This is essential for refreshing data or implementing new data transformations in a project. Returns: Output from the dbt run command as text (this command does not support JSON output format) """ command = ["run"] if models: command.extend(["-s", models]) if selector: command.extend(["--selector", selector]) if exclude: command.extend(["--exclude", exclude]) if full_refresh: command.append("--full-refresh") # 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="run") @mcp.tool() async def dbt_test( models: Optional[str] = Field( default=None, description="Specific models to test, 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: """Run dbt tests. An AI agent should use this tool when it needs to validate data quality and integrity by running tests defined in a dbt project. This helps ensure that data transformations meet expected business rules and constraints before being used for analysis or reporting. Returns: Output from the dbt test command as text (this command does not support JSON output format) """ command = ["test"] 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="test") @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 ) @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") @mcp.tool() async def dbt_debug( 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: """Run dbt debug to validate the project setup. An AI agent should use this tool when it needs to troubleshoot configuration issues, check database connectivity, or verify that all project dependencies are properly installed. This is essential for diagnosing problems before attempting to run models or tests. Returns: Output from the dbt debug command as text (this command does not support JSON output format) """ command = ["debug"] # 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="debug") @mcp.tool() async def dbt_deps( 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: """Install dbt package dependencies. An AI agent should use this tool when it needs to install or update external packages that the dbt project depends on. This ensures that all required modules, macros, and models from other packages are available before running the project. Returns: Output from the dbt deps command as text (this command does not support JSON output format) """ command = ["deps"] # 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="deps") @mcp.tool() async def dbt_seed( selector: Optional[str] = Field( default=None, description="Named selector to use" ), exclude: Optional[str] = Field( default=None, description="Seeds 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: """Load CSV files as seed data. An AI agent should use this tool when it needs to load initial data from CSV files into the database. This is essential for creating reference tables, test datasets, or any static data that models will depend on. Returns: Output from the dbt seed command as text (this command does not support JSON output format) """ command = ["seed"] # The --no-print flag is not supported by dbt Cloud CLI # We'll rely on proper parsing to handle any print macros if selector: command.extend(["--selector", selector]) if exclude: command.extend(["--exclude", exclude]) result = await execute_dbt_command(command, project_dir, profiles_dir) # Use the centralized result processor return await process_command_result(result, command_name="seed") @mcp.tool() async def dbt_show( models: str = Field( description="Specific model to show. For model references, use standard dbt syntax like 'model_name'. For inline SQL, use the format 'select * from {{ ref(\"model_name\") }}' to reference other models." ), 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)" ), limit: Optional[int] = Field( default=None, description="Limit the number of rows returned" ), output: Optional[str] = Field( default="json", description="Output format (json, table, etc.)" ) ) -> str: """Preview the results of a model. An AI agent should use this tool when it needs to preview data from a specific model without materializing it. This helps inspect transformation results, debug issues, or demonstrate how data looks after processing without modifying the target database. Returns: Output from the dbt show command, defaulting to JSON format if not specified """ # Use enhanced SQL detection is_inline_sql, sql_type = is_inline_sql_query(models) # If it's SQL, check for security risks if is_inline_sql: has_risk, risk_reason = contains_mutation_risk(models) if has_risk: logger.warning(f"Security risk detected in SQL: {risk_reason}") error_result = { "success": False, "output": f"Security validation failed: {risk_reason}. For security reasons, mutation operations are not allowed.", "error": "SecurityValidationError", "returncode": 1 } return await process_command_result( error_result, command_name="show", include_debug_info=True ) logger.info(f"dbt_show called with models={models}, is_inline_sql={is_inline_sql}") # If it's inline SQL, strip out any LIMIT clause as we'll handle it with the --limit parameter if is_inline_sql: # Use regex to remove LIMIT clause from the SQL original_models = models models = re.sub(r'\bLIMIT\s+\d+\b', '', models, flags=re.IGNORECASE) logger.info(f"Stripped LIMIT clause: {original_models} -> {models}") # For inline SQL, use the --inline flag with the SQL as its value command = ["show", f"--inline={models}", "--output", output or "json"] # Only add --limit if the inline type is WITH or SELECT (select_inline vs meta_inline) if limit and sql_type in ["WITH", "SELECT"]: command.extend(["--limit", str(limit)]) logger.info(f"Executing dbt command: {' '.join(command)}") # Don't use --quiet for inline SQL to ensure we get error messages result = await execute_dbt_command(command, project_dir, profiles_dir) logger.info(f"Command result: success={result['success']}, returncode={result.get('returncode')}") if isinstance(result["output"], str): logger.info(f"Output (first 100 chars): {result['output'][:100]}") elif isinstance(result["output"], (dict, list)): logger.info(f"Output structure: {json.dumps(result['output'])[:100]}") # Check for specific error patterns in the output if not result["success"] or ( isinstance(result["output"], str) and any(err in result["output"].lower() for err in ["error", "failed", "syntax", "exception"]) ): logger.warning(f"Error detected in output: {result['output'][:200]}") error_result = { "success": False, "output": f"Error executing inline SQL\n{result['output']}", "error": result["error"], "returncode": result["returncode"] } return await process_command_result( error_result, command_name="show", include_debug_info=True ) else: # For regular model references, check if the model exists first check_command = ["ls", "-s", models] check_result = await execute_dbt_command(check_command, project_dir, profiles_dir) # If the model doesn't exist, return the error message if not check_result["success"] or "does not match any enabled nodes" in str(check_result["output"]): error_result = { "success": False, "output": f"Model does not exist or is not enabled\n{check_result['output']}", "error": check_result["error"], "returncode": check_result["returncode"] } return await process_command_result( error_result, command_name="show", include_debug_info=True ) # If the model exists, run the show command with --quiet and --output json command = ["show", "-s", models, "--quiet", "--output", output or "json"] if limit: command.extend(["--limit", str(limit)]) result = await execute_dbt_command(command, project_dir, profiles_dir) # Use the centralized result processor return await process_command_result( result, command_name="show", output_formatter=show_formatter, include_debug_info=True ) @mcp.tool() async def dbt_build( models: Optional[str] = Field( default=None, description="Specific models to build, 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)" ), full_refresh: bool = Field( default=False, description="Whether to perform a full refresh" ) ) -> str: """Run build command (seeds, tests, snapshots, and models). An AI agent should use this tool when it needs to execute a comprehensive build process that runs seeds, snapshots, models, and tests in the correct order. This is ideal for complete project deployment or ensuring all components work together. Returns: Output from the dbt build command as text (this command does not support JSON output format) """ command = ["build"] if models: command.extend(["-s", models]) if selector: command.extend(["--selector", selector]) if exclude: command.extend(["--exclude", exclude]) if full_refresh: command.append("--full-refresh") # 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="build") logger.info("Registered all dbt tools")
• src/cli.py:385-410 (helper)

  CLI wrapper function run_dbt_build that mirrors the MCP tool logic for command-line usage, mapped to the 'build' command in the CLI command_map.
  async def run_dbt_build(models=None, selector=None, exclude=None, project_dir=".", profiles_dir=None, full_refresh=False): """Run build command.""" command = ["build"] if models: command.extend(["-s", models]) if selector: command.extend(["--selector", selector]) if exclude: command.extend(["--exclude", exclude]) if full_refresh: command.append("--full-refresh") from src.command import execute_dbt_command result = await execute_dbt_command(command, project_dir, profiles_dir) if not result["success"]: error_msg = f"Error executing dbt build: {result['error']}" if result["output"]: error_msg += f"\nOutput: {result['output']}" return error_msg return json.dumps(result["output"]) if isinstance(result["output"], (dict, list)) else str(result["output"])