MCP协议简介

1. Model Context Protocol (模型上下文协议，简称：MCP) 是由Anthropic 2024年底提出的开放协议，
2. 旨在标准化AI大模型与外部数据源和工具的连接。
3. 官网称 MCP 是 AI 应用的 USB-C 端口，提供一种将 AI 模型连接到不同数据源、工具的标准化方法。目前，包括Block、Apollo、Replit等知名企业已开始采用 MCP 协议，显示出其在 AI 领域的重要地位和发展潜力。
4. 简单来说，MCP 是一种客戶端-服务器架构的协议，允许 LLM 应用程序（如 Claude、各种 IDE 等）通过标准化的接口访问外部数据和功能。
5. 这解决了 LLM 在实际应用中常⻅的一些痛点：
   a. 无法直接访问实时数据（如天气、股票行情等）；
   b. 无法执行外部操作（如发送邮件、控制设备等）；
   c. 无法访问用戶的本地文件或其他私有数据（如性别、年龄、收入等）。
6. 通过MCP，这些限制得到了优雅的解决，同时保持了安全性和可扩展性。所以也有人说，MCP是 AI 的 “Type-C”，统一接口接入各种工具

MCP 主要架构

刚有提到，MCP 采用客戶端-服务器架构，主要包含以下几个组件：

1. MCP 主机（Host）：如Claude Desktop、IDE或其他AI工具，通过MCP访问数据。
2. MCP 客戶端（Client）：与服务器保持 1:1 连接的协议客戶端。
3. MCP 服务器（Server）：轻量级程序，通过标准化的MCP 协议公开特定功能。
4. 本地数据源（Local Data Source）：计算机上的文件、数据库和服务，MCP 服务器可以安全访问这些内容。
5. 远程服务（Remote Service）：通过互联网可用的外部系统（例如通过API），MCP 服务器可以连接这些服务

整体的大致内容如下：

最左侧是一个用户，我们的用户，通过我们例如dify平台这个工具，调用dify平台的服务端，告诉它，要查询杭州的天气，这个是用户的需求，相当于，把这个需求告诉给了AI平台——dify客户端，然后AI工具将用户的语言进行转换，以及其他工具进行打包，整体交给AI大模型，这里的大模型是支持工具的大模型，然后大模型就开始进行思考，有没有能够查询天气的相关工具，如果没有，就应该是直接返回没有。AI客户端比如说dify接收到有查询天气的工具之后，就开始调用这个工具，这个时候是不会再去调用AI模型的；之后这个工具就开始搜索关键字，比如说搜索到了“杭州”这两个关键字，然后把这两个关键字传递给我们远程的数据服务，这里比如说天气网等第三方服务，然后天气网就把信息返回给我们的工具，我们这个工具再把信息返回给我们的dify平台，dify平台接收到信息后，会再次把信息传递给我们的大模型，最后，我们的大模型组织完语言后，返回给用户

案例

接下来，我们可以举出几个案例

查询天气的MCP，使用python实现

首先我们创建一个python目录和文件

如果我们不借助MCP，我们也能够正常的实现给自己的微信好友发送消息

from wxauto import WeChat
import time

"""
微信发送消息-测试

pip install  wxauto
"""

wx = WeChat()
time.sleep(2000)  # 等待微信初始化

# 定位到文件传输助手
if wx.ChatWith("文件传输助手"):
    # 发送文本
    wx.SendMsg("2025年05月21号222222！")

    # 发送文件（多个文件用列表）
    # wx.SendFiles([r"C:\test.txt", r"C:\test.xlsx"])
else:
    print("未找到目标会话")

然后，我们也可以自己正常通过接口获取某一个地方的天气

import httpx
import urllib.parse
import asyncio

"""
获取城市天气信息-测试
"""

