乐于分享
好东西不私藏

手把手教你搭建 MCP 服务器:让 AI 助手连接你的数据库只需 10 分钟

手把手教你搭建 MCP 服务器:让 AI 助手连接你的数据库只需 10 分钟

2026 年,AI Agent 已经不是新鲜事。但让 AI 真正干活的秘密武器,是一个叫 MCP 的协议。今天带你从 0 搭一个,附完整代码和踩坑记录。


为什么突然所有人都在聊 MCP?

如果你还在用"复制提示词 → 粘贴到 ChatGPT → 手动整理结果"的方式用 AI,那你已经在 2026 年落伍了。

现在最火的玩法是:让 AI 直接连上你的数据库、文件系统、API,自己查数据、自己分析、自己出报告。 而这一切的核心桥梁,就是 MCP(Model Context Protocol)

MCP 是 Anthropic 在 2024 年底开源的协议,2026 年已经成了 AI Agent 生态的事实标准。Claude、GPT、Cursor、各类 Agent 框架都支持它。简单说:有了 MCP,AI 就能"伸手"去操作外部工具。

举个例子:

  • 连接 MySQL → AI 自动写 SQL 查数据、生成报表
  • 连接 GitHub → AI 自动看 PR、写 Code Review
  • 连接企业内部 API → AI 自动拉数据、跑流程

今天这篇,我带你从 0 搭一个 MCP 服务器,连上本地 SQLite 数据库,让 AI 能查询你的数据。全程实操,代码可复现。


准备工作

环境要求

  • **Python 3.10+**(推荐 3.12)
  • pip
     包管理器
  • 一个支持 MCP 的客户端
    (Claude Desktop、Cursor、或任意 MCP 兼容 Agent)
  • SQLite
    (Mac/Linux 自带,Windows 需单独安装)

安装 MCP SDK

打开终端,执行:

pip install mcp

就这一个包,官方 SDK,不到 2MB。


第一步:创建项目结构

mkdir mcp-sqlite-democd mcp-sqlite-demotouch server.py

目录很简单,就一个 server.py 文件。MCP 不需要复杂的项目结构。


第二步:初始化测试数据库

先造一个本地 SQLite 数据库,放一些测试数据,方便后面验证。

# init_db.py - 初始化数据库import sqlite3conn = sqlite3.connect("test.db")cursor = conn.cursor()# 建表cursor.execute("""CREATE TABLE IF NOT EXISTS sales (    id INTEGER PRIMARY KEY AUTOINCREMENT,    product TEXT NOT NULL,    amount REAL NOT NULL,    sale_date TEXT NOT NULL)""")# 插入测试数据test_data = [    ("iPhone 15", 5999.00, "2026-08-01"),    ("MacBook Pro", 14999.00, "2026-08-02"),    ("AirPods Pro", 1899.00, "2026-08-02"),    ("iPad Air", 4399.00, "2026-08-03"),    ("iPhone 15", 5999.00, "2026-08-03"),]cursor.executemany(    "INSERT INTO sales (product, amount, sale_date) VALUES (?, ?, ?)",    test_data)conn.commit()conn.close()print("数据库初始化完成,插入了 5 条测试数据。")

运行:

python init_db.py

第三步:编写 MCP 服务器(核心步骤)

这是全文最重要的部分。把以下代码完整复制到 server.py

# server.py - MCP 服务器:连接 SQLite,暴露查询能力给 AIfrom mcp.server import Serverfrom mcp.server.stdio import stdio_serverfrom mcp.types import Tool, TextContentimport sqlite3import json# 创建 MCP 服务器实例app = Server("sqlite-assistant")# 数据库路径(改成你自己的路径)DB_PATH = "test.db"@app.list_tools()async def list_tools() -> list[Tool]:    """告诉 AI 客户端:我提供了哪些工具"""    return [        Tool(            name="query_sales",            description="查询销售数据。可以按产品名筛选,也可以查全部。",            inputSchema={                "type": "object",                "properties": {                    "product": {                        "type": "string",                        "description": "要查询的产品名称,如 'iPhone 15'。留空则查全部。"                    }                },                "required": []            }        ),        Tool(            name="get_sales_summary",            description="获取销售汇总统计:总销售额、各产品销量、最近日期。",            inputSchema={                "type": "object",                "properties": {},                "required": []            }        )    ]@app.call_tool()async def call_tool(name: str, arguments: dict) -> list[TextContent]:    """处理 AI 发来的工具调用请求"""    conn = sqlite3.connect(DB_PATH)    cursor = conn.cursor()    if name == "query_sales":        product = arguments.get("product", "")        if product:            cursor.execute(                "SELECT * FROM sales WHERE product LIKE ? ORDER BY sale_date DESC",                (f"%{product}%",)            )        else:            cursor.execute("SELECT * FROM sales ORDER BY sale_date DESC")        rows = cursor.fetchall()        columns = [desc[0] for desc in cursor.description]        # 格式化为 AI 易读的结构        results = [dict(zip(columns, row)) for row in rows]        conn.close()        return [TextContent(            type="text",            text=json.dumps(results, ensure_ascii=False, indent=2)        )]    elif name == "get_sales_summary":        # 总销售额        cursor.execute("SELECT SUM(amount) FROM sales")        total = cursor.fetchone()[0] or 0        # 各产品销量        cursor.execute("""            SELECT product, COUNT(*) as count, SUM(amount) as total            FROM sales GROUP BY product ORDER BY total DESC        """)        by_product = cursor.fetchall()        # 最近日期        cursor.execute("SELECT MAX(sale_date) FROM sales")        latest = cursor.fetchone()[0]        conn.close()        summary = {            "total_amount": total,            "by_product": [                {"product": r[0], "count": r[1], "total": r[2]}                for r in by_product            ],            "latest_date": latest        }        return [TextContent(            type="text",            text=json.dumps(summary, ensure_ascii=False, indent=2)        )]    else:        conn.close()        return [TextContent(type="text", text=f"未知工具:{name}")]async def main():    async with stdio_server() as (read_stream, write_stream):        await app.run(read_stream, write_stream)if __name__ == "__main__":    import asyncio    asyncio.run(main())

