做一个数据 MCP，让 AI 直连数据库

打开学院的数据门户（内网地址，连上校园网才能访问），每个数据集一张卡片：服务器、端口、用户名、密码写得清清楚楚，然后一行「数据表 略」，再贴一段可以复制的 pandas 连接示例。这一页信息对人是够用的，对 AI 是断的。

断在四处，一处一处说。

第一，AI 不知道怎么连数据库。门户上的地址、账号、密码，你得自己复制下来、粘贴到对话框里告诉 AI；下次换一个数据集，这套动作再来一遍。

第二，AI 不知道库里有哪些表。门户的「数据表」一栏只写着一个「略」字，什么表、什么字段都没有列。比如高频股票数据库里有 5237 张表，表名就是 6 位股票代码——你不说，AI 连从哪张表查起都无从下手。

第三，AI 不知道数据里的坑。这些坑只有用过的人才知道：高频行情的价格是后复权的，直接拿来算收益会算错；时间字段存错了位置，取出来的时间是错的；有的表有 3 亿多行，一句不带条件的查询就能把服务器拖垮。AI 不知情，照查不误——查错了，它自己也不知道错在哪里。

第四，查数和用数是分开的。你在数据库软件里把数查出来、导成表格，再粘贴到对话框里给 AI 分析。聊天记录一长，前面贴过的数据它就记不清了；它说一句「再帮我看看隔壁那张表」，你又得切回数据库重新导一次。

于是日子就过成了这样：打开浏览器看表结构 → 打开 DBeaver 写 SQL → 导出 CSV → 粘贴给 AI → AI 说「再查一下隔壁表」→ 又回到第一步。一晚上就耗在几个窗口之间来回切换，人成了数据库和 AI 之间的搬运工。上一篇讲过：MCP 的作用是把「AI 的能力」做成标准插座，AI 缺什么就给它插什么。这次 AI 缺的是数据，我们就做一个数据 MCP——把插座直接插到数据库上。

一、数据 MCP 要解决的四个新问题

下面几个名词都在上一篇里逐个讲过，这里只点名复习：Host 是你正在用的那个 AI 应用（豆包办公、WorkBuddy、Trae……），它内部装着一个 MCP Client（专职传话的协议层）；Client 负责启动 MCP Server——也就是你要写的那个程序，它提供工具、真正去连数据库；stdio 是 Client 和 Server 之间的通信方式（两个进程之间的管道，不走网络、不开端口）；双方来回传递的消息都写成 JSON-RPC 格式（一种「一问一答」的标准报文，问题和答案都按固定字段组织）。哪个词陌生，就回上一篇的对应小节细读，不影响继续往下看。数据场景给这道「能力题」加了四个新的约束，它们决定了本文的全部设计：

设计目标浓缩成一句话：学生在对话框里说「用中国专利数据统计各年份的申请量」，其余全部自动发生——名称解析、连库、读结构、写 SQL、渲染表格，模型自己完成，人只看结果。

二、项目全貌：三个文件加一个环境

整个项目小得可以一眼看完：一个数据目录、一个服务端、一个启动脚本，加上隔离的 Python 环境。

intern-data-mcp/
├── server.py          # MCP 服务端全部逻辑（约 630 行）
├── datasets.json      # 数据集目录：连接 + 中文名 + 别名 + 踩坑笔记
├── requirements.txt   # mcp / psycopg / pymssql / pymongo / pandas
├── run.sh             # 统一启动脚本（所有客户端都指向它）
├── configs/           # 各客户端的现成配置片段
│   ├── workbuddy.json / trae.json / cursor.json
│   ├── claude_desktop.json / codex.toml
├── README.md          # 使用说明
└── GUIDE.md           # 构建指南
      

一句话架构：客户端说中文 → MCP 解析成连接参数 → 连真实数据库取数 → 渲染成 Markdown 表格返回。全程只读，三层防护。与上一篇的日历 MCP 相比，语言从 TypeScript 换成了 Python（数据科学生态更顺手），场景从「写系统日历」换成了「读数据库」——协议层一模一样，这正是 MCP 的意义：换个 Server，客户端什么都不用改。

三、五个工具：给 AI 的一条「数据探索阶梯」

拿到一个陌生数据库，人的探索顺序是「先看有什么，再看结构，再取数，需要时才写代码」。工具设计应该对齐这条路径，而不是做一个大而全的 query(anything)——小工具的 description 更清晰，模型的选择更准，出错时也知道自己该退回哪一级。

上一篇说过 description 是「给模型看的手册」，这一篇多了一个新知识点：Server 级的 instructions。它相当于 Server 一进场就递上的自我介绍，告诉模型这个 Server 是干什么的、典型工作流是什么——模型还没调任何工具，就已经知道「数据以中文名称标识，先 list 再 describe 再 run_sql」。五个工具加上这段话，探索路径就长在了协议层里。

