1. 客户数据问答链路里，明文到底在哪一步失控
先说一个我实际排查过的场景。某 CRM 团队把 AI Agent 接到金仓数据库上做客户回访助手，客服在对话框里问「查一下张三最近三笔订单和联系方式」，Agent 生成 SQL、数据库返回结果、模型组织语言，最后回答「客户张*，手机号 138****5678」。产品经理看完演示很满意，觉得脱敏已经做完了。
问题在于：手机号13812345678、身份证110101199001011234这些明文，在模型输出掩码之前，已经完整地进入了模型上下文。它可能落在 Agent 的会话历史里、落在模型请求日志里、落在 Trace 的 span attribute 里、落在 Prompt Cache 里，甚至被下一轮工具调用当作参数再传一次。最终输出打了码，但明文早就扩散出去了。
这就是「脱敏发生得太晚」。真正要控制的边界不是模型输出，而是数据库明文到模型上下文之间那一段。金仓数据库（KingbaseES）兼容 PostgreSQL 协议，绝大多数 PostgreSQL 的权限模型、RLS、列级授权都能直接复用，所以我们可以把脱敏网关放在 MCP Server 这一层：数据库返回明文 → MCP Server 内部短暂持有 → 字段级脱敏 → 只有掩码值进入 Agent 和模型。
这篇文章要交付的是可跟做的三件事：一份字段分类元数据表结构、一份角色 × 字段类型的脱敏策略配置（JSON 可直接复制）、一个 MCP 查询工具的完整实现与验证动作。跑完之后你能用一张含手机号/身份证的样例表，确认模型侧只收到掩码值。
适合谁看：正在把金仓数据库接入 AI Agent 做客户数据问答的后端/数据工程师；已经在用 MCP 协议封装数据库工具、但脱敏还停留在 Prompt 层面的团队；以及需要给脱敏做评测指标的安全同学。
核心原则只有一句：模型不需要看到的数据，就不要先给模型。
2. 接入前的准备：TaoToken 与金仓数据库的边界划分
在动手写脱敏代码之前，先把链路上各个组件的职责划清楚，否则后面很容易把安全逻辑写散。
整条链路我建议这样分层：
金仓数据库负责「能不能取到」——通过列级 SELECT 授权和行级安全策略（RLS）限制查询账号的可见范围；MCP Server 负责「取到之后能看到多少」——字段分类 + 角色策略 + 掩码函数；Agent 的 System Prompt 负责「怎么解释给用户」——禁止还原、禁止推测隐藏字符。三层解决三个不同问题，谁也别越界。
模型入口这一侧，我用 TaoToken 做统一接入。它的作用是给 Agent 提供一个稳定的模型调用出口，模型对话、Coding Plan、API Keys 都在同一个控制台里管理。对脱敏这件事来说，它的价值在于：模型请求的入口是收敛的，你可以在这条出口上统一做日志脱敏和审计，而不用每个 Agent 各自接一家模型、各自留一份明文日志。
具体要准备的东西：
- 一个 TaoToken 账号，在控制台创建 API Key（地址：https://taotoken.net/api-keys ，注意 API 域名是 https://taotoken.net/api ，不要加多余路径）
- 模型 ID 选一个你常用的对话模型即可，Agent 场景建议选指令遵循稳定的
- 金仓数据库一个只读账号，只授予必要的列级 SELECT 权限
- Python 环境 + MCP SDK，用来写查询工具
这里有个容易踩的坑：很多人把 MCP Server 的数据库账号配成高权限用户，图省事。一旦账号拥有 BYPASSRLS 或是表所有者，RLS 策略会被直接绕过，前面所有脱敏设计都白做。金仓数据库里同样要遵守最小权限原则，查询账号只给SELECT且限定列。
关于模型侧接入，如果你只是验证脱敏效果，用模型对话页面手动发几次请求就够了；如果要做长期的 Agent 编码或自动化问答，建议走 Coding Plan，配额和调用方式更适合持续跑。接入文档在 https://taotoken.net/doc ，里面有 Base URL、鉴权和请求格式的完整说明。
3. 可复制的脱敏规则配置与 MCP 工具实现
这一节是全文的核心，所有配置都可以直接复制改路径使用。
3.1 字段分类元数据表
不要依赖 Agent 临时猜「mobile 看起来像手机号」，而是在元数据里显式维护。在金仓数据库里建一张分类表：
CREATE TABLE kfs_data_classification ( schema_name varchar(128), table_name varchar(128), column_name varchar(128), sensitivity_level varchar(32), data_category varchar(64), masking_rule varchar(64), owner_team varchar(128), updated_at timestamp DEFAULT now(), PRIMARY KEY (schema_name, table_name, column_name) ); INSERT INTO kfs_data_classification (schema_name, table_name, column_name, sensitivity_level, data_category, masking_rule, owner_team) VALUES ('crm','customer','customer_name','PII','NAME','name_partial','crm-team'), ('crm','customer','mobile','PII_HIGH','MOBILE','mobile_standard','crm-team'), ('crm','customer','email','PII','EMAIL','email_standard','crm-team'), ('crm','customer','id_card','PII_HIGH','ID_CARD','id_card_strict','crm-team'), ('crm','customer','address','PII_HIGH','ADDRESS','address_city_only','crm-team'), ('crm','customer','bank_card','FINANCIAL','BANK_CARD','bank_card_strict','risk-team'), ('crm','customer','city','INTERNAL','CITY','passthrough','crm-team');3.2 角色 × 字段类型策略矩阵
策略配置用 JSON 维护，放在 MCP Server 能读到的路径，比如config/masking_policy.json：
{ "policy_version": "2026-08-01", "default_action": "redact", "roles": { "customer_service": { "NAME": "name_partial", "MOBILE": "mobile_standard", "EMAIL": "email_standard", "ID_CARD": "id_card_strict", "ADDRESS": "address_city_only", "BANK_CARD": "bank_card_strict", "CITY": "passthrough" }, "risk_analyst": { "NAME": "passthrough", "MOBILE": "mobile_extended", "EMAIL": "email_extended", "ID_CARD": "id_card_extended", "ADDRESS": "address_district", "BANK_CARD": "bank_card_extended", "CITY": "passthrough" } } }注意default_action: redact这一行。它的含义是：分类未知的字段默认拒绝或全遮罩，而不是默认透传。假设 DBA 新增了customer.face_token，元数据分类还没同步，系统不能因为「没匹配到规则」就把原值返回。安全系统里未知状态要 fail closed。
3.3 掩码函数
def mask_mobile(value: str) -> str: if not value: return value return value[:3] + "****" + value[-4:] def mask_email(value: str) -> str: if not value or "@" not in value: return "***" local, domain = value.split("@", 1) return local[:1] + "***@" + domain def mask_id_card(value: str) -> str: if not value: return value return value[:4] + "*" * max(0, len(value) - 8) + value[-4:] def mask_name(value: str) -> str: if not value: return value if len(value) == 1: return value + "*" return value[0] + "*" * (len(value) - 1)这里有个细节：单字姓名王如果按value[0] + "*"处理会输出王*，实际上保留了完整姓名。所以单字要单独处理成王*之外的形式，或者直接返回*。这类边界值必须进回归测试。
3.4 MCP 查询工具
from mcp.server.fastmcp import FastMCP import json mcp = FastMCP("kfs-customer-query") POLICY = json.load(open("config/masking_policy.json")) CLASSIFICATION = load_classification() # 从 kfs_data_classification 读取 def mask_row(row: dict, role: str, table: str) -> dict: role_policy = POLICY["roles"].get(role, {}) masked = {} for col, val in row.items(): meta = CLASSIFICATION.get((table, col)) if meta is None: masked[col] = "***" # 未知字段默认遮罩 continue category = meta["data_category"] rule = role_policy.get(category, POLICY["default_action"]) masked[col] = apply_rule(rule, val) return masked @mcp.tool() def query_customer_readonly(sql: str, role: str, max_rows: int = 100) -> dict: rows = execute_readonly_sql(sql, max_rows) masked_rows = [mask_row(r, role, "customer") for r in rows] return { "ok": True, "masked": True, "policy_version": POLICY["policy_version"], "role": role, "row_count": len(masked_rows), "rows": masked_rows }关键点：execute_readonly_sql返回的明文只在函数内部短暂存在，masked_rows才是 MCP Tool 的返回值。明文允许存在于数据库 Buffer、网络连接、MCP Server 受控内存；明文不允许进入 Agent Context、模型请求、MCP Client 日志、前端页面。
3.5 模型侧接入配置
Agent 调用模型时，Base URL 指向 TaoToken 的 API 地址，Key 用控制台创建的 API Key，Model ID 填你选的对话模型。三件套缺一不可，配置示例：
{ "base_url": "https://taotoken.net/api", "api_key": "sk-你的Key", "model": "你的模型ID" }如果你用的是 Claude Code 这类编码 Agent，接入方式在 https://taotoken.net/doc 里有对应说明，Base URL 和 Key 的填法一致。
4. 验证请求：确认模型侧只收到掩码值
配置写完必须验证，否则你不知道脱敏到底生效没有。
4.1 准备样例数据
CREATE TABLE crm.customer ( customer_id bigint PRIMARY KEY, customer_name varchar(64), mobile varchar(20), email varchar(128), id_card varchar(32), address varchar(256), bank_card varchar(32), city varchar(64) ); INSERT INTO crm.customer VALUES (1,'张三','13812345678','zhangsan@example.com','110101199001011234','上海市浦东新区XX路88号','6222021234567890123','上海市'), (2,'李四','13998765432','lisi@example.com','310101199202022345','北京市朝阳区YY路66号','6222029876543210987','北京市');4.2 直接调用 MCP 工具验证
result = query_customer_readonly( sql="SELECT customer_name, mobile, email, id_card FROM crm.customer WHERE customer_id = 1", role="customer_service" ) print(json.dumps(result, ensure_ascii=False, indent=2))期望输出：
{ "ok": true, "masked": true, "policy_version": "2026-08-01", "role": "customer_service", "row_count": 1, "rows": [ { "customer_name": "张*", "mobile": "138****5678", "email": "z***@example.com", "id_card": "1101**********1234" } ] }看到masked: true和掩码值，说明 MCP 出口这一层生效了。
4.3 验证模型侧收到的内容
这一步最关键。在 Agent 里问「查一下张三的联系方式和证件信息」，然后在 TaoToken 控制台的请求日志里检查实际发给模型的 payload。你要确认的是：payload 里出现的是138****5678而不是13812345678，是1101**********1234而不是完整身份证号。
如果日志里还能搜到明文，说明脱敏层没接在正确位置——很可能你脱敏的是模型输出，而不是 MCP 返回值。
4.4 角色差异化验证
同一个查询换risk_analyst角色再跑一次，应该得到不同的掩码粒度：
result = query_customer_readonly( sql="SELECT customer_name, mobile, id_card FROM crm.customer WHERE customer_id = 1", role="risk_analyst" )风控角色下mobile应该是1381234****这类更宽的可见范围，customer_name可能是明文。如果两个角色返回完全一样，说明策略矩阵没生效。
4.5 未知字段验证
临时加一列不在分类表里的字段，比如SELECT customer_id, wechat_id FROM crm.customer，确认返回的是***而不是原值。这是 fail closed 的验证。
5. 常见报错排查：401、local proxy failed 与 reading choices
脱敏链路跑起来之后，报错通常集中在模型接入侧和 MCP 侧两块。下面是我实际遇到过的几类。
401 Unauthorized。最常见的原因是 API Key 没带对或者 Base URL 写错。检查两点：Key 是不是从 https://taotoken.net/api-keys 创建的、有没有多余空格；Base URL 是不是https://taotoken.net/api，不要写成带/v1或其他后缀的形式。如果 Key 刚创建就报 401，确认一下是不是复制时截断了。
local proxy failed。这个报错通常出现在 Agent 通过本地代理转发请求的场景。排查顺序：先确认 MCP Server 进程本身能正常访问数据库；再确认模型请求的出口配置没有指向一个已经失效的本地端口。如果你在 Agent 配置里同时写了 MCP 的本地地址和模型的 Base URL，注意别把两者搞混——MCP 是工具调用通道，模型是推理通道，它们走不同的地址。
reading choices 相关报错。这类错误一般出现在解析模型响应时，响应体里没有预期的choices字段。原因可能是：请求被中间层拦截返回了错误页、模型 ID 填错导致返回了非预期结构、或者请求体格式不对。先用模型对话页面手动发一条最简单的请求，确认 Base URL + Key + Model ID 三件套本身是通的，再回到 Agent 里排查。
OAuth 相关报错。如果你用的是 Claude Code 这类需要 OAuth 流程的客户端，报错往往和回调地址、token 刷新有关。接入文档 https://taotoken.net/doc 里有对应客户端的配置说明，按文档填 Base URL 和 Key 即可，不要自己拼 OAuth 端点。
脱敏没生效但没报错。这种最危险。排查：MCP Tool 的返回值里有没有masked: true；分类表里对应字段的data_category是不是拼错了；策略 JSON 里角色名和代码里传的role参数是否一致。我见过一次是策略里写customer_service，代码里传的是customerservice，结果走了default_action全遮罩，看起来「更安全」但其实是配置错误。
审计日志里出现明文。检查你的日志语句，是不是把before和after都打进去了。审计应该记录column=mobile action=MASK rule=mobile_standard，而不是before=13812345678 after=138****5678。否则脱敏系统只是把敏感信息从数据库搬到了日志系统。
6. 把脱敏做成可评测的工程能力
脱敏不是「替换几个星号」，它需要可评测。至少同时看两个方向的错误：漏脱敏率（应脱敏却返回明文的比例）和误脱敏率（不该脱敏却被处理的比例）。订单号13812345678只是长得像手机号，如果被正则误伤成138****5678，业务就出问题了。元数据驱动比纯正则稳定，原因就在这里。
最重要的指标是明文进入模型比例：发送给模型的敏感明文字段数 / 数据库返回的敏感明文字段数，目标应该是 0%。这个指标可以直接在 TaoToken 的请求日志侧做统计，因为模型入口是收敛的。
还有几个容易忽略的风险：字符串拼接绕过列级规则（SELECT mobile || '-' || id_card AS detail返回的列名是detail，按列名识别就会漏），JSON/TEXT 大字段里嵌套 PII，以及脱敏后仍可能被重识别（城市 + 年龄 + 客户等级 + 罕见订单组合起来可能定位到个人）。这些需要在后续治理里逐步补上，但第一步永远是：先把明文挡在模型之外。
如果你要长期跑这套 Agent 问答链路，建议把模型调用统一走 Coding Plan，配额和审计都更省心；只是验证脱敏效果的话，模型对话页面手动发几次请求就够。接入细节看文档，Key 在控制台创建。最后留一句我踩过坑之后的体会：不要把敏感明文交给模型再期待它保密，先在 MCP Server 把数据变成模型可以安全使用的形态，再交给 Agent 推理。
