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

MCP（Model Context Protocol）

java 2025/5/7 2:28:24

一、MCP概念

MCP（Model Context Protocol，模型上下文协议）是由 Anthropic 提出并于 2024 年 11 月开源的一种通信协议， 其实，从本质上来讲，MCP 的灵感部分来源于 USB-C 的类比：如同 USB-C 通过统一接口连接多种设备，MCP 旨在为 AI 应用提供一个“即插即用”的上下文管理框架。 更为准确而言，MCP 的核心思想是将模型与外部系统之间的通信抽象为一个客户端-服务器架构，通过标准化的接口，实现上下文的动态传递和工具的灵活调用。

二、主要功能

  • 上下文共享：应用程序可以通过 MCP 向模型提供所需的上下文信息（如文件内容、数据库记录等），增强模型的理解能力。
  • 工具暴露：MCP 允许应用程序将功能（如文件读写、API 调用）暴露给模型，模型可以调用这些工具完成复杂任务。
  • 可组合的工作流：开发者可以利用 MCP 集成多个服务和组件，构建灵活、可扩展的 AI 工作流。
  • 安全性：通过本地服务器运行，MCP 避免将敏感数据上传至第三方平台，确保数据隐私。

三、MCP架构

MCP 采用客户端-服务器架构：

  • MCP 客户端（Client）：通常是 AI 应用程序（如 Claude Desktop 或其他 LLM 工具），负责发起请求并与服务器通信。
  • MCP 服务器（Server）：轻量级程序，负责暴露特定的数据源或工具功能，并通过标准化协议与客户端交互。

通信格式：基于 JSON-RPC 2.0，支持请求、响应和通知三种消息类型，确保通信的标准化和一致性。

四、疑问：

简单了解完这个MCP之后，可能会有点懵，并不是知道有啥用，我是有这个疑问的，我的问题是和Prompt助手有啥区别，和Agent智能体有啥区别？那我们自定义一个Prompt助手。比如我要查询浙江嘉兴的天气，然后调用大模型，大模型内部是可以联网搜索的。也可以进行输出。那我MCP调用天气接口，也可以输出，这不是多余？大模型还有一个Agent 智能体，它的功能更厉害。MCP这不就是多余？带着问题查阅资料。还是了解的不深入。三者的概念如下：

1、Prompt 助手：

  • 这是最简单的一个概念，就是你给大模型（比如 ChatGPT）输入一个 提示词（Prompt），让它根据这个提示词来生成答案。
  • 你直接输入的 一句话或几句提示词 就是大模型的指令，比如：“你是一个电商客服助手，请回答客户的退货问题”。
  • 功能：这个助手只能 根据给定的提示词生成内容，是最简单的模型交互方式。

2、MCP（Model Control Platform）：

  • MCP 是 一个平台，它不仅仅帮助你管理和调用大模型，还能对 多个大模型的行为进行调度、控制和优化。
  • 它的作用是 帮助你配置和控制大模型的执行，比如设置提示词、控制上下文、调用多个模型或工具，甚至在需要时管理 不同任务的工作流程。
  • 功能：它提供了一个集中管理的框架，使得大模型的调用更加灵活、自动化。

3、Agent（智能体）：

  • Agent 是一个 更高级的智能体，它不仅仅是调用大模型来生成内容，而是具备 自我决策、任务规划、执行 等能力。
  • Agent 可以理解和处理复杂任务，感知环境、执行任务、与外界交互，甚至能 主动决策。它不再是被动回答问题，而是 根据情况做出下一步的决策并行动。
  • 例如，一个智能客服 Agent，不仅能回答问题，还能 自动分析用户需求、生成任务（如退货申请），并根据任务执行进一步操作（如查询订单、更新系统状态）。

Prompt 助手和MCP

我都可以自己写 Prompt，接个大模型 API，比如 DeepSeek甚至加上工具调用，干嘛还需要 MCP？

你说的没错：你完全可以不用 MCP！

比如你直接写代码这样用 👇：

prompt = f"""
你是一个天气助手，帮我查询天气。
请调用这个天气 API：https://xxx.com/weather?city=北京
并告诉我今天天气如何。
"""
response = 调用_deepseek(prompt)

这样做确实 简单直接，自己控制逻辑，灵活自由。这个方式我们称为：

「手动构造 Prompt + 手动调模型 + 可选接 API 工具」

适合轻量简单的项目，开发者对流程有掌控。

当你一旦碰到下面这些 “麻烦事儿”，你就知道 MCP 的价值了：

1. 提示词 & 工具调用的统一管理（核心）