四、一步步搭建

下面九步是完整路线。代码只给关键骨架，重点是「每一步解决什么问题」——理解了为什么，代码是水到渠成的事。

1. 建隔离环境。数据驱动依赖较重（三种数据库客户端加 pandas），务必用独立 venv，不污染系统 Python：

   python3 -m venv ~/envs/intern-data-mcp
   ~/envs/intern-data-mcp/bin/pip install mcp "psycopg[binary]" pymssql pymongo pandas numpy
             

   官方 SDK 包名就是 mcp；psycopg[binary] 带上二进制轮子，省去本地编译。

2. 探数据——整个项目最重要的一步。写 MCP 之前先当「数据侦探」：用 DBeaver、MongoDB Compass 或一段临时脚本，把每个库的表名、行数、字段、样本值全摸一遍。门户页面没写的真相都在这一步现形，例如：

   数据集	探测发现
   高频股票数据	5237 张表、表名即股票代码；OHLC 是后复权价；trade_time 有存储缺陷，09:30 被存成 00:09:30
   中国专利数据	patents 3883 万行，中文列名须加双引号；日期多为文本，好在「申请年份」是数值列
   企查查数据	company_change_info 3.23 亿行——任何不带 WHERE 的查询都会拖垮服务器
   EPO 专利数据	SQL Server 方言：用 top 不用 limit；大表排序实测 77 秒
   东方财富股吧	guba_comments 2.52 亿条，Mongo 查询必须带 filter 与 limit

   这些坑不写下来，每个 AI 客户端、每个学生都要重新踩一遍。它们是下一步的原材料。

3. 写 datasets.json：数据通讯录 + 踩坑笔记。把探测结果结构化。每个数据集一个条目，中文名给学生看，别名给模型容错，坑按条写入 caveats：

   {
     "sources": {                          // 三个库的连接（密码可用环境变量覆盖）
       "pg_main":    { "kind": "postgres", "host": "…", "port": 5432, "user": "readonly_…", "password": "…" },
       "mssql_epo":  { "kind": "mssql",    "host": "…", "port": 1433, "user": "…", "password": "…" },
       "mongo_main": { "kind": "mongodb",  "host": "…", "port": 27017, "user": "…", "password": "…", "auth_source": "admin" }
     },
     "datasets": [
       {
         "name": "高频股票数据",             // 中文名：学生在对话里直接说它
         "aliases": ["hf_stockdata", "stock_hf", "高频股票", "高频数据"],
         "kind": "postgres", "source": "pg_main", "database": "stock_hf",
         "table_hint": "每只 A 股一张表，表名即 6 位代码，引用需加双引号：public.\"000001\"",
         "caveats": [
           "【重要】OHLC 均为后复权价，原始价 = close_price / post_adjustment_factor",
           "【重要】trade_time 存储有缺陷：09:30 存成了 00:09:30，真实时间需还原"
         ]
       }
       // …共 10 个条目
     ]
   }
             

   设计要点有两个。sources 与 datasets 分离：3 个库的连接写一次，10 个数据集引用它——以后加数据集只改 JSON，不改代码。caveats 是一等公民：describe_dataset 会把坑原样吐给 AI，等于把老学生的经验装进了协议层。

4. 搭 Server 骨架，守住 stdio 纪律。官方 Python SDK 里起一个 Server 非常短：

   from mcp.server.mcpserver import MCPServer
   
   server = MCPServer(
       name="intern-data", title="…内部数据", version="1.0.0",
       instructions=("数据以中文名称标识，例如：高频股票数据、中国专利数据……"
                     "典型流程：list_datasets() → describe_dataset(名称) → run_sql(名称, SQL)。"
                     "全部为只读访问。"),
   )
   
   def log(*a):
       print("[intern-data]", *a, file=sys.stderr, flush=True)   # 关键：走 stderr
             

   上一篇踩过的头号坑在这里同样致命：stdio 模式下 stdout 只能输出协议 JSON，任何 print 调试都会污染管道、让客户端解析失败。日志一律走 stderr。

5. 名称解析 resolve()：说人话就能取数的核心。两级匹配——先精确比对中文名 / 英文别名 / 库名，失败再做包含匹配（用户说「用一下高频股票数据那个库」也能命中）。实测输出：

   resolve('高频股票数据')  ->  高频股票数据
   resolve('stock_hf')      ->  高频股票数据      # 英文别名
   resolve('股吧')           ->  东方财富股吧数据   # 模糊说法
   resolve('EPO')           ->  EPO 专利数据
             

   解析失败时不抛异常，而是返回一句友好提示加 list_datasets() 指引——模型读到提示会自己去查目录，对话不会中断。

