analyze_impact

Instructions

Implementation Reference

• src/dbt_core_mcp/tools/analyze_impact.py:19-32 (handler)

  The _implementation function that executes the analyze_impact tool logic. It ensures state is initialized, then delegates to state.manifest.analyze_impact(name, resource_type).

  async def _implementation(ctx: Context | None, name: str, resource_type: str | None, state: DbtCoreServerContext, force_parse: bool = True) -> dict[str, Any]:
      """Implementation function for analyze_impact tool.
  
      Separated for testing purposes - tests call this directly with explicit state.
      The @tool() decorated analyze_impact() function calls this with injected dependencies.
      """
      # Initialize state if needed (metadata tool uses force_parse=True)
      await state.ensure_initialized(ctx, force_parse)
  
      # Delegate to manifest helper for downstream impact calculation
      try:
          return state.manifest.analyze_impact(name, resource_type)  # type: ignore
      except ValueError as e:
          raise ValueError(f"Impact analysis error: {e}")

• src/dbt_core_mcp/tools/analyze_impact.py:35-78 (schema)

  The @dbtTool() decorated analyze_impact function with full schema: parameter definitions, return type documentation, and examples. This is the MCP tool entry point.

  @dbtTool()
  async def analyze_impact(
      ctx: Context,
      name: str,
      resource_type: str | None = None,
      state: DbtCoreServerContext = Depends(get_state),
  ) -> dict[str, Any]:
      """Analyze the impact of changing any dbt resource with auto-detection.
  
      This unified tool works across all resource types (models, sources, seeds, snapshots, etc.)
      showing all downstream dependencies that would be affected by changes. Provides actionable
      recommendations for running affected resources.
  
      Args:
          name: Resource name. For sources, use "source_name.table_name" or just "table_name"
              Examples: "stg_customers", "jaffle_shop.orders", "raw_customers"
          resource_type: Optional filter to narrow search:
              - "model": Data transformation models
              - "source": External data sources
              - "seed": CSV reference data files
              - "snapshot": SCD Type 2 historical tables
              - "test": Data quality tests
              - "analysis": Ad-hoc analysis queries
              - None: Auto-detect (searches all types)
  
      Returns:
          Impact analysis with:
          - List of affected models by distance
          - Count of affected tests and other resources
          - Total impact statistics
          - Resources grouped by distance from changed resource
          - Recommended dbt command to run affected resources
          - Human-readable impact assessment message
          If multiple matches found, returns all matches for LLM to process.
  
      Raises:
          ValueError: If resource not found
  
      Examples:
          analyze_impact("stg_customers") -> auto-detect and show impact
          analyze_impact("jaffle_shop.orders", "source") -> impact of source change
          analyze_impact("raw_customers", "seed") -> impact of seed data change
      """
      return await _implementation(ctx, name, resource_type, state)

