MCP 协议原理与自定义服务器开发实战：从零搭建 AI 工具调用基础设施

去年年底我接了个需求：让公司内部 AI 助手能直接查询工单系统、操作数据库、发钉钉消息。

调研一圈后发现，每个工具写一套 HTTP API 再让 AI 调，维护成本和安全性都是灾难。直到看到 Anthropic 开源的 Model Context Protocol（MCP）——一个让 AI 应用与外部工具通信的标准化协议。

用了两个月，踩了不少坑，今天把这些经验写下来。

MCP 是什么，解决什么问题

一句话：MCP 是 AI 应用的 USB-C 接口。

USB-C 之前，每个外设都要专门的接口和驱动。AI 集成外部工具也一样——没有 MCP 之前，每接一个数据源或工具，就得写定制的集成代码、改 prompt、调解析逻辑。

MCP 定义了三种核心角色：

Host：用户与 AI 交互的界面（比如 Claude Desktop、VS Code 扩展）
Client：与 MCP Server 建立一对一连接的会话层
Server：暴露具体工具、数据资源和计算能力

数据流向：User → Host → Client → Server(工具)

Host 不需要知道 Server 背后是什么，双方遵守 MCP 协议就能通信。

协议底层基于 JSON-RPC 2.0，传输层目前支持两种：

stdio：通过子进程的标准输入/输出通信，适合本地开发
SSE (Server-Sent Events)：基于 HTTP，适合远程部署

先跑一个官方 Demo 感受一下

动手之前，先看一个能跑的东西。创建一个最简单的 MCP Server，暴露一个天气查询工具：

# weather_server.py
from mcp.server.fastmcp import FastMCP

# 创建一个 MCP Server 实例
mcp = FastMCP("Weather Server")

# 用装饰器注册一个工具
@mcp.tool()
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    # 这里假装调了天气 API
    data = {
        "北京": "晴，15-25°C",
        "上海": "多云，18-27°C",
        "深圳": "阵雨，22-30°C",
    }
    return data.get(city, f"暂无 {city} 的天气数据")

if __name__ == "__main__":
    mcp.run(transport="stdio")

用的是 fastmcp——MCP 官方提供的简化封装，比直接操作底层 Server 对象少写 80% 的样板代码。如果从 2025 年才开始用 MCP，直接用 FastMCP 就好，不用纠结底层 API。

跑起来：

pip install mcp
python weather_server.py

看起来什么都没发生，因为 stdio 模式等着被父进程启动。用 MCP Inspector 来调试：

npx @modelcontextprotocol/inspector python weather_server.py

打开浏览器访问 http://localhost:5173，可以看到 Server 暴露的 get_weather 工具，在输入框里传参数试一下。

看到返回的 JSON 了吗？左侧是工具列表，右侧是调用记录和返回结果，Host 与 Server 的通信过程一目了然。

注意: 第一次启动 Inspector 会从 npm 拉包，建议提前科学上网，或者用 npm config set registry https://registry.npmmirror.com 切到国内镜像。

从零开发一个 MySQL 查询 MCP Server

Demo 跑通了，来个真有用的——写一个能查 MySQL 的 MCP Server。AI 直接用自然语言查数据库，不用手写 SQL。

第一步：项目结构

mkdir mcp-mysql-server && cd mcp-mysql-server
pip install mcp pymysql

第二步：实现 Server

选了 FastMCP + pymysql 的组合：

# server.py
import pymysql
from pymysql.cursors import DictCursor
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("MySQL Assistant")

# 数据库连接配置
DB_CONFIG = {
    "host": "127.0.0.1",
    "port": 3306,
    "user": "root",
    "password": "your_password",
    "database": "your_database",
    "charset": "utf8mb4",
}


def get_conn():
    """获取数据库连接"""
    return pymysql.connect(**DB_CONFIG, cursorclass=DictCursor)


