模型上下文协议(MCP)详解
MCP(ModelContextProtocol)是一种标准化协议,用于连接AI模型与外部工具和上下文数据。它采用客户端-服务器架构,通过JSON格式交换数据,支持本地stdio和远程HTTP/SSE两种通信方式。MCP可实现AI助手访问日历、数据库查询、3D设计等多种功能。开发者可使用Python构建MCP服务器,示例展示了如何创建天气查询服务,包括本地和远程两种运行模式。该协议类似于"
MCP(Model Context Protocol)模型上下文协议是一种将模型连接到工具和上下文的标准方法。MCP 提供了一个所有人都可以使用的标准,而不是让每个 AI 应用开发者都创建自己的自定义工具和数据连接方式。Anthropic 将其描述为“AI 应用的 USB-C 接口”。正如USB-C提供了一种连接电子设备的标准化方式一样,MCP也提供了一种将人工智能应用连接到外部系统的标准化方式。

MCP可以实现哪些功能?
-
代理人可以访问您的 Google 日历和 Notion,充当更加个性化的 AI 助手。
-
Claude Code 可以使用 Figma 设计生成整个 Web 应用程序。
-
企业聊天机器人可以连接到组织内的多个数据库,使用户能够通过聊天分析数据。
-
AI模型可以在Blender上创建3D设计,并使用3D打印机将其打印出来。
MCP的工作原理
MCP 采用客户端-服务器架构。换句话说,MCP 客户端向 MCP 服务器发送请求,服务器则响应请求,提供对工具、资源和提示的访问权限。
MCP是AI应用访问工具和上下文的程序。包含三个关键部分:
-
连接——客户端初始化连接,服务器响应,客户端通过初始化通知进行确认。
-
交换——请求、响应和通知以 JSON 格式交换。
-
终止方式有 3 种:客户端关闭、传输断开或发生错误。
MCP客户端
MCP客户端内置于AI应用程序中,主要负责以下事项:
-
发现服务器功能
-
从服务器接收数据
-
管理LLM工具执行
MCP服务器
MCP 服务器是为 AI 应用提供的独立服务。它们监听客户端请求,并根据自身的能力做出响应。
MCP 服务器可以提供三种主要类型的功能:
服务器可以位于本地或者远程,这可以通过两种默认的传输机制来实现:
-
stdio(本地): 服务器作为子进程运行,并通过标准 I/O 流与客户端通信。
-
HTTP 与服务器发送事件 (SSE)(远程) :用于客户端到服务器通信的 HTTP POST 请求或用于服务器到客户端流传输的 SSE
实例:使用 Python 构建 MCP 服务器
系统要求:
-
已安装 Python 3.10 或更高版本。
-
您必须使用 Python MCP SDK 1.2.0 或更高版本。
首先,安装uv并设置 Python 项目和环境(之后请务必重启终端,以确保uv命令生效):
# macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh
# Windows
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
执行curl -LsSf https://astral.sh/uv/install.sh | sh之后,会有一下提示,表示安装成功。

默认安装到了/home/test/.local/bin(以我自己环境为例)目录了,并且环境变量已经写入到了/home/test/.bashrc,可以执行source /home/test/.bashrc立即生效

# 环境变量生效
source /home/test/.bashrc
创建并设置项目,本示例以linux系统为例,Windows详见官网:https://modelcontextprotocol.io/docs/develop/build-server
# 创建一个新的项目目录
uv init weather
cd weather
# 创建虚拟环境
uv venv
source .venv/bin/activate
# 安装依赖
uv add "mcp[cli]" httpx
# 创建服务器文件
touch weather.py
示例来源于官网(增加了远程):
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
# Initialize FastMCP server
mcp = FastMCP("weather")
# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"
# 辅助函数,用于查询和格式化 API 的数据
async def make_nws_request(url: str) -> dict[str, Any] | None:
"""Make a request to the NWS API with proper error handling."""
headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers, timeout=30.0)
response.raise_for_status()
return response.json()
except Exception:
return None
def format_alert(feature: dict) -> str:
"""Format an alert feature into a readable string."""
props = feature["properties"]
return f"""
Event: {props.get("event", "Unknown")}
Area: {props.get("areaDesc", "Unknown")}
Severity: {props.get("severity", "Unknown")}
Description: {props.get("description", "No description available")}
Instructions: {props.get("instruction", "No specific instructions provided")}
"""
##########################################TOOL
# 工具执行处理器负责实际执行每个工具的逻辑
@mcp.tool()
async def get_alerts(state: str) -> str:
"""Get weather alerts for a US state.
Args:
state: Two-letter US state code (e.g. CA, NY)
"""
url = f"{NWS_API_BASE}/alerts/active/area/{state}"
data = await make_nws_request(url)
if not data or "features" not in data:
return "Unable to fetch alerts or no alerts found."
if not data["features"]:
return "No active alerts for this state."
alerts = [format_alert(feature) for feature in data["features"]]
return "\n---\n".join(alerts)
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
"""Get weather forecast for a location.
Args:
latitude: Latitude of the location
longitude: Longitude of the location
"""
# First get the forecast grid endpoint
points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
points_data = await make_nws_request(points_url)
if not points_data:
return "Unable to fetch forecast data for this location."
# Get the forecast URL from the points response
forecast_url = points_data["properties"]["forecast"]
forecast_data = await make_nws_request(forecast_url)
if not forecast_data:
return "Unable to fetch detailed forecast."
# Format the periods into a readable forecast
periods = forecast_data["properties"]["periods"]
forecasts = []
for period in periods[:5]: # Only show next 5 periods
forecast = f"""
{period["name"]}:
Temperature: {period["temperature"]}°{period["temperatureUnit"]}
Wind: {period["windSpeed"]} {period["windDirection"]}
Forecast: {period["detailedForecast"]}
"""
forecasts.append(forecast)
return "\n---\n".join(forecasts)
# 本地MCP服务
def local_main():
# Initialize and run the server
mcp.run(transport="stdio")
# 基于HTTP的MCP服务
app = mcp.sse_app
def remote_main():
uvicorn.run(app, host="0.0.0.0", port=8001)
if __name__ == "__main__":
#local_main() # 本地MCP服务
remote_main() # 远程MCP服务
运行:
# 运行MCP示例服务(本地)
uv run mcp dev example.py
# 运行基于HTTP的MCP服务,就会监听到指定的ip和端口
python example.py
如果运行报以下错,核心问题是系统找不到 npx 命令(npx 是 Node.js 自带的一个工具,随 npm 一起安装),这说明你的环境中没有正确安装 Node.js 和 npm,或者它们没有被添加到系统的环境变量(PATH)中,ubuntu下使用命令直接安装即可。

# 安装nodejs
sudo apt install nodejs
# 安装npm
sudo apt install npm
更多推荐




所有评论(0)