[POC] MCP Apps: interactive lineage via get_lineage

.tool-versions

@@ -1,4 +1,4 @@
-nodejs 20.17.0
+nodejs 22.18.0
 uv 0.11.8
 task 3.43.2
 pnpm 10.15.1

src/dbt_mcp/apps/register.py

@@ -0,0 +1,30 @@
+import httpx
+from mcp.server.fastmcp import FastMCP
+
+from dbt_mcp.config.config import AppsConfig
+
+
+def register_app_resource(
+    dbt_mcp: FastMCP,
+    config: AppsConfig,
+    *,
+    app_name: str,
+) -> None:
+    """Register a ``ui://`` resource that serves an MCP App's single-file HTML.
+
+    The app is built as a self-contained ``index.html`` (all JS/CSS inlined) and
+    published to the CDN at ``<cdn_base>/<app_name>/index.html``. On read, the
+    server fetches that HTML and returns it directly, so the host renders a fully
+    self-contained app with no further external requests.
+    """
+    uri = f"ui://dbt-mcp/{app_name}"
+    app_url = f"{config.cdn_base}/{app_name}/index.html"
+
+    @dbt_mcp.resource(
+        uri=uri,
+        name=app_name,
+        mime_type="text/html;profile=mcp-app",
+    )
+    def get_app_ui() -> str:
+        with httpx.Client() as client:
+            return client.get(app_url).raise_for_status().text

src/dbt_mcp/config/config.py

@@ -80,6 +80,11 @@ class LspConfig:
     lsp_client_provider: LocalLSPClientProvider
 
 
+@dataclass
+class AppsConfig:
+    cdn_base: str
+
+
 @dataclass
 class Config:
     disable_tools: list[ToolName]
@@ -98,6 +103,7 @@ class Config:
     admin_api_config_provider: DefaultAdminApiConfigProvider
     credentials_provider: CredentialsProvider
     lsp_config: LspConfig | None
+    apps_config: AppsConfig
     # Lazy: invoking the provider runs `dbt --version`, which can take several
     # seconds when many adapters are installed. Resolution is deferred until a
     # product_docs tool actually needs the version, then cached for the
@@ -260,5 +266,6 @@ def load_config(enable_proxied_tools: bool = True) -> Config:
         admin_api_config_provider=admin_api_config_provider,
         credentials_provider=credentials_provider,
         lsp_config=lsp_config,
+        apps_config=AppsConfig(cdn_base=settings.cdn_base),
         dbt_version_provider=dbt_version_provider,
     )

src/dbt_mcp/config/settings.py

@@ -100,6 +100,12 @@ class DbtMcpSettings(BaseSettings):
         False, alias="DBT_MCP_ENABLE_MCP_SERVER_METADATA"
     )
 