@mcp.tool()
def query(sql: str) -> list:
    """执行 SQL 查询并返回结果，仅支持 SELECT 语句"""
    sql_stripped = sql.strip().upper()
    if not sql_stripped.startswith("SELECT"):
        return [{"error": "仅允许执行 SELECT 查询"}]

    conn = get_conn()
    try:
        with conn.cursor() as cursor:
            cursor.execute(sql)
            rows = cursor.fetchall()
            # 将 Decimal、datetime 等类型转成字符串方便 JSON 序列化
            result = []
            for row in rows:
                result.append({k: str(v) if not isinstance(v, (int, float, type(None))) else v
                               for k, v in row.items()})
            return result
    except Exception as e:
        return [{"error": str(e)}]
    finally:
        conn.close()


if __name__ == "__main__":
    mcp.run(transport="stdio")

做了两个防御措施：

只允许 SELECT，防止 SQL 注入
把 Decimal、datetime 转成字符串，避免 JSON 序列化报错

你以为够了？不够。实际上线第一个版本时，我问 AI"这个月新增了多少用户"，它生成的 SQL 把整个表 SELECT * 了一次，返回了 10 万行数据——直接把 MCP 通信撑爆了。

第三步：限流和分页

被生产环境毒打后加上的：

import time

# 请求计数器，简单限流
_request_count = 0
_last_reset = time.time()


def check_rate_limit():
    global _request_count, _last_reset
    now = time.time()
    if now - _last_reset > 60:
        _request_count = 0
        _last_reset = now
    _request_count += 1
    if _request_count > 30:
        raise Exception("每分钟最多执行 30 次查询，请稍后再试")


@mcp.tool()
def query_with_limit(sql: str, limit: int = 200) -> list:
    """执行 SQL 查询，自动加 LIMIT 防止返回过多数据"""
    check_rate_limit()

    sql_stripped = sql.strip().upper()
    if not sql_stripped.startswith("SELECT"):
        return [{"error": "仅允许执行 SELECT 查询"}]

    # 自动加 LIMIT
    if "LIMIT" not in sql_stripped.upper():
        sql = sql.rstrip(";") + f" LIMIT {limit}"

    conn = get_conn()
    try:
        with conn.cursor(cursor=DictCursor) as cursor:
            cursor.execute(sql)
            rows = cursor.fetchall()
            result = []
            for row in rows:
                result.append({k: str(v) if not isinstance(v, (int, float, type(None))) else v
                               for k, v in row.items()})
            return {"count": len(result), "data": result}
    except Exception as e:
        return {"error": str(e)}
    finally:
        conn.close()

核心改动：自动加 LIMIT，每分钟查询次数限制。数据量再大也撑不死。

第四步：搭配 Claude Desktop 使用

要让 MCP Server 被 Claude 识别，编辑配置文件：

{
  "mcpServers": {
    "mysql-assistant": {
      "command": "python",
      "args": ["D:/projects/mcp-mysql-server/server.py"],
      "env": {}
    }
  }
}

配置路径看客户端：

Claude Desktop：%APPDATA%/Claude/claude_desktop_config.json（Windows）
VS Code 扩展：.vscode/mcp.json

重启 Claude Desktop，在聊天框里直接说"查一下本月订单数量"，Claude 会调用 MCP Server 去查数据库。之前在 Inspector 里看到的 JSON 通信，现在在后台静默完成。

第五步：部署成远程服务

本地跑 stdio 没问题，部署到服务器必须用 SSE。这里用 FastAPI + sse-starlette：

# remote_server.py
from mcp.server.fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.routing import Route

mcp = FastMCP("MySQL Assistant Remote")

# 注册工具（同上文的 query_with_limit）
@mcp.tool()
def query_with_limit(sql: str, limit: int = 200) -> dict:
    # ... 实现同上 ...
    pass

# 挂载到 Starlette 应用
app = Starlette(routes=[
    Route("/mcp", endpoint=mcp.sse_app()),
])

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

客户端配置指向 SSE 端点：

