搞不懂 MCP？那就动手搓一个... - 文章 - 开发者社区

picture.image

猴哥的第 172 期分享，欢迎追看

MCP 火了有一阵了，市面上相关教程也不少，无奈看完，依旧一头雾水。

原因无他，没动手而已。

这两天实操了一把，发现 MCP 确实以一种更优雅的方式，帮我解决了之前难以搞定的问题。

今天从实战出发，聊聊：如何把 MCP 接入自己的 LLM ，希望给有需要的朋友，一点点参考。

本文将首先回答：

万能的 LLM 为啥干啥啥不行？
Function Call 是怎么诞生的？
MCP 解决了 Function Call 的什么痛点？

然后，以实操的方式回答：

MCP Server 有哪些实现方式？
MCP Client 和 Server 是如何通信的？
如何把 MCP 接入 LLM？

LLM 的局限性

各大厂商竞相发布 LLM，即便 benchmark 指标刷上天，拉到现实场景中，依然各种拉跨。

因为大模型训练完，它的信息量就不再更新。

比如用户要查个天气，就必须借助高德/百度地图等外部 API 来获取最新数据。

笔者在微信 AI 机器人的实现中，就是首先让 LLM 识别用户意图，然后再调用外部 API，获取相关消息后，进行回答：

picture.image

为了实现各种功能，必须在提示词中写明所有意图标签，然后逐一开发对应功能：

picture.image

实际你会发现，即便非常明确告诉它这些意图，意图识别也不是 100% 正确。

当然，解决上述问题的另外一个思路是Function call，这也是聊 MCP 之前，绕不开的话题。

Function call 的局限性

Function call（下文简称 FC）早在 2023 年，就由 OpenAI 提出，主要就是为了解决上述 LLM 的局限性。

你在 Coze/Dify 等智能体平台上，看到的各种插件，本质就是基于 FC 实现的。

但 FC 的缺点也很明显 ：必须要 LLM 本身支持，然后在 OpenAI 的接口中传入 tools 参数：

picture.image

但是不同 LLM 训练时，因为各种实现方式不一样，导致输入/输出结构、触发结构都不一样。

所以，换个 LLM 就得适配上面的 tools，实在是太难受。

只是 Coze/Dify 等平台，做了大量适配工作，才让你感到用起来很爽。

但是，即便是生态丰富的 Coze，也不是万能的，总有一些个性化需求，平台是满足不了的。

还是绕不开要定制化开发，然后面临头疼的问题：一顿操作写了个接口，换个 LLM 居然理解不了。

所以，MCP，它来了！

MCP 是啥

MCP 全称 Model Context Protocol，模型上下文协议。

就像 Type-C 为设备连接各种外设提供了标准化，MCP 为 LLM 连接各种工具提供了统一方式。

3.1 整体架构

Anthropic 公司最早提出 MCP 时的一张图，非常形象刻画出了它的整体架构：

picture.image

图中有几个关键概念：

MCP hosts ：应用程序，比如各种 AI 工具或 IDE，最新版 Trae 就已接入
MCP clients ：客户端组件，和服务端维持连接
MCP server ：服务端，通过 MCP 协议提供各种功能
Local data ：本地资源，比如本地文件、数据库等
Remote services ：远程服务，通过网络访问

抛开这些概念，MCP 和 Function call 在应用层面有什么区别？

答案是没区别，还是上面这张图，只是其中的 tools 交给 MCP client 来托管：

picture.image

下面，我们动手实操一番，相信你就能完全掌握 MCP 了。

3.2 MCP 服务端

服务端有两种：

一是有很多别人实现好了的 server，完全能够满足你的需求，拿来即用；
二是需要自己开发的 server；

如何接入别人的 server？

目前各大模型厂商都在拥抱 MCP 生态，比如：

阿里云 ModelScope：https://modelscope.cn/mcp：

picture.image

mcp.so：https://mcp.so/zh，也是一个聚合平台，收集了优秀的 MCP 服务器和客户端。

picture.image

以高德地图提供的 MCP 为例：

picture.image

这里就需要有本地安装好 node 环境，本质上就是通过如下命令，启动了 server 服务：

  
npx -y @amap/amap-maps-mcp-server

当然，也有通过云端服务提供的 server，比如这个 12306 的 MCP：

picture.image

如何自己开发 server？

基于 FastMCP 框架，和 FastAPI 用法非常类似。

假设我们要开发一个计算器，只需三步：

首先，创建一个名为 Calculator 的服务：

  
mcp = FastMCP("Calculator")

然后，添加一个 tool，实现具体功能：

  
def calculator(python\_expression: str) -> dict:  
    """For mathamatical calculation, always use this tool to calculate the result of a python expression. `math` and `random` are available."""  
    result = eval(python\_expression)  
    return {"success": True, "result": result}

