用MCP构建数据血缘追踪Server：让AI成为企业数据的寻根大师

别再问数据对不对了：实战 MCP 数据血缘追踪 Server，让 AI 成为企业数据的「寻根大师」

早上十点，业务负责人拿着日报找到我：「日报里 GMV 怎么和财务差了两千万？这个数到底哪来的？」我打开调度平台，翻出凌晨 00:30 的跑批日志，顺着血缘链路一层一层往上追，查出是某个上游表在 0 点抽取时漏了两个小时的分区。全程花了四十分钟，而这样的「寻根问底」，数据团队每周都要重复好几次。

传统血缘平台不是没有，但它是「人找数」的工具：你得自己打开系统、搜表、看链路图、读加工逻辑，才能得出一个结论。真正高频的「解释数据口径」这件事，系统帮不上忙。我这段时间做的事，就是不重做血缘平台，而是用 MCP（Model Context Protocol）把血缘查询能力包装成一套标准工具，交给 Claude、Cursor 这类 AI 客户端调用。业务方直接问自然语言问题，AI 调工具拿到血缘链路，再端回一段能听懂的解释。这篇文章就把这个 Server 从设计、选型到落地完整拆一遍，适合数据平台、数据治理工程师，也适合对 MCP 感兴趣的后端同学参考。

1. 先把问题摆清楚：血缘系统「能看不能问」，缺的是让 AI 查数据的协议

1.1 数据团队真正花时间的地方，不是建血缘，而是「讲口径」

你去看任何一个数据团队的工作群，被问得最多的永远是这几类问题：「日报这个数怎么算的」「上个月的数和这个月为什么不可比」「DS 想改一张表，会影响哪几个报表」。这些问题背后有一个共同点：都需要查血缘。

但查血缘只是过程，最终要交付的是一段人能听懂的解释。数据开发接到问题的真实操作是：打开血缘平台、输入表名、看链路图、点开每个节点看加工逻辑、再到调度平台确认依赖关系，最后自己组织语言回复业务方。血缘平台把「查找过程」工具化了，却没有把「解释输出」工具化。MCP 恰好能补上这后半段。

我做一个直白的类比。血缘平台像一本很厚的族谱，你想知道某个人的祖辈是谁，得自己翻目录、翻页码、梳理关系；而 MCP Server 像是一个族谱查询电话，你打电话过去问，对面帮你翻完书，直接告诉你「这个人的祖父是某某，中间经过两次过继」。传统血缘系统的数据都还在，缺的是一个能把查询结果变成对话答案的中间层。

1.2 MCP 本质上是什么：一个「工具插座」标准

MCP 是 Anthropic 在 2024 年底开源的一个应用层协议，全称 Model Context Protocol。它基于 JSON-RPC 2.0，规定了几件事：客户端怎么发现服务端有哪些工具可用（tools/list）、客户端怎么调用这些工具（tools/call）、服务端怎么把结构化结果返回给客户端。

你可以把它理解成 USB-C 接口。以前每个 AI 功能都要自己定制集成方式：插件、函数调用、私有 API，各管各的；MCP 定义了统一接口，只要服务端实现了这个协议，所有支持 MCP 的 AI 客户端就能直接使用。目前 Claude Desktop、Cursor、Windsurf，以及 Spring AI、各种 Agent 框架都已经原生支持 MCP。你写一个 Server，不需要关心前端对话体验、模型管理、权限 UI，这些由 AI 客户端负责。

1.3 为什么血缘查询适合做成 MCP 工具，而不是自研聊天机器人

有人问我：直接做一个数据问答聊天机器人不就行了？自研聊天机器人要处理模型接入、对话管理、提示词、多轮记忆、权限控制、前端界面，投入很大。而血缘查询本身是典型的「结构化查询 + 可解释输出」场景：工具固定（查表、查字段、查链路）、返回结果固定（路径、加工逻辑）。这种场景非常适合做成函数调用，让大模型按需编排。

另一个重要原因是生态红利。GitHub 官方 MCP Server、蓝湖/Mastergo 的设计稿 MCP，都是「把原本只有系统里能查的东西变成 AI 能调的工具」。血缘 Server 走的是同一条路。你做完一个 Server，Claude Desktop 能用、Cursor 能用、自研 Agent 也能接，复用性远比自己写的私有接口高。

1.4 顺手澄清一个概念：MCP 和 Computer Use 不是一回事

很多朋友把 MCP 和 Computer Use 搞混。Computer Use 是让模型像人一样看屏幕、移动鼠标、敲键盘，解决的是「操作 GUI」的问题；MCP 是让模型以结构化方式调用后端数据和服务能力，解决的是「访问系统能力」的问题。做数据血缘查询，铁定走 MCP 路线：我们要的是「能查到链路并返回结构化结果」，而不是让模型打开血缘平台可视化界面去点按钮。想清楚这一点，方向就不会跑偏。

2. 整体设计：不重造一个血缘平台，只做血缘查询网关

2.1 先把边界划清楚

我给自己定的第一条原则：Server 只读，不承担血缘采集、SQL 解析、可视化展示这些重活。血缘平台负责把血缘关系算出来，我负责把它们变成 AI 能查询和理解的接口。

这个 Server 实际只做四件事：

1. 连接已有的血缘元数据源（可以是关系库，也可以是 DataHub / Atlas 的 API）。
2. 把血缘查询逻辑封装成 MCP 工具。
3. 把图结构血缘转成 LLM 友好的文本格式。
4. 按用户身份做权限过滤和审计。

这样做的好处是足够轻。不需要实现调度解析器，不需要写采集器，先把「查询对话化」跑通，再逐步扩展。我见过不少团队一上来就想自研一套带血缘解析的 MCP Server，结果卡在解析器上，半年没上线。先做网关，价值最快。

2.2 规划三个核心工具

工具粒度不需要太细，我最终只保留了三个，覆盖日常 80% 的问题：

工具太多会让模型选错，工具太少又覆盖不了场景。这三个工具从「向上溯源」「解释口径」「向下评估」三个维度切分，业务方的日常问题基本都能落到其中一个上。

2.3 技术选型：我为什么用 Python + FastMCP

选型这件事我纠结过一阵。官方 TypeScript SDK 很完善，但最后我还是选了 Python + FastMCP 库，原因有三点：

第一，数据团队的主力语言基本是 Python。血缘解析生态（sqlglot、DataHub 的 Python client、Apache Atlas 的 Python client）都集中在 Python 侧，后续如果要接 SQL 解析器，Python 最顺手。

第二，FastMCP 的写法非常轻，本质上就是用装饰器包一个函数。你定义一个普通函数，加一行@mcp.tool()，它就变成了一个 MCP 工具，参数校验交给 Pydantic。对于数据工程师来说，心智负担极低。

第三，Python 生态里连接各种元数据源都很方便。连 SQLite 可以只用标准库，连 PostgreSQL 用 psycopg，连 DataHub 有现成 client，后面接什么都能复用。

如果你团队是纯前端，用官方 TypeScript SDK 实现同样的逻辑也没问题，协议是同一套，只是语言不同。

2.4 血缘数据从哪来：一个最小可用的元数据模型

生产环境的血缘来源很杂：最常见的是解析 ETL SQL 自动生成，比如用 sqlglot 解析代码，找出每个目标表的字段来自哪些源表字段；其次是调度平台（Airflow、DolphinScheduler）的任务依赖 API；还有一些是人工登记的补充血缘。

但无论来源多复杂，落到存储上都是一张「节点表 + 关系表」的结构。我这里给出一个最小可用的关系模型，后面所有代码都基于它：

-- 表节点 CREATE TABLE table_nodes ( id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE, -- dws_trade_gmv_di layer TEXT, -- ods / dwd / dws / ads biz_owner TEXT, -- 业务负责人 description TEXT -- 表注释，LLM 靠这个理解业务 ); -- 字段节点 CREATE TABLE column_nodes ( id INTEGER PRIMARY KEY, table_id INTEGER NOT NULL REFERENCES table_nodes(id), name TEXT NOT NULL, -- gmv data_type TEXT, -- decimal(20,2) description TEXT, -- 字段注释 UNIQUE(table_id, name) ); -- 血缘边 CREATE TABLE lineage_edges ( id INTEGER PRIMARY KEY, src_type TEXT NOT NULL, -- 'table' 或 'column' src_id INTEGER NOT NULL, dst_type TEXT NOT NULL, dst_id INTEGER NOT NULL, transform_type TEXT, -- join / filter / aggregate / direct transform_desc TEXT, -- 'sum(gmv)', "filter status='paid'" created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );

这套模型用同一张 edges 表表达表级和字段级血缘，表与表之间的关系可以建模成「后端字段指向前端字段」的聚合形式。为了让后文讲得具体，我用一组企业里最常见的零售数据链路做示例：

ods_trade_order.amount → dwd_trade_order_detail.gmv （过滤 status='paid'，口径定义） → dws_trade_gmv_di.gmv （按 dt 聚合 sum(gmv)） → ads_trade_report.gmv （报表直连汇总表）

这四层正好对应 ODS → DWD → DWS → ADS 的数仓分层结构。需要注意的是，表注释和字段注释非常重要，因为 LLM 完全靠元数据来理解业务语义，注释越全，回答质量越高。

3. 动手实现：三个核心 MCP 工具的代码级拆解

3.1 项目骨架与运行方式

先看项目结构，保持最简单：

lineage_mcp_server/ ├── server.py # MCP 工具定义 ├── store.py # 血缘查询存储层 └── requirements.txt # fastmcp、sqlite 用标准库

安装依赖：

pip install fastmcp

requirements.txt 里其实就一行fastmcp，数据库先用 Python 内置的 sqlite3，零配置，最适合本地跑通。启动方式：

python server.py

FastMCP 默认走 stdio 模式，Claude Desktop 里这样配置：

{ "mcpServers": { "lineage": { "command": "python", "args": ["/path/to/lineage_mcp_server/server.py"] } } }

Cursor 里则在 MCP 配置界面填同样的命令。客户端启动时自动调用 tools/list 发现工具，后续的 tools/call 会映射到我们定义的 Python 函数上。

3.2 query_lineage：从图里走出链路

这是最基础的工具：输入一张表名或字段名，返回上游或下游路径。核心逻辑就是广度优先遍历，同时用 visited 集合防止环状血缘导致死循环。先看存储层：

# store.py import sqlite3 class LineageStore: def __init__(self, db_path="lineage.db"): self.db_path = db_path def _conn(self): conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row return conn def find_node(self, name): """按名称找到节点，兼容表名和字段名""" conn = self._conn() row = conn.execute( "SELECT id, name, layer, description FROM table_nodes WHERE name = ?", (name,), ).fetchone() if row: return ("table", row["id"], row) row = conn.execute( """SELECT c.id, c.name, c.description, t.name AS table_name FROM column_nodes c JOIN table_nodes t ON c.table_id = t.id WHERE c.name = ?""", (name,), ).fetchall() # 如果字段名重复，返回所有候选 return [("column", r["id"], r) for r in row] if row else None

然后在 Server 里定义工具：

# server.py from fastmcp import FastMCP from store import LineageStore mcp = FastMCP("Data Lineage Server") store = LineageStore("lineage.db") @mcp.tool() def query_lineage(node_name: str, direction: str = "upstream", depth: int = 3) -> str: """查询血缘链路。node_name 传表名或字段名； direction 传 upstream（上游来源）或 downstream（下游影响）； depth 控制追溯深度，默认 3 层。""" result = store.trace(node_name, direction=direction, max_depth=depth) return format_paths(result)

trace 方法的关键实现，就是一层层向外扩展：

# store.py def trace(self, node_name, direction="upstream", max_depth=3): start = self.find_node(node_name) if not start: return {"error": f"节点 {node_name} 不存在"} paths = [] seen = set() def expand(node_key, path): if len(path) > max_depth: return node_type, node_id, meta = node_key edge_key = f"{node_type}:{node_id}:{direction}" if edge_key in seen: return seen.add(edge_key) paths.append({"node_type": node_type, "meta": dict(meta), "path": list(path)}) # 继续查上游/下游 conn = self._conn() if direction == "upstream": rows = conn.execute( """SELECT src_type, src_id, transform_desc FROM lineage_edges WHERE dst_type = ? AND dst_id = ?""", (node_type, node_id), ).fetchall() else: rows = conn.execute( """SELECT dst_type, dst_id, transform_desc FROM lineage_edges WHERE src_type = ? AND src_id = ?""", (node_type, node_id), ).fetchall() for r in rows: nxt = self.get_node(r[0], r[1]) if nxt: expand(nxt, path + [r[2]]) # path 里带上加工逻辑描述 expand(start, []) return paths

返回给 LLM 的文本我用箭头表达链路关系，比如查ads_trade_report.gmv的上游，输出：

ads_trade_report.gmv ← dws_trade_gmv_di.gmv (transform: SUM(gmv) GROUP BY dt) ← dwd_trade_order_detail.gmv (transform: filter status='paid') ← ods_trade_order.amount (原始字段)

这里我刻意不用 JSON 嵌套，而是用类似文件路径的箭头文本。模型读这种文本的理解成本远低于读多层嵌套结构，回答的时候也不容易出错。format_paths 函数就是把这些路径拼成上面的格式。

3.3 explain_column：把「口径」翻成人话

query_lineage 返回的是链路，但业务方真正要的是「口径解释」。explain_column 这个工具专门做这件事：从目标字段出发，回溯上游，沿途收集每个节点的表注释、字段注释、转换描述，最后拼成一段「数据旅程」。

@mcp.tool() def explain_column(table_name: str, column_name: str) -> str: """解释某个业务指标字段的完整加工链路。 适合回答『这个数怎么算的』『口径是什么』这类问题。 table_name 传表名，column_name 传字段名。""" steps = store.explain_chain(table_name, column_name) if not steps: return f"未找到 {table_name}.{column_name} 的血缘链路" lines = [f"字段 {table_name}.{column_name} 的血缘解释：", ""] for idx, step in enumerate(steps, 1): lines.append(f"第 {idx} 步：{step}") return "\n".join(lines)

explain_chain 的实现，就是从目标字段向上游逐层回溯，把节点元数据和转换描述拼成列表。核心数据结构是这样：

def explain_chain(self, table_name, column_name): # 先从字段节点定位 start = self.find_column(table_name, column_name) if not start: return [] chain = [] seen = set() cur = start while cur: node_id = cur["id"] if node_id in seen: break seen.add(node_id) # 查上游边 conn = self._conn() rows = conn.execute( """SELECT src_type, src_id, transform_desc FROM lineage_edges WHERE dst_type='column' AND dst_id = ?""", (node_id,), ).fetchall() if not rows: chain.append(f"原始来源：{cur['table_name']}.{cur['name']}，无上游，此为源头字段") break # 同一层可能有多个上游，全部展开 for r in rows: src = self.get_node(r["src_type"], r["src_id"]) transform = r["transform_desc"] or "直接映射" if src: chain.append(f"{src['table_name']}.{src['name']}（{src['description'] or '无注释'}）→ 经过转换：{transform} → {cur['table_name']}.{cur['name']}") # 继续向上追溯（只取第一个上游的主线，避免分支爆炸） cur = self.get_node(rows[0]["src_type"], rows[0]["src_id"]) return chain

假设库里存的注释是：ods_trade_order.amount注释「订单金额」，dwd_trade_order_detail.gmv注释「订单支付金额（已剔除未支付）」且 transform_desc 是filter status='paid'，dws_trade_gmv_di.gmv注释「按天汇总GMV」且 transform_desc 是SUM(gmv) GROUP BY dt。那这个工具返回的就是：

字段 ads_trade_report.gmv 的血缘解释： 第 1 步：dws_trade_gmv_di.gmv（按天汇总GMV）→ 经过转换：直接映射 → ads_trade_report.gmv 第 2 步：dwd_trade_order_detail.gmv（订单支付金额，已剔除未支付）→ 经过转换：SUM(gmv) GROUP BY dt → dws_trade_gmv_di.gmv 第 3 步：ods_trade_order.amount（订单金额）→ 经过转换：filter status='paid' → dwd_trade_order_detail.gmv 第 4 步：原始来源：ods_trade_order.amount，无上游，此为源头字段

业务方看了这段，基本就能理解「这个 GMV 只包含已支付订单，按天去重汇总」。模型拿到这些文本后，再组织成自然语言回答用户，效果非常接近一个懂数仓的分析师。

3.4 impact_analysis：让 AI 告诉你「改这张表会炸哪些报表」

第三个工具面向数据开发和数据运维：准备改一张表之前，先问影响范围。实现上就是从指定节点向下游遍历，返回所有受影响的路径。

@mcp.tool() def impact_analysis(table_name: str, max_depth: int = 4) -> str: """分析某张表变更会影响到哪些下游表/报表。 适合回答『这张表能改吗』『下游有哪些依赖』。 max_depth 控制下游追溯深度。""" downstream = store.trace(table_name, direction="downstream", max_depth=max_depth) if not downstream: return f"未找到 {table_name} 的下游依赖" result = [f"表 {table_name} 的下游影响路径（最深 {max_depth} 层）：", ""] for item in downstream[:20]: # 限制返回条数 meta = item["meta"] path = " → ".join(str(p if p else "直接映射") for p in item["path"]) if item["path"] else "直连" result.append(f"- {meta['name']}（{meta.get('description') or '无注释'}）") result.append("") result.append(f"共找到 {len(downstream)} 条路径，仅显示前 20 条。") return "\n".join(result)

实际使用场景很典型：DBA 要改dws_trade_gmv_di的字段类型，先在对话里问一句「影响分析 dws_trade_gmv_di」，AI 马上列出下游所有报表，并标注哪张表是 ADS 层的核心报表、负责人是谁。省去了手动打开血缘平台逐层点开的流程。

4. 让 AI 用好这个 Server，重点在工具输出和描述

4.1 MCP 的输出不是给人看的，是给模型「阅读」的

这是我做这个项目最大的认知转变：MCP Server 返回的结果，普通终端用户通常看不到原始内容，它是被塞进模型上下文里的。所以 Server 返回的每一段文本，都必须「自带上下文」。

什么意思？比如 query_lineage 返回dws_trade_gmv_di.gmv ← dwd_trade_order_detail.gmv，如果节点信息里不带注释，模型只知道「一个 gmv 来自另一个 gmv」，它无法解释这个数怎么算的。但如果你在节点后面补上该字段是对已支付订单金额的按天汇总，模型就有素材了。

对比一下两种返回：

• 没注释的返回（坏）：ads_trade_report.gmv ← dws_trade_gmv_di.gmv
• 带注释的返回（好）：ads_trade_report.gmv（报表GMV，负责人：商业分析部）← dws_trade_gmv_di.gmv（按天汇总GMV，口径为已支付订单）

后者不需要模型瞎猜，也不给模型「脑补」的机会。血缘平台里的元数据越干净，AI 回答的质量越高。所以说白了，MCP Server 做的事是「把元数据翻译成模型能读懂的上下文」。

4.2 工具描述是模型决定调不调用的唯一线索

FastMCP 里，@mcp.tool()函数下面的 docstring 会被转换成 tools/list 返回给模型，模型根据这段描述决定「当前问题该不该调这个工具、参数怎么填」。描述写得差，工具白做。

亲身对比：

• 坏描述：「查询血缘」
• 好描述：「查询某张表或字段的上游/下游血缘链路。direction 传 upstream（向上溯源）或 downstream（向下影响）。当用户问『这个数从哪来』『影响了哪些下游报表』时使用」

后者包含了触发条件（什么时候用）、方向参数（怎么填），模型基本不会用错。我在调试时发现，参数说明里明确写「当用户问…时使用」，能显著提升工具的命中率。

4.3 实测：在 Cursor / Claude Desktop 中追问一个报表字段的血缘

我在 Claude Desktop 里配好 Server 后，做了这么一次实测。我输入：

帮我查一下 ads_trade_report 的 gmv 字段上游血缘，并解释每一步加工逻辑。

模型内部调用了 explain_column("ads_trade_report", "gmv")，拿到 Server 返回的链路文本后，给出的最终回答是：

ads_trade_report 的 gmv 字段来自数仓汇总表 dws_trade_gmv_di，该表按天对订单明细做 SUM(gmv) 聚合；再往上游，订单明细表的 gmv 取自订单主表，且在抽取时过滤了 status='paid' 的订单。所以这个 GMV 的口径是「已支付订单金额的日累计」，如果和财务对不上，优先检查订单抽取任务是否遗漏分区。

这个回答里，黑体部分是模型基于 Server 返回的元数据重新组织的，括号里的排查建议是模型自己补充的。真实体验是，我从「自己翻血缘再解释」变成了「AI 直接给我解释」，时间从半小时压缩到一分钟。

5. 从 0 到 1 踩过的坑：传输模式、环状血缘、上下文长度

5.1 stdio 与 Streamable HTTP：本地跑通不等于远程可用

FastMCP 默认跑在 stdio 模式，Python 进程和 AI 客户端在同一台机器上，通过标准输入输出通信。本地调试没问题，但团队成员分散、需要集中部署场景，就得切到 Streamable HTTP 模式，让客户端通过网络访问 Server。

一个常见的坑：很多人在 Claude Desktop 本地配好了 stdio Server，以为生产环境也能这样用。实际上多用户访问、权限控制、审计这些需求，都必须走远程 HTTP 模式。FastMCP 切远程的方式很简单：

if __name__ == "__main__": mcp.run() # 默认 stdio # 远程部署时改用下面这行 # mcp.run(transport="streamable-http")

但注意，远程模式要配鉴权，否则整个血缘库裸奔在网络上。我见过直接在公网暴露 MCP Server 的例子，这个真的要避免。

5.2 环状血缘和自引用导致死循环

数据仓库里的血缘图不是一棵干净的树。现实中的 ETL 经常出现 A → B → A 的循环依赖，或者一张表更新自身产生的自引用边。如果用简单的递归去遍历，直接爆栈或死循环。

我第一次实现 query_lineage 时没加 visited，结果一个带环的测试数据让 Server 卡死了。后来在 expand 函数里加了：

seen = set() def expand(node_key, path): edge_key = f"{node_type}:{node_id}:{direction}" if edge_key in seen: return seen.add(edge_key) ...

并且给 max_depth 设置默认值 3、上限 6。这样即使遇到环，最多走到第 6 层就会停。生产环境里，血缘图的深度通常不会超过 6 层，这个上限足够用。

5.3 一次性返回几十条路径，模型开始「脑补」不存在的血缘

这是个很有意思的问题。血缘查询返回的路径一多，比如某个下游分析返回 40 条链路，全部塞进上下文之后，模型在处理时偶尔会说出链路里不存在的节点。我后来分析，原因是模型面对超长列表时，倾向于补全「看起来合理」的路径，哪怕源文本里根本没有。

解决方案是限制每次返回的条数和描述长度。我在 impact_analysis 里限制前 20 条，并在结尾明确写「共找到 N 条路径，仅显示前 20 条」。这样模型知道自己看到的不是全貌，就不会自作主张地编造。

经验值：给模型的文本控制在 1500 token 以内最稳。超过这个量，优先做摘要、截断，别一股脑全丢给模型。

5.4 工具参数的坑和中文注释

FastMCP 使用 Pydantic 做参数校验，函数签名里没标注类型的参数是不会作为工具参数暴露的。我第一次写def query_lineage(node_name, direction="upstream", depth=3)，node_name 没标: str，结果工具调用一直报参数缺失。速查：所有参数都必须有类型注解。

