开源模子应用落地-模子上下文协议(MCP)-从数据孤岛到万物互联(一)
一、前言当开发者还在为每个AI工具编写痴肥的API适配器时,一场关于「毗连」的技术革命已悄然降临。模子上下文协议(MCP)正在用一套全新的交互语法,重新界说人工智能与物理天下的对话方式。MCP协议如同为AI系统装上了“万能接口”,通过标准化的客户端-服务器架构,将大语言模子(LLM)与数据库、API乃至物理装备无缝毗连。它不仅是数据孤岛的破壁者——让及时更新的企业文档、代码仓库直接成为模子的“记忆延伸”;更是行动力的赋能器——从自动生成销售报告到远程操控IoT装备,AI得以突破捏造界限,真正成为生产力工具。这种“知行合一”的特性,正在重新界说人机协作的深度与广度。
二、术语先容
2.1.MCP
MCP协议(Model Context Protocol,模子上下文协议)是由Anthropic公司于2024年11月推出的开放标准协议,旨在为大型语言模子(LLM)与外部数据源、工具之间创建同一的通信接口。其核心作用在于解决AI模子因数据孤岛限制难以发挥潜力的问题,通过标准化交互方式提升AI应用的扩展性和实用性。
一、MCP协议的核心界说
1.1. 协议定位
MCP被类比为AI领域的“USB-C接口”,通过解耦LLM与外部系统的集成,实现一次开发即可动态毗连多种服务(如数据库、API、当地文件系统等)。例如,开发者无需为每个新工具单独定制接口,只需遵循MCP标准即可实现AI与工具的即插即用。
https://i-blog.csdnimg.cn/direct/13216a4fedc04fb9abc19f2e5a1812bf.png
1.2. 技术架构
采用客户端-服务器模子,包罗三个核心组件:
[*]MCP主机:运行AI模子的环境(如Claude桌面应用);
[*]MCP客户端:负责与服务器通信并转发请求;
[*]MCP服务器:提供详细功能服务的轻量级步调(如文件管理或API调用)。
二、MCP协议的核心作用
2.1. 同一交互标准
标准化JSON-RPC 2.0协议支持当地(标准输入输出)与远程(HTTP/SSE)通信,兼容Python、Java等多种编程语言,低落开发复杂度。例如,当地通信保障医疗数据安全传输,远程通信支持云端服务扩展。
2.2. 动态扩展本领
通过资源(Resource)、工具(Tool)、提示(Prompt)三大核心概念,AI模子可自动发现新工具并及时交互。例如毗连GitHub时,AI可自动提交代码或修复BUG,无需人工预界说接口。
2.3. 安全与权限控制
内置权限验证机制,用户需手动批准敏感操纵(如文件删除或邮件发送),确保数据隐私。在企业场景中,该特性尤为紧张,如金融分析时需严格管控市场数据访问权限。
https://i-blog.csdnimg.cn/direct/e9f164adec02471d9a76108079550fd8.png
2.2.STDIO vs SSE
STDIO和SSE是两种核心的传输机制,分别实用于差别的通信场景
(一)STDIO(标准输入输出)
界说与实现:
STDIO模式通过标准输入(stdin)和标准输出(stdout)实现当地历程间通信。客户端将MCP服务端作为子历程启动,两边通过JSON-RPC消息进行交互:
[*]客户端通过stdin发送请求,服务端通过stdout返回相应。
[*]消息以换行符分隔,且需严格遵守JSON-RPC 2.0格式。
作用与特点:
[*]高效低延迟:实用于当地开发或单一呆板环境,避免了网络传输开销。
[*]简单易用:无需网络配置,得当快速集成命令行工具或调试。
[*]局限性:仅支持当地通信,无法扩展至分布式场景。
应用场景:
[*]当地AI助手与工具集成(如天气查询服务)。
[*]开发阶段的快速原型验证。
(二)SSE(Server-Sent Events,服务端推送变乱)
界说与实现
SSE模式基于HTTP协议,服务端通过单向流式通信自动推送数据至客户端:
[*]服务端需提供两个端点:SSE端点(用于客户端毗连)和HTTP POST端点(用于客户端发送消息)。
[*]消息以SSE变乱流形式传输,客户端通过长毗连吸收及时更新。
作用与特点
[*]及时性:支持服务端自动推送(如股票行情、及时日志),实用于需持续更新的场景。
[*]分布式支持:可跨物理呆板部署,允许多客户端并发毗连。
[*]协议限制:仅服务端到客户端单向通信,双向交互需结合HTTP POST请求。
应用场景:
[*]远程AI代理系统(如跨区域部署的智能客服)。
[*]必要及时数据流的应用(如IoT装备监控)。
对比与选择发起
https://i-blog.csdnimg.cn/direct/db961255a2ec4542a285aa52519a5b8e.png
2.3.UV
是由Rust编写的高性能包管理工具,主要用于项目的依赖管理和打包。
UV 作为 Python 生态中的新一代包管理工具,相比传统的 pip 具有显著优势,主要表现在以下几个方面:
一、极速性能(10-100 倍提升)
UV 的核心竞争力在于其 Rust 语言实现的高性能引擎。以下为关键性能对比:
[*]依赖安装:安装大型项目(如 Trio)时,UV 耗时约 0.06 秒,而 pip 必要 1.56 秒,速度提升 26 倍。
[*]捏造环境创建:UV 创建捏造环境仅需 4.1 毫秒,而 pip 的 venv 需 1.5 秒,速度提升 365 倍。
[*]依赖剖析:通过 PubGrub 算法优化和并行处理,UV 能快速解决复杂依赖冲突,尤其在处理大规模依赖树时优势明显。
二、功能集成(一站式解决方案)
UV 整合了多个工具链,解决了 pip 需配合其他工具(如 virtualenv、pip-tools、pipx 等)的碎片化问题
https://i-blog.csdnimg.cn/direct/7283b828618046d7b9f7b60f66e1518f.png
三、兼容性与易用性
[*]无缝迁移:支持现有 requirements.txt 和 pyproject.toml,无需修改项目配置即可直接更换 pip。
[*]跨平台支持:兼容 Windows、macOS、Linux,并通过 全局缓存机制 减少重复下载,节省磁盘空间。
[*]开发者友好:错误信息更清晰,包罗依赖冲突的详细跟踪。支持 pip 的高级功能(如可编辑安装、Git 依赖、当地包安装)
三、前置条件
3.1.底子环境及前置条件
1. 操纵系统:无限
3.2.安装UV
以windows系统为例
(一)直接利用pip安装
pip install uv (二)利用脚本安装
irm https://astral.sh/uv/install.ps1 | iex 3.3.初始化及安装依赖
# 初始化项目
uv init mcp_test
# 切换目录
cd mcp_test
# 创建虚拟环境
uv venv -p 3.10
# 激活虚拟环境
.venv\Scripts\activate
# 安装依赖
uv add "mcp" 四、技术实现
4.1.MCP Server实现
# -*-coding:utf-8 -*-
from mcp.server.fastmcp import FastMCP
DEAULT_TRANSPORT = "stdio"
# Create an MCP server
mcp = FastMCP("Demo")
# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
print(f'a: {a},b: {b}')
return a + b
# Add a dynamic greeting resource
@mcp.resource(uri="greeting://",name="greeting")
def get_greeting() -> str:
"""Get a personalized greeting"""
return f"Hello, world!"
@mcp.resource(uri="greeting://{name}",name="greeting")
def get_greeting(name: str) -> str:
"""Get a personalized greeting"""
return f"Hello, {name}!"
@mcp.prompt()
def review_code(code: str) -> str:
return f"Please review this code:\n\n{code}"
if __name__ == "__main__":
# Initialize and run the server
print(f"start mcp server,transport: {DEAULT_TRANSPORT}...")# 记录服务启动日志
# mcp.run(transport='sse')
mcp.run(transport=DEAULT_TRANSPORT)
调用:
直接在pycharm中运行
https://i-blog.csdnimg.cn/direct/1fa527ffc35942b189201f9280f73023.png
4.2.STDIO MCP Client实现
# -*-coding:utf-8 -*-
import sys
import traceback
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
# Create server parameters for stdio connection
server_params = StdioServerParameters(
command="python",
args=["mcp_server.py"]
)
async def run():
async with stdio_client(server_params,errlog=sys.stderr) as (read, write):
async with ClientSession(
read, write
) as session:
# Initialize the connection
await session.initialize()
# List available prompts
prompts = await session.list_prompts()
print(f'prompts: {prompts}')
# Get a prompt
prompt = await session.get_prompt(
"review_code", arguments={"code": "11111"}
)
print(f'prompt: {prompt}')
# List available resources
resources = await session.list_resources()
print(f'resources: {resources}')
resource = await session.read_resource('greeting://test')
print(f'resource: {resource}')
# List available tools
tools = await session.list_tools()
print(f'tools: {tools}')
# Call a tool
result = await session.call_tool("add", arguments={"a": 3, "b": 2})
print(f'result: {result}')
if __name__ == "__main__":
import asyncio
asyncio.run(run()) 调用:
直接在pycharm中运行
https://i-blog.csdnimg.cn/direct/e4a579452bdf47e6ba7ff46d6188beb6.png
4.3.SSE MCP Client实现
# -*-coding:utf-8 -*-
from mcp import ClientSession
from mcp.client.sse import sse_client
async def run():
async with sse_client(url='http://localhost:8000/sse') as (read, write):
async with ClientSession(
read, write
) as session:
# Initialize the connection
await session.initialize()
# List available prompts
prompts = await session.list_prompts()
print(f'prompts: {prompts}')
# Get a prompt
prompt = await session.get_prompt(
"review_code", arguments={"code": "11111"}
)
print(f'prompt: {prompt}')
# List available resources
resources = await session.list_resources()
print(f'resources: {resources}')
resource = await session.read_resource('greeting://test')
print(f'resource: {resource}')
# List available tools
tools = await session.list_tools()
print(f'tools: {tools}')
# Call a tool
result = await session.call_tool("add", arguments={"a": 3,"b": 2})
print(f'result: {result}')
if __name__ == "__main__":
import asyncio
asyncio.run(run()) 调用:
直接在pycharm中运行,必要先修改MCP Server中的transport,调整为sse方式。
https://i-blog.csdnimg.cn/direct/6c6f5e7452eb43b89db519cef7f14ef1.png
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。
页:
[1]