浅析Model Context Protocol (MCP)协议：概念、优势与实战应用

---

前言

         在当今快速发展的AI技术领域，大语言模型（LLM）的应用越来越广泛。然而，如何高效地利用这些模型并将其集成到实际应用中，仍然是一个挑战。MCP（大模型上下文协议）正是为了解决这一问题而设计的。它通过定义一种标准化的通信机制，使得客户端和服务器能够高效地交互，充分利用大语言模型的强大能力。本文将详细介绍MCP协议的相关概念、优势以及实战应用。

---

一、什么是MCP（Model Context Protocol）

（一）MCP 的定义

         MCP（Model Context Protocol，模型上下文协议） ，2024年11月底，由 Anthropic 推出的一种开放标准，旨在统一大型语言模型（LLM）与外部数据源和工具之间的通信协议。它类似于AI的“通用翻译器”，使AI能够安全、可控地访问文件、应用或网络服务，并执行具体任务。

（二）MCP 的优势

         MCP的优势在于它提供了一种标准化的通信机制，使得AI模型能够与不同的API和数据源无缝交互。它旨在替换碎片化的Agent代码集成，从而使AI系统更可靠、更有效。通过MCP，服务商可以基于协议推出自己的AI能力，支持开发者快速构建更强大的AI应用。

（三）MCP 与Function Calling 的区别

• MCP（Model Context Protocol），模型上下文协议
• Function Calling，函数调用

         上一篇文章我把Function Calling讲了一下，感兴趣可以去https://blog.csdn.net/pdsu_Zhe/article/details/146288112
         Function Calling指的是AI模型根据上下文自动执行函数的机制，不同的模型有不同的Function Calling实现。而MCP是一个标准协议，它通过定义统一的通信协议和架构，使得AI系统能够无缝地与多种数据源和服务进行交互。MCP的优势在于：一是开放标准利于服务商开发API，二是避免开发者重复造轮子，可利用现有MCP服务增强Agent。主要总结为以下几个方面：

Function Call	MCP
✔️简单直接： 容易理解和实现	✔️ 更完整的功能：工具、资源和提示三位一体
✔️广泛支持： 几乎所有LLM都兼容	✔️ 标准化协议：JSON-RPC通信标准
✔️标准化结构 ： JSON格式一致	✔️ 通用性：不绑定特定的AI模型
❌功能单一 只能调用函数	✔️ 双向通信：支持事件和通知
❌ 缺乏标准协议：每家实现不完全一致	✔️ 细粒度权限：明确的控制机制
❌只支持同步调用 ： 缺乏事件机制	❌ 相对复杂：实现和理解成本较高
适用场景：简单应用、快速原型、直接与LLM API交互	适用场景：企业级应用、需要支持多种AI的产品、复杂交互场景

---

二、MCP的工作原理

（一）MCP的架构

         MCP基于客户端-服务器模型，共分五个部分：

• MCP Hosts：运行AI模型的应用程序，如Cursor、Claude Desktop。
• MCP Clients：在Hosts内负责与服务器通信的模块。
• MCP Servers：通过标准化协议为Clients提供工具、数据和上下文。
• Local Data Sources：本地文件、数据库等直接可访问的数据。
• Remote Services：外部API或服务，如GitHub或Slack。

官方的 MCP 架构图：

（二）MCP的核心架构

         MCP的核心架构包括MCP Hosts、MCP Clients和MCP Servers。MCP Servers是整个协议的灵魂，它告诉AI Agent当前有哪些可用服务和数据，AI Agent再通过函数调用执行具体任务。

（三）MCP的工作流程

1. 建立连接：客户端向服务器发起连接，创建通信链路。
2. 发送请求：客户端构建并发送包含具体需求的消息。
3. 解析执行：服务器接收请求，解析并执行相关操作。
4. 返回响应：服务器将操作结果封装后反馈给客户端。
5. 关闭连接：任务完成后，连接被关闭以结束会话。
   

---

三、MCP协议的实战应用

         本实战以MCP官网给出的例子进行天气查询并且结合GLM-4-Flash大模型生成答案。
官方教程：https://modelcontextprotocol.io/quickstart/server

（一）项目结构

         官方教程给出的方案是用uv来管理Python程序。
         uv是由Astral公司开发的一个高性能的Python包管理工具，旨在成为“Python的Cargo”。它提供了快速、可靠且易用的包管理体验，支持依赖管理、虚拟环境管理、单文件脚本运行，甚至Python本身的安装和管理。
         uv是一个统一接口，可以管理Python项目、命令行工具、单文件脚本，甚至Python本身。它是一个二进制文件，支持多种方式安装而不依赖于Rust和Python环境）。

# uv安装方式：
pip install uv

