用 MCP 把数据库开放给智能体

这篇适合谁，读完能做到什么

适合已经了解 MCP 基本概念，想把真实数据库交给智能体使用的开发者。读完你应该能看懂一套多门店零售数据库的表结构，知道怎样用行级安全（Row Level Security）限制每个门店只能看到自己的数据，并能照着搭出一个带语义搜索的商品库。

先说清楚：这一篇主要讲数据库

这篇的场景是虚构的零售系统 Zava Retail。MCP 服务器要连接的是 PostgreSQL 数据库，所以这一部分讲的是数据库本身：表结构、权限、搜索函数、审计、备份。真正的 MCP 服务器（用 FastMCP 实现查询工具）放在后面^[1]。把数据库先设计好，智能体通过 MCP 调用时才不会越权，也不会因为数据量变大而搜索缓慢。

准备：PostgreSQL 与 pgvector

数据库使用 PostgreSQL，并加装 pgvector 扩展，用来做语义搜索，让顾客可以用自然语言找商品^[1]^[2]。启用扩展需要三个扩展：uuid-ossp、pgcrypto 和 vector。下面这段 SQL 启用它们，并检查 vector 是否已安装：

-- Enable required extensions
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
CREATE EXTENSION IF NOT EXISTS "pgcrypto";
CREATE EXTENSION IF NOT EXISTS "vector";

-- Verify vector extension installation
SELECT * FROM pg_extension WHERE extname = 'vector';

第一步：门店就是租户

这套表结构采用共享数据库、共享 schema 的多租户模式，并用行级安全隔离数据^[1]。租户的主表是 retail.stores，示例数据里有四个门店：seattle、redmond、bellevue 和 online^[1]。customers、products、sales_transactions 和 product_embeddings 这几张表都带有 store_id 列，用来标明数据属于哪个门店^[1]。

这里有一个容易忽略的细节：sales_transaction_items（订单明细）表没有 store_id 列。它的策略改为检查 transaction_id 是否属于当前门店的某笔销售，即通过关联来判断归属^[1]。

商品的 SKU 在门店内唯一，约束是 unique_sku_per_store UNIQUE (store_id, sku)，也就是说不同门店可以有相同的 SKU^[1]。

第二步：用行级安全限制访问

开启行级安全之后，每张表都要配一条策略。下面是客户表的策略，它把角色 mcp_user 限制在当前门店之内^[1]：

CREATE POLICY customers_store_isolation ON retail.customers
    FOR ALL
    TO mcp_user
    USING (store_id = current_setting('app.current_store_id', true));

策略里的 current_setting('app.current_store_id', true) 读取的是会话变量。门店由数据库函数设置，而不是由智能体自己声明，这样一个智能体即使在提示词里要求查看其他门店，数据库层也会挡住^[1]。这是这一课最值得记住的设计：安全边界放在数据库，而不是放在模型的指令里。

第三步：商品向量与语义搜索

product_embeddings 表保存的内容包括：被向量化的原始文本、一个 vector(1536) 类型的向量列，以及模型名称。源码注释说 1536 维对应 OpenAI 的 text-embedding-3-small 模型^[1]。

同一商品对同一个模型只能有一条向量，由 (product_id, embedding_model) 上的唯一约束保证^[1]。向量索引使用 HNSW，并采用 vector_cosine_ops，源码把它描述为快速的近似搜索^[1]。

搜索函数 retail.search_products_by_similarity 接收一个 1536 维的查询向量，默认相似度阈值为 0.7，默认最多返回 20 条^[1]。它把相似度算作 1 - (pe.embedding <=> search_embedding)，只保留达到阈值的结果，并且只返回当前门店中处于启用状态的商品，最后按向量距离排序^[1]。

这个函数声明为 SECURITY DEFINER，并把执行权限授予 mcp_user^[1]。这意味着函数以定义者的权限运行，所以函数体内必须自己做好门店过滤，源码正是这样做的。

第四步：创建 mcp_user 角色

智能体连接数据库时使用专门的 mcp_user 角色，不要用超级用户。下面的代码创建角色并授予 schema 使用权限^[1]：

-- Create MCP application role
CREATE ROLE mcp_user LOGIN;

-- Grant schema usage
GRANT USAGE ON SCHEMA retail TO mcp_user;

源码还给该角色授予了 retail 下所有表的 SELECT、INSERT、UPDATE、DELETE 权限，并通过 ALTER DEFAULT PRIVILEGES 让以后新建的表也自动授权^[1]。注意，授予的是表级权限，真正限定可见范围的仍是前面的行级策略。

门店上下文由 retail.set_store_context 函数设置。它先检查门店存在且处于启用状态，无效或停用的门店会直接抛出异常；通过后用 set_config 写入 app.current_store_id，并往审计日志写一条记录^[1]。

第五步：审计日志

retail.audit_log 表记录表名、操作类型、用户名、门店、记录 ID，以及修改前后的值（以 JSON 保存）^[1]。customers、products 和 sales_transactions 上都挂了审计触发器，每次插入、更新或删除都会写一条日志^[1]。这对于合规检查很有用，也能看出智能体到底改过什么。

第六步：生成测试数据

示例数据脚本使用 asyncpg 连接数据库，用 Faker 生成假数据，一次生成 2000 名顾客、500 件商品和 5000 笔销售^[1]。顾客的会员等级分为 bronze、silver、gold 和 platinum，权重分别为 50、30、15 和 5^[1]。生成器对小计加 8% 的税，并给约 20% 的明细行随机打折^[1]。

第七步：维护与备份

维护函数 retail.perform_maintenance 对主要表执行 ANALYZE，对 customers 和 products 执行带 ANALYZE 的 VACUUM，并对全文搜索索引和向量索引执行 REINDEX CONCURRENTLY^[1]。源码给的定时方式是 cron，例如 0 2 * * 0 psql -d retail_db -c "SELECT retail.perform_maintenance();"^[1]。这个表达式里第一个 0 是分钟，2 是小时，最后的 0 表示星期日，所以它会在每周日凌晨两点执行。

备份脚本使用 pg_dump 的自定义格式，压缩后用 pg_restore --list 校验，并删除超过 30 天（RETENTION_DAYS=30）的旧备份^[1]。核心命令如下：

pg_dump \
    --host="$DB_HOST" \
    --port="$DB_PORT" \
    --username="$DB_USER" \
    --dbname="$DB_NAME" \
    --verbose \
    --clean \
    --create \
    --if-exists \
    --format=custom \
    --file="$BACKUP_FILE"

恢复脚本先用 createdb 创建目标库，再用 pg_restore --clean --if-exists 恢复，最后统计 retail schema 下的表数量，用来确认恢复是否完整^[1]。

常见问题和限制

行级安全依赖会话变量。如果应用层没有调用 set_store_context，策略里 current_setting 读到的就不是有效门店，查询结果会为空，所以连接池的使用方式要和门店上下文一起设计。
搜索函数用 SECURITY DEFINER 运行，函数体中的门店过滤不能省。
向量维度和模型必须一致。表里的列是 vector(1536)，换成别的嵌入模型时要同步修改维度，并重新生成向量。
示例里的数字（2000 名顾客、8% 税率、30 天保留期）只是演示设定，正式环境要根据业务和合规要求调整。