DBT Core MCP Server

Instructions

Implementation Reference

• src/dbt_core_mcp/dbt/manifest.py:573-719 (handler)

  Core handler implementing the analyze_impact tool logic: locates the target resource, traverses all downstream dependencies using the manifest's child_map, categorizes affected models/tests/others by distance, sorts by distance/name, generates type-specific dbt run recommendations, and provides impact assessment message.

  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:1434-1477 (registration)

  MCP tool registration for 'analyze_impact' using FastMCP @app.tool() decorator. Defines input schema (name: str required, resource_type: str optional), comprehensive docstring with usage examples, ensures server initialization, and delegates to toolImpl_analyze_impact.

  async def analyze_impact(
      ctx: Context,
      name: str,
      resource_type: str | None = None,
  ) -> 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
      """
      await self._ensure_initialized_with_context(ctx)
      return await self.toolImpl_analyze_impact(name, resource_type)

• src/dbt_core_mcp/server.py:605-610 (handler)

  Server-side toolImpl handler that initializes dbt components if needed and delegates core logic to ManifestLoader.analyze_impact, with error wrapping.

  async def toolImpl_analyze_impact(self, name: str, resource_type: str | None = None) -> dict[str, Any]:
      """Implementation for analyze_impact tool."""
      try:
          return self.manifest.analyze_impact(name, resource_type)  # type: ignore
      except ValueError as e:
          raise ValueError(f"Impact analysis error: {e}")

• src/dbt_core_mcp/dbt/manifest.py:720-770 (helper)

  Recursive helper function to compute all downstream dependents from the manifest's child_map, tracking distance and avoiding cycles with seen set. Critical for impact analysis.

  def get_downstream_nodes(self, unique_id: str, max_depth: int | None = None, current_depth: int = 0) -> list[dict[str, Any]]:
      """Get all downstream dependents of a node recursively.
  
      Args:
          unique_id: The unique identifier of the node
          max_depth: Maximum depth to traverse (None for unlimited)
          current_depth: Current recursion depth (internal use)
  
      Returns:
          List of dictionaries with downstream node info:
          {"unique_id": str, "name": str, "type": str, "distance": int}
      """
      if not self._manifest:
          raise RuntimeError("Manifest not loaded. Call load() first.")
  
      if max_depth is not None and current_depth >= max_depth:
          return []
  
      child_map = self._manifest.get("child_map", {})
      children = child_map.get(unique_id, [])
  
      downstream: list[dict[str, Any]] = []
      seen: set[str] = set()
  
      for child_id in children:
          if child_id in seen:
              continue
          seen.add(child_id)
  
          node = self.get_node_by_unique_id(child_id)
          if node:
              resource_type = node.get("resource_type", "unknown")
              downstream.append(
                  {
                      "unique_id": child_id,
                      "name": node.get("name", ""),
                      "type": resource_type,
                      "distance": current_depth + 1,
                  }
              )
  
              # Recurse
              if max_depth is None or current_depth + 1 < max_depth:
                  grandchildren = self.get_downstream_nodes(child_id, max_depth, current_depth + 1)
                  for gc in grandchildren:
                      if gc["unique_id"] not in seen:
                          seen.add(str(gc["unique_id"]))
                          downstream.append(gc)
  
      return downstream

• src/dbt_core_mcp/dbt/manifest.py:213-303 (helper)

  Helper to locate target resource by name across all types (models/sources/etc.), supports source_name.table format and table_name fallback, returns multiple_matches dict if ambiguous for LLM handling.

  def get_resource_node(self, name: str, resource_type: str | None = None) -> dict[str, Any]:
      """
      Get a resource node by name with auto-detection across all resource types.
  
      This method searches for resources across models, sources, seeds, snapshots, tests, etc.
      Designed for LLM consumption - returns all matches when ambiguous rather than raising errors.
  
      Args:
          name: Resource name. For sources, can be "source_name.table_name" or just "table_name"
          resource_type: Optional filter (model, source, seed, snapshot, test, analysis).
                        If None, searches all types.
  
      Returns:
          Single resource dict if exactly one match found, or dict with multiple_matches=True
          containing all matching resources for LLM to process.
  
      Raises:
          RuntimeError: If manifest not loaded
          ValueError: If resource not found (only case that raises)
  
      Examples:
          get_resource_node("customers") -> single model dict
          get_resource_node("customers", "source") -> single source dict
          get_resource_node("customers") with multiple matches -> {"multiple_matches": True, ...}
      """
      if not self._manifest:
          raise RuntimeError("Manifest not loaded. Call load() first.")
  
      # Validate resource_type if provided
      valid_types = {"model", "source", "seed", "snapshot", "test", "analysis"}
      if resource_type is not None and resource_type not in valid_types:
          raise ValueError(f"Invalid resource_type '{resource_type}'. Must be one of: {', '.join(sorted(valid_types))}")
  
      matches: list[dict[str, Any]] = []
  
      # For sources, try "source_name.table_name" format first
      if "." in name and (resource_type is None or resource_type == "source"):
          parts = name.split(".", 1)
          if len(parts) == 2:
              # Search sources dict directly
              sources_dict = self._manifest.get("sources", {})
              for _, source in sources_dict.items():
                  if isinstance(source, dict) and source.get("source_name") == parts[0] and source.get("name") == parts[1]:
                      matches.append(dict(source))
                      break
  
      # Search nodes (models, tests, snapshots, seeds, analyses, etc.)
      nodes = self._manifest.get("nodes", {})
      for unique_id, node in nodes.items():
          if not isinstance(node, dict):
              continue
  
          node_type = node.get("resource_type")
          node_name = node.get("name")
  
          # Type filter if specified
          if resource_type is not None and node_type != resource_type:
              continue
  
          if node_name == name:
              matches.append(dict(node))
  
      # Search sources by table name only (fallback when no dot in name)
      if resource_type is None or resource_type == "source":
          sources = self._manifest.get("sources", {})
          for unique_id, source in sources.items():
              if not isinstance(source, dict):
                  continue
  
              if source.get("name") == name:
                  # Avoid duplicates if already matched via source_name.table_name
                  if not any(m.get("unique_id") == unique_id for m in matches):
                      matches.append(dict(source))
  
      # Handle results based on match count
      if len(matches) == 0:
          type_hint = f" of type '{resource_type}'" if resource_type else ""
          raise ValueError(f"Resource '{name}'{type_hint} not found in manifest")
      elif len(matches) == 1:
          # Single match - return the resource directly
          return matches[0]
      else:
          # Multiple matches - return all with metadata for LLM to process
          return {
              "multiple_matches": True,
              "name": name,
              "match_count": len(matches),
              "matches": matches,
              "message": f"Found {len(matches)} resources named '{name}'. Returning all matches for context.",
          }