你可能有几十个、上百个 Prompt，甚至每个 Prompt 要配套绑定：

  • 用哪个模型（DeepSeek？GPT-4？）
  • 调用哪个工具（天气API？数据库查询？）
  • 输出什么格式？
  • 多轮对话还是单轮？

MCP 就像是一个「大脑 + 控制面板」，统一配置、管理、更新所有 Prompt 和模型行为，你不用手动维护这些逻辑。

2. 工具链统一调用

你现在可能只用天气 API，但很快可能会变成：

  • 天气 API
  • 订单系统 API
  • 数据库查询
  • PDF 文件读取
  • URL 网页爬取

这些“工具”的调用要规范格式、统一输出，还要告诉模型“怎么用”，一堆麻烦逻辑。

MCP 可以自动为大模型注入工具描述，帮你管理工具接入、调用流程。

Agent和MCP

先来个超简版回答：

Agent 是“大脑 + 行为逻辑”，而 MCP 是“大脑调用模型和工具的遥控器”。

它们不是互相替代，而是配合使用的。

打个超通俗的比喻：

Agent 就像是「项目经理」：
  • 它能规划任务、分析问题、安排流程（例如：处理退货请求 → 查询订单状态 → 判断规则 → 生成处理意见）
MCP 就像是「员工管理系统 + 工具房」：
  • 项目经理（Agent）不会自己干活，但它需要：
    • 某个工具员去查天气（MCP 控制天气 API）
    • 某个翻译员处理英文文档（MCP 分配任务给 DeepSeek）
    • 某个统计员生成图表（MCP 让 GPT-4 + 工具链处理）

所以 —— Agent 是 “决策大脑”， MCP 是 “执行调度器”，两者搭配才能把任务干漂亮！

五、MCP工作流程图：

在这里插入图片描述

六、代码：

客户端：

import asyncio
import os
import json
from typing import Optional
from contextlib import AsyncExitStackfrom openai import OpenAI  
from dotenv import load_dotenvfrom mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client# 加载 .env 文件，确保 API Key 受到保护
load_dotenv()class MCPClient:def __init__(self):"""初始化 MCP 客户端"""self.exit_stack = AsyncExitStack()self.openai_api_key = os.getenv("OPENAI_API_KEY")  # 读取 OpenAI API Keyself.base_url = os.getenv("BASE_URL")  # 读取 BASE YRLself.model = os.getenv("MODEL")  # 读取 modelif not self.openai_api_key:raise ValueError(" 未找到 OpenAI API Key，请在 .env 文件中设置 OPENAI_API_KEY")self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url) # 创建OpenAI clientself.session: Optional[ClientSession] = Noneself.exit_stack = AsyncExitStack()        async def connect_to_server(self, server_script_path: str):"""连接到 MCP 服务器并列出可用工具"""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"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_transportself.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))await self.session.initialize()# 列出 MCP 服务器上的工具。这里会拿到mcp server所有的工具response = await self.session.list_tools()tools = response.toolsprint("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])     async def process_query(self, query: str) -> str:"""使用大模型处理查询并调用可用的 MCP 工具 (Function Calling)"""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.nametool_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.contentreturn content.message.contentasync def chat_loop(self):"""运行交互式聊天循环"""print("\n MCP 客户端已启动！输入 'quit' 退出")while True:try:query = input("\n你: ").strip()if query.lower() == 'quit':breakresponse = await self.process_query(query)  # 发送用户输入到 OpenAI APIprint(f"\n OpenAI: {response}")except Exception as e:print(f"\n⚠️ 发生错误: {str(e)}")async def cleanup(self):"""清理资源"""await self.exit_stack.aclose()async def main():if len(sys.argv) < 2:print("Usage: python client.py <path_to_server_script>")sys.exit(1)print(sys.argv[0])print(sys.argv[1])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())

服务端：