• src/dbt_core_mcp/dbt/manifest.py:585-730 (helper)

  The ManifestLoader.analyze_impact() method which performs the actual impact analysis: finds downstream dependents, categorizes by type (model/test/other), groups by distance, builds recommendations, and returns the structured result.

  def analyze_impact(
      self,
      name: str,
      resource_type: str | None = None,
  ) -> dict[str, Any]:
      """
      Analyze the impact of changing a resource across all resource types.
  
      Shows all downstream dependencies that would be affected by changes,
      including models, tests, and other resources. Provides actionable
      recommendations for running affected resources.
  
      Args:
          name: Resource name. For sources, use "source_name.table_name" or just "table_name"
          resource_type: Optional filter (model, source, seed, snapshot, test, analysis).
                        If None, auto-detects resource type.
  
      Returns:
          Dictionary with impact analysis:
          {
              "resource": {...},  # The target resource info
              "impact": {
                  "models_affected": [...],  # Downstream models by distance
                  "models_affected_count": int,
                  "tests_affected_count": int,
                  "other_affected_count": int,
                  "total_affected": int
              },
              "affected_by_distance": {
                  "1": [...],  # Immediate dependents
                  "2": [...],  # Second-level dependents
                  ...
              },
              "recommendation": str,  # Suggested dbt command
              "message": str  # Human-readable impact assessment
          }
  
          If multiple matches found, returns:
          {"multiple_matches": True, "matches": [...], "message": "..."}
  
      Raises:
          RuntimeError: If manifest not loaded
          ValueError: If resource not found
  
      Examples:
          analyze_impact("stg_customers") -> impact of changing staging model
          analyze_impact("jaffle_shop.orders", "source") -> impact of source change
          analyze_impact("raw_customers", "seed") -> impact of seed change
      """
      if not self._manifest:
          raise RuntimeError("Manifest not loaded. Call load() first.")
  
      # Get the resource (auto-detect if resource_type not specified)
      resource = self.get_resource_node(name, resource_type)
  
      # Handle multiple matches - return for LLM to process
      if resource.get("multiple_matches"):
          return resource
  
      # Extract unique_id for impact traversal
      unique_id = resource.get("unique_id")
      if not unique_id:
          raise ValueError(f"Resource '{name}' does not have a unique_id")
  
      # Get all downstream dependencies (no depth limit for impact)
      downstream = self.get_downstream_nodes(unique_id, max_depth=None)
  
      # Categorize by resource type
      models_affected: list[dict[str, Any]] = []
      tests_affected: list[dict[str, Any]] = []
      other_affected: list[dict[str, Any]] = []
      affected_by_distance: dict[str, list[dict[str, Any]]] = {}
  
      for dep in downstream:
          dep_type = str(dep["type"])
          distance = str(dep["distance"])
  
          # Group by distance
          if distance not in affected_by_distance:
              affected_by_distance[distance] = []
          affected_by_distance[distance].append(dep)
  
          # Categorize by type
          if dep_type == "model":
              models_affected.append(dep)
          elif dep_type == "test":
              tests_affected.append(dep)
          else:
              other_affected.append(dep)
  
      # Sort models by distance for better readability
      models_affected_sorted = sorted(models_affected, key=lambda x: (int(x["distance"]), str(x["name"])))
  
      # Build recommendation based on resource type
      resource_name = resource.get("name", name)
      current_resource_type = resource.get("resource_type")
  
      if current_resource_type == "source":
          # For sources, recommend running downstream models
          if len(models_affected) == 0:
              recommendation = f"dbt test -s source:{resource.get('source_name')}.{resource_name}"
          else:
              recommendation = f"dbt run -s {resource_name}+"
      elif current_resource_type == "seed":
          # For seeds, recommend seeding + downstream
          if len(models_affected) == 0:
              recommendation = f"dbt seed -s {resource_name} && dbt test -s {resource_name}"
          else:
              recommendation = f"dbt seed -s {resource_name} && dbt run -s {resource_name}+"
      else:
          # For models, snapshots, etc.
          if len(models_affected) == 0:
              recommendation = f"dbt run -s {resource_name}"
          else:
              recommendation = f"dbt run -s {resource_name}+"
  
      # Build result
      result: dict[str, Any] = {
          "resource": {
              "name": resource_name,
              "unique_id": unique_id,
              "resource_type": current_resource_type,
              "package_name": resource.get("package_name"),
          },
          "impact": {
              "models_affected": models_affected_sorted,
              "models_affected_count": len(models_affected),
              "tests_affected_count": len(tests_affected),
              "other_affected_count": len(other_affected),
              "total_affected": len(downstream),
          },
          "affected_by_distance": affected_by_distance,
          "recommendation": recommendation,
      }
  
      # Add helpful message based on impact size
      if len(models_affected) == 0:
          result["message"] = "No downstream models affected. Only this resource needs to be run/tested."
      elif len(models_affected) <= 3:
          result["message"] = f"Low impact: {len(models_affected)} downstream model(s) affected."
      elif len(models_affected) <= 10:
          result["message"] = f"Medium impact: {len(models_affected)} downstream models affected."
      else:
          result["message"] = f"High impact: {len(models_affected)} downstream models affected. Consider incremental changes."
  
      return result

• src/dbt_core_mcp/server.py:161-182 (registration)

  Tools are dynamically registered in _register_tools() via discover_tools_in_package which scans all modules in dbt_core_mcp.tools for @dbtTool()-decorated functions.

  def _register_tools(self) -> None:
      """Dynamically register all dbt Core MCP tools."""
      from .tools import discover_tools_in_package, get_tool_metadata
  
      tool_functions = discover_tools_in_package("dbt_core_mcp.tools")
      for tool_func in tool_functions:
          metadata = get_tool_metadata(tool_func, default=None)
          if metadata:
              allowed_keys = {
                  "name",
                  "description",
                  "tags",
                  "enabled",
                  "icons",
                  "annotations",
                  "meta",
              }
              tool_kwargs = {key: value for key, value in metadata.items() if key in allowed_keys}
              self.app.tool(**tool_kwargs)(tool_func)
              logger.info("Registered tool metadata for %s: %s", tool_func.__name__, metadata)
          else:
              self.app.tool()(tool_func)

• src/dbt_core_mcp/tools/__init__.py:17-17 (helper)

  The dbtTool() decorator used to mark the analyze_impact function as a discoverable MCP tool.

  def dbtTool(**metadata: Any) -> Callable[[F], F]: