前言业务同学想看个数要在群里 数据同学排期、写 SQL、跑数、贴图一个来回半天就没了。让 AI 直连数仓自然语言直接问是数据分析场景里收益最直观的 AI 落地之一。但「连上」只是第一步——真正决定这件事能不能用起来的是权限、口径和大结果处理。本篇是 Agent 基座系列第四篇在第二篇 MCP Server 的基础上把连接讲透权限部分直接复用上一篇刚搭好的网关虚拟 key 体系——调模型的钥匙和查数仓的钥匙是同一把。一、整体架构AI 和数仓之间隔了几层先看全貌一个生产可用的链路长这样AI 客户端Claude Code / 自研助手 ↕ MCP 协议stdio / HTTP 查询 Server工具定义、SQL 校验、行数守卫 ↕ 权限层只读账号、行级过滤、审计 ↕ 数仓Trino / Hive / ClickHouse ...各层职责层解决的问题不做会怎样连接层MCP标准化工具协议任意客户端接入每个客户端重复开发Server 层SQL 校验、行数守卫、schema 探索AI 拿到脏结果、超大结果权限层只读 行级过滤 审计越权查数、出事无法追溯治理层指标口径、表注释AI 答数口径错业务不信任二、工具设计的三种姿势给 AI 暴露数仓能力有三种做法可控性和灵活性此消彼长姿势形态优点缺点裸 SQLrun_query(sql)最灵活长尾需求全覆盖SQL 不可控审计难schema 全暴露固定业务动作city_monthly_sales(city, month)最可控口径固化在 server工具穷举不完新需求要改代码Schema 探索三件套list_tables/describe_table/run_queryAI 自己找表看字段覆盖面和可控性平衡依赖表注释质量第二篇用的是第二种固定业务动作适合高频固定报表。但真实使用中长尾需求极多——「按商圈看复购」「分时段看缺货」根本穷举不完。生产环境的推荐形态是第三种把 schema 的探索权交给 AI把执行的安全权留在 server。三、实战Schema 探索型数仓 Server核心思路AI 像一个新来的数据分析师——先看有哪些表list_tables再看表结构describe_table确认字段后自己写 SQLrun_query。全程只读行数有守卫。# dw_mcp_server.pyimportosimportreimporttrinofrommcp.server.fastmcpimportFastMCP mcpFastMCP(dw-query)ALLOWED_SCHEMAS{dws,dim,ads}# 白名单只开放集市层不碰 ODS 原始层MAX_ROWS500WRITE_KEYWORDSre.compile(r\b(insert|update|delete|drop|create|alter|truncate|grant)\b,re.I)defget_conn():returntrino.dbapi.connect(hostos.environ[TRINO_HOST],portint(os.environ.get(TRINO_PORT,8080)),useros.environ[TRINO_USER],cataloghive)mcp.tool()deflist_tables()-str:列出数仓中可查询的表库名.表名 表注释。 开始查数前先调用了解有哪些数据可用。curget_conn().cursor()cur.execute( SELECT table_schema || . || table_name, comment FROM information_schema.tables WHERE table_schema IN %s ,(tuple(ALLOWED_SCHEMAS),))return\n.join(f{s}--{cor无注释}fors,cincur.fetchall())mcp.tool()defdescribe_table(table_name:str)-str:查看表结构字段名、类型、注释。写 SQL 前必须先调用 确认字段真实名称和格式不要凭猜测写 SQL。 Args: table_name: 完整表名格式 库名.表名如 dws.sales_order_di schema,_,tabletable_name.partition(.)ifschemanotinALLOWED_SCHEMAS:# 白名单校验returnf库{schema}不在开放范围可选{ALLOWED_SCHEMAS}curget_conn().cursor()cur.execute( SELECT column_name, data_type, comment FROM information_schema.columns WHERE table_schema %s AND table_name %s ORDER BY ordinal_position ,(schema,table))rowscur.fetchall()ifnotrows:returnf表{table_name}不存在先 list_tables 确认表名return\n.join(f{n}{t}--{cor}forn,t,cinrows)mcp.tool()defrun_query(sql:str)-str:执行只读 SELECT 查询并返回结果。 规则只允许 SELECT优先写聚合查询拿汇总数明细只在前几条样例时使用 大范围查询请加 WHERE 过滤和 GROUP BY 聚合。 Args: sql: 完整的 SELECT 语句字段名以 describe_table 的结果为准 ifWRITE_KEYWORDS.search(sql):# 只读校验return仅支持 SELECT 查询写操作已被拒绝ifnotsql.lstrip().lower().startswith(select):return仅支持 SELECT 查询iflimitnotinsql.lower():# 没写 LIMIT 就注入一个sqlf\nLIMIT{MAX_ROWS}curget_conn().cursor()cur.execute(sql)cols[d[0]fordincur.description]rowscur.fetchall()# 顶格返回 结果可能被截断明确告知 AI引导它改聚合head | .join(cols)body\n.join( | .join(str(v)forvinr)forrinrows)warn(f\n(已达{MAX_ROWS}行上限结果可能不完整——f如需总数/汇总请改写为聚合查询)iflen(rows)MAX_ROWSelsef\n(共{len(rows)}行))returnhead\nbodywarnif__name____main__:mcp.run()三个工具的 description 都写了触发时机和规则「写 SQL 前必须先调用」「优先聚合」——这是第一篇就强调的description 就是提示词模型照着做。run_query返回末尾的截断警告是关键设计让 AI 知道数据可能不完整主动换聚合写法而不是拿半截数据硬算。四、权限怎么做AI 才看不到不该看的数据先立一条铁律安全边界不能建立在提示词上。在系统提示里写「只查自己城市的数据」没有防护力——提示词可以被对话诱导绕过模型本身也可能出错。权限必须在 server 端用代码强制。分四步实现第一步只读账号 库白名单挡写操作和越库双保险数仓侧建专用账号只授 SELECTserver 侧ALLOWED_SCHEMAS白名单 run_query的写关键词校验。白名单只开集市层dws/dim/ads原始层ods和未治理的表一概不见。第三节代码已包含这一层。第二步识别发起人行级权限的前提行级权限先要回答「这次查询是谁发起的」。身份传递按部署形态分两种远程 HTTP server推荐客户端每次请求携带 tokenserver 用 token 换取用户身份身份无法伪造天然支持多人共用本地 stdio server配置文件写 user 标识只适合个人自用——配置可被修改多人场景不能依赖它。token 不用新造一套直接复用上一篇网关的虚拟 key。数仓 server 收到请求后拿 key 向网关校验或查网关的 key 表key_alias就是发起人身份。这样把两套权限收敛成一套员工入职领一把虚拟 key调模型用它、查数仓也用它——max_budget管花费scope 管数据范围离职时吊销这一把 key模型和数据的访问同时收回不存在「模型权限收了、数仓权限忘了」的缝隙。拿到身份后取该用户的数据可见范围scopedefget_scope(user:str)-dict:发起人的数据可见范围。来源权限表或公司权限系统的 API。# 返回形如 {city: [杭州, 宁波], store_type: [直营]}returnfetch_scope_from_auth_service(user)scope 的结构建议是「维度 → 允许值列表」由权限系统维护、server 只消费。这样人员转岗时改权限系统的数据即可server 零改动第八节的维护联动也依赖这条链路。第三步行级过滤的两种实现方案 ASQL 改写注入推荐AI 提交 SQL 后、执行前server 把权限条件 AND 进语句。关键是用 SQL 解析库改写不做字符串拼接——拼接挡不住子查询、CTE、UNION 这些绕过路径AST 级改写才能对每个查询块统一生效。以 sqlglot 为例importsqlglotfromsqlglotimportexpdefinject_row_filter(sql:str,scope:dict)-str:把 scope 条件注入 SQL 的每个查询块。astsqlglot.parse_one(sql,readtrino)forcol,allowedinscope.items():condexp.In(thisexp.column(col),# 生成 city IN (...)expressions[exp.Literal.string(v)forvinallowed])forselectinast.find_all(exp.Select):select.where(cond,copyFalse)# 与已有 WHERE 自动 AND 合并returnast.sql(dialecttrino)# run_query 中校验通过后、执行前# sql inject_row_filter(sql, get_scope(current_user))这个方案的核心性质注入发生在执行前AI 不可见也不可绕过——AI 写的 SQL 永远不含权限条件它甚至不知道过滤的存在。生产环境还需处理列别名、JOIN 同名列等细节骨架就是这些。方案 B权限视图数仓侧为每个权限角色建过滤好的视图server 白名单只开视图-- 每个角色一个视图行级条件固化在视图里CREATEVIEWdws.sales_order_hzASSELECT*FROMdws.sales_order_diWHEREcity杭州;优点是零改写逻辑、绝对可靠缺点是视图数量 表 × 角色容易膨胀且 AI 看到sales_order_hz这类带后缀的表名会困惑。适合角色少而稳定的场景。改写注入A权限视图B实现位置server 端代码数仓侧 DDL权限变更改 scope 数据即可要新建/改视图AI 看到的真实表名无感视图名有感知主要风险改写要覆盖全语法路径视图膨胀、漏建第四步审计留痕每次run_query落一条记录时间、发起人、原始 SQL改写前、改写后 SQL、返回行数、耗时。既是安全事件的回溯依据说服安全团队放行的前提也是第八节错数排查的证据链。defaudit(user,raw_sql,final_sql,rows,ms):每次查询落审计。存审计表或结构化日志字段固定便于回查。append_audit_log({ts:now(),user:user,raw:raw_sql,rewritten:final_sql,rows:rows,ms:ms})五、口径AI 答数准不准取决于你喂不喂口径权限解决「能不能看」口径解决「看得对不对」。同一个「销售额」含不含退款含不含赠品口径不一致AI 算得再对也是错的。给 AI 一份指标口径字典把核心指标的定义计算公式、包含/排除规则、适用维度写成一份文档作为 resource 或独立工具暴露mcp.tool()defmetric_definitions()-str:查询核心指标口径定义。回答任何涉及销售额、毛利、 活跃门店等指标的问题前必须先调用确认口径。returnPath(metrics.md).read_text()# 指标字典团队维护指标字典由数据团队维护改口径只改文档不用动代码。静默陷阱清单第二篇踩坑清单里的三个坑在直连数仓场景是重灾区这里再强调一次整数除法比率计算先CAST AS DOUBLE日期格式先LIMIT 3看一眼真实存储格式再写条件静默截断网关行数上限不报错server 必须自己设上限并告知。排查心法AI 算出的数不对先查数据口径再怀疑模型。实践中多数「AI 算错了」最后都定位到这三个坑或口径文档缺失。—### 六、大结果集限行、分批与合并第三节的行数守卫只堵住了一半的路if limit not in sql只兜底「没写 LIMIT」的查询AI 主动写一句LIMIT 1000000就穿防了。而业务真会这么干——「把这张表拉出来看看」背后就是几百万行一次拉全量查询 server 和数仓协调节点一起被拖趴所有人的查询跟着卡。三个补丁1.守卫要钳制不能只兜底解析 SQL 把超上限的 LIMIT 改写回来而不是只在缺省时注入python def clamp_limit(sql: str) - str: AI 写了超大 LIMIT 也压回上限。与第四节同款 sqlglot 改写。 ast sqlglot.parse_one(sql, readtrino) for lim in ast.find_all(exp.Limit): if int(lim.expression.name) MAX_ROWS: lim.set(expression, exp.Literal.number(MAX_ROWS)) return ast.sql(dialecttrino)2.真需要全量分批拉取游标翻页聚合替代不了分析型任务——全量门店逐家核对、给几百个对象跑分明细就是要拉。正确姿势是 keyset 分页每批不超过上限用上一批末尾的排序键当游标而不是 OFFSET深翻页时 OFFSET 越大扫描越慢python last_key, merged None, [] while True: sql (fSELECT id, ... FROM dws.sales_order_di {WHERE id str(last_key) if last_key else } fORDER BY id LIMIT {MAX_ROWS}) rows fetch(sql) # 每批 ≤ MAX_ROWS merged rows if len(rows) MAX_ROWS: break # 不满一批 到头了 last_key rows[-1][0] # 游标推进3.合并要有纪律-拼齐再算AI 最容易犯的错是拿第一批就出结论——批次没拉完数字必然偏小。分批任务的说明里写明「先拉完全部批次再汇总」 -总数校验动手前先SELECT COUNT(*)拿基准批次累加和基准对不上说明漏了或重了 -排序键唯一游标列必须唯一不唯一就再加一列否则翻页会漏行。对话型问答 90% 的场景仍然是聚合优先下一节分批合并是为数不多「确实需要全量明细」的任务型场景准备的——两条路分清楚守卫才既保得住服务器又不误伤真需求。—### 七、成本与体验优化schema 描述做缓存describe_table的结果基本不变server 内做 LRU 缓存同一张表的重复探索不再打数仓。表结构变更时失效。引导聚合优先run_query的 description 和截断警告都在推同一件事优先拿汇总明细只抽样。一次拉几万行明细既撑上下文又烧 token而业务问题 90% 只要一个数。表注释当产品做AI 选表选字段完全靠注释。注释缺失的表对 AI 来说等于不存在。上这套方案前先把白名单内表的字段注释补齐这件事的投入产出比最高。—### 八、后期维护上线才是开始数仓 server 不是交付即结束的项目——数仓在变、人在变、口径在变server 必须跟着变否则会以「静默答错数」的方式劣化。五个维护动作表结构变更要同步新表上线评估是否加白名单、补注释老表改名或下线describe_table会突然报不存在。两个配套机制describe 缓存设 TTL小时级或手动刷新入口表结构变更后缓存就是毒数据监控工具报错率——报错率突增第一反应查「是不是表结构变了」而不是查代码。口径字典是活文档指标口径变更比如 GMV 定义调整必须当天同步口径文档否则 AI 拿着旧口径一本正经地答错全程无任何异常信号。建议把口径文档更新纳入数仓变更流程改口径的变更单必须包含文档同步这一步。权限是动态的入职/转岗/离职行级过滤的 scope 要与人事流程联动离职当天回收表级权限回收要及时从白名单摘除。一个常见教训某张表权限被上游回收后依赖它的查询开始报权限错误排查方向全在「代码哪里坏了」实际是权限变了。权限变更要通知到 server 维护者这条通道不通排查成本会成倍放大。常驻进程的保活与监控server 挂掉所有用户的 AI 查数一起失效属于高优先级故障。按第二篇的方案配保活脚本nohup 循环拉起另加两个监控项数仓连接健康检查、工具级报错率。能自动拉起的自动拉起拉不起来的告警到人。建立错数反馈闭环用户报「AI 算的数不对」走固定排查路径先查口径 → 三个静默陷阱整数除法/日期格式/静默截断→ 再怀疑模型。每个确认的错数 case 都要沉淀成资产口径缺失就补口径文档表没注释就补注释工具描述误导就改 description。错数案例是这套系统最好的迭代燃料——上线后头一个月修的都是这类问题修一个少一类。参考资料MCP 官方文档Trino Python Client本系列第二篇《MCP 协议与 Server 手写实战详解》本系列第三篇《企业级 AI 底层搭建详解》虚拟 key 体系