MCP协议入门教程：用Python从零搭建MCP Server连接数据库（2026完整实战）

ai_liu_de_hua

485人浏览 · 2026-03-23 15:27:04

ai_liu_de_hua · 2026-03-23 15:27:04 发布

MCP（Model Context Protocol）是 2026 年 AI 工具链绕不开的关键词。从 Cursor 到 Claude Desktop，再到 VS Code 生态的各种插件，几乎所有主流 AI 开发工具都已经支持 MCP 协议。

但很多开发者还停留在「听过但没实操」的阶段。本文将从零开始，手把手带你搭建一个能查询数据库的 MCP Server，并在 Claude Desktop 和 Cursor 中实际跑通。全程附完整可运行代码，跟着做就能上手。

一、MCP 协议是什么？为什么你需要关注它

Model Context Protocol（模型上下文协议） 由 Anthropic 于 2024 年底提出，目标是解决一个实际问题：每个 AI 工具调用外部能力的方式都不一样，开发者疲于适配。

举个例子：

让 Claude 查数据库，要写一套 Function Calling 的 schema
让 GPT 读本地文件，又是另一套格式
换个 AI 客户端，接口全要重写

MCP 的思路很直接——定义一套通用协议，工具写一次，所有支持 MCP 的客户端都能调用。

打个比方：MCP 对于 AI 工具，相当于 USB 接口对于外设设备。标准统一了，生态才能真正繁荣起来。

二、MCP 架构核心概念详解

MCP 的架构由三个角色组成：

┌─────────────┐     MCP 协议     ┌─────────────┐
│  MCP Client │ ◄──────────────► │  MCP Server │
│ (AI 应用端)  │                  │ (工具服务端) │
└─────────────┘                  └─────────────┘
       │
       │ 调用大模型
       ▼
┌─────────────┐
│  LLM API    │
└─────────────┘

三个角色说明：

MCP Client：发起调用的 AI 应用，比如 Cursor、Claude Desktop、或者你自己写的 Agent 程序
MCP Server：提供具体能力的服务端，比如查数据库、读文件、调第三方 API
传输层：支持 stdio（本地进程通信）和 HTTP+SSE（远程网络服务）两种模式

MCP Server 可以暴露三类能力：

能力类型	说明	典型场景
Tools（工具）	可执行的函数	查询数据库、发邮件、调接口
Resources（资源）	可读取的数据源	文件内容、表结构、配置信息
Prompts（提示模板）	预定义对话模板	标准化的查询/分析流程

本教程重点实战 Tools 部分，这也是日常开发中用得最多的。

三、实战：Python 搭建数据库查询 MCP Server

3.1 环境搭建

推荐使用 uv 包管理器，速度快、依赖干净：

# 创建项目
uv init mcp-demo
cd mcp-demo

# 安装依赖
uv add mcp sqlalchemy

如果还没装 uv，一行搞定：curl -LsSf https://astral.sh/uv/install.sh | sh

3.2 第一步：Hello World 级别的 MCP Server

先确保链路跑通。创建 server.py：

# server.py —— 最小 MCP Server 示例
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

