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应用访问工具和上下文的程序。包含三个关键部分:

  1. 连接——客户端初始化连接,服务器响应,客户端通过初始化通知进行确认。

  2. 交换——请求、响应和通知以 JSON 格式交换。

  3. 终止方式有 3 种:客户端关闭、传输断开或发生错误。

MCP客户端

MCP客户端内置于AI应用程序中,主要负责以下事项:

  • 发现服务器功能

  • 从服务器接收数据

  • 管理LLM工具执行

MCP服务器

      MCP 服务器是为 AI 应用提供的独立服务。它们监听客户端请求,并根据自身的能力做出响应

      MCP 服务器可以提供三种主要类型的功能:

  1. 资源:可供客户端读取的类文件数据(例如 API 响应或文件内容)

  2. 工具:LLM 可以调用的函数(需经用户批准)

  3. 提示:预先编写的模板,可帮助用户完成特定任务

服务器可以位于本地或者远程,这可以通过两种默认的传输机制来实现:

  • 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

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