另一个坑是src_type、dst_type这类字段名，在 SQL 里没问题，但如果未来接 DataHub 这类系统，对方的字段名是sourceType、destinationType，映射的时候容易漏。建议在设计存储层时就用小写下划线命名，读写两侧都做一层字段映射。

关于中文注释：自定义 JSON 序列化时记得设置ensure_ascii=False，否则返回中文注释会变成\uXXXX转义序列，模型理解起来就困难了。另外库里的字段注释如果为空，模型会把节点当成未知对象，最好在查询时补一个默认描述「暂无注释」，至少不会让模型猜测。

6. 企业落地：权限、审计、缓存一个都不能少

6.1 权限：MCP Server 怎么知道「谁在问」

MCP 协议本身是客户端到服务端的模型上下文通道，但多用户场景下，服务端需要知道当前请求来自谁，才能决定返回哪些血缘可见范围。

本地 stdio 模式不好处理用户态，因为进程是在本机启动的，天然只能服务一个用户；远程 Streamable HTTP 模式可以在请求头里带用户身份。企业落地我建议的架构是：MCP Server 后面套一层 API 网关，网关负责统一认证（对接公司 SSO），再把用户身份注入请求体传给 Server，Server 内部的查询逻辑再按身份过滤。

简化版的权限模型不用太复杂，一张用户表 + 权限配置表就够了：

CREATE TABLE user_permissions ( user_id TEXT NOT NULL, role TEXT NOT NULL, -- 'admin' / 'analyst' / 'viewer' allow_layer TEXT, -- 'ods,dwd' 或 '*' allow_tables TEXT -- 逗号分隔的表白名单，空则不限 );

查询血缘时，根据身份过滤节点。实现很简单，但非常有效——至少能保证普通用户查不到未授权的核心业务表元数据。

6.2 脱敏与审计

血缘查询本身不返回数据值，但表名、字段名、注释照样是敏感信息。有些业务表的字段注释里会写「客户手机号」「渠道分成比例」这类字眼，在生成返回文本时要做一层过滤。

我在 Server 里维护了一个敏感词列表：字段注释或表注释命中敏感词时，把注释替换成「已脱敏」。成本很低，但能避免把不该暴露的业务语义泄露给低权限用户。

审计日志也是必须的。我会在每次 tools/call 时记录：

{ "time": "2025-01-12T10:30:00Z", "user": "zhangsan", "client": "claude-desktop", "tool": "explain_column", "params": { "table_name": "ads_trade_report", "column_name": "gmv" } }

存到单独的表里。出了数据争议的时候，能追溯「谁在什么时间通过 AI 助手查了哪条血缘」，这在企业环境里是合规要求的刚需。

6.3 缓存与图存储扩展

血缘数据的特点是「变化慢」，不像业务数据每秒都在变，ETL 调度关系可能一个月才动一次。所以血缘查询非常适合做缓存。

本地 Demo 用 SQLite 没问题，但生产环境建议直接上 PostgreSQL / MySQL，血缘图大了之后可以考虑 Neo4j 这种图数据库。查询热点集中在「某张表的上游/下游」上，缓存 key 直接设计成query_lineage:ads_trade_gmv_di:upstream:3，缓存 30 秒到 1 分钟，命中率相当高。

一个细节：MCP Server 的定位是只读查询，所以「刷新缓存」不能做成 MCP 工具（否则任何人有权限调）。正确做法是血缘采集系统在更新元数据后，主动调内部的缓存失效接口，或者在 Server 里做一个定时刷新的后台任务，把缓存刷新和查询分开。

我个人实际落地之后的体会是：MCP Server 本身的代码量并不大，真正花时间的反而是把血缘元数据整理干净、把注释补齐。这一步做完，Server 的价值才会真正发挥出来。上线之后，数据团队被问「这个数哪来的」的频率肉眼可见地下降，因为业务方已经学会直接问 AI 了。后续值得继续做的方向也不少：让 AI 直接生成 SQL 再反向解析血缘入库，把调度平台的任务依赖自动同步成血缘边，还可以把这个 Server 接入内部 Agent，让它在周报里自动解释异常指标。这几个方向我都准备接着往下做，回头有进展再继续分享。