import json
import httpx
from typing import Any, Optional
from typing import Union
from mcp.server.fastmcp import FastMCP# 初始化 MCP 服务器
mcp = FastMCP("WeatherServer")# OpenWeather API 配置
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
API_KEY = "xxxxxxxxxxxx"  # OpenWeather的API Key
USER_AGENT = "weather-app/1.0"async def fetch_weather(city: str) -> Optional[dict[str, Any]]:"""从 OpenWeather API 获取天气信息。:param city: 城市名称（需使用英文，如 Beijing）:return: 天气数据字典；若出错返回包含 error 信息的字典"""params = {"q": city,"appid": API_KEY,"units": "metric","lang": "zh_cn"}headers = {"User-Agent": USER_AGENT}async with httpx.AsyncClient() as client:try:response = await client.get(OPENWEATHER_API_BASE, params=params, headers=headers, timeout=30.0)response.raise_for_status()return response.json()  # 返回字典类型except httpx.HTTPStatusError as e:return {"error": f"HTTP 错误: {e.response.status_code}"}except Exception as e:return {"error": f"请求失败: {str(e)}"}def format_weather(data: Union[dict[str, Any], str]) -> str:"""将天气数据格式化为易读文本。:param data: 天气数据（可以是字典或 JSON 字符串）:return: 格式化后的天气信息字符串"""# 如果传入的是字符串，则先转换为字典if isinstance(data, str):try:data = json.loads(data)except Exception as e:return f"无法解析天气数据: {e}"# 如果数据中包含错误信息，直接返回错误提示if "error" in data:return f"⚠️ {data['error']}"city = data.get("name", "未知")country = data.get("sys", {}).get("country", "未知")temp = data.get("main", {}).get("temp", "N/A")humidity = data.get("main", {}).get("humidity", "N/A")wind_speed = data.get("wind", {}).get("speed", "N/A")# weather 可能为空列表，因此用 [0] 前先提供默认字典weather_list = data.get("weather", [{}])description = weather_list[0].get("description", "未知")return (f" {city}, {country}\n"f" 温度: {temp}°C\n"f" 湿度: {humidity}%\n"f" 风速: {wind_speed} m/s\n"f" 天气: {description}\n")@mcp.tool()
async def query_weather(city: str) -> str:"""输入指定城市的英文名称，返回今日天气查询结果。:param city: 城市名称（需使用英文）:return: 格式化后的天气信息"""data = await fetch_weather(city)return format_weather(data)if __name__ == "__main__":mcp.run(transport='stdio')

整体代码流程：

启动服务端以及客户端。

client.py 启动这个 MCP 服务，并初始化与它的连接。

用户输入：

北京天气如何

OpenAI 模型识别出你可能需要调用工具 query_weather，提取参数 city="Beijing"，并自动调用 MCP 服务器上的这个函数。

MCP 服务器调用 OpenWeather API，获取结果后格式化输出。

大模型收到工具返回的结果后，再做一次生成，最终返回用户：

效果如下：

 OpenAI: 目前北京的天气情况如下：- **温度**：21.94°C，属于较为舒适的范围，适合户外活动。
- **湿度**：10%，湿度较低，空气可能比较干燥，建议多喝水并注意保湿。
- **风速**：4.2 m/s，风速适中，不会让人感到过于寒冷或不适。
- **天气状况**：晴天，阳光明媚，适合出行和户外运动。总体来说，今天的天气非常宜人，非常适合外出游玩或进行其他户外活动。不过由于湿度较低，记得随身携带水，并做好皮肤保湿措施。

在这里插入图片描述

http://www.xdnf.cn/news/1967.html

相关文章：

  • AlarmClock4.8.4(官方版)桌面时钟工具软件下载安装教程
  • Zephyr kernel Build System (CMake)介绍
  • MySQL引擎分类与选择、SQL更新底层实现、分库分表、读写分离、主从复制 - 面试实战
  • 数字浪潮下的算力担当：GPU 服务器的多元应用、核心价值
  • 技术探索之路：从自我认知到成长规划
  • 实现层归一化
  • 数据结构------C语言经典题目（7）
  • 【T-MRMSM】文本引导多层次交互多尺度空间记忆融合多模态情感分析
  • 基于cesium实现鼠标移动动态绘制矩形和圆
  • Rust 学习笔记：函数和控制流
  • React 中什么时候用事件总线
  • 微信小程序直传阿里云 OSS 实践指南（V4 签名 · 秒传支持 · 高性能封装）
  • ROS1、ROS2如何把预编译好的二进制文件封装成功能包？
  • 【Django】新增字段后兼容旧接口 This field is required
  • 代码随想录:数组
  • 如何实现Android屏幕和音频采集并启动RTSP服务？
  • 如何使用@KafkaListener实现从nacos中动态获取监听的topic
  • 【Hive入门】Hive数据导出完全指南：从HDFS到本地文件系统的专业实践
  • 利用JMeter代理服务器方式实现高效压测
  • Leetcode 2845 题解
  • C++_数据结构_详解红黑树
  • 微软官网Win10镜像下载快速获取ISO文件
  • 第18章：MCP在创作领域中的应用
  • Java集成Redisson实现分布式锁（实战）
  • 学生管理系统微服务方式实现
  • WebUI可视化：第3章：Gradio入门实战
  • FerretDB：基于PostgreSQL的MongoDB替代产品
  • 2、Ubuntu 环境下安装RabbitMQ
  • PDFMathTranslate：基于LLM的PDF文档翻译及双语对照的工具【使用教程】
  • Golang | 迭代器模式