{
  "mcpServers": {
    "mysql-assistant-remote": {
      "type": "sse",
      "url": "https://your-server.com/mcp"
    }
  }
}

注意: 远程部署一定要加认证！我见过直接把 MCP 暴露在公网的，任何能访问这个 URL 的人都能操作你的数据库。最简单的方案是在 Nginx 层加 Basic Auth 或 IP 白名单。

协议细节：理解 tool/resource/prompt 三大能力

MCP Server 不只是暴露工具给 AI 调用，它有三种能力模型。

1. Tool（工具）——最常用

Tool 就是可被 AI 调用的函数。刚才的 query_with_limit 就是一个 Tool。AI 决定何时调用、传什么参数。

关键点：

函数签名 + 文档字符串 = AI 理解你的工具的唯一途径
参数名和类型提示要语义化，AI 靠这个推断参数含义
返回值必须是 JSON 可序列化的

2. Resource（资源）——暴露数据

Resource 让 AI 读取外部数据，像读文件一样自然：

@mcp.resource("config://database")
def get_db_config() -> str:
    """返回数据库配置信息（脱敏后）"""
    return f"数据库: {DB_CONFIG['database']}, 主机: {DB_CONFIG['host']}"

Resource URI 是自定义的，这里用了 config://database。AI 在对话中"看到"有这个资源可用，会自动读取来理解上下文。

3. Prompt（提示模板）——引导对话

Prompt 相当于预定义的指令模板，让 AI 知道特定场景下该怎么表现：

@mcp.prompt()
def sql_expert() -> str:
    return """你是一个 SQL 专家助手。当你需要查询数据库时:
1. 先理解用户要什么数据
2. 写出合适的 SQL
3. 用 query_with_limit 工具执行
4. 用中文解释结果
"""

三种能力怎么选：

想让 AI 主动做事 → Tool（调接口、写数据、发通知）
想让 AI 获取上下文 → Resource（读文档、查配置、看日志）
想让 AI 按套路出牌 → Prompt（设定角色、规范行为）

踩坑记录

坑 1：JSON 序列化失败

第一个版本查询返回值里有 datetime 和 Decimal，直接 JSON 序列化报错。解决方式前面说了——递归转字符串。

坑 2：AI 读不懂工具说明

我第一次注册工具只写了个名字：

@mcp.tool()
def execute_sql(sql):
    pass

AI 用它来更新数据、删表，导致了一个小事故。后来发现文档字符串就是你的 API 文档，AI 完全依赖它理解工具用途。改成：

@mcp.tool()
def query_readonly(sql: str) -> list:
    """仅用于只读查询。如果你需要修改数据（INSERT/UPDATE/DELETE），不要使用此工具。"""

加上"只读"的函数名和明确限制的文档字符串后，AI 再没误用过。

坑 3：一次性返回大量数据

自动 LIMIT + 次数限流，再加一个体积限制：

import sys

result_str = json.dumps(result)
if sys.getsizeof(result_str) > 1024 * 1024:  # 超过 1MB
    return {"error": "返回数据超过 1MB，请缩小查询范围"}

坑 4：stdio 模式的进程管理

stdio 模式下，Host 启动 Server 的子进程。Server 挂掉了，Host 不会自动重启。生产环境建议加进程守护：

# 用 systemd 或 supervisor 守护
supervisorctl start mcp-mysql

或者在代码里加心跳检测，但这需要底层 API 了，FastMCP 不直接支持。

进阶：多 Server 组合与代理模式

一个 MCP Host 可以同时连接多个 Server。比如我本地同时跑三个 Server：

MySQL Server（数据库查询）
DingTalk Server（钉钉消息通知）
Redis Server（缓存查询）

AI 自动根据任务选择调用哪个 Server 的哪个工具。

但 Server 多了，启动加载会变慢。我的做法是用一个代理 Server 做路由：

