MCP 协议实战（上）：什么是 MCP，怎么跑起来

智能客服"只会说不会做"是 Function Call 硬编码之痛。本文从零讲清 MCP 的三层架构、JSON-RPC 通信流程，对比 Function Call 给出选型依据，手把手带你用 Python 搭建天气查询和数据库查询两个 MCP Server，最终接入阿里云百炼与 Qwen 智能体，跑通第一轮真实工具调用。

---

---

开篇：一个让你抓狂的场景

去年我做智能客服项目时，遇到了一个让人崩溃的问题：

Qwen 大模型能流利地回答用户问题——"您的订单目前是什么状态呢？我帮您查一下。"然后……就没有然后了。它查不了。

大模型就像一个口才极好的实习生：什么都知道，但什么工具都不会用。不能调数据库、不能查物流接口、不能操作文件系统。你把订单号给它，它只能凭空编一个物流状态。

团队最初用 Function Call 解决。很快发现，10 个业务系统、20 个接口，意味着要写 200 个适配函数。更崩溃的是，换一个模型（从 Qwen 换到 GPT）又要重写一遍。

这就是 MCP 要解决的问题。

---

一、MCP 一句话理解

MCP（Model Context Protocol）= 大模型和外部工具的"通用 USB 接口"

过去：每个模型（Qwen、GPT、Claude）接同一个数据库，要写三套代码。

现在：写一套 MCP Server，所有支持 MCP 的模型都能直接调用。

以前：Qwen → 适配层A → 数据库
      GPT → 适配层B → 数据库
      Claude → 适配层C → 数据库

现在：Qwen ↘
      GPT   →  MCP Server → 数据库
      Claude ↗

MCP 由 Anthropic 在 2024 年底开源，2026 年已成为 AI 应用开发的事实标准。阿里云百炼、Cursor、通义千问等国内主流平台均已原生支持。

二、核心概念：三层架构

MCP 定义了三个角色：

角色	职责	举例
Host（宿主）	运行 AI 应用的平台	阿里云百炼、Claude Desktop、Cursor
Client（客户端）	与 MCP Server 通信的协议客户端	百炼内置的 MCP 客户端
Server（服务端）	提供具体工具能力的程序	你写的天气查询 Server、数据库 Server

一个 Host 可以连接多个 MCP Server，每个 Server 提供一组工具。大模型自主决定何时调用哪个工具。

三、通信流程

一次完整的工具调用经历四个阶段：

1. 初始化：Client 与 Server 握手，交换能力信息
2. 发现工具：Client 调用 tools/list，获取 Server 提供的所有工具清单
3. 模型决策：大模型根据用户问题，从工具清单中选择合适的工具
4. 调用执行：Client 调用 tools/call，Server 执行并返回结果

所有通信基于 JSON-RPC 2.0，请求和响应都是标准 JSON 格式。

四、MCP vs Function Call：一张表看懂区别

维度	Function Call	MCP
工具定义	内嵌在每次请求的 prompt 中	独立运行，通过协议发现
跨模型复用	每个模型写一套适配	一套 Server，所有模型通用
工具生态	孤岛，项目间难以共享	MCP 服务广场，开箱即用
部署方式	随应用一起部署	独立进程，STDIO / HTTP / 云函数
维护成本	接口变更需改多处	只改 Server，所有 Host 自动同步

一句话总结：Function Call 是"一次性筷子"，MCP 是"不锈钢餐具"。

五、动手实操：天气查询 MCP Server

下面用 Python 从零搭建一个天气查询 MCP Server。完整代码不到 60 行。

5.1 环境准备

pip install mcp

5.2 完整代码

# weather_server.py
import json
import asyncio
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
from mcp.server.stdio import stdio_server

# 1. 创建 MCP Server 实例
server = Server("weather-server")

