dbt_test
Validate data quality and integrity by running tests defined in a dbt project to ensure data transformations meet business rules before analysis.
Instructions
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)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| models | No | Specific models to test, using the dbt selection syntax | |
| selector | No | Named selector to use | |
| exclude | No | Models to exclude | |
| project_dir | No | ABSOLUTE PATH to the directory containing the dbt project (e.g. '/Users/username/projects/dbt_project' not '.') | . |
| profiles_dir | No | Directory containing the profiles.yml file (defaults to project_dir if not specified) |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/tools.py:87-132 (handler)The core handler function for the MCP 'dbt_test' tool. Decorated with @mcp.tool(), it defines input parameters using Pydantic Field (serving as schema), constructs the 'dbt test' command with optional selectors/excludes, executes it via execute_dbt_command, and processes the result.
@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") - src/server.py:88-89 (registration)Registration of all MCP tools (including 'dbt_test') by calling register_tools(mcp) on the FastMCP server instance.
# Register tools register_tools(mcp) - src/tools.py:25-25 (registration)The register_tools function where all @mcp.tool() decorated handlers, including dbt_test, are defined and thus registered with the MCP server.
def register_tools(mcp: FastMCP) -> None: - src/tools.py:17-17 (helper)Import of helper functions used by dbt_test handler: execute_dbt_command (executes the dbt command), process_command_result (formats output).
from src.command import execute_dbt_command, parse_dbt_list_output, process_command_result