实战解析MCP-使用本地的Qwen-2.5模型-AI协议的未来？

文章目录

目录

文章目录

前言

一、MCP是什么？

1.1MCP定义

1.2工作原理

二、为什么要MCP？

2.1 打破碎片化的困局

2.2 实时双向通信，提升交互效率

2.3 提高安全性与数据隐私保护

三、MCP 与 LangChain 的区别

3.1 目标定位不同

3.2 实现方式的差异

3.3 使用场景的侧重点

四、MCP 的未来发展前景

4.1 行业内外的热烈讨论

4.2 开放生态与标准共识

4.3 跨平台与多模式部署

五、实战

5.1 服务端

5.2 客户端

总结

---

前言

近年来，随着大语言模型（LLM）在各类应用中的广泛使用，我们逐渐意识到：仅靠单一模型的能力，很难满足实际应用中对数据、工具、环境等多样化需求的不断增长。就在这种背景下，Anthropic 推出的模型上下文协议（Model Context Protocol，简称 MCP）悄然登场，它被誉为“为 AI 装上 USB-C 接口”的革命性标准，为 AI 工具整合带来了全新的思路。本文将深入探讨 MCP 是什么、为什么要使用 MCP，以及 MCP 与 LangChain 等其他技术的核心区别和应用前景。最后，我会用本地的qwen模型实战完成MCP的应用。

---

一、MCP是什么？

1.1MCP定义

模型上下文协议（MCP）是一种开放标准协议，旨在为大语言模型与外部工具、数据源之间建立统一、标准化的通信接口。简单来说，MCP 就像一个“万能适配器”，只需一次整合，就能让 AI 模型（例如 Anthropic 的 Claude）连接上各式各样的数据接口与工具，而不必为每个数据源单独开发对接代码。这种设计理念不仅大大降低了开发难度，还为不同平台间的互操作性奠定了基础。

“MCP 通过统一的通信协议，让模型能够与外部数据源和工具实现无缝对接，就像 USB-C 接口让各种设备共享充电和数据传输功能一样”。

1.2工作原理

MCP 的核心架构基于客户端—服务器模式，主要由三个部分组成：

• MCP 主机（Host）：一般为 AI 应用程序或桌面端工具，例如 Claude 桌面版、IDE 插件等，它负责发起请求。
• MCP 客户端（Client）：集成在主机内部，通过标准化协议与服务端建立稳定连接，发送请求并接收响应。
• MCP 服务器（Server）：负责对外提供具体的数据、工具或提示信息。它可以连接到本地资源（如文件、数据库）或远程服务（如第三方 API）。

这种设计保证了数据的动态传输和实时交互，支持双向通信，使得 AI 不再是单向的信息接收者，而可以主动触发操作和获取实时反馈。

二、为什么要MCP？

2.1 打破碎片化的困局

在传统的开发过程中，AI 应用与每个外部工具或数据源的对接往往都是孤立的。每个 API 都有不同的认证方式、数据格式和错误处理机制，这不仅增加了开发者的负担，也导致系统整体的集成性和扩展性大打折扣。MCP 则通过一次标准化的整合，解决了这一“每扇门都有一把不同钥匙”的问题。通过 MCP，开发者可以将各种工具和数据源统一接入，大幅提升开发效率。

2.2 实时双向通信，提升交互效率

传统 API 往往采用单向的请求—响应模式，模型仅能被动等待数据返回。而 MCP 支持双向实时通信，这种机制不仅使得数据查询更加迅速，还允许模型主动触发操作。例如，在需要实时获取天气信息或查询本地文件内容的场景中，MCP 可以让 AI 模型通过与外部工具的双向对话，获得更准确的上下文数据，从而生成更贴切的回答

2.3 提高安全性与数据隐私保护

在许多企业级应用场景中，数据隐私和安全性是首要考量。传统做法中，数据往往需要上传到云端进行处理，这在安全性和隐私保护上存在隐患。而 MCP 的设计允许数据在本地或企业内部网络中流转，避免将敏感信息暴露到公共云端。同时，通过统一的协议标准，MCP 可以在不同工具间实施统一的安全策略，确保各方访问权限受控。

三、MCP 与 LangChain 的区别

近年来，LangChain 作为一款开源框架，也在大语言模型整合工具方面受到广泛关注。那么，MCP 与 LangChain 到底有何区别？

3.1 目标定位不同

• MCP：作为一个开放的标准协议，MCP 侧重于提供一种统一的通信接口，使 AI 模型能够通过一次整合接入成千上万的外部数据源和工具。其核心在于标准化、动态发现和双向通信，让开发者可以构建灵活、安全且高效的 AI Agent。
• LangChain：则更像是一个上层应用框架，它为开发者提供了大量现成的工具和模块，帮助他们快速构建 AI Agent。LangChain 的优势在于成熟的生态和丰富的示例，但在面对不同平台和服务时，其接入方式可能仍然存在一定的碎片化问题。

