MCP 服务器怎么连数据库：连接池、只读约束与 SQL 安全边界

MCP 服务器连数据库的安全底线只有一条：给 AI 的账号必须是只读账号，并且在数据库协议层就把写入挡死，而不是靠提示词告诉模型「请不要修改数据」。提示词能被注入绕过，数据库权限不能。下面是从账号、语句、超时到审计的完整落地做法。

四道防线，缺一道都不算安全

很多教程只讲第二道，那是本末倒置。真正不可绕过的是第一道——数据库端就拒绝，就算工具被完全攻破也写不进去。

第一步：建一个真正的只读账号

-- PostgreSQL
CREATE USER mcp_readonly WITH PASSWORD 'strong_password_here';
GRANT CONNECT ON DATABASE appdb TO mcp_readonly;
GRANT USAGE ON SCHEMA public TO mcp_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO mcp_readonly;
ALTER USER mcp_readonly SET default_transaction_read_only = ON;

-- 核验：只应看到 SELECT
SELECT grantee, privilege_type
FROM information_schema.role_table_grants
WHERE grantee = 'mcp_readonly';

MySQL 同理：GRANT SELECT ON appdb.* TO 'mcp_readonly'@'%';，再加上 SET SESSION TRANSACTION READ ONLY 的习惯。注意别偷懒用主库管理员账号「先跑通再说」，这是 MCP 数据库事故最常见的起点。

第二步：工具层再拦一次写语句

import re
import sqlite3
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("db-readonly")

WRITE_PATTERN = re.compile(
    r"\b(INSERT|UPDATE|DELETE|DROP|ALTER|TRUNCATE|CREATE|GRANT|COPY|ATTACH)\b",
    re.IGNORECASE,
)

MAX_ROWS = 200
TIMEOUT_MS = 5000

@mcp.tool()
def query(sql: str, params: list | None = None) -> dict:
    """执行只读 SQL 查询。只支持 SELECT，自动加 LIMIT，带超时。"""
    stripped = sql.strip().rstrip(";")
    if ";" in stripped:            # 堵住多语句
        return {"error": "不支持多语句"}
    if WRITE_PATTERN.search(stripped):
        return {"error": "只允许 SELECT 查询"}
    if not stripped.lower().startswith("select"):
        return {"error": "只允许 SELECT 查询"}

    conn = sqlite3.connect("file:app.db?mode=ro", uri=True, timeout=TIMEOUT_MS / 1000)
    conn.execute(f"PRAGMA query_only = ON")
    try:
        cur = conn.execute(f"{stripped} LIMIT {MAX_ROWS}", params or [])
        cols = [d[0] for d in cur.description]
        rows = cur.fetchall()
        return {"columns": cols, "rows": rows, "truncated": len(rows) == MAX_ROWS}
    finally:
        conn.close()

关键点：参数化（? / $1 占位符）而不是字符串拼接；显式拒绝多语句；强制追加 LIMIT；连接本身用只读模式打开。

第三步：连接池与并发，别让 AI 把连接打满

数据库 MCP 服务器通常是长驻进程，每次查询新建连接会迅速耗尽数据库连接数。要点：

• 用连接池：池大小按「并发客户端数 × 2 」设，通常 5–10 就够，别给 50 。
• 设置 statement_timeout：PostgreSQL 建议 5 秒，超时直接返回错误让模型改写查询，而不是干等到锁表。
• 限制返回结果：除了 LIMIT，还要限制单行体积（比如 text 字段截断到 2000 字符），否则一条大字段就能吃掉几十 K token 。
• 只读副本优先：能连从库就连从库，把 AI 的查询压力和主库彻底隔离。

第四步：把 schema 喂给模型，减少无效往返

模型不知道表结构时，会先猜、报错、再猜，白白多花三四轮调用。做法是在工具描述里直接写关键表和字段，或者单独提供一个 describe_schema 工具：

@mcp.tool()
def describe_schema() -> dict:
    """返回关键表的字段名、类型与含义，供生成 SQL 前查阅。"""
    return {
        "orders": {
            "id": "int, 订单主键",
            "user_id": "int, 下单用户",
            "status": "enum(pending|paid|refunded)",
            "created_at": "timestamp",
        },
        "users": {"id": "int", "email": "string", "created_at": "timestamp"},
    }

注意只暴露非敏感表。含手机号、身份证、密码哈希的表和字段，要么不给 SELECT 权限，要么在 schema 里直接不列出。

审计：每一次查询都要留痕

• 把执行的 SQL 、参数、耗时、行数写到 stderr 或独立审计日志（不要回传给模型）。
• 记录客户端标识与会话 ID，便于事后追责。
• 不要在工具返回值里回显完整连接串或凭据。

三个常见错误

1. 用主库管理员账号跑通就上线：一次提示注入就可能是删库。
2. 不设超时：模型写出一个笛卡尔积 JOIN，整库被拖慢。
3. 把 schema 和查询结果全量塞进上下文：几十张表的 DDL 会吃掉大量 token，只给关键表。

权限边界的更多细节见 MCP 安全风险怎么防：提示注入、权限边界与 URL 校验；服务端骨架怎么搭可以看 用 TypeScript 写 MCP Server；错误该分几类返回，参考 MCP 错误处理怎么做；并发压不上去时再看 MCP Server 并发上不去怎么办。