async def get_weather(location: str) -> str:
    """Get weather for a location."""
    # 1.获取城市代码
    url = "http://weather.cma.cn/api/autocomplete?q=" + urllib.parse.quote(location)

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, timeout=10.0)
            response.raise_for_status()
            data = response.json()
            if data["code"] != 0:
                print(url)
                print(data)
                return "系统错误，请稍后重试"
            location_code = ""
            for item in data["data"]:
                str_array = item.split("|")
                if (
                        str_array[1] == location
                        or str_array[1] + "市" == location
                        or str_array[2] == location
                ):
                    location_code = str_array[0]
                    break
            if location_code == "":
                return "没找到该位置的信息"
            print('城市代码为：', location_code)
            # 2.查询城市天气
            url = f"http://weather.cma.cn/api/now/{location_code}"
            response = await client.get(url, timeout=10.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            print(url)
            return "系统错误，请稍后重试"


if __name__ == "__main__":
    # Initialize and run the server
    result = asyncio.run(get_weather('北京'))
    print(result)

那么，我们怎么通过MCP协议去实现呢？
首先我们需要使用pip命令去安装pip install mcp[cli]
总体来讲就是
第一步：FastMCP
pip install mcp[cli]
第二步：运行mcp服务
1.实例化FastMCP对象
2.函数添加装饰器mcp.tool
3.函数添加描述
4.Main 方法中，调用run 方法

from mcp.server import FastMCP
from wxauto import WeChat
import time

"""
微信发送消息mcp服务
pip install  wxauto
"""

wx = WeChat()
time.sleep(2)  # 等待微信初始化

# 1.实例化FastMCP对象
mcp = FastMCP(host='192.168.10.52', port=8000)


# mcp = FastMCP(host='127.0.0.1',port=8000)
# mcp = FastMCP(port=8000)
# 测试
# curl http://192.168.10.52:8000/tools


@mcp.tool(name="wechat_server")
def wechat_server(msg: str, who: str) -> str:
    """发送微信文本消息"""
    print(f'需要发送的消息为：{msg}')
    if wx.ChatWith(who):
        # 发送文本
        wx.SendMsg(msg)
        return 'success'
    else:
        print("未找到目标会话")
        return 'fail'


if __name__ == '__main__':
    # wechat_server('是的是的水电费', '文件传输助手')
    # wechat_server('你好', '樊老师')
    mcp.run(transport="sse")

"""
第一步：FastMCP
pip install   mcp[cli]
第二步：运行mcp服务
 1.实例化FastMCP对象 
 2.函数添加装饰器mcp.tool
 3.函数添加描述
 4.Main 方法中，调用run 方法

"""

import urllib
import httpx
from mcp.server import FastMCP

"""
查询天气mcp服务
"""

# 1.实例化FastMCP对象
mcp = FastMCP(host='192.168.10.52', port=8001)


# mcp = FastMCP(host='127.0.0.1',port=8001)
# mcp = FastMCP(port=8001)
# 测试
# curl http://192.168.10.52:8001/tools


@mcp.tool(name="wechat_server")
async def wechat_server(location: str) -> str:
    """根据城市名称获取天气信息."""

    print(f'查询的城市名称为：{location}')
    # 1.获取城市代码
    url = "http://weather.cma.cn/api/autocomplete?q=" + urllib.parse.quote(location)

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, timeout=10.0)
            response.raise_for_status()
            data = response.json()
            if data["code"] != 0:
                return "系统错误，请稍后重试"
            location_code = ""
            for item in data["data"]:
                str_array = item.split("|")
                if (
                        str_array[1] == location
                        or str_array[1] + "市" == location
                        or str_array[2] == location
                ):
                    location_code = str_array[0]
                    break
            if location_code == "":
                return "没找到该位置的信息"
            print('城市代码为：', location_code)
            # 2.查询城市天气
            url = f"http://weather.cma.cn/api/now/{location_code}"
            response = await client.get(url, timeout=10.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return "系统错误，请稍后重试"


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

我们可以看到，运行之后，相当于是启动了一个服务

也就是在使用过程中，我们需要使用@mcp.tool()进行装饰，然后对函数进行添加描述
这个时候，我们运行我们py文件，我们就相应的可以启动我们这个工具了，也就是说，我们的MCP工具已经开发好，我们可以进行调用使用了

那么，怎么使用呢
如果以agent为例，可以经过以下步骤进行尝试
第一步：我们需要安装MCP插件
第二步：配置MCP，也就是，将我们的MCP服务配置到工具中
第三步：制定智能体，调用工具
第四步：大模型的提示词编写
这里有机会再给大家进行详细介绍_期待中