手写 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( namequery_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(typetext, textjson.dumps(result, ensure_asciiFalse))] except Exception as e: return [TextContent(typetext, textf查询失败{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 自觉是靠代码硬约束。跑起来看看启动 Serverpython 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 通过标准输入输出和客户端通信适合本地用。如果想做成远程服务比如公司内部所有人共享需要用 SSEServer-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 mcp1.0.0别追新新版本可能 breaking change。最后说两句MCP 这套协议让我觉得有意思的地方在于它不是又一个 JSON Schema 格式之争而是真的在解决「AI 怎么跟现有系统交互」这个实际问题。50 行代码就能让 AI 安全操作数据库门槛确实低。下一步我打算写个 MCP Server 连公司内部 API 网关把几十个内部接口包装成 AI 工具。到时候再分享。