深入理解 MCP 协议：从底层通信到 MySQL 实战接入

2401_83537300

582人浏览 · 2026-06-09 21:59:33

2401_83537300 · 2026-06-09 21:59:33 发布

深入理解 MCP 协议：从底层通信到 MySQL 实战接入

最近在做一个课设，想让 AI 能直接查询我本地的 MySQL 数据库。网上翻了一圈，全是"MCP 是 AI 的 USB-C 接口"这类比喻，看完还是不知道怎么用。花了三天踩坑，把底层逻辑搞清楚了，写下来给自己留个记录。

先说结论：MCP 解决的是一个被忽视已久的根本问题

为什么需要 MCP？

在 MCP 出现之前，如果你想让 ChatGPT 或者任何 LLM 能查你的数据库，你得这么做：

自己写一套 Function Calling 的接口定义
把数据库操作封装成函数
把函数 schema 塞进 prompt
解析模型返回的 JSON，手动执行

每接入一个新工具，就要重新写一套。换个模型（比如从 GPT 换成 DeepSeek），又要重新适配。

这就好比家里每个电器都有一个独立遥控器，需要逐个编写接口；而 MCP 则像全屋智能中枢，通过统一协议接入所有设备。

听起来是个比喻，但工程上的含义很实在：写一次 MCP Server，任何支持 MCP 的 AI 客户端都能直接用。

MCP 的三层架构，用人话说清楚

MCP 的核心思想是将 LLM 的核心推理能力与外部功能的实现细节解耦，通过定义清晰的主机（Host）、客户端（Client）和服务器（Server）架构实现灵活、可扩展且安全的 LLM 应用生态。

Host / Client / Server，三个词一出来，大多数人就晕了。我当时也是。后来我用一个具体场景理解了：

你（用户）
  ↓ 提问
Claude Desktop（Host，宿主应用）
  ↓ 转发请求
MCP Client（内嵌在 Host 里，负责通信）
  ↓ 标准化协议
MCP Server（你自己写的，封装了数据库/文件/API）
  ↓ 执行
MySQL / 本地文件 / 第三方 API

关键点：MCP Server 是你来写的，是把你的工具"翻译"成 AI 能懂的标准接口。

Host 是 Claude Desktop、Cursor 这类应用，它们已经帮你实现了 MCP Client，你不需要管。你只需要写 Server，告诉 AI “我有哪些工具、每个工具怎么用”。

底层通信：JSON-RPC 2.0，没有那么神秘

我踩的第一个坑就是以为 MCP 是某种复杂的私有协议。

其实不是。MCP 协议使用 JSON-RPC 2.0 作为消息传输格式。

JSON-RPC 2.0 是一个极简的远程调用协议，一条请求长这样：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "query_database",
    "arguments": {
      "sql": "SELECT * FROM users WHERE id = 1"
    }
  }
}

Server 返回：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "[{\"id\": 1, \"name\": \"张三\", \"email\": \"zhangsan@example.com\"}]"
      }
    ]
  }
}

就这么简单。AI 说"我要调用 query_database，参数是这条 SQL"，Server 执行完把结果返回去，AI 再把结果转成自然语言告诉你。

理解了这一层，后面的东西就都清晰了。

我踩的坑：传输方式的选择

MCP 支持两种传输方式，这里是我卡了最久的地方。

1. stdio（标准输入输出）

Server 作为一个本地子进程运行，Host 通过标准输入输出和它通信。

# Server 启动方式
if __name__ == "__main__":
    import asyncio
    from mcp.server.stdio import stdio_server
    asyncio.run(stdio_server(server))

优点：配置简单，本地开发首选。
缺点：只能本地用，无法远程访问。

2. HTTP + SSE（Streamable HTTP）

新的传输规范中，服务端作为独立进程运行，支持多客户端连接，基于 HTTP POST/GET 请求实现，可选用服务器推送事件（SSE）实现多消息流式传输。

# Server 以 HTTP 方式启动
if __name__ == "__main__":
    import uvicorn
    from mcp.server.fastmcp import FastMCP
    app = FastMCP("my-server")
    uvicorn.run(app.get_asgi_app(), host="0.0.0.0", port=8000)

我踩的坑：我一开始用的是老版本的 SSE 实现，结果发现协议已经改了，连接一直断。后来看了官方 changelog 才知道，2025年初协议把传输层做了破坏性更新，旧的 SSE 方案已经废弃，要用新的 Streamable HTTP。

教训：用 MCP 之前先确认 SDK 版本，pip show mcp 看一眼，0.9.0 以上才支持新传输协议。

写一个真实可用的 MCP Server（查 MySQL）

说了这么多，来看实际代码。这是我课设里用的版本，精简过，核心逻辑都在。

环境准备

