implement get_lineage tool for complete lineage across dbt resources (#502)

## Summary

A User building a catalog-like experience with dbt and Snowflake via
Claude Desktop requested full lineage capabilities
#110:

> "Get the full lineage of a model, not only its children/parents. To
parse the full lineage with the official MCP it would require quite a
few calls to the tool to parse through all layers, which would be nice
to get the server doing async instead providing the full list as a
return."

---


## What Changed

Added a new `get_lineage` tool that enables efficient lineage traversal
across all dbt resource types using the Discovery API's `lineage()`
endpoint.

### Key Implementation Details

**Query Approach:**
- Uses Discovery API's `lineage()` endpoint to fetch the entire lineage
graph in a single API call
- Requests minimal fields: `name`, `uniqueId`, `resourceType`, and
`parentIds`
- Pre-builds a children adjacency map for  descendant lookups

**Why Not Use Selector Syntax?**

The selector parameters (`select: "model+"`, `exclude: "test_*"`) does
not seem to be available in the public API:

## Test Case
<img width="1264" height="783" alt="image"
src="https://github.com/user-attachments/assets/ee045546-9979-48e2-b31a-7780f85c81e1"
/>


Search for any exposures:
<img width="1264" height="833" alt="image"
src="https://github.com/user-attachments/assets/8ba63f3d-cc99-4e7f-a63a-811881a6f847"
/>


## Why

Implementing to close #110

## Related Issues
<!-- Link any related issues using #issue_number -->
Closes ##110
Related to ##110


## Checklist
- [x] I have performed a self-review of my code
- [x] I have made corresponding changes to the documentation (in
https://github.com/dbt-labs/docs.getdbt.com) if required -- Mention it
here dbt-labs/docs.getdbt.com#8233
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes

## Additional Notes
<!-- Any additional information that would be helpful for reviewers -->






## Related Issues
<!-- Link any related issues using #issue_number -->
Closes #
Related to #


## Checklist
- [x] I have performed a self-review of my code
- [x] I have made corresponding changes to the documentation (in
https://github.com/dbt-labs/docs.getdbt.com) if required -- Mention it
here dbt-labs/docs.getdbt.com#8233
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes

## Additional Notes
<!-- Any additional information that would be helpful for reviewers -->

---------

Co-authored-by: Devon Fulcher <24593113+DevonFulcher@users.noreply.github.com>

Author: Muizzkolapo

.changes/unreleased/Enhancement or New Feature-20260107-133417.yaml

@@ -0,0 +1,3 @@
+kind: Enhancement or New Feature
+body: Adding get_lineage discovery mcp tool.
+time: 2026-01-07T13:34:17.425226Z

README.md

@@ -35,6 +35,7 @@ The dbt MCP server architecture allows for your agent to connect to a variety of
 - `get_all_sources`
 - `get_exposure_details`
 - `get_exposures`
+- `get_lineage`
 - `get_macro_details`
 - `get_mart_models`
 - `get_model_children`

src/dbt_mcp/discovery/client.py

@@ -8,7 +8,7 @@
 
 from dbt_mcp.config.config_providers import ConfigProvider, DiscoveryConfig
 from dbt_mcp.discovery.graphql import load_query
-from dbt_mcp.errors import InvalidParameterError
+from dbt_mcp.errors import InvalidParameterError, ToolCallError
 from dbt_mcp.gql.errors import raise_gql_error
 
 DEFAULT_PAGE_SIZE = 100
@@ -302,6 +302,9 @@ class GraphQLQueries:
         }
     """)
 
+    # Lineage query
+    GET_FULL_LINEAGE = load_query("get_full_lineage.gql")
+
 
 class MetadataAPIClient:
     def __init__(self, config_provider: ConfigProvider[DiscoveryConfig]):
@@ -667,3 +670,121 @@ async def fetch_details(
         if not edges:
             return []
         return [e["node"] for e in edges]
+
+
+class LineageResourceType(StrEnum):
+    """Resource types supported by the lineage API."""
+
+    MODEL = "Model"
+    SOURCE = "Source"
+    SEED = "Seed"
+    SNAPSHOT = "Snapshot"
+    EXPOSURE = "Exposure"
+    METRIC = "Metric"
+    SEMANTIC_MODEL = "SemanticModel"
+    SAVED_QUERY = "SavedQuery"
+    TEST = "Test"
+
+
+class LineageFetcher:
+    """Fetcher for lineage data. Returns nodes connected to the target."""
+
+    def __init__(self, api_client: MetadataAPIClient):
+        self.api_client = api_client
+
+    async def fetch_lineage(
+        self,
+        unique_id: str,
+        depth: int,
+        types: list[LineageResourceType] | None = None,
+    ) -> list[dict]:
+        """Fetch lineage graph filtered to nodes connected to unique_id.
+
+        Args:
+            unique_id: The dbt unique ID of the resource to get lineage for.
+            types: List of resource types to include. If None, includes all types.
+
+        Returns:
+            List of nodes connected to unique_id (upstream + downstream).
+        """
+        if depth <= 0:
+            raise ToolCallError("Depth must be greater than 0")
+        config = await self.api_client.config_provider.get_config()
+        type_filter = [
+            t.value for t in (types if types is not None else LineageResourceType)
+        ]
+        variables = {
+            "environmentId": config.environment_id,
+            "types": type_filter,
+            # uniqueId removed - not used by GraphQL
+        }
+
+        result = await self.api_client.execute_query(
+            GraphQLQueries.GET_FULL_LINEAGE, variables
+        )
+        raise_gql_error(result)
+
+        all_nodes = (
+            result.get("data", {})
+            .get("environment", {})
+            .get("applied", {})
+            .get("lineage", [])
+        )
+
+        # Filter to connected nodes only
+        return self._filter_connected_nodes(all_nodes, unique_id, depth)
+
+    def _filter_connected_nodes(
+        self, nodes: list[dict], target_id: str, depth: int
+    ) -> list[dict]:
+        """Return only nodes connected to target_id (upstream and downstream).
+
+        Uses BFS to find all nodes reachable from target in both directions.
+        """
+        node_map = {
+            n["uniqueId"]: n
+            for n in nodes
+            if (resource_type := n.get("resourceType"))
+            and isinstance(resource_type, str)
+            # Filtering out macros because they have large
+            # dependency graphs that aren't always useful.
+            and resource_type.strip().lower() != "macro"
+        }
+
+        if target_id not in node_map:
+            return []
+
+        # BFS to find all connected nodes
+        connected = {target_id}
+        queue = [(target_id, 0)]
+
+        while queue:
+            current_id, current_depth = queue.pop(0)
+            node = node_map.get(current_id)
+            if not node:
+                continue
+
+            # Stop traversing beyond the depth limit
+            if current_depth >= depth:
+                continue
+
+            # Traverse upstream (parents)
+            for parent_id in node.get("parentIds", []):
+                if parent_id not in connected and parent_id in node_map:
+                    connected.add(parent_id)
+                    queue.append((parent_id, current_depth + 1))
+
+            # Traverse downstream (children)
+            for candidate in nodes:
+                candidate_id = candidate.get("uniqueId")
+                if not candidate_id or candidate_id not in node_map:
+                    continue
+                if (
+                    current_id in candidate.get("parentIds", [])
+                    and candidate_id not in connected
+                ):
+                    connected.add(candidate_id)
+                    queue.append((candidate_id, current_depth + 1))
+
+        # Return in original order
+        return [node_map[uid] for uid in connected]

src/dbt_mcp/discovery/graphql/get_full_lineage.gql

@@ -0,0 +1,14 @@
+query GetFullLineage($environmentId: BigInt!, $types: [ResourceNodeType!]!) {
+  environment(id: $environmentId) {
+    applied {
+      lineage(filter: { types: $types }) {
+        name
+        uniqueId
+        resourceType
+        ... on LineageNodeWithParents {
+          parentIds
+        }
+      }
+    }
+  }
+}

src/dbt_mcp/discovery/tools.py

@@ -8,6 +8,8 @@
 from dbt_mcp.discovery.client import (
     AppliedResourceType,
     ExposuresFetcher,
+    LineageFetcher,
+    LineageResourceType,
     MetadataAPIClient,
     ModelsFetcher,
     PaginatedResourceFetcher,
@@ -35,6 +37,13 @@
     "This is not required if `unique_id` is provided. "
     "Only use name when `unique_id` is unknown.",
 )
+TYPES_FIELD = Field(
+    default=None,
+    description="List of resource types to include in lineage results. "
+    "If not provided, includes all types. "
+    "Valid types: Model, Source, Seed, Snapshot, Exposure, Metric, SemanticModel, SavedQuery, Test.",
+)
+DEPTH_FIELD = Field(default=5, description="The depth of the lineage graph to return.")
 
 
 @dataclass
@@ -43,6 +52,7 @@ class DiscoveryToolContext:
     exposures_fetcher: ExposuresFetcher
     sources_fetcher: SourcesFetcher
     resource_details_fetcher: ResourceDetailsFetcher
+    lineage_fetcher: LineageFetcher
 
     def __init__(self, config_provider: ConfigProvider[DiscoveryConfig]):
         api_client = MetadataAPIClient(config_provider=config_provider)
@@ -83,6 +93,7 @@ def __init__(self, config_provider: ConfigProvider[DiscoveryConfig]):
             ),
         )
         self.resource_details_fetcher = ResourceDetailsFetcher(api_client=api_client)
+        self.lineage_fetcher = LineageFetcher(api_client=api_client)
 
 
 @dbt_mcp_tool(
@@ -174,6 +185,24 @@ async def get_model_health(
     return await context.models_fetcher.fetch_model_health(name, unique_id)
 
 
+@dbt_mcp_tool(
+    description=get_prompt("discovery/get_lineage"),
+    title="Get Lineage",
+    read_only_hint=True,
+    destructive_hint=False,
+    idempotent_hint=True,
+)
+async def get_lineage(
+    context: DiscoveryToolContext,
+    unique_id: str,
+    types: list[LineageResourceType] | None = TYPES_FIELD,
+    depth: int = DEPTH_FIELD,
+) -> list[dict]:
+    return await context.lineage_fetcher.fetch_lineage(
+        unique_id=unique_id, types=types, depth=depth
+    )
+
+
 @dbt_mcp_tool(
     description=get_prompt("discovery/get_exposures"),
     title="Get Exposures",
@@ -340,6 +369,7 @@ async def get_test_details(
     get_model_parents,
     get_model_children,
     get_model_health,
+    get_lineage,
     get_exposures,
     get_exposure_details,
     get_all_sources,

src/dbt_mcp/prompts/discovery/get_lineage.md

@@ -0,0 +1,99 @@
+Retrieves the lineage graph for a dbt resource.
+
+Returns all nodes connected to the specified resource, including both upstream dependencies (ancestors) and downstream dependents (descendants).
+
+**Parameters:**
+- `unique_id`: **Required** - Full unique ID of the resource (e.g., "model.my_project.customers")
+- `types`: *Optional* - List of resource types to include in results. If not provided, includes all types.
+  - Valid types: `Model`, `Source`, `Seed`, `Snapshot`, `Exposure`, `Metric`, `SemanticModel`, `SavedQuery`, `Test`
+- `depth`: *Optional* - The depth of the lineage graph to return (default: 5). Controls how many levels upstream and downstream to traverse from the target node.
+
+**Returns:**
+A list of all nodes in the connected subgraph, where each node contains:
+- `uniqueId`: The resource's unique identifier
+- `name`: The resource name
+- `resourceType`: The type of resource (Model, Source, etc.)
+- `parentIds`: List of unique IDs that this resource directly depends on
+
+**Example Response:**
+```json
+[
+  {
+    "uniqueId": "source.raw.users",
+    "name": "users",
+    "resourceType": "Source",
+    "parentIds": []
+  },
+  {
+    "uniqueId": "model.stg_customers",
+    "name": "stg_customers",
+    "resourceType": "Model",
+    "parentIds": ["source.raw.users"]
+  },
+  {
+    "uniqueId": "model.customers",
+    "name": "customers",
+    "resourceType": "Model",
+    "parentIds": ["model.stg_customers"]
+  }
+]
+```
+
+**Usage Examples:**
+```python
+# Get complete lineage (all connected nodes, all types, default depth of 5)
+get_lineage(unique_id="model.analytics.customers")
+
+# Get lineage filtered to only models and sources
+get_lineage(unique_id="model.analytics.customers", types=["Model", "Source"])
+
+# Get only immediate neighbors (depth=1)
+get_lineage(unique_id="model.analytics.customers", depth=1)
+
+# Get deeper lineage for comprehensive analysis
+get_lineage(unique_id="model.analytics.customers", depth=10)
+```
+
+**Traversing the Graph:**
+
+The graph is represented by parent-child relationships. To navigate:
+
+**Finding Upstream Dependencies (Parents):**
+```python
+# Direct: What does this node depend on?
+target_node = find_node_by_id(result, "model.customers")
+direct_parents = target_node["parentIds"]
+# Result: ["model.stg_customers"]
+```
+
+**Finding Downstream Dependents (Children):**
+```python
+# Search: What depends on this node?
+target_id = "model.customers"
+direct_children = [
+    node for node in result
+    if target_id in node.get("parentIds", [])
+]
+# Result: nodes that list "model.customers" in their parentIds
+```
+
+**Understanding the Results:**
+
+- The target node is always included in the response
+- All returned nodes are connected to the target (no disconnected nodes)
+- To get full lineage, omit the `types` parameter
+- To reduce payload size, specify relevant `types`
+
+**Common Use Cases:**
+
+1. **Impact Analysis**: "What will break if I change this model?"
+   - Look at downstream dependents (nodes that have target in `parentIds`)
+
+2. **Dependency Tracking**: "What does this model depend on?"
+   - Look at upstream dependencies (`parentIds` of the target node)
+
+3. **Data Lineage**: "Show the complete data flow for this entity"
+   - Use all returned nodes to build a complete graph
+
+4. **Finding Tests**: "What tests exist for this model and its dependencies?"
+   - Filter results where `resourceType == "Test"`

src/dbt_mcp/tools/tool_names.py

@@ -30,6 +30,7 @@ class ToolName(Enum):
     GET_MODEL_PARENTS = "get_model_parents"
     GET_MODEL_CHILDREN = "get_model_children"
     GET_MODEL_HEALTH = "get_model_health"
+    GET_LINEAGE = "get_lineage"
     GET_ALL_SOURCES = "get_all_sources"
     GET_SOURCE_DETAILS = "get_source_details"
     GET_EXPOSURES = "get_exposures"

src/dbt_mcp/tools/toolsets.py

@@ -61,6 +61,7 @@ class Toolset(Enum):
         ToolName.GET_MODEL_PARENTS,
         ToolName.GET_MODEL_CHILDREN,
         ToolName.GET_MODEL_HEALTH,
+        ToolName.GET_LINEAGE,
         ToolName.GET_ALL_SOURCES,
         ToolName.GET_SOURCE_DETAILS,
         ToolName.GET_EXPOSURES,