+    # MCP Apps settings
+    cdn_base: str = Field(
+        "https://cloud-ui.cdn.getdbt.com/dbt-ui/mcp-apps",
+        alias="DBT_MCP_CDN_BASE",
+    )
+
     # Tracking settings
     do_not_track: str | None = Field(None, alias="DO_NOT_TRACK")
     send_anonymous_usage_data: str | None = Field(

src/dbt_mcp/contract/snapshot.py

@@ -269,16 +269,28 @@ async def generate_snapshot() -> ContractSnapshot:
             resources = []
         for resource in resources:
             uri = str(resource.uri)
+            mime_type = getattr(resource, "mimeType", None)
+            # MCP App UIs are released independently on a CDN (owned by the
+            # frontend repo), so their rendered content is intentionally
+            # out-of-contract: hashing it here would couple this repo to bytes it
+            # does not own and flag every independent UI release as a contract
+            # change. We still guard the resource's existence and shape
+            # (uri/name/mime/_meta); the server<->app interface is guarded by the
+            # linked tool's input/output schema and _meta.resourceUri.
+            is_mcp_app = bool(mime_type) and "profile=mcp-app" in mime_type
+            content_sha256 = (
+                None if is_mcp_app else await _read_resource_hash(dbt_mcp, uri)
+            )
             resource_contracts.append(
                 ResourceContract(
                     uri=uri,
                     name=getattr(resource, "name", None),
-                    mime_type=getattr(resource, "mimeType", None),
+                    mime_type=mime_type,
                     meta=_normalize(
                         getattr(resource, "meta", None)
                         or getattr(resource, "_meta", None)
                     ),
-                    content_sha256=await _read_resource_hash(dbt_mcp, uri),
+                    content_sha256=content_sha256,
                 )
             )
 

src/dbt_mcp/discovery/tools.py

@@ -3,7 +3,7 @@
 from typing import Annotated
 
 from mcp.server.fastmcp import FastMCP
-from pydantic import Field
+from pydantic import BaseModel, Field
 
 from dbt_mcp.config.config_providers import ConfigProvider, DiscoveryConfig
 from dbt_mcp.discovery.client import (
@@ -246,23 +246,61 @@ async def get_model_performance(
     )
 
 
+class LineageNode(BaseModel):
+    unique_id: str
+    name: str
+    resource_type: str
+
+
+class LineageEdge(BaseModel):
+    source: str
+    target: str
+
+
+class LineageGraph(BaseModel):
+    type: str = "lineage_graph"
+    root_id: str
+    nodes: list[LineageNode]
+    edges: list[LineageEdge]
+
+
 @dbt_mcp_tool(
     description=get_prompt("discovery/get_lineage"),
     title="Get Lineage",
     read_only_hint=True,
     destructive_hint=False,
     idempotent_hint=True,
+    structured_output=True,
+    meta={"ui": {"resourceUri": "ui://dbt-mcp/get-lineage"}},
 )
 async def get_lineage(
     context: DiscoveryToolContext,
     unique_id: str = UNIQUE_ID_REQUIRED_FIELD,
     types: list[LineageResourceType] | None = TYPES_FIELD,
     depth: int = DEPTH_FIELD,
-) -> list[dict]:
+) -> LineageGraph:
     config = await context.config_provider.get_config()
-    return await context.lineage_fetcher.fetch_lineage(
+    nodes = await context.lineage_fetcher.fetch_lineage(
         unique_id=unique_id, types=types, depth=depth, config=config
     )
+    node_ids = {n["uniqueId"] for n in nodes}
+    return LineageGraph(
+        root_id=unique_id,
+        nodes=[
+            LineageNode(
+                unique_id=n["uniqueId"],
+                name=n["name"],
+                resource_type=n["resourceType"],
+            )
+            for n in nodes
+        ],
+        edges=[
+            LineageEdge(source=parent_id, target=n["uniqueId"])
+            for n in nodes
+            for parent_id in n.get("parentIds", [])
+            if parent_id in node_ids
+        ],
+    )
 
 
 @dbt_mcp_tool(

src/dbt_mcp/mcp/server.py

@@ -15,6 +15,7 @@
 from dbt_mcp.dbt_admin.tools import register_admin_api_tools
 from dbt_mcp.dbt_cli.tools import register_dbt_cli_tools
 from dbt_mcp.dbt_codegen.tools import register_dbt_codegen_tools
+from dbt_mcp.apps.register import register_app_resource
 from dbt_mcp.discovery.tools import register_discovery_tools
 from dbt_mcp.discovery.tools_multiproject import register_multiproject_discovery_tools
 from dbt_mcp.errors.common import MissingHostError
@@ -381,4 +382,7 @@ async def create_dbt_mcp(config: Config) -> FastMCP:
         multi_project_mcp=multi_project_dbt_mcp,
         single_project_mcp=single_project_dbt_mcp,
     )
+    # MCP App UI resources are served by the dispatcher itself (it only routes
+    # tool calls, not resources). The bundle is fetched from the CDN on read.
+    register_app_resource(tool_dispatcher, config.apps_config, app_name="get-lineage")
     return tool_dispatcher

src/dbt_mcp/tools/fields.py

@@ -27,7 +27,7 @@
 )
 
 DEPTH_FIELD = Field(
-    default=5,
+    default=2,
     description="The depth of the lineage graph to return. "
     "Controls how many levels to traverse from the target node."
     "A depth of 1 returns only direct parents/children."

tests/mocks/config.py

@@ -1,4 +1,5 @@
 from dbt_mcp.config.config import (
+    AppsConfig,
     Config,
     DbtCliConfig,
     DbtCodegenConfig,
@@ -177,6 +178,7 @@ async def get_credentials(self):
     semantic_layer_config_provider=MockSemanticLayerConfigProvider(),
     admin_api_config_provider=MockAdminApiConfigProvider(),
     lsp_config=mock_lsp_config,
+    apps_config=AppsConfig(cdn_base="https://example.test/mcp-apps"),
     disable_tools=[],
     enable_tools=None,  # None means not set, [] would mean allowlist mode
     disabled_toolsets=set(),

tests/unit/contract/contract_snapshot.json

@@ -12,7 +12,15 @@
     "product_docs",
     "semantic_layer"
   ],
-  "resources": [],
+  "resources": [
+    {
+      "content_sha256": null,
+      "meta": null,
+      "mime_type": "text/html;profile=mcp-app",
+      "name": "get-lineage",
+      "uri": "ui://dbt-mcp/get-lineage"
+    }
+  ],
   "server_instructions": "Use these tools to interact with dbt resources (typically via dbt Platform):\n\n- Understand how the project is structured: what exists in the environment, how objects depend on one another, and where to look when something looks wrong or slow.\n- Reason over governed business meaning -- named measures and breakdowns the project maintains -- and answer questions with validated aggregates or the warehouse logic behind them when that helps.\n- Work with platform automation when available: see what runs on a schedule, inspect past outcomes, act on failures or reruns, and dig into logs and outputs.\n- Assist with engineering work on dbt projects: take action on the project and reason about the underlying SQL.\n\nExample data-oriented questions (users may not use dbt vocabulary):\n\n- Revenue, ARR, bookings, pipeline, or quota: \"how are we doing this quarter?\"; \"split it by region or product\"; \"this total doesn't match finance.\"\n- Customers, users, signups, or retention: \"how many active ...?\" \"is churn getting worse?\" \"cohorts or segments -- whatever we already track.\"\n- Orders, inventory, SKUs, or fulfillment: \"what's selling?\" \"stock or backorder questions\" when they only describe the business problem.\n- Funnel, marketing, or web activity: \"conversion from visit to purchase\" \"campaign performance\" with loose wording and no metric names.\n- Trust and definitions: \"where does this dashboard number come from?\" \"two reports disagree -- help me find why\" \"what's the official definition of ...?\"\n- Quality and freshness: \"is this table up to date?\" \"missing yesterday's data\" \"something looks stale in our reporting.\"\n- Orchestration tied to data: \"did last night's refresh finish?\" \"prod load failed and Finance is blocked\" without run or job IDs.\n- Engineering on a dbt project: \"help me with this model,\" \"what's wrong with my project?\"",
   "tools": [
     {
@@ -834,7 +842,7 @@
         },
         "properties": {
           "depth": {
-            "default": 5,
+            "default": 2,
             "description": "The depth of the lineage graph to return. Controls how many levels to traverse from the target node.A depth of 1 returns only direct parents/children.A depth of 0 returns the entire lineage graph.",
             "title": "Depth",
             "type": "integer"
@@ -867,23 +875,87 @@
         "title": "get_lineageArguments",
         "type": "object"
       },
-      "meta": null,
+      "meta": {
+        "ui": {
+          "resourceUri": "ui://dbt-mcp/get-lineage"
+        }
+      },
       "name": "get_lineage",
       "output_schema": {
+        "$defs": {
+          "LineageEdge": {
+            "properties": {
+              "source": {
+                "title": "Source",
+                "type": "string"
+              },
+              "target": {
+                "title": "Target",
+                "type": "string"
+              }
+            },
+            "required": [
+              "source",
+              "target"
+            ],
+            "title": "LineageEdge",
+            "type": "object"
+          },
+          "LineageNode": {
+            "properties": {
+              "name": {
+                "title": "Name",
+                "type": "string"
+              },
+              "resource_type": {
+                "title": "Resource Type",
+                "type": "string"
+              },
+              "unique_id": {
+                "title": "Unique Id",
+                "type": "string"
+              }
+            },
+            "required": [
+              "unique_id",
+              "name",
+              "resource_type"
+            ],
+            "title": "LineageNode",
+            "type": "object"
+          }
+        },
         "properties": {
-          "result": {
+          "edges": {
             "items": {
-              "additionalProperties": true,
-              "type": "object"
+              "$ref": "#/$defs/LineageEdge"
             },
-            "title": "Result",
+            "title": "Edges",
+            "type": "array"
+          },
+          "nodes": {
+            "items": {
+              "$ref": "#/$defs/LineageNode"
+            },
+            "title": "Nodes",
             "type": "array"
+          },
+          "root_id": {
+            "title": "Root Id",
+            "type": "string"
+          },
+          "type": {
+            "default": "lineage_graph",
+            "title": "Type",
+            "type": "string"
           }
         },
         "required": [
-          "result"
+          "root_id",
+          "nodes",
+          "edges"
         ],
-        "title": "get_lineageOutput",
+        "title": "LineageGraph",
         "type": "object"
       },
       "title": "Get Lineage",

tests/unit/contract/test_contract_snapshot.py

@@ -80,6 +80,25 @@ async def test_committed_snapshot_passes_claude_connector_lint():
     assert lint_claude_connector(snapshot) == []
 
 
+async def test_mcp_app_resource_content_is_out_of_contract(monkeypatch):
+    """MCP App UIs are released independently on the CDN, so their rendered
+    content is not hashed into the contract -- and generating the snapshot must
+    not fetch it over the network."""
+    import dbt_mcp.contract.snapshot as snapshot_mod
+
+    async def _fail(*_args, **_kwargs):
+        raise AssertionError("mcp-app resource content must not be fetched")
+
+    monkeypatch.setattr(snapshot_mod, "_read_resource_hash", _fail)
+
+    snapshot = await generate_snapshot()
+    app_resources = [
+        r for r in snapshot.resources if r.mime_type and "profile=mcp-app" in r.mime_type
+    ]
+    assert app_resources, "expected at least one mcp-app resource (get-lineage)"
+    assert all(r.content_sha256 is None for r in app_resources)
+
+
 def test_classify_no_change():
     snap = _snap([_tool("a")])
     assert classify_change(snap, snap) == ("none", [])