pip install mcp pymysql

完整代码

import pymysql
import json
from mcp.server.fastmcp import FastMCP

# 初始化 MCP Server
mcp = FastMCP("mysql-assistant")

# 数据库配置
DB_CONFIG = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "password": "your_password",
    "database": "your_db",
    "charset": "utf8mb4"
}

def get_connection():
    return pymysql.connect(**DB_CONFIG)

# ========== 定义工具 ==========

@mcp.tool()
def query_data(sql: str) -> str:
    """
    执行 SELECT 查询语句，返回查询结果。
    只允许 SELECT，不允许增删改操作。
    
    Args:
        sql: 要执行的 SELECT SQL 语句
    
    Returns:
        JSON 格式的查询结果
    """
    # 安全校验：只允许 SELECT
    sql_upper = sql.strip().upper()
    if not sql_upper.startswith("SELECT"):
        return "错误：只允许执行 SELECT 查询语句"
    
    try:
        conn = get_connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        cursor.execute(sql)
        results = cursor.fetchall()
        cursor.close()
        conn.close()
        
        if not results:
            return "查询结果为空"
        
        return json.dumps(results, ensure_ascii=False, default=str)
    
    except pymysql.Error as e:
        return f"数据库错误：{str(e)}"


@mcp.tool()
def get_table_schema(table_name: str) -> str:
    """
    获取指定数据表的字段结构信息。
    在写 SQL 之前先调用这个工具了解表结构。
    
    Args:
        table_name: 表名
    
    Returns:
        表的字段名、类型、是否为空等信息
    """
    try:
        conn = get_connection()
        cursor = conn.cursor()
        cursor.execute(f"DESCRIBE `{table_name}`")
        columns = cursor.fetchall()
        cursor.close()
        conn.close()
        
        result = f"表 {table_name} 的结构：\n"
        result += "字段名 | 类型 | 允许空 | 键 | 默认值\n"
        result += "-" * 60 + "\n"
        for col in columns:
            result += " | ".join(str(c) for c in col) + "\n"
        return result
    
    except pymysql.Error as e:
        return f"获取表结构失败：{str(e)}"


@mcp.tool()
def list_tables() -> str:
    """
    列出当前数据库中所有的表名。
    不知道有哪些表时，先调用这个。
    """
    try:
        conn = get_connection()
        cursor = conn.cursor()
        cursor.execute("SHOW TABLES")
        tables = [row[0] for row in cursor.fetchall()]
        cursor.close()
        conn.close()
        return "数据库中的表：\n" + "\n".join(f"- {t}" for t in tables)
    except pymysql.Error as e:
        return f"获取表列表失败：{str(e)}"


# ========== 启动 ==========
if __name__ == "__main__":
    mcp.run()  # 默认 stdio 模式，用于本地 Claude Desktop

配置到 Claude Desktop

在 claude_desktop_config.json 里加上：

{
  "mcpServers": {
    "mysql-assistant": {
      "command": "python",
      "args": ["/绝对路径/mysql_mcp_server.py"]
    }
  }
}

重启 Claude Desktop，然后直接问它：

“帮我看看 users 表里注册时间最早的 10 个用户”

它会自动调用 list_tables → get_table_schema → query_data，最后给你一个完整答案。不需要你写任何 SQL。

一个容易忽视的设计细节：工具描述比代码更重要

我刚开始写的时候，工具描述就一句话，结果 AI 经常调错工具或者传错参数。

后来我意识到：AI 是通过读你写的 docstring 来决定"该不该调用这个工具、怎么调用"的。 描述越精确，AI 的行为越准确。

# ❌ 模糊的描述
@mcp.tool()
def query(sql: str) -> str:
    """执行SQL"""
    ...

# ✅ 清晰的描述
@mcp.tool()
def query_data(sql: str) -> str:
    """
    执行 SELECT 查询语句，返回查询结果。
    只允许 SELECT，不允许增删改操作。
    在查询前请先用 get_table_schema 了解表结构，避免写出错误的字段名。
    
    Args:
        sql: 完整的 SELECT SQL 语句，例如：SELECT * FROM users LIMIT 10
    """
    ...

这个细节在所有 MCP 教程里几乎没人提，但它直接决定了 Agent 的实际可用性。

MCP 和 Function Calling 的本质区别

有人会问：这和 OpenAI 的 Function Calling 有什么区别？

本质区别只有一个：Function Calling 是模型私有的，MCP 是跨模型的标准。

维度	Function Calling	MCP
标准化程度	各家模型实现不同	统一开放协议
复用性	换模型要重写	一次编写，到处使用
工具发现	需要手动在 prompt 中声明	Server 自动暴露工具列表
部署方式	嵌入应用代码	独立进程，可远程部署