手写 MCP Server 连数据库：50 行代码让 AI 学会查 SQL
![]()
这事得从我上个月接的一个需求说起
产品经理跑过来说："能不能让公司的 AI 客服直接查数据库？用户问'我的订单到哪了'，AI 自己跑 SQL 查了回复。"
我第一反应是——开玩笑吧？让 AI 直接操作数据库，这不等于把厨房刀递给三岁小孩？
后来研究了一下 MCP 协议（Model Context Protocol），发现 Anthropic 已经把这套东西标准化了。AI 不直接操作数据库，而是通过一个 MCP Server 做中间层，暴露有限的、可控的工具接口。AI 只能调你定义好的工具，不能随便执行 DROP TABLE。
下面是我折腾出来的最小实现，50 行 Python，跑通全流程。
MCP 到底是个啥
别急着去搜定义。MCP 本质上就是个协议，让大模型能调用外部工具。类比一下：
传统 API 是「前端调后端」——人类写的代码去调 HTTP 接口。
MCP 是「AI 调工具」——大模型根据你的需求，自己去决定调哪个函数、传什么参数。
你的数据库、文件系统、搜索引擎，都可以通过 MCP Server 暴露给 AI。AI 把它们当工具用，而不是自己瞎猜答案。
环境准备
只依赖三个东西：Python 3.10+、一个 SQLite 数据库、还有mcp库。
pip install mcp httpx # 建个测试数据库 sqlite3 test.db <<EOF CREATE TABLE orders ( id INTEGER PRIMARY KEY, user_id INTEGER, product TEXT, amount REAL, status TEXT, created_at TEXT ); INSERT INTO orders VALUES (1, 1001, 'MacBook Pro', 14999.00, 'shipped', '2026-06-28'), (2, 1001, 'AirPods', 1299.00, 'pending', '2026-07-01'), (3, 1002, 'iPad Air', 4799.00, 'delivered', '2026-06-25'); EOF这里踩了个坑：mcp库有两个版本，老版是mcp，新版 SDK 叫mcp-sdk。我用的是新版，别搞混了。
50 行 MCP Server
直接上代码：
"""database_mcp_server.py - 让 AI 安全查数据库的 MCP Server""" import sqlite3, json from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent server = Server("db-query") DB_PATH = "test.db" def query_db(sql: str) -> list: """安全执行 SELECT 查询，非 SELECT 自动拒绝""" sql = sql.strip().upper() if not sql.startswith("SELECT"): raise ValueError("只允许 SELECT 查询") conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute(sql) rows = [dict(row) for row in cur.fetchall()] conn.close() return rows @server.list_tools() async def list_tools(): return [ Tool( name="query_orders", description="查询订单数据，支持 WHERE 条件过滤", inputSchema={ "type": "object", "properties": { "sql": { "type": "string", "description": "SQL SELECT 语句，例如 SELECT * FROM orders WHERE user_id = 1001" } }, "required": ["sql"] } ) ] @server.call_tool() async def call_tool(name: str, arguments: dict): if name != "query_orders": raise ValueError(f"未知工具: {name}") sql = arguments.get("sql", "") try: result = query_db(sql) return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False))] except Exception as e: return [TextContent(type="text", text=f"查询失败：{str(e)}")] async def main(): async with stdio_server() as (read, write): await server.run(read, write, server.create_initialization_options()) if __name__ == "__main__": import asyncio asyncio.run(main())核心就两个函数：
list_tools()— 告诉 AI 你有什么工具、参数长什么样（用的是 JSON Schema，大模型原生理解）call_tool()— AI 调工具时的入口，在这里做参数校验和安全检查
注意query_db()里我硬写了只允许 SELECT，这就回答了开头产品经理的问题——AI 只能读，不能写。安全性不是靠 AI 自觉，是靠代码硬约束。
跑起来看看
启动 Server：
python database_mcp_server.py正常启动后不会有任何输出，它通过标准输入输出和客户端通信。
这时候你需要一个 MCP Client。最简单的测试方式是直接装一个支持 MCP 的桌面客户端，比如 Claude Desktop 或者 Continue.dev。在它们的配置里加上：
{ "mcpServers": { "db-query": { "command": "python", "args": ["database_mcp_server.py"] } } }配好之后，你就可以问 AI "查一下 1001 用户的订单"、"哪个订单金额最大"、"已发货的订单有哪些"。AI 会自己决定调哪个 SQL、怎么过滤，你只管看结果。
几个你肯定也会踩的坑
1. stdio 模式 vs SSE 模式
上面用的是 stdio 模式，Server 通过标准输入输出和客户端通信，适合本地用。如果想做成远程服务（比如公司内部所有人共享），需要用 SSE（Server-Sent Events）模式，代码量多 20 行左右。
2. 大模型生成 SQL 不一定对
实测 Claude 和 GPT-4 生成 SQL 的准确率大概 85%，简单的 WHERE 过滤没问题，复杂的 JOIN/子查询可能翻车。建议在生产环境加一层 SQL 校验——比如先用 EXPLAIN 跑一遍确认语法正确，再正式执行。
3. 上下文窗口
如果你表有几十万行，AI 不可能看完所有数据。好的做法是让工具返回聚合结果（COUNT、SUM、TOP 10），而不是裸查全表。工具设计决定了 AI 的使用上限。
4. MCP 协议版本还在快速迭代
写这篇文章时 MCP 规范是 2025-03-26 草案，已经在变。建议锁版本——pip install mcp==1.0.0，别追新，新版本可能 breaking change。
最后说两句
MCP 这套协议让我觉得有意思的地方在于，它不是又一个 JSON Schema 格式之争，而是真的在解决「AI 怎么跟现有系统交互」这个实际问题。50 行代码就能让 AI 安全操作数据库，门槛确实低。
下一步我打算写个 MCP Server 连公司内部 API 网关，把几十个内部接口包装成 AI 工具。到时候再分享。