# proxy_server.py
@mcp.tool()
def route_request(server_name: str, tool_name: str, params: dict) -> dict:
    """将请求路由到对应的后端 MCP Server"""
    servers = {
        "mysql": mysql_client,
        "dingtalk": dingtalk_client,
        "redis": redis_client,
    }
    server = servers.get(server_name)
    if not server:
        return {"error": f"未知的 Server: {server_name}"}
    return server.call_tool(tool_name, params)

只暴露一个端口给 AI，后端 Server 独立部署、独立扩容。

延伸思考

MCP 目前还处于早期阶段，有几个方向值得关注：

认证与授权：MCP 协议目前没有标准化的认证机制。生产环境需要在传输层自己解决（Nginx 认证、API Key、OAuth2 Proxy）。
Server 发现：当你有几十个 MCP Server 时，怎么让 Host 知道自己该连谁？社区有 MCP Registry 的提议，但还没有统一标准。
工具调用链：多个 Tool 组合成一个工作流（比如先查订单再发通知），目前需要 AI 自己规划。如果能声明式定义工作流，会更可靠。
Streaming 响应：大数据量场景下，Tool 返回流式数据比一次性返回更友好。MCP 协议已经有 streaming 的雏形，但生态支持还在路上。

如果你正准备接 MCP，我的建议是：从小处开始，先接一个非关键工具跑通流程，再逐步扩展到核心业务。别想一步到位搭完所有基础设施——MCP 还在快速演进，适度超前部署即可。

你可能感兴趣的文章

来源：每日教程， 每日一例，深入学习实用技术教程，关注公众号TeachCourse
转载请注明出处： https://teachcourse.cn/4085.html ，谢谢支持！

资源分享

分类：Android 标签：

MCP 协议原理与自定义服务器开发实战：从零搭建 AI 工具调用基础设施

MCP 协议原理与自定义服务器开发实战：从零搭建 AI 工具调用基础设施

MCP 是什么，解决什么问题

先跑一个官方 Demo 感受一下

从零开发一个 MySQL 查询 MCP Server

第一步：项目结构

第二步：实现 Server

第三步：限流和分页

第四步：搭配 Claude Desktop 使用

第五步：部署成远程服务

协议细节：理解 tool/resource/prompt 三大能力

1. Tool（工具）——最常用

2. Resource（资源）——暴露数据

3. Prompt（提示模板）——引导对话

踩坑记录

坑 1：JSON 序列化失败

坑 2：AI 读不懂工具说明

坑 3：一次性返回大量数据

坑 4：stdio 模式的进程管理

进阶：多 Server 组合与代理模式

延伸思考

你可能感兴趣的文章

资源分享

Python框架JWT实现登录token生成和接口校验

php详细介绍正则表达式实际用法

OpenClaw CLI 命令参考手册

ubuntu系统一键快速安装shadowsocks-libev服务完整教程，小白也会用

Android Studio启动模拟器提示“HAXM is not installed”的解决办法

Ubuntu系统16

mysql重新启动失败

关于WordPress后台设置中的WordPress地址和站点地址的解读

Android开发之混淆高级教程01

ArrayMap方法解析

赞助2.00

MCP 协议原理与自定义服务器开发实战：从零搭建 AI 工具调用基础设施

MCP 是什么，解决什么问题

先跑一个官方 Demo 感受一下

从零开发一个 MySQL 查询 MCP Server

第一步：项目结构

第二步：实现 Server

第三步：限流和分页

第四步：搭配 Claude Desktop 使用

第五步：部署成远程服务

协议细节：理解 tool/resource/prompt 三大能力

1. Tool（工具）——最常用

2. Resource（资源）——暴露数据

3. Prompt（提示模板）——引导对话

踩坑记录

坑 1：JSON 序列化失败

坑 2：AI 读不懂工具说明

坑 3：一次性返回大量数据

坑 4：stdio 模式的进程管理

进阶：多 Server 组合与代理模式

延伸思考

你可能感兴趣的文章

资源分享

选择：

赞助2.00