代码解读:

  • @app.list_tools()
    :注册了两个工具——query_sales(按产品查销售)和 get_sales_summary(自动汇总统计)
  • @app.call_tool()
    :处理 AI 发来的调用请求,连 SQLite 执行 SQL,返回 JSON
  • stdio_server
    :通过标准输入输出和 AI 客户端通信,不需要起 HTTP 服务

第四步:配置 AI 客户端连接 MCP 服务器

方式一:Claude Desktop(推荐新手)

找到 Claude Desktop 的配置文件:

  • Mac
    ~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows
    %APPDATA%\Claude\claude_desktop_config.json

添加以下配置:

{  "mcpServers": {    "sqlite-assistant": {      "command": "python3",      "args": ["/绝对路径/mcp-sqlite-demo/server.py"],      "cwd": "/绝对路径/mcp-sqlite-demo/"    }  }}

⚠️ 踩坑提醒args 里的路径必须是绝对路径,相对路径会报错。cwd 也要设对,否则找不到 test.db

重启 Claude Desktop,你会看到左下角多了一个 🔌 图标,说明 MCP 服务器已连接。

方式二:Cursor

在 Cursor 设置 → MCP 中,点击 "Add MCP Server",填入:

  • Type
    : Command
  • Command
    python3 /绝对路径/mcp-sqlite-demo/server.py

第五步:测试效果

打开 Claude Desktop,输入:

帮我查一下所有销售数据,按日期倒序排列。

AI 会自动调用 query_sales 工具,返回类似这样的结果:

[  {"id": 5, "product": "iPhone 15", "amount": 5999.00, "sale_date": "2026-08-03"},  {"id": 4, "product": "iPad Air", "amount": 4399.00, "sale_date": "2026-08-03"},  {"id": 3, "product": "AirPods Pro", "amount": 1899.00, "sale_date": "2026-08-02"},  {"id": 2, "product": "MacBook Pro", "amount": 14999.00, "sale_date": "2026-08-02"},  {"id": 1, "product": "iPhone 15", "amount": 5999.00, "sale_date": "2026-08-01"}]

再试试:

给我一份销售汇总报告,包含总销售额和各产品排名。

AI 会自动调用 get_sales_summary,生成结构化的汇总数据。

关键点:你不需要告诉 AI 用哪个工具、写什么 SQL。你只说人话,MCP 协议负责让 AI "看见"可用工具,自己决定调用哪个、传什么参数。


真实踩坑记录

搭建过程中我遇到了几个坑,分享给大家:

坑 1:路径问题导致服务器启动失败

报错FileNotFoundError: test.db not found

原因:MCP 服务器的工作目录默认是 /,不是你的项目目录。

解决:在配置里加上 "cwd": "/绝对路径/mcp-sqlite-demo/",或者在代码里用绝对路径连接数据库。

坑 2:中文乱码

报错:返回的 JSON 里中文变成 \uXXXX

解决json.dumps() 加上 ensure_ascii=False 参数(代码里已加)。

坑 3:工具被 AI "看见"但没被调用

现象:AI 不调用工具,直接瞎编答案。

原因:工具的 description 写得不够清楚,AI 不知道什么时候该用它。

解决:把 description 写得越具体越好。比如 "查询销售数据。可以按产品名筛选" 比 "查询数据库" 好得多。AI 是根据 description 决定是否调用工具的。

坑 4:SQLite 被锁

报错database is locked

原因:MCP 服务器进程和你的手动查询同时操作一个 SQLite 文件。

解决:加 timeout 参数:sqlite3.connect(DB_PATH, timeout=10),或者用 WAL 模式:cursor.execute("PRAGMA journal_mode=WAL")


进阶:接入 MySQL / 企业内部 API

把 SQLite 换成 MySQL 只需要改连接方式:

# pip install mysql-connector-pythonimport mysql.connectorconn = mysql.connector.connect(    host="your-mysql-host",    port=3306,    user="your-user",    password="your-password",    database="your-db")

其他代码不用动。这就是 MCP 的魅力——协议是统一的,后端接什么都行。

接入企业内部 API 同理,把 SQL 查询替换成 requests.get("https://your-api/endpoint") 就行。


总结:MCP 的价值在哪?

对比维度
没有 MCP
有 MCP
AI 能力边界
只能聊天,不能"动手"
直接操作数据库、API、文件系统
集成成本
每个工具单独写适配
统一协议,一次配置,所有 MCP 兼容客户端通用
安全性
API Key 可能泄露在上下文
凭证留在 MCP 服务器本地,AI 只拿到执行结果
复用性
换一个 AI 就要重写
同一个 MCP 服务器,Claude、Cursor、GPT 都能用

MCP 的本质是给 AI 装上了"手"。以前 AI 只能说,现在 AI 能做。2026 年下半年,不会用 MCP 就像 2023 年不会写提示词一样。

下一步建议

  1. 先用本教程的 SQLite 版本跑通
  2. 替换成你真实的数据源(MySQL、内部 API)
  3. 配合 Claude Desktop 或 Cursor,让 AI 自动查数据、写报告
  4. 逐步添加更多工具(增删改查),打造你的 AI 工具箱

本文代码已开源,欢迎复现。如果觉得有用,转发给你身边还在手动复制粘贴 AI 回复的朋友 😄