当前位置：首页 > news >正文

【MCP】从0到1实现一个MCP Client

news 2025/5/11 8:04:49

如果连人间真爱都不知，千万年的修行还有什么用？

《青蛇》

01 MCP回顾

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

上一篇文章【MCP】从0到1实现一个MCP Server 我们介绍了如何从0~1实现一个自定义的 MCP Server，感兴趣的朋友可以点击文章链接回顾下或者查看下。

02 MCP Client本质

为什么我们需要 MCP Client

因为我们在拿到一个 MCP Server 的时候没办法直接使用，必须结合 LLM 才能发挥这个 Server 的能力，所以我们需要一个基于 LLM 的 Client 去承接这个 Server。此时，MCP Client就出现了。

MCP Client 的本质

纵观市面上所有的，基于 LLM 的应用或者说客户端，无不是聊天对话机器人形式。再到具体的应用，比如说Cline，比如说Cursor等等，也是对话形式，coze、dify、n8n工具的本质也是对话。再往底层说，LLM 的本质就是输入输出。所以，如果我们要自定义一个 MCP Client，那必然也是对话形式。

03 快速实现简单的MCP Client

开始构建可与所有 MCP 服务器集成的，专属我们自己的客户端。在本教程中，我们将学习如何构建连接到 MCP 服务器的 LLM 支持的聊天机器人客户端。

这里我们使用 Python 实现MCP Client的编码与调用，Python版本为3.10或更高，MCP SDK 版本为 1.2.0 或更高。

官方文档建议使用 uv 作为包管理工作，那这次我就用一下吧。

这里默认各位朋友是掌握了 Python 语言的，所以我省略了环境的配置，也省略了项目的创建。我的项目长这样：

在这里插入图片描述

跟 MCP Server 的开发一样，就一个 main.py 文件，所有逻辑都在这一个文件里面。

我的代码基本逻辑来源于官方Demo：https://gist.github.com/zckly/f3f28ea731e096e53b39b47bf0a2d4b1 ，官方使用的 LLM 是Anthropic，我的代码里用的是OpenAI的SDK，做了一些微小的改动。完整代码如下：

import json
import os
import asyncio
from typing import Optional
from contextlib import AsyncExitStackfrom mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_clientfrom openai import OpenAI
from dotenv import load_dotenv# 加载.env文件中的环境变量
load_dotenv()class MCPClient:def __init__(self):# 初始化会话和客户端对象self.session: Optional[ClientSession] = None # 用于保存 MCP 客户端会话self.exit_stack = AsyncExitStack()  # 用于管理异步资源的生命周期# 初始化 OpenAI 客户端self.client = OpenAI(base_url=os.getenv("BASE_URL"),api_key=os.getenv("API_KEY"),  # 从环境变量中获取 API 密钥)async def connect_to_server(self, server_script_path: str):"""连接到 MCP 服务器参数:server_script_path: 服务器脚本路径 (.py 或 .js)"""is_python = server_script_path.endswith('.py')  # 判断是否为 Python 脚本is_js = server_script_path.endswith('.js') # 判断是否为 JavaScript 脚本if not (is_python or is_js):raise ValueError("Server script must be a .py or .js file")command = "python" if is_python else "node" # 根据文件类型选择执行命令server_params = StdioServerParameters(command=command,args=[server_script_path],env=None)# 启动标准输入输出客户端stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))self.stdio, self.write = stdio_transport# 创建并初始化 MCP 客户端会话self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))await self.session.initialize() # 初始化会话# 列出可用工具response = await self.session.list_tools()tools = response.toolsprint("\nConnected to server with tools:", [tool.name for tool in tools]) # 打印可用工具名称async def process_query(self, query: str) -> str:"""处理用户查询，使用 LLM 和可用工具进行响应"""messages = [{"role": "user","content": query # 用户输入的查询内容}]# 获取可用工具列表并构造工具调用格式response = await self.session.list_tools()available_tools =[{"type": "function","function": {"name":tool.name, # 工具名称"description": tool.description,  # 工具描述"parameters": tool.inputSchema # 输入参数格式}}for tool in response.tools]# 调用 LLM 模型生成回复response = self.client.chat.completions.create(model=os.getenv("MODEL"),  # 使用的模型名称messages=messages, # 对话历史tools=available_tools # 可用工具列表)# 处理模型回复并执行工具调用（如有）tool_results = []final_text = []for choice in response.choices:message = choice.messageis_function_call = message.tool_callsif is_function_call:tool_call = message.tool_calls[0]tool_name = tool_call.function.nametool_args = json.loads(tool_call.function.arguments)# 执行工具调用result = await self.session.call_tool(tool_name, tool_args)tool_results.append({"call": tool_name, "result": result})final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")  # 记录调用信息# 将工具调用结果加入对话历史if message.content and hasattr(message.content, 'text'):messages.append({"role": "assistant","content": message.content})messages.append({"role": "user","content": result.content[0].text})# 再次调用 LLM 获取最终回复response = self.client.chat.completions.create(model=os.getenv("MODEL"),messages=messages,)final_text.append(response.choices[0].message.content) # 添加模型最终回复else:final_text.append(message.content) # 直接添加模型回复return "\n".join(final_text)  # 返回最终回复内容async def chat_loop(self):"""运行交互式聊天循环"""print("\nMCP Client Started!")print("Type your queries or 'quit' to exit.")while True:try:query = input("\nQuery: ").strip() # 获取用户输入if query.lower() == 'quit': # 如果输入 quit 则退出循环breakresponse = await self.process_query(query)  # 处理用户查询print("\n" + response) # 打印回复except Exception as e:print(f"\nError: {str(e)}")async def cleanup(self):"""Clean up resources"""await self.exit_stack.aclose()  # 关闭所有异步资源async def main():if len(sys.argv) < 2:print("Usage: python main.py <path_to_server_script>")sys.exit(1)# 创建 MCP 客户端client = MCPClient()try:# 将客户端链接到 MCP Serverawait client.connect_to_server(sys.argv[1])# 开启聊天循环await client.chat_loop()finally:# 程序终止，清理资源await client.cleanup()if __name__ == "__main__":import sys# 异步运行主函数asyncio.run(main())