DBT Core MCP Server

Instructions

Implementation Reference

• src/dbt_core_mcp/server.py:1386-1432 (registration)

  FastMCP tool registration for 'get_lineage', defines input parameters serving as schema and calls toolImpl_get_lineage after initialization.
  async def get_lineage( ctx: Context, name: str, resource_type: str | None = None, direction: str = "both", depth: int | None = None, ) -> dict[str, Any]: """Get lineage (dependency tree) for any dbt resource with auto-detection. This unified tool works across all resource types (models, sources, seeds, snapshots, etc.) showing upstream and/or downstream dependencies with configurable depth. Args: name: Resource name. For sources, use "source_name.table_name" or just "table_name" Examples: "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) direction: Lineage direction: - "upstream": Show where data comes from (parents) - "downstream": Show what depends on this resource (children) - "both": Show full lineage (default) depth: Maximum levels to traverse (None for unlimited) - depth=1: Immediate dependencies only - depth=2: Dependencies + their dependencies - None: Full dependency tree Returns: Lineage information with upstream/downstream nodes and statistics. If multiple matches found, returns all matches for LLM to process. Raises: ValueError: If resource not found or invalid direction Examples: get_lineage("customers") -> auto-detect and show full lineage get_lineage("customers", "model", "upstream") -> where customers model gets data get_lineage("jaffle_shop.orders", "source", "downstream", 2) -> 2 levels of dependents """ await self._ensure_initialized_with_context(ctx) return await self.toolImpl_get_lineage(name, resource_type, direction, depth)
• src/dbt_core_mcp/server.py:598-604 (handler)

  Primary handler method for the get_lineage tool, delegates core logic to ManifestLoader.get_lineage and handles errors.
  async def toolImpl_get_lineage(self, name: str, resource_type: str | None = None, direction: str = "both", depth: int | None = None) -> dict[str, Any]: """Implementation for get_lineage tool.""" try: return self.manifest.get_lineage(name, resource_type, direction, depth) # type: ignore except ValueError as e: raise ValueError(f"Lineage error: {e}")
• src/dbt_core_mcp/dbt/manifest.py:472-572 (helper)

  Core lineage computation logic: finds resource by name/type, traverses upstream/downstream dependencies using dbt's parent_map/child_map with configurable depth, returns structured lineage with stats.
  def get_lineage( self, name: str, resource_type: str | None = None, direction: str = "both", depth: int | None = None, ) -> dict[str, Any]: """ Get lineage (dependency tree) for any resource type with auto-detection. This unified method works across all resource types (models, sources, seeds, etc.) and provides upstream, downstream, or bidirectional dependency traversal. 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. direction: Lineage direction: - "upstream": Show where data comes from (parents) - "downstream": Show what depends on this resource (children) - "both": Show full lineage (default) depth: Maximum levels to traverse (None for unlimited) - depth=1: Immediate dependencies only - depth=2: Dependencies + their dependencies - None: Full dependency tree Returns: Dictionary with lineage information: { "resource": {...}, # The target resource info "upstream": [...], # List of upstream dependencies (if direction in ["upstream", "both"]) "downstream": [...], # List of downstream dependents (if direction in ["downstream", "both"]) "stats": { "upstream_count": int, "downstream_count": int, "total_dependencies": int } } If multiple matches found, returns: {"multiple_matches": True, "matches": [...], "message": "..."} Raises: RuntimeError: If manifest not loaded ValueError: If resource not found or invalid direction Examples: get_lineage("customers") -> auto-detect and show full lineage get_lineage("customers", "model", "upstream") -> show where customers model gets data get_lineage("customers", direction="downstream", depth=2) -> 2 levels of dependents """ if not self._manifest: raise RuntimeError("Manifest not loaded. Call load() first.") # Validate direction valid_directions = {"upstream", "downstream", "both"} if direction not in valid_directions: raise ValueError(f"Invalid direction '{direction}'. Must be one of: {', '.join(sorted(valid_directions))}") # 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 lineage traversal unique_id = resource.get("unique_id") if not unique_id: raise ValueError(f"Resource '{name}' does not have a unique_id") # Build lineage based on direction result: dict[str, Any] = { "resource": { "name": resource.get("name"), "unique_id": unique_id, "resource_type": resource.get("resource_type"), "package_name": resource.get("package_name"), } } upstream: list[dict[str, Any]] = [] downstream: list[dict[str, Any]] = [] if direction in ("upstream", "both"): upstream = self.get_upstream_nodes(unique_id, max_depth=depth) result["upstream"] = upstream if direction in ("downstream", "both"): downstream = self.get_downstream_nodes(unique_id, max_depth=depth) result["downstream"] = downstream # Add statistics result["stats"] = { "upstream_count": len(upstream), "downstream_count": len(downstream), "total_dependencies": len(upstream) + len(downstream), } return result
• src/dbt_core_mcp/dbt/manifest.py:421-471 (helper)

  Supporting utility for recursive upstream (parents) dependency traversal from a node's unique_id.
  def get_upstream_nodes(self, unique_id: str, max_depth: int | None = None, current_depth: int = 0) -> list[dict[str, Any]]: """Get all upstream dependencies 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 upstream 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 [] parent_map = self._manifest.get("parent_map", {}) parents = parent_map.get(unique_id, []) upstream: list[dict[str, Any]] = [] seen: set[str] = set() for parent_id in parents: if parent_id in seen: continue seen.add(parent_id) node = self.get_node_by_unique_id(parent_id) if node: resource_type = node.get("resource_type", "unknown") upstream.append( { "unique_id": parent_id, "name": node.get("name", ""), "type": resource_type, "distance": current_depth + 1, } ) # Recurse if max_depth is None or current_depth + 1 < max_depth: grandparents = self.get_upstream_nodes(parent_id, max_depth, current_depth + 1) for gp in grandparents: if gp["unique_id"] not in seen: seen.add(str(gp["unique_id"])) upstream.append(gp) return upstream
• src/dbt_core_mcp/dbt/manifest.py:720-770 (helper)

  Supporting utility for recursive downstream (children) dependency traversal from a node's unique_id.
  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