3.2 实现方式的差异

• MCP：要求服务端和客户端按照统一的 JSON-RPC 或 SSE 等标准协议进行通信，实现上相对底层和标准化。它更注重与底层系统的整合，强调的是“写一次，接入万次”的理念。
• LangChain：则更偏向于为开发者提供高层次的抽象和即插即用的组件，其实现方式可能因平台不同而略有差异，需要开发者在具体场景中进行适配和扩展。

3.3 使用场景的侧重点

• MCP：适用于那些对数据安全、实时交互和统一接口要求较高的场景，如企业内部系统集成、敏感数据处理以及需要跨平台动态调用工具的应用。
• LangChain：则更适合快速开发原型和构建简单 AI Agent，在开发者社区中已积累了丰富的案例和资源，但在面对大规模、复杂系统时，可能需要额外整合措施来弥补标准化不足的问题。

---

四、MCP 的未来发展前景

4.1 行业内外的热烈讨论

自 MCP 问世以来，各大厂商和开发者社区对其前景展开了激烈讨论。LangChain 的大佬们甚至就此展开了辩论——一部分人认为 MCP 是未来 AI 工具整合的必由之路，能够大幅降低开发者成本并实现真正的“AI原生”体验；另一部分则持怀疑态度，认为目前的标准还不够成熟，仍有许多细节需要打磨。

4.2 开放生态与标准共识

MCP 的最大亮点在于其开放性和标准化。通过建立统一的协议，不仅能让单个厂商如 Anthropic 推动自己的生态，还能吸引更多公司和开源社区参与进来。正如一些业内专家所说，“MCP 可能成为未来 AI 工具整合的通用标准”，这种共识如果达成，将极大地促进 AI 生态系统的健康发展。

4.3 跨平台与多模式部署

未来，MCP 不仅可以在本地和企业内部网络中运行，还能通过 WebSocket、HTTP 等网络协议实现远程部署。这意味着，AI 模型无论是在云端还是边缘设备，都能通过 MCP 统一对接外部工具和数据，实现无缝协作和实时交互。这种跨平台的灵活性，将是 MCP 在实际应用中大放异彩的重要因素。

五、实战

5.1 服务端

from mcp.server.fastmcp import FastMCPmcp = FastMCP("FileWriter")@mcp.tool()
def write_to_txt(filename: str, content: str) -> str:"""将指定内容写入文本文件并且保存到本地。参数:filename: 文件名（例如 "output.txt"）content: 要写入的文本内容返回:写入成功或失败的提示信息"""try:with open(filename, "w", encoding="utf-8") as f:f.write(content)return f"成功写入文件 {filename}。"except Exception as e:return f"写入文件失败：{e}"if __name__ == "__main__":mcp.run(transport='stdio')  # 默认使用 stdio 传输

5.2 客户端

