DC娱乐网

一文理解并掌握MCP的开发

最近在探索如何使用大模型来改进自动化测试,在入门MCP的过程中遇到了不少困扰,比如个人感觉MCP协议中的Host、Cli

最近在探索如何使用大模型来改进自动化测试,在入门MCP的过程中遇到了不少困扰,比如个人感觉MCP协议中的Host、Client、Server等概念特别具有迷惑性,让你搞不清它们到底是什么,还有网上的很多示例教程都是直接基于Trae、Cline、Claude这种IDE来讲解的,这些集成了AI的IDE隐藏了太多的细节,让人很难掌握整个应用的全局。本文通过尽量简单的示例代码让你能够快速开发一个MCP工具并演示如何使用大模型进行调用,运行本文示例不需要使用特殊的IDE。

MCP简介

大语言模型LLM就像一个超级大脑,拥有人类无法企及的知识库、强大的分析推理能力。然而,它就像是一个只会纸上谈兵的将军,说起来头头是道,却什么也不会做。一旦LLM能够调用各种工具,它就成为了智能体,不仅会说还能做,那将会产生无数可能性,难怪微软CEO纳德拉预测所有传统意义的软件应用将走向终结。

MCP协议就是大模型和各类工具沟通的规范和标准,没有MCP之前也能做到让大模型去调用工具,但是都是各自为战,很难产生合力,一旦统一了标准,就会出现互相共享的MCP服务市场,大家都可以共建共荣智能体的生态,这也是为什么MCP如此火爆的原因。

以UI自动化测试为例,传统的自动化测试需要测试人员编写自动化测试脚本来操作页面元素,在实际工作中你会发现这本身就是一个不小的挑战,测试人员很少有时间、有能力去完成这些工作,后续还要面临无穷无尽的维护工作。接入了大模型之后,可以直接使用自然语言来写用例,大模型通过识别UI页面中的特征可以精准定位元素并进行相应的操作和结果检查,这完全是可行的,如果有兴趣可以去了解下midscene.js。

MCP协议架构官方给出的就是下面这张图,包含Host、Client、Server关键角色,我觉得这些命名太有迷惑性了,让你搞不清它们到底怎么回事,我试着说明下自己的理解,不一定完全准确。

Host就是你的整个应用,并不是什么物理主机或者虚拟机什么的,它包含了Client和Server;Server负责提供工具、资源和提示词(LLM需要理解提示词才能知道Server提供什么能力以及如何调用它们);Client负责和Server通信,并将结果提供给Host。

整体来说,MCP就是大模型和工具之间的桥梁,工具通过提供结构化(json格式)的提示词告知大模型工具的功能描述、工具调用名称和参数等信息,大模型通过学习这些提示词就能自主决定调用哪个工具以及传入什么参数和获取工具响应。

选择模型

在正式进行mcp开发之前,你首先得有一个可用的模型,这里选择使用阿里通义千问大模型,有一定的免费额度。访问阿里云百炼平台https://bailian.console.aliyun.com/按照指引一步步注册申请API-Key即可。

API-Key申请成功之后,可以通过文档中提供的测试代码验证大模型是否可以正常使用。

后续的开发中需要指定大模型的型号,可以在模型广场中进行选择,Code列为具体的模型类型,点击详情可以查看模型token的消耗情况。如果后续希望使用大模型进行浏览器自动化操作需要选择多模态的大模型,因为大模型是通过接收浏览器页面截图来识别元素的,文本模型是不支持的。

MCP服务开发

接下来开始构建自己的MCP server,这里还是以官方的入门示例为参考,但是会尽量去掉干扰内容,比如不会使用uv工具(使用默认的pip工具),大模型也改为使用通义千问。

首先创建mcp-demo目录,执行python -m venv venv创建虚拟环境venv并激活,创建weather.py文件,为简化代码这里不调用查询天气API,仅模拟返回结果。

pip install "mcp[cli]"安装mcp官方sdk,注意这里需要python3.10以上版本

weather.py的代码如下,此时运行python weather.py就会启动mcp server,client就可以通过标准输入输出与之通信,但是一般不这么做,后面会介绍如何启动mcp server,至此就完成一个简单的MCP Server开发。

import asyncioimport randomfrom mcp.server.fastmcp import FastMCP# 初始化MCP实例mcp = FastMCP("weather")# 使用装饰器将方法注册为mcp工具,主要目的是提取方法名、方法描述、入参生成结构化描述信息给大模型@mcp.tool()async def get_alerts(state: str) -> str: """查询城市的天气预警信息 Args: state: 城市名称 """ alerts_list = ["龙卷风", "泥石流", "冰雹"] return f"天气预警结果为:{random.choice(alerts_list)}"@mcp.tool()async def get_forecast(latitude: float, longitude: float) -> str: """查询地区天气预报. Args: latitude: 经度 longitude: 纬度 """ weathers_list = ["多云", "小雨", "中雨", "大雨", "大雾", "晴天"] return f"天气预报结果为:{random.choice(weathers_list)}"if __name__ == "__main__": # 调试代码 # print(asyncio.run(get_alerts("上海"))) # print(asyncio.run(get_forecast(1, -1))) # 启动mcp server,使用标准输入输出进行通信 mcp.run(transport="stdio")MCP客户端开发

接下来我们开发client来连接mcp server并创建对话循环,实现大模型调用前面开发的查询天气的工具,特别要说明的是这里开发的客户端是可以连接任意mcp server的。

由于使用的是通义千问大模型,根据阿里云百炼的官方文档示例代码可知需要安装openai依赖库(pip install openai),因为它使用与openai兼容的api。相同目录下创建client.py,代码如下:

import osimport jsonimport asynciofrom typing import Optionalfrom contextlib import AsyncExitStackfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom openai import OpenAIclass MCPClient: def __init__(self): # Initialize session and client objects self.session: Optional[ClientSession] = None self.exit_stack = AsyncExitStack() # 将自己申请的通义大模型API_key加入到环境变量中 self.openai_client = OpenAI( api_key=os.getenv("OPEN_API_KEY"), base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" ) async def connect_to_server(self, server_script_path: str): """连接一个MCP server Args: server_script_path: mcp server的脚本路径 """ is_python = server_script_path.endswith('.py') is_js = server_script_path.endswith('.js') 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_client(server_params)会通过开启进程的方式启动mcp server, stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params)) self.stdio, self.write = stdio_transport # client session通过标准输入输出与mcp server通信 self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write)) await self.session.initialize() # client可以通过与mcp server通信获取可用的工具信息 response = await self.session.list_tools() tools = response.tools print("\n成功连接包含工具的服务:", [tool.name for tool in tools]) async def process_query(self, query: str) -> str: """使用通义千问大模型和可用工具执行查询""" # 结构化用户的提问 messages = [ { "role": "user", "content": query } ] # 根据mcp server获取的可用工具生成结构化的提示词 response = await self.session.list_tools() # 基于openai的格式要求 available_tools = [{ "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": tool.inputSchema } } for tool in response.tools] # 打印信息用于了解中间过程 print("messages", messages) print("available_tools", available_tools) # 此步调用大模型,会返回用户提问的分析结果以及选择调用的工具及实参 # model为选择的大模型类型,对应阿里云百炼中选择的模型型号 response = self.openai_client.chat.completions.create( model="qwen-vl-max-latest", messages=messages, tools=available_tools, temperature=0.1, tool_choice="auto", ) qwen_response = response.choices[0].message print("qwen_response", qwen_response) # 处理响应和调用mcp工具 final_text = [] assistant_message_content = [] # 大模型没有选择调用工具直接返回答案 if qwen_response.content: final_text.append(qwen_response.content) assistant_message_content.append(qwen_response.content) # 大模型选择调用工具 if qwen_response.tool_calls: for tool_call in qwen_response.tool_calls: # 解析调用工具方法名和实参 tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments) # 通过client向mcp server执行工具调用 result = await self.session.call_tool(tool_name, tool_args) final_text.append(f"[Calling tool {tool_name} with args {tool_args}]") # 更新对话历史,也就是维持上下文信息 assistant_message_content.append({ "type": "tool_use", "id": tool_call.id, "name": tool_name, "input": tool_args }) # 向提示词中加入调用工具信息,content最好为字符串 messages.append({ "role": "assistant", "content": json.dumps(assistant_message_content) }) # 向提示词中加入工具调用结果,content最好为字符串 messages.append({ "role": "user", "content": json.dumps([ { "type": "tool_result", "tool_use_id": tool_call.id, "content": result.content[0].text } ]) }) # 打印信息用于了解中间过程 print("messages_2", messages) print("available_tools_2", available_tools) # 再次调用模型处理工具调用返回结果生成最终答案 # 为什么这里不直接返回工具调用的结果作为答案,大模型还需要做什么呢? # 比如用户的提示可能是中文或英文,大模型需要对工具调用结果做出翻译之后再返回匹配的结果 response = self.openai_client.chat.completions.create( model="qwen-vl-max-latest", messages=messages, tools=available_tools, temperature=0.1, tool_choice="auto", ) final_response = response.choices[0].message print("final_response", final_response) if final_response.content: final_text.append(final_response.content) return "\n".join(final_text) async def chat_loop(self): """运行一个交互式的对话循环""" print("\nMCP Client Started!") print("输入你的问题或者输入'quit'退出.") while True: try: query = input("\n提问: ").strip() if query.lower() == 'quit': break response = await self.process_query(query) print("\n" + response) except Exception as e: print(f"\nError: {str(e)}") async def cleanup(self): """清理资源""" await self.exit_stack.aclose() async def main(): if len(sys.argv) < 2: print("使用: python client.py <path_to_server_script>") sys.exit(1) client = MCPClient() try: await client.connect_to_server(sys.argv[1]) await client.chat_loop() finally: await client.cleanup()if __name__ == "__main__": import sys asyncio.run(main())测试

完成了客户端的开发,接下来就可以进行测试了,cd到项目根目录运行python client.py .\weather.py,结果显示输入了如下四个问题的返回信息,通过分析其中的日志信息基本上就可以大致了解大模型做了什么。

纽约天气如何?查询上海天气预警信息what's weather in New York?what's the weather alert in Beijing?

使用Trae测试你开发的MCP Server

恭喜完成了一个全流程的mcp开发过程,虽然它非常简单,但是此时相信你应该对mcp有了一个比较清晰的理解了吧。接下来,也许你还有兴趣试试在AI IDE能不能调用我们开发的这个超简单mcp server,这里IDE就类似前面开发的client,我们就简单演示下如何使用字节Trae来调用我们开发的这个天气预报的mcp工具。

安装并启动trae,右上角配置选择MCP

选择手动添加MCP Servers

这一步的关键是MCP server的配置,前面介绍过执行python weather.py就会启动mcp server,这里的python就是对应虚拟环境下的python.exe的全路径,args就是weather.py的全路径,所谓的mcp server配置其实就是包含了名称、启动命令和启动参数的JSON串而已。

创建成功之后可以显示提供了两个工具get_alerts和get_forecast

在对话框中选择Builder with MCP

由于Trae使用的大模型本身就能查询天气信息,因此如果直接问纽约天气如何它会直接给出答案,重新提问要求大模型调用工具之后,返回答案中发现确实调用了get_alerts方法。

参考文献

[1] MCP官方文档https://modelcontextprotocol.io/quickstart/server