server = Server("demo-server")

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="hello",
            description="打个招呼，验证 MCP 连接是否正常工作",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {
                        "type": "string",
                        "description": "你的名字"
                    }
                },
                "required": ["name"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "hello":
        return [TextContent(
            type="text",
            text=f"你好 {arguments['name']}！MCP Server 连接正常 🎉"
        )]
    raise ValueError(f"未知工具: {name}")

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

这就是一个最基本的 MCP Server：注册了一个 hello 工具，AI 调用时返回一句问候。结构清晰，容易理解。

3.3 第二步：加入数据库查询功能

接下来实现真正有用的功能——让 AI 助手直接查询 SQLite 数据库。

创建完整的 server.py：

# server.py —— 带数据库查询的 MCP Server
import json
import sqlite3
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

server = Server("db-query-server")
DB_PATH = "demo.db"

def init_db():
    """创建示例数据库和测试数据"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            price REAL NOT NULL,
            category TEXT,
            stock INTEGER DEFAULT 0
        )
    """)
    sample_data = [
        ("机械键盘", 299.0, "外设", 150),
        ("4K 显示器", 2499.0, "显示器", 30),
        ("无线鼠标", 149.0, "外设", 200),
        ("降噪耳机", 899.0, "音频", 80),
        ("USB-C 扩展坞", 199.0, "配件", 120),
    ]
    cursor.executemany(
        "INSERT OR IGNORE INTO products (name, price, category, stock) VALUES (?, ?, ?, ?)",
        sample_data
    )
    conn.commit()
    conn.close()

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="query_products",
            description="按条件查询产品信息。支持分类筛选、价格上限过滤和多种排序方式。",
            inputSchema={
                "type": "object",
                "properties": {
                    "category": {
                        "type": "string",
                        "description": "产品分类，可选值：外设、显示器、音频、配件"
                    },
                    "max_price": {
                        "type": "number",
                        "description": "价格上限，只返回不超过该价格的产品"
                    },
                    "sort_by": {
                        "type": "string",
                        "enum": ["price_asc", "price_desc", "stock_desc"],
                        "description": "排序规则：price_asc 价格升序 / price_desc 价格降序 / stock_desc 库存降序"
                    }
                }
            }
        ),
        Tool(
            name="get_stats",
            description="获取产品库的汇总统计：总数量、平均价格、价格区间、总库存及各分类分布",
            inputSchema={"type": "object", "properties": {}}
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    try:
        if name == "query_products":
            query = "SELECT * FROM products WHERE 1=1"
            params = []

            if category := arguments.get("category"):
                query += " AND category = ?"
                params.append(category)

            if max_price := arguments.get("max_price"):
                query += " AND price <= ?"
                params.append(max_price)

            sort_map = {
                "price_asc": "price ASC",
                "price_desc": "price DESC",
                "stock_desc": "stock DESC"
            }
            if sort_by := arguments.get("sort_by"):
                query += f" ORDER BY {sort_map.get(sort_by, 'id')}"

            cursor.execute(query, params)
            rows = [dict(row) for row in cursor.fetchall()]

            return [TextContent(
                type="text",
                text=json.dumps(rows, ensure_ascii=False, indent=2)
            )]

        elif name == "get_stats":
            cursor.execute("""
                SELECT 
                    COUNT(*) as total,
                    ROUND(AVG(price), 2) as avg_price,
                    MIN(price) as min_price,
                    MAX(price) as max_price,
                    SUM(stock) as total_stock
                FROM products
            """)
            stats = dict(cursor.fetchone())

            cursor.execute("""
                SELECT category, COUNT(*) as count 
                FROM products GROUP BY category
            """)
            stats["categories"] = {
                row["category"]: row["count"] for row in cursor.fetchall()
            }

            return [TextContent(
                type="text",
                text=json.dumps(stats, ensure_ascii=False, indent=2)
            )]

        raise ValueError(f"未知工具: {name}")
    finally:
        conn.close()

async def main():
    init_db()
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

3.4 第三步：配置 AI 客户端连接

Claude Desktop 配置方法：

编辑配置文件（macOS 路径示例）：

# macOS
~/Library/Application Support/Claude/claude_desktop_config.json

# Windows
%APPDATA%\Claude\claude_desktop_config.json

写入以下内容：

{
  "mcpServers": {
    "db-query": {
      "command": "uv",
      "args": ["run", "server.py"],
      "cwd": "/your/path/to/mcp-demo"
    }
  }
}

重启 Claude Desktop，界面上会出现工具图标。试着问它：「帮我查一下 500 块以内的外设有什么」——它会自动调用你的 MCP Server，查询数据库后用自然语言回答。

Cursor 配置：进入 Settings → MCP Servers，添加同样的配置即可。

四、进阶：HTTP+SSE 模式支持远程访问

stdio 模式限于本地使用。如果需要多台设备或远程连接同一个 MCP Server，可以切换到 HTTP+SSE 传输模式：

# server_http.py
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route, Mount
import uvicorn

server = Server("db-query-server")
sse = SseServerTransport("/messages/")

# list_tools 和 call_tool 与前面相同，这里省略

async def handle_sse(request):
    async with sse.connect_sse(
        request.scope, request.receive, request._send
    ) as streams:
        await server.run(
            streams[0], streams[1],
            server.create_initialization_options()
        )

app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ]
)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

