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

《青蛇》

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：，官方使用的 LLM 是Anthropic，我的代码里用的是OpenAI的SDK，做了一些微小的改动。完整代码如下：

python

体验AI代码助手

代码解读

复制代码

import json import os import asyncio from typing import Optional from contextlib import AsyncExitStack from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from 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.tools print("\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.message is_function_call = message.tool_calls if is_function_call: tool_call = 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) 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 则退出循环 break response = 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 Server await client.connect_to_server(sys.argv[1]) # 开启聊天循环 await client.chat_loop() finally: # 程序终止，清理资源 await client.cleanup() if __name__ == "__main__": import sys # 异步运行主函数 asyncio.run(main())

关键组件解释

1. 客户端初始化

• MCPClient 类使用会话管理和 API 客户端进行初始化
• 使用 AsyncExitStack 进行适当的资源管理
• 使用OpenAI sdk编写 LLM 使用逻辑

1. 服务器连接

• 支持 Python 和 Node.js 服务器
• 验证服务器脚本类型
• 建立适当的沟通渠道（控制台的输入输出）
• 初始化会话并列出可用工具

1. 查询处理

• 维护对话上下文
• 处理 OpenAI 的响应和工具调用
• 管理 OpenAI 和 tools 之间的消息流
• 将结果合并为连贯的响应

1. 交互式界面

• 提供简单的命令行界面
• 处理用户输入并显示响应
• 包括基本错误处理
• 允许正常退出

1. 资源管理

• 正确清理资源
• 连接问题的错误处理
• 正常关闭过程

这里我们通过uv run main.py <path/to/server.py>运行我们的python代码。

当我们输入“获取alice的用户信息”，按回车键发送后，模型将调用工具，并做如下回复

04 总结

至此，我们就从 0~1 完成了一个简单的，属于我们自己的 MCP Client。

整个 client 执行逻辑如下，当我们提交查询时：

• 客户端从服务器获取可用工具的列表
• 您的查询将与工具描述一起发送给 LLM
• LLM 决定使用哪些工具（如果有）
• 客户端通过服务器执行请求的工具调用
• 结果将发送回 LLM
• LLM 提供自然语言响应
• 此时将向用户显示响应

这生活已经很难了，难能可贵的是您能在百忙之中抽空阅读这些文章。如果能给到您一点小小的帮助，也是我非常喜闻乐见的😁

作者：imtoken
链接：https://www.chinaqicheng.com/post/7346589651801994280
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。