### 项目结构
MCP-CLIENT/
│
├── mcp-client/            # 客户端代码目录
│   ├── .venv/            # 客户端虚拟环境，用于隔离Python依赖
│   ├── .env             # 客户端环境变量文件，存储敏感信息如API密钥
│   ├── .python-version    # Python版本指定文件
│   ├── client.py         # 客户端主程序文件
│   ├── pyproject.toml    # 客户端项目配置文件
│   ├── README.md        # 客户端项目说明文件
│   └── uv.lock          # 客户端依赖锁定文件
│
├── weather/               # 服务端代码目录
│   ├── .venv/            # 服务端虚拟环境，用于隔离Python依赖
│   ├── .python-version   # Python版本指定文件，存储敏感信息如API密钥
│   ├── pyproject.toml   # 服务端项目配置文件
│   ├── README.md         # 服务端项目说明文件
│   └── uv.lock           # 服务端依赖锁定文件
│   └── weather.py       # 服务端主程序文件
│
└── 外部库/               # 项目依赖的第三方库（具体内容未展示）
└── 临时文件和控制台       # IDE生成的临时文件和控制台输出区域

---

（二）服务端 (weather.py)

1. 环境准备

# Create a new directory for our project
uv init weather
cd weather

# Create virtual environment and activate it
uv venv
.venv\Scripts\activate

# Install dependencies
uv add mcp httpx

# Create our server file
echo. > weather.py

---

2. 初始化 FastMCP 服务器

# 创建了一个名为 "weather" 的 FastMCP 服务器实例。
mcp = FastMCP("weather")

3. 定义常量

# 定义了用于访问天气API的基础URL和用户代理字符串。
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

4. 定义异步函数 make_nws_request

# 这个函数用于向美国国家气象局（NWS）的API发送异步请求，并处理可能的异常。

async def make_nws_request(url: str) -> dict[str, Any] | None:
    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

5. 定义 MCP 工具函数 get_alerts 和 get_forecast

# 使用 @mcp.tool() 装饰器定义了两个MCP工具函数，get_alerts 和 get_forecast
# 它们分别用于获取特定州的天气警报和给定位置的天气预报。
@mcp.tool()
async def get_alerts(state: str) -> str:
    ...

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    ...

---

（三）客户端 (client.py)

1. 环境准备

# Create project directory
uv init mcp-client
cd mcp-client

# Create virtual environment
uv venv

# Activate virtual environment
# On Windows:
.venv\Scripts\activate

# Install required packages
uv add mcp openai python-dotenv

# Create our main file
echo. >  client.py

2. .env文件设置

---

3. 加载环境变量

# 从.env文件加载环境变量，这些变量用于配置OpenAI API的密钥和基础URL。
load_dotenv()  # load environment variables from .env
api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_API_BASE"]

4. 定义 MCPClient 类

# 定义了一个MCPClient类，它初始化会话和客户端对象，并创建一个OpenAI实例。
class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.openai = OpenAI(api_key=api_key, base_url=base_url)

5. 定义异步方法 connect_to_server

# 这个方法连接到MCP服务器，设置命令行参数，初始化客户端会话，并列出服务器上可用的工具。
async def connect_to_server(self, server_script_path: str):
    ...

6. 定义异步方法 process_query

# 这个方法处理来自用户的查询，使用OpenAI模型生成响应，并处理任何由模型调用的工具。
async def process_query(self, query: str) -> str:
    ...

---

（四）运行

python client.py D:\程序\MCP-CLIENT\weather\weather.py

输出：

(mcpserver) D:\程序\MCP-CLIENT\mcp-client>python client.py D:\程序\MCP-CLIENT\weather\weather.py

Connected to server with tools: [['get_alerts', 'Get weather alerts for a US state.\n\n    Args:\n        state: Two-letter US state code (e.g. CA, NY)\n    ', {'properties': {'state': {'title': 'State', 'type': 'string'}}, 'required': ['state'], 'title': 'get_alertsArguments', 'type': 'object'}], ['get_forecast', 'Get weather forecast for a location.\n\n    Args:\n        latitude: Latitude of the location\n        longitude: Longitude of the location\n    ', {'properties': {'latitude': {'title': 'Latitude', 'type': 'number'}, 'longitude': {'title': 'Longitude', 'type': 'number'}}, 'required': ['latitude', 'longitude'], 'title': 'get_forecastArguments', 'type': 'object'}]]

MCP Client Started!
Type your queries or 'quit' to exit.