6. 连接管理：三个工厂 + 只读上锁。每种库一个工厂函数，连接进程内缓存避免反复握手。最值得抄的一行是 PostgreSQL 连接参数：

   conn = psycopg.connect(
       host=s["host"], port=s["port"], user=s["user"], password=s["password"],
       dbname=dbname, autocommit=True,
       options=f"-c statement_timeout={STMT_TIMEOUT_MS} -c default_transaction_read_only=on",
   )
             

   连接一建立就是只读事务、且语句 120 秒超时——防线设在连接层，不依赖调用方自觉。SQL Server 没有 LIMIT，run_sql 会自动把查询包成 select top(N) * from (…) as _sub，失败再回退 fetchmany 截断——方言差异在 Server 内部消化，模型无感。

7. 语句层只读校验 assert_readonly()。先剥掉字符串常量与注释，再查写关键字，最后要求首词在白名单里：

   def assert_readonly(sql: str) -> None:
       bare = _strip_literals(sql)          # 剥字符串/注释，防误伤
       if _WRITE_KW.search(bare):           # insert|update|delete|drop|… 
           raise ValueError(f"只允许只读查询, 检测到关键字 {…}")
       head = bare.strip().split(None, 1)
       if head[0].lower() not in ("select", "with", "show", "explain", "table", "values"):
           raise ValueError(f"只允许 SELECT / WITH 查询, 收到 {head[0]!r}。")
             

   为什么要先剥字符串？看实测就明白——select 'delete' as x 是完全合法的查询，不剥会误拦；而 delete from public.company 必须拦：

   放行: select * from t
   放行: select 'delete' as x          # 字符串里的 'delete' 不误伤
   拦截: delete from public.company   -> 检测到关键字 'DELETE'
   拦截: update company set a=1        -> 检测到关键字 'UPDATE'
   放行: with x as (select 1) select * from x
             

8. 注册五个工具。用装饰器把函数暴露成 MCP 工具，重点是 description 与错误处理：

   @server.tool(
       name="run_sql",
       description="用 SQL 查询指定的内部数据集(PostgreSQL 或 SQL Server), 返回结果表格。"
                   "仅允许只读 SELECT/WITH 语句。dataset 传中文数据名称即可, 例如 '中国专利数据'。",
   )
   def run_sql(dataset: str, sql: str, limit: int = 100) -> str:
       ds, msg = resolve_or_msg(dataset)
       if ds is None:
           return msg
       try:
           return exec_sql(ds, sql, limit)
       except Exception as e:
           return ("❌ 查询失败 … 排查建议: 先用 describe_dataset 核对表名与字段名; "
                   "注意方言差异 (PostgreSQL 用 limit, SQL Server 用 top)。")
             

   注意两点。错误也不抛异常，而是降级成一段带排查建议的文本——模型读到「先核对表名」「注意方言」能自己改 SQL 重试，这是给 AI 设计返回值的通用技巧：把错误信息写成给模型看的调试指南。run_python 预注入连接对象：pg()、mssql()、mongo()、query_sql()、read_pg()、read_mssql()、pd、np 全部注入执行环境，代码里只写分析逻辑不写连接代码；print 的内容被捕获返回，最后一个表达式若是 DataFrame 会自动打印成表格。担心代码执行风险，可以设 INTERN_DATA_ALLOW_PYTHON=0 整体关闭。

9. 测试：单元级 + 协议级。单元级直接 import server 调 resolve() / assert_readonly()（上面那些输出就是这么来的）。协议级才是决定性验证——用 SDK 自带的客户端库起一个真 Client，完整走一遍握手：

   from mcp import ClientSession, StdioServerParameters
   from mcp.client.stdio import stdio_client
   
   async with stdio_client(StdioServerParameters(
           command="~/envs/intern-data-mcp/bin/python", args=["server.py"])) as (r, w):
       async with ClientSession(r, w) as s:
           info  = await s.initialize()          # 握手
           tools = await s.list_tools()          # 发现工具
           res   = await s.call_tool("run_sql",  # 调用工具
                    {"dataset": "模糊性歧义数据",
                     "sql": "select count(*) as n from public.ambiguity_000001"})
             

   真实运行输出：

   [initialize] server = 'intern-data' v1.0.0
   [tools/list] 5 个工具:
     - list_datasets / describe_dataset / run_sql / mongo_find / run_python
   [tools/call] run_sql 输出:
       | n |
       | --- |
       | 2545 |        # ambiguity_000001 表，0.47 秒
             

   这一步通过，任何 MCP 客户端都能用你的 Server——因为协议是标准的，剩下的只是配置。

五、安全设计：只读是三层防线，不是一句承诺

「让 AI 直连数据库」听上去危险，所以安全必须是设计出来的结构，而不是 README 里的一句保证。这一篇的方案是纵深防御：