# 2. 注册工具：定义一个"查询天气"的能力
@server.list_tools()
async def list_tools():
    return [
        {
            "name": "get_weather",
            "description": "查询指定城市的天气信息",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如'北京'、'上海'"
                    }
                },
                "required": ["city"]
            }
        }
    ]

# 3. 实现工具逻辑
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_weather":
        city = arguments.get("city", "")
        # 这里可以替换为真实 API 调用
        weather_data = {
            "北京": "晴，25°C，湿度 40%",
            "上海": "多云，28°C，湿度 65%",
            "深圳": "阵雨，30°C，湿度 80%"
        }
        result = weather_data.get(city, f"暂不支持查询 {city} 的天气")
        return {"content": [{"type": "text", "text": result}]}

# 4. 启动 Server（STDIO 模式）
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationCapabilities(
                sampling=None,
                experimental=None,
                roots=None
            ),
            NotificationOptions()
        )

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

5.3 代码逐段解释

段落	作用
创建 Server 实例	给 Server 起个名字，用于标识
list_tools()	告诉大模型"我能干什么"——名称、描述、参数格式
call_tool()	实际执行逻辑——大模型决定调用后，这段代码真正运行
stdio_server()	通过标准输入输出与 Client 通信，最简单的方式

5.4 本地测试

python weather_server.py

Server 启动后会等待 JSON-RPC 请求。可以用 mcp CLI 工具测试：

mcp dev weather_server.py

六、对接国内大模型：阿里云百炼

本地跑通之后，把它接入国内大模型平台。

6.1 部署到云函数

阿里云函数计算 FC 完美支持 MCP Server 部署：

# 安装 Serverless Devs 工具
npm install -g @serverless-devs/s

# 初始化项目
s init mcp-server-python

# 部署
s deploy

6.2 在百炼 MCP 服务广场注册

1. 登录 阿里云百炼控制台
2. 进入「MCP 服务」→「自定义服务」
3. 填入云函数地址，完成注册

6.3 与 Qwen 智能体联动

在百炼创建智能体时，在「工具」选项中勾选刚刚注册的天气 Server。然后问智能体：

“北京今天天气怎么样？”

智能体会自动识别意图，调用你的 MCP Server，返回真实天气数据。

七、进阶实操：数据库查询 MCP Server

天气查询太简单？来一个更实用的——让大模型直接查数据库。

# db_server.py（核心片段）
import pymysql
from mcp.server import Server

server = Server("mysql-server")

# 安全：只允许 SELECT，禁止 DROP/DELETE/UPDATE
ALLOWED_TABLES = ["orders", "users", "products"]

@server.list_tools()
async def list_tools():
    return [{
        "name": "query_database",
        "description": f"查询数据库，仅限以下表：{', '.join(ALLOWED_TABLES)}",
        "inputSchema": {
            "type": "object",
            "properties": {
                "sql": {
                    "type": "string",
                    "description": "SELECT 查询语句"
                }
            },
            "required": ["sql"]
        }
    }]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "query_database":
        sql = arguments.get("sql", "").strip().lower()

        # 安全校验：只允许 SELECT
        if not sql.startswith("select"):
            return {"content": [{"type": "text",
                "text": "错误：仅允许 SELECT 查询"}]}

        # 表名校验
        table = sql.split("from")[1].split()[0].strip("`")
        if table not in ALLOWED_TABLES:
            return {"content": [{"type": "text",
                "text": f"错误：仅允许查询 {ALLOWED_TABLES}"}]}

        # 执行查询
        conn = pymysql.connect(host="localhost", user="readonly",
                               password="***", database="business")
        cursor = conn.cursor()
        cursor.execute(sql)
        rows = cursor.fetchall()
        conn.close()

        return {"content": [{"type": "text",
                "text": json.dumps(rows, ensure_ascii=False, default=str)}]}

安全要点：

• 数据库账号用只读权限
• SQL 白名单校验
• 禁止所有写操作

---

感谢阅读，记得点赞、关注、收藏，欢迎各位评论区交流！！！