最后，启动服务：

  
if \_\_name\_\_ == "\_\_main\_\_":  
    mcp.run(transport="stdio")

3.3 MCP 客户端

参考：https://modelcontextprotocol.io/quickstart/client

有了 Server，问题来了：我如何把它们接入自己的 LLM 使用？

答案是：还需要一个 client，负责以 MCP 协议 ，和所有的 server 进行通信！

MCP 协议 又怎么理解？

MCP 包括两种标准传输协议：

标准输入/输出（stdio） ：通过标准输入和输出流进行通信。这对于本地集成和命令行工具特别有用。
服务器发送事件（SSE） ：通过HTTP POST请求进行客户端到服务器的通信。

为了方便开发，官方开源了不同语言的 sdk：

不过从 v1.8.0 版本开始，“Streamable HTTP” 取代了 “SSE” 协议，优势体现在：

灵活性和标准化：用标准的 HTTP POST 和 GET 请求进行通信。
会话管理：服务端为每个客户端分配一个会话ID，方便处理多个客户端连接。

由于之前的服务都是基于 SSE，所以对 SSE 的支持并没有移除。

下文以 Python 为例展开。

3.4 MCPClient开发

之前，我们接入 LLM 的是 OpenAI 的接口：

  
self.openai = OpenAI(api\_key=xxx, base\_url=xxx)  
response = self.openai.chat.completions.create(  
            model=self.model,  
            messages=messages  
        )

现在要为它接入各种 MCP Server，为此需要封装一个 MCPClient。

具体流程如下：

step 1 ：定义 server 配置文件，格式如下：

  
{  
    "mcpServers": {  
        "filesystem": {  
            "command": "npx",  
            "args": [  
                "-y",  
                "@modelcontextprotocol/server-filesystem",  
                "/root/projects/"  
            ]  
        },  
        "calculator": {  
            "command": "python",  
            "args": [  
                "calculator.py"  
            ]  
        },  
        "time": {  
            "command": "docker",  
            "args": [  
                "run",  
                "-i",  
                "--rm",  
                "mcp/time"  
            ]  
        },  
        "amap-maps": {  
            "url": "https://mcp.api-inference.modelscope.cn/sse/xxx"  
        }  
    }  
}

step 2 ：初始化客户端实例：

  
class MCPClient:  
    def \_\_init\_\_(self):  
        # 初始化会话和客户端对象  
        self.session\_list: List[ClientSession] = []  
        self.available\_tools: List = []  
        self.tool\_session\_map: Dict = {}  
        self.exit\_stack = AsyncExitStack() # 退出堆栈  
        self.openai = OpenAI(api\_key=os.getenv("OPENAI\_API\_KEY"), base\_url=os.getenv("OPENAI\_BASE\_URL"))   
        self.model = "qwen-plus"  
        self.mcp\_server\_config = self.parse\_mcp\_server("./mcp\_server\_config.json")

step 3 ：连接服务器：根据不同协议，分别连接到 stdio 和 sse 服务器：

  
async def connect\_server(self):  
        for sever\_name in self.mcp\_server\_config:  
            server = self.mcp\_server\_config[sever\_name]  
            if server.get('url'):  
                session, tools = await self.connect\_sse\_server(sever\_name)  
            else:  
                session, tools = await self.connect\_stdio\_server(sever\_name)  
            self.session\_list.append(session)  
            self.available\_tools.append(tools)  
        # 构建 tool name 到 session 的映射  
        self.tool\_session\_map = {  
            tool['function']['name']: session for session, tools in zip(self.session\_list, self.available\_tools) for tool in tools  
        }

step 4 ：获取可用工具列表：以 stdio 服务为例：

  
async def connect\_stdio\_server(self, server\_name: str):  
        """连接到 stdio MCP 服务器"""  
        server = self.mcp\_server\_config[server\_name]  
        # 创建 StdioServerParameters 对象  
        server\_params = StdioServerParameters(  
            command=server['command'],  
            args=server['args'],  
            env=None  
        )  
        # 使用 stdio\_client 创建与服务器的 stdio 传输  
        read, write = await self.exit\_stack.enter\_async\_context(stdio\_client(server\_params))  
        # 创建 ClientSession 对象，用于与服务器通信  
        session = await self.exit\_stack.enter\_async\_context(ClientSession(read, write))  
        # 初始化会话  
        await session.initialize()  
        # 列出可用工具  
        response = await session.list\_tools()  
        tools = self.get\_tools(response)  
        return session, tools

