snapshot_models
Execute dbt snapshots to capture slowly changing dimensions (SCD Type 2) by tracking historical record changes over time, including when records were first seen, changed, or deleted.
Instructions
Execute dbt snapshots to capture slowly changing dimensions (SCD Type 2).
Snapshots track historical changes over time by recording:
When records were first seen (valid_from)
When records changed or were deleted (valid_to)
The state of records at each point in time
Unlike models and seeds, snapshots are time-based and should be run on a schedule (e.g., daily or hourly), not during interactive development.
Args: select: dbt selector syntax (e.g., "snapshot_name", "tag:daily") exclude: Exclude specific snapshots
Returns: Snapshot results with status and captured changes
Examples: snapshot_models() # Run all snapshots snapshot_models(select="customer_history") # Run specific snapshot snapshot_models(select="tag:hourly") # Run snapshots tagged 'hourly'
Note: Snapshots do not support state-based selection (select_state_modified*) because they are time-dependent, not change-dependent.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| select | No | ||
| exclude | No |
Implementation Reference
- src/dbt_core_mcp/server.py:1189-1230 (handler)The handler function that executes the 'dbt snapshot' command with optional select and exclude parameters, parses results from run_results.json, and returns success/error status with details.async def toolImpl_snapshot_models( self, select: str | None = None, exclude: str | None = None, ) -> dict[str, Any]: """Implementation of snapshot_models tool.""" # Build command args args = ["snapshot"] if select: args.extend(["-s", select]) if exclude: args.extend(["--exclude", exclude]) # Execute logger.info(f"Running DBT snapshot with args: {args}") result = await self.runner.invoke(args) # type: ignore if not result.success: error_msg = str(result.exception) if result.exception else "Snapshot failed" response = { "status": "error", "message": error_msg, "command": " ".join(args), } # Include dbt output for debugging if result.stdout: response["dbt_output"] = result.stdout if result.stderr: response["stderr"] = result.stderr return response # Parse run_results.json for details run_results = self._parse_run_results() return { "status": "success", "command": " ".join(args), "results": run_results.get("results", []), "elapsed_time": run_results.get("elapsed_time"), }
- src/dbt_core_mcp/server.py:1686-1717 (registration)Registers the 'snapshot_models' tool using FastMCP @app.tool() decorator. Includes comprehensive docstring describing usage, args, returns, and examples. Calls the handler after ensuring initialization.async def snapshot_models( ctx: Context, select: str | None = None, exclude: str | None = None, ) -> dict[str, Any]: """Execute dbt snapshots to capture slowly changing dimensions (SCD Type 2). Snapshots track historical changes over time by recording: - When records were first seen (valid_from) - When records changed or were deleted (valid_to) - The state of records at each point in time Unlike models and seeds, snapshots are time-based and should be run on a schedule (e.g., daily or hourly), not during interactive development. Args: select: dbt selector syntax (e.g., "snapshot_name", "tag:daily") exclude: Exclude specific snapshots Returns: Snapshot results with status and captured changes Examples: snapshot_models() # Run all snapshots snapshot_models(select="customer_history") # Run specific snapshot snapshot_models(select="tag:hourly") # Run snapshots tagged 'hourly' Note: Snapshots do not support state-based selection (select_state_modified*) because they are time-dependent, not change-dependent. """ await self._ensure_initialized_with_context(ctx) return await self.toolImpl_snapshot_models(select, exclude)