dbt_test
Validate data quality and integrity by running tests in a dbt project. Ensures data transformations meet business rules and constraints for accurate analysis and reporting.
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 |
|---|---|---|---|
| exclude | No | Models to exclude | |
| models | No | Specific models to test, using the dbt selection syntax | |
| profiles_dir | No | Directory containing the profiles.yml file (defaults to project_dir if not specified) | |
| project_dir | No | ABSOLUTE PATH to the directory containing the dbt project (e.g. '/Users/username/projects/dbt_project' not '.') | . |
| selector | No | Named selector to use |
Implementation Reference
- src/tools.py:87-132 (handler)The MCP tool handler for 'dbt_test'. This async function defines the tool logic, input schema via Pydantic Fields, and executes the 'dbt test' command with optional selectors (models, selector, exclude), project_dir, and profiles_dir. It processes and returns the command 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)The registration of all MCP tools, including dbt_test, by calling register_tools(mcp) on the FastMCP server instance. The tools are defined and registered within the register_tools function in src/tools.py.# Register tools register_tools(mcp)
- src/tools.py:89-108 (schema)Input schema parameters for the dbt_test tool, defined using Pydantic Field with descriptions and defaults. Used by FastMCP to generate JSON schema for tool calls.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)" )