30 分钟手搓一个 MCP Server，让 Claude 直接查公司内网 SQL

说实话这事儿是被运营部门逼出来的。

我们公司有个内部订单系统，MySQL 库里挂着几千万行数据。运营组的小姑娘几乎每天都要问技术：”帮我查一下昨天 XX 渠道的退款单数””帮我拉一下 3 月 GMV 按城市分组”。一开始还好，后来一天能来十几次，我们这边排期排不过来，她那边也急，经常下班前还在等一张报表。

上个月我心一横，花了一个下午给她搭了个 MCP Server，直接让 Claude Desktop 接到内网 MySQL 上。现在她自己用自然语言问，Claude 帮她写 SQL、跑 SQL、甚至画个简单的 Markdown 表格出来。这篇想把整个过程完整写一遍——不是教科书式的”hello world”，是真真正正上了生产、正在跑的那版。

一、为什么选 MCP，不选 Function Calling

先澄清一个事儿。这个需求技术上确实可以用 Function Calling 直接做——写个 Python 脚本，调 Claude API，把 execute_sql 定义成一个 tool，让模型生成 SQL 然后你本地执行。我一开始也是这么想的。

但有几个问题绕不开：

• 运营小姑娘不是程序员，她用的是 Claude Desktop 客户端，不可能让她打开终端跑你的脚本
• 公司之后还要接第二个工具（查 CRM）、第三个工具（发企业微信通知），每加一个都得改脚本
• 权限控制散在代码里，谁能查什么表、哪些字段要脱敏，改一次要重新部署

MCP 正好解决这几件事。它是协议层的东西，Server 一次写好，任何支持 MCP 的客户端都能接——Claude Desktop、Cursor、Continue、Cline，甚至自研的 Agent。关于 MCP 和 Function Calling 到底差在哪，我在 Tool与MCP 分类 里还有一篇专门拆过，这里就不展开了。

二、技术栈选型：为什么是 FastMCP

Anthropic 官方 SDK 是 mcp 这个包，能用但偏底层，你得自己处理 JSON-RPC 的细节。社区里有个叫 FastMCP 的封装，用起来就跟写 Flask 差不多，一个装饰器搞定一个 tool。生产项目我一律用 FastMCP，开发效率差了不止一个量级。

1

pip install fastmcp pymysql python-dotenv

就这么简单。Python 3.10+ 就行。

三、最小可运行版本：20 行代码接通数据库

先上最原始的版本，让整个链路跑通：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# server.py
import os
import pymysql
from fastmcp import FastMCP
from dotenv import load_dotenv

load_dotenv()
mcp = FastMCP("internal-sql")

def get_conn():
    return pymysql.connect(
        host=os.getenv("DB_HOST"),
        user=os.getenv("DB_USER"),
        password=os.getenv("DB_PASS"),
        database=os.getenv("DB_NAME"),
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
    )

@mcp.tool()
def query_sql(sql: str) -> list[dict]:
    """执行只读 SQL 查询，返回结果行列表"""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(sql)
            return cur.fetchall()

if __name__ == "__main__":
    mcp.run()

配上 .env：

1
2
3
4

DB_HOST=10.0.1.15
DB_USER=readonly_bot
DB_PASS=xxx
DB_NAME=orders

在 Claude Desktop 的配置文件里加一段（Mac 是 ~/Library/Application Support/Claude/claude_desktop_config.json）：

1
2
3
4
5
6
7
8

{
  "mcpServers": {
    "internal-sql": {
      "command": "python",
      "args": ["/path/to/server.py"]
    }
  }
}

重启 Claude Desktop，你就能在输入框看到一个小锤子图标，里面有 query_sql 这个工具。问它”帮我查最近 7 天退款最多的前 10 个订单”，它就会自己写 SQL 调用 tool，然后把结果整理成表格给你。

这一版能跑，但离生产还差十万八千里。下面是我真实踩过的坑。

四、坑一：权限——这玩意儿敢不敢让 Claude 随便写 SQL

这是我第一版上线当天就翻车的地方。

Claude 很聪明，但它不知道你公司的”业务约束”。运营小姑娘问了一句”把测试订单都删掉”，它还真就写了个 DELETE FROM orders WHERE ...。还好我当时用的是只读账号，SQL 执行直接报了权限错误。

教训是三层防御一个都不能少：

第一层，数据库账号只给 SELECT 权限。这是地板，永远不能塌。

1
2
3
4

CREATE USER 'readonly_bot'@'%' IDENTIFIED BY 'xxx';
GRANT SELECT ON orders.* TO 'readonly_bot'@'%';
-- 敏感表直接不给
REVOKE SELECT ON orders.user_identity FROM 'readonly_bot'@'%';

第二层，代码里白名单校验 SQL。用 sqlparse 解析，只允许 SELECT：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import sqlparse

def is_readonly(sql: str) -> bool:
    parsed = sqlparse.parse(sql)
    if not parsed:
        return False
    stmt_type = parsed[0].get_type()
    return stmt_type == "SELECT"