Query: 帮我查询一下旧金山的天气
{'content': None, 'refusal': None, 'role': 'assistant', 'annotations': None, 'audio': None, 'function_call': None, 'tool_calls': [{'id': 'call_-8891224313239820525', 'function': {'arguments': '{"latitude": 37.7749, "longitude": -122.4194}', 'name': 'get_forecast'}, 'type': 'function', 'index': 0}]}
{'content': '旧金山的天气预报如下：\n\n今晚：\n温度：49°F\n风速：9 英里/小时 南风\n预报：部分多云，最低气温约为49°F。南风，风速约为9英里/小时，阵风高达22英里/小时。\n\n周日：\n温度：62°F\n风速：9至23率为70%。新的降雨量可能不到十分之一英寸。\n\n周一夜间：\n温度：45°F\n风速：7至18 英里/小时 西北风\n预报：晴朗，最低气温约为45°F。西北风，风速7至18英里/小时，阵风高达28英里/小时。', 'refusal': None, ' None, 'tool_calls': None}

旧金山的天气预报如下：

今晚：
温度：49°F
风速：9 英里/小时 南风
预报：部分多云，最低气温约为49°F。南风，风速约为9英里/小时，阵风高达22英里/小时。

周日：
温度：62°F
风速：9至23 英里/小时 南风
预报：上午11点后降雨。大部分时间多云。最高气温接近62°F，下午气温降至约60°F。南风，风速9至23英里/小时，阵风高达35英里/小时。降水概率为100%。新的降雨量可能在十分之一到四分之一英寸之间。

周日夜间：
温度：50°F
风速：7至24 英里/小时 东南偏西南风
预报：降雨。多云。最低气温约为50°F，夜间气温升至约52°F。东南偏西南风，风速7至24英里/小时，阵风高达37英里/小时。降水概率为100%。新的降雨量可能在四分之一到半英寸之间。

周一：
温度：58°F
风速：7至18 英里/小时 西风
预报：下午5点前有可能降雨。大部分时间晴朗，最高气温接近58°F。西风，风速7至18英里/小时，阵风高达28英里/小时。降水概率为70%。新的降雨量可能不到十分之一英寸。

周一夜间：
温度：45°F
风速：7至18 英里/小时 西北风
预报：晴朗，最低气温约为45°F。西北风，风速7至18英里/小时，阵风高达28英里/小时。

---

四、源码

1. 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"


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')}
"""


@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)


if __name__ == "__main__":
    # Initialize and run the server
    mcp.run(transport="stdio")

---

2. Weather.py

import asyncio
from typing import Optional
from contextlib import AsyncExitStack
import os
import json

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()  # load environment variables from .env

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_API_BASE"]


# What are the weather alert in California
class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

        self.openai = OpenAI(api_key=api_key, base_url=base_url)

    async def connect_to_server(self, server_script_path: str):
        """Connect to an MCP server

        Args:
            server_script_path: Path to the server script (.py or .js)
        """
        is_python = server_script_path.endswith(".py")
        is_js = server_script_path.endswith(".js")
        if not (is_python or is_js):
            raise ValueError("Server script must be a .py or .js file")

        command = (
            "D:\程序\MCP-CLIENT\mcp-client\\.venv\\Scripts\\python.exe"
            if is_python
            else "node"
        )
        server_params = StdioServerParameters(
            command=command, args=[server_script_path], env=None
        )

        stdio_transport = await self.exit_stack.enter_async_context(
            stdio_client(server_params)
        )
        self.stdio, self.write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(
            ClientSession(self.stdio, self.write)
        )

        await self.session.initialize()

        # List available tools
        response = await self.session.list_tools()
        tools = response.tools
        print(
            "\nConnected to server with tools:",
            [[tool.name, tool.description, tool.inputSchema] for tool in tools],
        )

    async def process_query(self, query: str) -> str:
        """Process a query using Claude and available tools"""
        messages = [{"role": "user", "content": query}]

        response = await self.session.list_tools()

        available_tools = []

        for tool in response.tools:
            tool_schema = getattr(
                tool,
                "inputSchema",
                {"type": "object", "properties": {}, "required": []},
            )

            openai_tool = {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool_schema,
                },
            }
            available_tools.append(openai_tool)

        # Initial Claude API call
        model_response = self.openai.chat.completions.create(
            model="glm-4-flash",
            max_tokens=1000,
            messages=messages,
            tools=available_tools,
        )

        # Process response and handle tool calls
        tool_results = []
        final_text = []

        messages.append(model_response.choices[0].message.model_dump())
        print(messages[-1])
        if model_response.choices[0].message.tool_calls:
            tool_call = model_response.choices[0].message.tool_calls[0]
            tool_args = json.loads(tool_call.function.arguments)

            tool_name = tool_call.function.name
            result = await self.session.call_tool(tool_name, tool_args)
            tool_results.append({"call": tool_name, "result": result})
            final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")

            messages.append(
                {
                    "role": "tool",
                    "content": f"{result}",
                    "tool_call_id": tool_call.id,
                }
            )

            # Get next response from Claude
            response = self.openai.chat.completions.create(
                model="glm-4-flash",
                max_tokens=1000,
                messages=messages,
            )

            messages.append(response.choices[0].message.model_dump())
            print(messages[-1])

        return messages[-1]["content"]

    async def chat_loop(self):
        """Run an interactive chat loop"""
        print("\nMCP Client Started!")
        print("Type your queries or 'quit' to exit.")

        while True:
            try:
                query = input("\nQuery: ").strip()

                if query.lower() == "quit":
                    break

                response = await self.process_query(query)
                print("\n" + response)

            except Exception as e:
                print(f"\nError: {str(e)}")

    async def cleanup(self):
        """Clean up resources"""
        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)

    client = MCPClient()
    try:
        await client.connect_to_server(sys.argv[1])
        await client.chat_loop()
    finally:
        await client.cleanup()


if __name__ == "__main__":
    import sys

    asyncio.run(main())

3. /.env

OPENAI_API_KEY="Your_OPENAI_API_KEY"
OPENAI_API_BASE="https://open.bigmodel.cn/api/paas/v4/"

---