缪斯：MCP 开发指南：从入门到实战

缪斯：MCP 开发指南：从入门到实战 缪斯：MCP 开发指南：从入门到实战 Modified April 7, 2025 Code block Python Copy :return: 文件内容或错误消息 """ try: with open(file name, "r", encoding="utf 8") as file: return file.read() except FileNotFoundError: return f"文件'{file name}'未找到" except Exception as e: return f"读取文件失败: {str(e)}" @mcp.tool() async def write file(file name: str, content: str) str: """ 写入文件内容 :param file name: 文件名 :param content: 文件内容 :return: 写入成功消息 """ try: with open(file name, "w", encoding="utf 8") as file: file.write(content) return f"文件'{file name}'写入成功" except Exception as e: return f"写入文件失败: {str(e)}" if name == " main ": 以标准 I/O 方式运行 MCP 服务器 mcp.run(transport="stdio") 服务端代码详解 服务端代码的逻辑主要是： 初始化服务 启动一个名为 filesystem 的MCP服务 Code block Plain Text Copy mcp = FastMCP("filesystem") 创建tool方法 通过@mcp.tool()装饰器注册三个 MCP 服务器工具：read file 、write file，create file，能够被客户端调用。 注意：方法注释一定要写上，内容包含：tool的功能，入参，出参，他是服务端tool的说明书，是大模型能否成功调用该方法的关键。 服务端入口代码 以标准 I/O 方式运行 MCP 服务器 Code block Plain Text Copy mcp.run(transport="stdio") 通过 mcp.run(transport='stdio') 启动 MCP 服务器，采用标准 I/O 通信方式，等待客户 端调用，适合客户端和服务端在同一台电脑上通信。 stdio 模式是一种本地进程间通信（IPC，Inter Process Communication）的方式，它需要服务端程序作为子进程运行，并通过标准输入输出 （ stdin / stdout ）进行数据交换。 因此，当我们指定 transport='stdio' 运行 MCP 服务器时，需要先启动服务端程序server.py ，然后再启动客户端程序 client.py 。 这样客户端才能和服务端通信。 启动服务 我们可以在启动一个命令行运行服务端程序： Code block Python Copy uv run server.py 修改客户端使其能与服务端通信 客户端核心逻辑 前面，我们写了客户端代码，但是只是说接通了在线或本地大模型，客户端若需要与服务端通信，就需要修改客户端代码了，客户端的核心逻辑就要调整为： 1、启动并初始化 MCP 客户端 2、连接MCP服务端 2、列出 MCP 服务器上的工具 3、运行交互式聊天循环，处理用户对话 • • 将用户输入发送给 OpenAI 模型 • • 如果模型想调用 MCP 工具（Function Calling），就执行 call tool • • 将结果重新发给模型，并返回最终回答 修改原有客户端代码 在原有基础上修改代码，只需要增加或修改一下代码即可： 增加连接服务端的方法 Code block Python Copy async def connect to server(self, server script path: str): """ 连接到 MCP 服务器 """ 检查文件类型是否为 Python 或 JavaScript is python = server script path.endswith('.py') is js = server script path.endswith('.js') if not (is python or is js): raise ValueError("不支持的文件类型") 根据文件类型选择命令 command = "python" if is python else "node" 设置服务器启动参数 server params = StdioServerParameters(command=command, args=[server script path], env=None) 启动 MCP 服务器并建立通信 stdio transport = await self.exit stack.enter async context( stdio client(server params) ) 获取输入输出流 self.stdio, self.write = stdio transport 创建会话并初始化 self.session = await self.exit stack.enter async context( ClientSession(self.stdio, self.write) ) await self.session.initialize() 增加列出服务端工具列表的方法 Code block Python Copy async def list tools(self): """列出所有工具""" 列出 MCP 服务器上的工具 response = await self.session.list tools() tools = response.tools 输出已连接的工具列表 print("已连接到服务器，服务器支持以下工具:", [tool.name for tool in tools]) 修改用户对话逻辑 Code block Python Copy import json async def process query(self, query: str) str: """ 调用大模型处理用户输入并执行相关工具操作 """ 用户输入的消息 messages = [{"role": "user", "content": query}] 获取服务器上的工具列表 response = await self.session.list tools() 准备服务器上可用的工具 available tools = [{ "type": "function", "function": { "name": tool.name, "description": tool.description, "input schema": tool.inputSchema } } for tool in response.tools] print('服务端工具列表:', available tools) 调用大模型进行对话处理 response = self.client.chat.completions.create( model=self.model, messages=messages, tools=available tools ) 获取返回内容 content = response.choices[0] if content.finish reason == "tool calls": 如果模型返回要使用工具的请求，执行相应工具 tool call = content.message.tool calls[0] tool name = tool call.function.name tool args = json.loads(tool call.function.arguments) 执行工具 result = await self.session.call tool(tool name, tool args) print(f"\n\n[Calling tool {tool name} with args {tool args}]\n\n") 将模型返回的原始消息和工具执行的结果都添加到 messages 中 messages.append(content.message.model dump()) messages.append({ "role": "tool", "content": result.content[0].text, "tool call id": tool call.id, }) 将结果再返回给大模型，获取最终结果 response = self.client.chat.completions.create( model=self.model, messages=messages, ) return response.choices[0].message.content 如果没有工具调用请求，直接返回模型生成的内容 return content.message.content 入口方法也要做调整 Code block Python Copy async def main(): """启动并初始化 MCP 客户端并处理用户对话""" 启动并初始化 MCP 客户端 client = MCPClient() try: 连接到 MCP 服务器 print("正在连接到 MCP 服务器...") await client.connect to server('server.py') print("成功连接到 MCP 服务器") 列出 MCP 服务器上的工具 await client.list tools() 运行交互式聊天循环，处理用户对话 print("进入聊天模式，输入 'exit' 退出") await client.chat loop() except Exception as e: 处理连接或其他错误 print(f"发生错误: {str(e)}") finally: 清理资源 print("清理资源并退出...") await client.clean() 开头的导入内容也要对应做修改 Code block Python Copy import asyncio import os from openai import OpenAI from dotenv import load dotenv from contextlib import AsyncExitStack from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio client import json 客户端最终完整代码 Code block Python Copy 更多MCP服务器合集导航地址 号称最大的MCP集合导航站：https://mcp.so/ MCP集合：https://github.com/ahujasid/blender mcp 官方MCP合集：https://github.com/modelcontextprotocol/servers Github热门MCP导航：https://github.com/punkpeye/awesome mcp servers 结语 服务端代码详解 服务端代码的逻辑主要是： 初始化服务 启动一个名为 filesystem 的MCP服务 创建tool方法 通过@mcp.tool()装饰器注册三个 MCP 服务器工具：read file 、write file，create file，能够被客户端调用。 注意：方法注释一定要写上，内容包含：tool的功能，入参，出参，他是服务端tool的说明书，是大模型能否成功调用该方法的关键。 服务端入口代码 以标准 I/O 方式运行 MCP 服务器 通过 mcp.run(transport='stdio') 启动 MCP 服务器，采用标准 I/O 通信方式，等待客户 端调用，适合客户端和服务端在同一台电脑上通信。 stdio 模式是一种本地进程间通信（IPC，Inter Process Communication）的方式，它需要服务端程序作为子进程运行，并通过标准输入输出 （ stdin / stdout ）进行数据交换。 因此，当我们指定 transport='stdio' 运行 MCP 服务器时，需要先启动服务端程序server.py ，然后再启动客户端程序 client.py 。 这样客户端才能和服务端通信。 启动服务 我们可以在启动一个命令行运行服务端程序： 修改客户端使其能与服务端通信 客户端核心逻辑 前面，我们写了客户端代码，但是只是说接通了在线或本地大模型，客户端若需要与服务端通信，就需要修改客户端代码了，客户端的核心逻辑就要调整为： 1、启动并初始化 MCP 客户端 2、连接MCP服务端 2、列出 MCP 服务器上的工具 3、运行交互式聊天循环，处理用户对话 • • 将用户输入发送给 OpenAI 模型 • • 如果模型想调用 MCP 工具（Function Calling），就执行 call tool • • 将结果重新发给模型，并返回最终回答 修改原有客户端代码 在原有基础上修改代码，只需要增加或修改一下代码即可： 增加连接服务端的方法 增加列出服务端工具列表的方法 修改用户对话逻辑 入口方法也要做调整 开头的导入内容也要对应做修改 客户端最终完整代码 更多MCP服务器合集导航地址 号称最大的MCP集合导航站：https://mcp.so/ MCP集合：https://github.com/ahujasid/blender mcp 官方MCP合集：https://github.com/modelcontextprotocol/servers Github热门MCP导航：https://github.com/punkpeye/awesome mcp servers 结语 以上，我们通过一个极其简单的案例，让MCP客户端与服务端进行通信，并成功调用服务端的对应工具，帮我们完成了一件事情。 这意味着，AI可以通过MCP服务完成几乎任何事情，比如：操作电脑本地文件夹，帮我们浏览网页并获取内容存，然后存入本地等等很多场景，想象空间很大，而这一切都是由AI驱动完成的！ MCP的功能远不止此，它还支持SSE传输模式，实现服务器与客户端异地运行；提供Resources类资源接口和Prompt类提示词模板；通过标准化协议促进开发者协作，已有数千服务器可供调用。 快快实战起来吧，何以破局，唯有行动！ MCP是一个典型的CS架构，对于有编程基础的同学来说，很容易理解，因为开发中常见的MySql就是典型的CS架构，程序员日常开发过程中，会经常接触到CS架构的产品。 本文不做太多关于MCP是什么的解释，尽最可能减少无关的噪音，只需要知道，MCP（Model Context Protocol）是一个标准化协议，通过客户端 服务端架构，让AI模型能够安全地调用工具、访问外部数据源，并实时获取信息。 接下来我们通过极简单的案例，带大家手搓一个MCP客户端和服务端代码，来快速了解MCP是什么？：让AI根据你的输入自动规划并调用MCP服务端，给本地电脑创建一个文件，并写入一句话。 如果这个案例你跑通了，你就会对MCP有一个初步的且正确的认知了。后面，你再去看网上其他关于MCP的大段文字科普，或者眼花缭乱的客户端配置MCP服务器的教程，再或者通过Dify等工作流与MCP结合，就会有种拨云见日，一览众山小的感觉。 环境安装 下载并安装python 官网：https://www.python.org/ 这里就不深入讲解如何安装了，不懂的同学网上找找教程。 安装uv 1. uv介绍 MCP开发要求借助uv进行虚拟环境创建和依赖管理。uv 是一个Python 依赖管理工具，采用 Rust 编写， 功能类似于pip，venv，但它更快、更高效，并且可以更好地管理 Python 虚拟环境和依赖项。也就是说他兼有了创建虚拟环境和包管理工具的功能，可以平替pip，venv。 它完全兼容 pip ：支持 requirements.txt 和 pyproject.toml 依赖管理。 跨平台：支持 Windows、macOS 和 Linux。 2. uv安装 ctrl+r，打开命令行，输入一下命令安装uv uv常见使用命令可以自行问AI或百度，此处不再赘述。 案例场景概述 我们先从最简单的案例入手，创建client端，server端，然后联调通，对mcp建立起一个初步完整的认知即可，尽最大可能避免引入复杂的东西。 所以，我们的案例就是：通过手动创建的client运行起来后，来调用server端的逻辑，server端的逻辑就是：创建一个名为aaa.txt文件，写入 今天天气真好！ ，即可。 MCP客户端 初始化client项目 进入自己的代码目录下，创建一个文件夹： 可以看到一个完整的初始化项目目录：包含项目入口文件，依赖管理文件等。 创建MCP客户端虚拟环境 这里我们用cursor打开项目，进行后面的操作。 这里，uv会自动识别当前项目主目录，然后自动创建虚拟环境。 现在，我们通过add方法在虚拟环境中安装相关的库。 注意：如果网络问题，安装依赖失败，可以更改镜像源，或者使用科学上网工具。 新增client所需依赖 后面，为了支持调用大模型，读取env环境变量中的API KEY等信息，需要先安装如下依赖： 接入硅基流动的免费大模型 创建env文件 env文件用来存放大模型API Key等配置信息： 编写MCP客户端 具体也可以参考官方示例：https://modelcontextprotocol.io/quickstart/client 客户端代码如下： 运行client，测试大模型是否连接成功 MCP服务端 创建服务端文件 server.py：