import os
import asyncio
from typing import Optional
from contextlib import AsyncExitStack
import json
import tracebackfrom mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_clientfrom openai import OpenAI
from dotenv import load_dotenvload_dotenv()  # 加载环境变量从 .envclass MCPClient:def __init__(self):# 初始化会话和客户端对象self.session: Optional[ClientSession] = None # 会话对象self.exit_stack = AsyncExitStack() # 退出堆栈self.openai = OpenAI(api_key="EMPTY", base_url="") self.model="qwen-2.5-14b"def get_response(self, messages: list,tools: list):response = self.openai.chat.completions.create(model=self.model,max_tokens=1000,messages=messages,tools=tools,)return responseasync def get_tools(self):# 列出可用工具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]return available_toolsasync def connect_to_server(self, server_script_path: str):"""连接到 MCP 服务器参数:server_script_path: 服务器脚本路径 (.py 或 .js)"""is_python = server_script_path.endswith('.py')is_js = server_script_path.endswith('.js')if not (is_python or is_js):raise ValueError("服务器脚本必须是 .py 或 .js 文件")command = "python" if is_python else "node"# 创建 StdioServerParameters 对象server_params = StdioServerParameters(command=command,args=[server_script_path],env=None)# 使用 stdio_client 创建与服务器的 stdio 传输stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))# 解包 stdio_transport，获取读取和写入句柄self.stdio, self.write = stdio_transport# 创建 ClientSession 对象，用于与服务器通信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("
连接到服务器，工具列表:", [tool.name for tool in tools])async def process_query(self, query: str) -> str:"""使用 OpenAI 和可用工具处理查询"""# 创建消息列表messages = [{"role": "user","content": query}]# 列出可用工具available_tools = await self.get_tools()print(f"
可用工具: {json.dumps([t['function']['name'] for t in available_tools], ensure_ascii=False)}")# 处理消息response = self.get_response(messages, available_tools)# 处理LLM响应和工具调用tool_results = []final_text = []for choice in response.choices:message = choice.messageis_function_call = message.tool_calls# 如果不调用工具，则添加到 final_text 中if not is_function_call:final_text.append(message.content)# 如果是工具调用，则获取工具名称和输入else:#解包tool_callstool_name = message.tool_calls[0].function.nametool_args = json.loads(message.tool_calls[0].function.arguments)print(f"准备调用工具: {tool_name}")print(f"参数: {json.dumps(tool_args, ensure_ascii=False, indent=2)}")try:# 执行工具调用，获取结果result = await self.session.call_tool(tool_name, tool_args)print(f"
工具调用返回结果类型: {type(result)}")print(f"工具调用返回结果属性: {dir(result)}")print(f"工具调用content类型: {type(result.content) if hasattr(result, 'content') else '无content属性'}")# 安全处理contentcontent = Noneif hasattr(result, 'content'):if isinstance(result.content, list):content = "
".join(str(item) for item in result.content)print(f"将列表转换为字符串: {content}")else:content = str(result.content)print(f"工具调用content值: {content}")else:content = str(result)print(f"使用完整result作为字符串: {content}")tool_results.append({"call": tool_name, "result": content})final_text.append(f"[调用工具 {tool_name} 参数: {json.dumps(tool_args, ensure_ascii=False)}]")# 继续与工具结果进行对话if message.content and hasattr(message.content, 'text'):messages.append({"role": "assistant","content": message.content})# 将工具调用结果添加到消息messages.append({"role": "user", "content": content})# 获取下一个LLM响应print("获取下一个LLM响应...")response = self.get_response(messages, available_tools)# 将结果添加到 final_textcontent = response.choices[0].message.content or ""final_text.append(content)except Exception as e:print(f"
工具调用异常: {str(e)}")print(f"异常详情: {traceback.format_exc()}")final_text.append(f"工具调用失败: {str(e)}")return "
".join(final_text)async def chat_loop(self):"""运行交互式聊天循环（没有记忆）"""print("
MCP Client 启动!")print("输入您的查询或 'quit' 退出.")while True:try:query = input("
Query: ").strip()if query.lower() == 'quit':breakprint("
处理查询中...")response = await self.process_query(query)print("
" + response)except Exception as e:print(f"
错误: {str(e)}")print(f"错误详情: {traceback.format_exc()}")async def cleanup(self):"""清理资源"""await self.exit_stack.aclose() async def main():"""主函数：初始化并运行 MCP 客户端此函数执行以下步骤：1. 检查命令行参数是否包含服务器脚本路径2. 创建 MCPClient 实例3. 连接到指定的服务器4. 运行交互式聊天循环5. 在结束时清理资源用法：python client.py <path_to_server_script>"""# 检查命令行参数if len(sys.argv) < 2:print("Usage: python client.py <path_to_server_script>")sys.exit(1)# 创建 MCPClient 实例client = MCPClient()try:# 连接到服务器await client.connect_to_server(sys.argv[1])# 运行聊天循环await client.chat_loop()finally:# 确保在任何情况下都清理资源await client.cleanup()if __name__ == "__main__":import sysasyncio.run(main())

输出指令：写一首诗并且保存到本地。成功完成任务。

---

总结

随着 AI 技术的不断进步和应用场景的日益复杂，单一大语言模型的能力已无法满足实际需求。MCP 作为一种全新的开放标准协议，通过提供统一、标准化、双向实时的接口，为 AI 模型整合外部工具和数据源提供了革命性的解决方案。从打破碎片化困局、提升数据安全、到实现跨平台部署，MCP 的出现无疑将推动 AI 应用向更高层次发展。

虽然目前行业内对 MCP 的看法仍存在分歧——有支持者认为它将引领未来，有质疑者将其视为短暂的热潮——但不可否认的是，MCP 已经为开发者提供了一种全新的工具接入思路，其标准化和开放性特质，必将在未来的 AI 生态中发挥越来越重要的作用。

无论你是偏向于使用 MCP 构建企业级解决方案，还是更喜欢依赖 LangChain 快速开发原型，都可以看到这一领域正在经历一场深刻的变革。正如业内专家所言，“未来的 AI 应用生态，很可能就是在这两种思路的碰撞与融合中诞生的。”

希望这篇文章能够为你提供对 MCP 及其在 AI 领域中作用的全面认识。如果你有更多想法或疑问，欢迎在评论区交流探讨！