step 5 ：工具调用和响应处理

  
async def process\_query(self, query: str) -> str:  
        messages = [{"role": "user", "content": query}]  
        while True:  
            tool\_functions, response = self.get\_response(messages)  
            messages.extend(response)  
            if not tool\_functions:  
                break  
            for tool\_functions in tool\_functions:  
                tool\_call\_id = tool\_functions['id']  
                tool\_name = tool\_functions['function']['name']  
                tool\_args = json.loads(tool\_functions['function']['arguments'])  
                session = self.tool\_session\_map[tool\_name]  
                result = await session.call\_tool(tool\_name, tool\_args)  
                messages.append({  
                    "role": "tool",  
                    "tool\_call\_id": tool\_call\_id,  
                    "content": json.dumps({  
                        "result": [content.text for content in result.content],  
                        "meta": str(result.meta),  
                        "isError": str(result.isError)  
                    })  
                })  
        return messages

篇幅受限，这里只贴核心步骤的代码，需要完整代码的朋友，文末自取！

3.5 结果展示

问它当前时间：

  
messages = await client.process\_query("现在北京时间几点")

输出结果如下：

  
{'role': 'user', 'content': '现在北京时间几点'}  
{'role': 'assistant', 'content': None, 'tool\_calls': [{'id': 'call\_ffb7460aa192417cb0218e', 'type': 'function', 'function': {'name': 'get\_current\_time', 'arguments': '{"timezone": "Asia/Shanghai"}'}}]}  
{'role': 'tool', 'tool\_call\_id': 'call\_ffb7460aa192417cb0218e', 'content': '{"result": ["{\\n  \\"timezone\\": \\"Asia/Shanghai\\",\\n  \\"datetime\\": \\"2025-05-25T16:04:43+08:00\\",\\n  \\"is\_dst\\": false\\n}"], "meta": "None", "isError": "False"}'}  
{'role': 'assistant', 'content': '现在北京时间是2025年5月25日下午15点49分37秒。请注意，这个时间是基于标准时间，不适用夏令时。'}

可以看到，成功调用了 time 服务中的工具。

再来个双重调用的任务：

  
messages = await client.process\_query("现在北京时间几点，天气怎么样")

输出结果如下：

  
{'role': 'user', 'content': '现在北京时间几点，天气怎么样'}  
{'role': 'assistant', 'content': None, 'tool\_calls': [{'id': 'call\_18d31b5557e749fbb40b6f', 'type': 'function', 'function': {'name': 'get\_current\_time', 'arguments': '{"timezone": "Asia/Shanghai"}'}}]}  
{'role': 'tool', 'tool\_call\_id': 'call\_18d31b5557e749fbb40b6f', 'content': '{"result": ["{\\n  \\"timezone\\": \\"Asia/Shanghai\\",\\n  \\"datetime\\": \\"2025-05-25T16:06:17+08:00\\",\\n  \\"is\_dst\\": false\\n}"], "meta": "None", "isError": "False"}'}  
{'role': 'assistant', 'content': None, 'tool\_calls': [{'id': 'call\_64410ee2fe104b1b943037', 'type': 'function', 'function': {'name': 'maps\_weather', 'arguments': '{"city": "北京"}'}}]}  
{'role': 'tool', 'tool\_call\_id': 'call\_64410ee2fe104b1b943037', 'content': '{"result": ["{\\n  \\"city\\": \\n}"], "meta": "None", "isError": "False"}'}  
{'role': 'assistant', 'content': '现在北京时间是2025年5月25日16点06分17秒。\n\n北京今天的天气情况如下：\n- 白天天气：多云\n- 晚上天气：多云\n- 白天最高温度：28°C\n- 晚上最低温度：15°C\n- 白天风向：东\n- 晚上风向：东\n- 白天风力：1-3级\n- 晚上风力：1-3级\n\n这是未来几天的天气预报：\n- 2025年5月26日：白天和晚上都是多云，温度在17°C到29°C之间。\n- 2025年5月27日：白天和晚上都是阴，温度在20°C到31°C之间。\n- 2025年5月28日：白天多云，晚上阴，温度在20°C到31°C之间。'}

成功调用了time服务和高德地图服务。

最后，测试下本地搭建的数学计算服务：

  
messages = await client.process\_query("半径是8，计算圆的面积")

输出结果如下：

  
{'role': 'user', 'content': '半径是8，计算圆的面积'}  
{'role': 'assistant', 'content': None, 'tool\_calls': [{'id': 'call\_c247e31e964344c9922c3d', 'type': 'function', 'function': {'name': 'calculator', 'arguments': '{"python\_expression": "math.pi * 8**2"}'}}]}  
{'role': 'tool', 'tool\_call\_id': 'call\_c247e31e964344c9922c3d', 'content': '{"result": ["{\\n  \\"success\\": true,\\n  \\"result\\": 201.06192982974676\\n}"], "meta": "None", "isError": "False"}'}  
{'role': 'assistant', 'content': '圆的面积为 \\(201.06\\) 平方单位（结果四舍五入到小数点后两位）。'}