@mcp.tool()
def query_sql(sql: str) -> list[dict]:
    if not is_readonly(sql):
        raise ValueError("仅支持 SELECT 查询")
    # ... 原逻辑

第三层，敏感字段在返回层脱敏。手机号、身份证、邮箱这类东西，数据库里有，但返回给 Claude 之前要处理：

1
2
3
4
5
6
7

SENSITIVE_COLS = {"phone", "id_card", "email"}

def mask_row(row: dict) -> dict:
    return {
        k: (v[:3] + "****" + v[-2:] if k in SENSITIVE_COLS and isinstance(v, str) and len(v) > 5 else v)
        for k, v in row.items()
    }

这三层加起来，运营组再怎么折腾也不会把线上数据搞挂。

五、坑二：超时——大表一跑就卡死整个 Claude

早期版本我没做超时控制。结果有一次她问”把去年所有订单按用户维度聚合”，Claude 写了个没加 LIMIT 的聚合 SQL，直接把 MCP Server 卡住了 40 秒。Claude Desktop 那边超时断开，整个会话都崩了。

修复方案两步走：

数据库层加 max_execution_time：

1
2
3
4
5
6
7
8
9
10
11

@mcp.tool()
def query_sql(sql: str, limit: int = 1000) -> list[dict]:
    if not is_readonly(sql):
        raise ValueError("仅支持 SELECT 查询")
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute("SET SESSION MAX_EXECUTION_TIME=10000")  # 10 秒
            if "limit" not in sql.lower():
                sql = f"SELECT * FROM ({sql}) t LIMIT {limit}"
            cur.execute(sql)
            return [mask_row(r) for r in cur.fetchall()]

行数上限硬编码。Claude 要是真给你返回 10 万行，上下文就爆了。默认 1000 行，需要更多的话让用户显式传参。

六、坑三：日志——出了事儿要能复盘

这个是我朋友血泪教训。他们那边有次数据被疑似”泄露”，老板要追查每一次查询，但 MCP Server 完全没打日志，只好翻 MySQL 的 general log，工作量巨大。

我的做法是每次调用都打结构化日志，包括 SQL、调用方、执行时间、返回行数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

import logging
import json
import time
from pathlib import Path

log_path = Path("/var/log/mcp-sql/audit.log")
log_path.parent.mkdir(parents=True, exist_ok=True)

audit = logging.getLogger("audit")
audit.addHandler(logging.FileHandler(log_path))
audit.setLevel(logging.INFO)

@mcp.tool()
def query_sql(sql: str, limit: int = 1000) -> list[dict]:
    t0 = time.time()
    try:
        if not is_readonly(sql):
            raise ValueError("仅支持 SELECT 查询")
        # ... 执行
        elapsed = time.time() - t0
        audit.info(json.dumps({
            "ts": int(t0),
            "sql": sql,
            "rows": len(result),
            "ms": int(elapsed * 1000),
            "status": "ok",
        }, ensure_ascii=False))
        return result
    except Exception as e:
        audit.info(json.dumps({
            "ts": int(t0),
            "sql": sql,
            "status": "error",
            "error": str(e),
        }, ensure_ascii=False))
        raise

日志按天切，配合 Loki 或者直接 grep，出事追责分分钟。

七、Docker 化：让它能在公司服务器上稳定跑

本地跑是给自己用的，上生产给全运营组用就得容器化。Dockerfile 非常朴素：

1
2
3
4
5
6
7
8
9
10

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY server.py .
COPY .env .

CMD ["python", "server.py", "--transport", "sse", "--port", "8080"]

MCP 官方支持两种 transport：stdio（本地进程间）和 SSE（HTTP 长连）。给团队用一定要选 SSE 模式，放内网服务器上，所有人客户端配一个 URL 就行：

1
2
3
4
5
6
7

{
  "mcpServers": {
    "internal-sql": {
      "url": "http://10.0.1.30:8080/sse"
    }
  }
}

八、上线之后的真实反馈

到现在用了快一个月。运营组的使用频率：日均 40-60 次查询，高峰期一小时 20 次。我这边完全不用管了，她自己跟 Claude 聊两句就拿到数据。

有几个没预料到的好处：

• 她开始学 SQL 了。看多了 Claude 生成的 SQL，她现在自己能读懂 JOIN 和 GROUP BY
• 需求被过滤了一轮。以前来找我们的需求十个有三个是”她想错了”，Claude 会在执行前复述一遍”你是想查 XX 对吧？”，经常她就发现自己问反了
• 跨部门传播。财务组听说了，现在在催我给她们也搞一个

下一步我打算把 CRM 客户查询、企业微信消息推送也做成 MCP Tool，塞进同一个 Server。想了解更多 MCP 实战案例的话，可以去 openclaw.cocoloop.cn 看看，那边有更多开源项目的分享。如果你对 Claude 在企业内部的深度集成感兴趣，claudecode.cocoloop.cn 也有不少相关内容。

写到这里差不多把我知道的坑都倒完了。MCP 这东西说难不难，三十分钟确实能跑起来；说简单也不简单，真要用在生产上，权限、超时、日志、脱敏，每一环都得踩一遍才放心。