再把兜底参数列全，它们都可以用环境变量调整而不必改代码：

六、run.sh：一条路径服务所有客户端

Server 要被五个客户端拉起，每个客户端的配置里都得写「启动命令」。如果各写各的，有人写 venv 的 python 路径、有人写绝对路径，以后 venv 一搬家就要改五处。解法是加一层薄薄的启动脚本，把「怎么启动」封装成一条固定路径：

#!/usr/bin/env bash
# intern-data MCP 统一启动脚本：所有客户端只需引用这一条路径
set -euo pipefail
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
VENV_PY="$HOME/envs/intern-data-mcp/bin/python"
exec "$VENV_PY" "$DIR/server.py" "$@"
      

别忘了 chmod +x run.sh。此后所有客户端配置里只出现这一行——command 指向 run.sh，环境怎么变都只改这一处脚本。一个小技巧：exec 让 python 直接替换 shell 进程，信号传递和进程树都干净。

七、挂载：五个客户端，一套 Server

原理上一篇讲透了：客户端读自己的配置文件，里面一条 command 指向你的启动脚本，stdio 管道接通即用。各家的差异只在配置文件的位置与格式（Codex 用 TOML）。通用 JSON 就一段：

{
  "mcpServers": {
    "intern-data": {
      "command": "/path/to/intern-data-mcp/run.sh"
    }
  }
}
      

WorkBuddy

配置文件 ~/.workbuddy/mcp.json，把上面的通用 JSON 合并进 mcpServers；然后到连接器管理页右上角「自定义连接器」找到 intern-data 点 Trust 启用。这是本项目的主力客户端——它的托管 Python 正是 venv 的来源。

Trae

两种方式任选。界面添加（推荐）：设置 → MCP → 添加 → 手动添加 → 粘贴通用 JSON → 确认。项目级共享：在项目根建 .trae/mcp.json 写入同样内容，再到设置里启用项目级 MCP——配置随仓库走，课题组人人可得。

豆包办公（桌面端）

与上一篇挂日历 MCP 的做法完全一样：设置 → MCP 连接器，以 stdio 方式新增，填入 run.sh 的路径。配好后直接在对话框说「用互动易数据看看 2023 年有多少条提问」。

Codex CLI

配置文件 ~/.codex/config.toml——注意Codex 用 TOML，且顶层键是 mcp_servers 而不是 mcpServers，这是最常见的手滑点：

[mcp_servers.intern_data]
command = "/path/to/intern-data-mcp/run.sh"
# 可选: startup_timeout_ms = 20000
# 可选: env = { INTERN_DATA_MAX_ROWS = "500" }
      

也可以用命令行一键添加：codex mcp add intern_data -- /path/to/intern-data-mcp/run.sh，用 codex mcp list 与 TUI 里的 /mcp 验证。特别注意 Codex 的沙箱：它默认把子进程关进无网络的沙箱，MCP server 连不上内网数据库时会超时——遇到「连不上」，在配置里放宽沙箱或确认 MCP 进程可访问网络。

Zcode 与其他任意 stdio 客户端

Zcode、Cursor、Claude Desktop、Cherry Studio、Cline……凡是支持 MCP stdio 的客户端，配置都是同一段通用 JSON，差别只是各自的 MCP 设置入口在哪。找不到入口时查该客户端的 MCP 文档，认准「自定义 MCP / MCP 服务器 / 命令行启动」一类字样即可。

常见坑速查

八、用起来：三个真实提问

挂载完成后，对话就是全部操作。三个真实例子，观察模型如何沿「探索阶梯」自己走完全程：

1. 「用中国专利数据统计各年份的申请量。」模型先 describe_dataset 读结构，caveats 里明明白白写着「日期多为文本，按年份筛选优先用『申请年份』数值列」——于是它写出 select "申请年份", count(*) from public.patents group by 1，避开文本日期的慢查询。坑还没踩就已被绕开。
2. 「查高频股票数据里 000001 在 2024 年 5 月 27 日的分钟收盘价，还原成原始价格。」模型在 describe 返回里读到「OHLC 为后复权价」的提示，转而用 run_python：read_pg("stock_hf", 'select … from public."000001" where …') 拿回 DataFrame，除以 post_adjustment_factor 再打印——一条龙不需要人解释任何背景。
3. 「从上市企业年报数据取平安银行 2020 年报正文的开头五百字。」模型用 mongo_find，filter 写 {"firmcode": "000001", "date": 2020}（firmcode 是字符串、date 只是年份——这些也写在 caveats 里），projection 排除无关字段，取回 content 截断展示。

背后的关键只有一条：隐性知识已经搬进了 datasets.json。以前这些坑靠口口相传，现在每个连上来的 AI 客户端第一次 describe 就自动继承——这也是数据 MCP 相比「贴连接串给 AI」最大的增值。