部署后，客户端配置中把 command 替换为 HTTP 地址即可远程连接。

五、开发踩坑经验总结

踩坑 1：Tool 描述质量直接影响调用准确率

大模型根据 description 字段判断什么时候调用、传什么参数。我早期把描述写成简单的「查询数据」，结果 AI 频繁误调用。

解决方案：把 description 当 API 文档写，明确说明适用场景、输入参数含义和返回格式。

踩坑 2：返回数据量太大导致效果变差

MCP Server 返回的内容直接进入大模型上下文窗口。如果一次返回几百条记录，不仅浪费 token，模型的理解和总结能力也会明显下降。

解决方案：在 Server 端实现分页，默认限制 20 条。对于统计类查询，直接返回聚合结果。

踩坑 3：多模型测试的效率问题

开发 MCP Server 的过程中，你可能需要用不同模型验证 Tool Calling 的兼容性。比如 Claude 准确率高、GPT-4o 响应快、DeepSeek 成本低。

我的做法是用统一的 API 入口，切换模型只需改一个参数：

from openai import OpenAI

client = OpenAI(
    api_key="your-api-key",
    base_url="https://api.ofox.cc/v1"
)

# 只改 model 参数就能测不同模型
for model in ["claude-sonnet-4-20250514", "gpt-4o", "deepseek-chat"]:
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": "查一下 500 以下的外设"}],
        tools=[{
            "type": "function",
            "function": {
                "name": "query_products",
                "description": "按条件查询产品信息",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "category": {"type": "string"},
                        "max_price": {"type": "number"}
                    }
                }
            }
        }]
    )
    print(f"{model}: {response.choices[0].message.tool_calls}")

一个 base_url 搞定多模型切换，不用装多个 SDK，也不用处理各家 API 的格式差异。

踩坑 4：安全防护不能忽视

MCP Server 相当于给 AI 开了一个操作你系统的入口，安全措施必须到位：

SQL 注入防护：始终使用参数化查询（? 占位符），绝不拼接 SQL 字符串
最小权限原则：生产环境使用只读数据库账号
参数白名单校验：对 AI 传入的参数做严格校验
调用审计日志：记录每一次 Tool 调用的入参和结果

import logging

logger = logging.getLogger("mcp-audit")

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    logger.info(f"Tool called: {name}, args: {arguments}")
    # ... 执行业务逻辑

六、MCP 生态现状（2026 年 3 月）

截至目前，MCP 生态已经相当完善：

客户端覆盖：Claude Desktop、Cursor、Windsurf、Continue、Cline 及主流 VS Code 插件
官方 Server：GitHub、Google Drive、Slack、PostgreSQL、文件系统等均有官方实现
社区生态：数百个第三方 Server，覆盖数据库、云服务、开发运维工具
多语言 SDK：Python、TypeScript、Go 均已成熟可用

个人判断：MCP 对 AI 应用开发者的意义，类似于 REST API 对 Web 开发者的意义。现在投入时间理解和实践 MCP，回报确定性很高。

总结

MCP 协议的核心逻辑可以归纳为三步：

注册工具（list_tools）—— 告诉 AI 你能提供什么能力
执行调用（call_tool）—— AI 决定调用时，执行对应逻辑
返回结果 —— 将结果以文本形式传回给 AI

掌握这个模式后，你可以快速扩展到更多场景：

对接公司内部系统（工单、监控、日志查询）
连接个人知识库（Notion、Obsidian）
控制智能家居设备
管理云服务器（查负载、部署、重启）

MCP 生态已经成熟，文档齐全，踩坑的人也都把经验分享出来了。如果你一直在观望，现在正是动手实践的好时机。

本文代码基于 mcp Python SDK 1.x 版本测试通过。有疑问欢迎评论区交流探讨。
MCP（Model Context Protocol）是 2026 年 AI 工具链绕不开的关键词。从 Cursor 到 Claude Desktop，再到 VS Code 生态的各种插件，几乎所有主流 AI 开发工具都已经支持 MCP 协议。