MCP协议解析：如何通过Model Context Protocol 实现高效的AI客户端与服务端交互

言简意赅的讲解MCP解决的痛点

在人工智能和机器学习的世界中，如何有效地连接客户端和服务器进行数据交换一直是个挑战。MCP（Model Context Protocol）应运而生，解决了传统方式中的很多问题，尤其是在数据权限获取和工具调用方面。

从最初的怀疑到现在的认可，MCP已经成为了越来越多人开发AI相关服务时的首选协议。它为客户端与服务端之间的交互提供了一种简单而高效的方式。今天，就让我们一起来深入了解MCP协议，以及如何使用它开发一个简单的天气服务。本文将用Claude 和python来完成展示。

MCP协议简介

MCP的核心目标是使AI客户端能够在没有直接访问底层数据的情况下，通过协议获取所需的数据或调用远程服务。这些服务被分为两类：资源和工具。

1. 资源（Resource）：这些是可以读取的数据，类似于REST API中的端点。
2. 工具（Tool）：这些是可以调用的功能，类似于远程过程调用（RPC）。

在MCP框架下，资源和工具之间有明确的区分。具体来说，资源用于提供数据，而工具则用于执行特定的操作。

MCP协议的调用流程

当客户端需要获取某些数据或调用某个工具时，它会通过MCP协议与服务器进行交互。这个过程可以分为几个步骤：

资源的调用流程：

1. 列出可用资源：客户端首先会调用 list_resources，查询服务器提供的所有资源。
2. 读取特定资源：在了解可用资源后，客户端可以选择一个资源，并通过 read_resource 请求特定资源的数据。例如，客户端可能请求 weather://London/current，获取伦敦当前的天气信息。
3. 返回数据：服务器会处理请求，调用相应的函数（例如 fetch_weather），并将处理后的数据返回给客户端。

工具的调用流程：

1. 列出可用工具：客户端通过 list_tools 查询服务器上可用的工具。
2. 调用工具：客户端可以选择一个工具并通过 call_tool 请求执行某个操作。例如，调用天气工具来获取某个城市的天气预报。

MCP的设计哲学使得每个功能都可以模块化、独立，且易于扩展。这也意味着在服务器端，每个函数仅负责执行单一任务，提高了代码的可维护性。

MCP协议的URI设计

MCP使用统一资源标识符（URI）来唯一标识资源，类似于REST API的URL。一个典型的URI格式可能是：

weather://London/current

• weather:// 是协议或方案部分，表明这是一个天气相关的资源。
• London 是主机部分，指定了查询的城市。
• /current 是路径部分，表示我们想要获取该城市的当前天气数据。

这种设计方式使得资源的层次结构清晰明了，支持多种不同的数据请求。例如，weather://Paris/forecast 可能代表巴黎的天气预报，weather://NewYork/current 则代表纽约的当前天气。

为什么MCP协议如此受欢迎？

1. 资源唯一标识：通过URI，客户端可以清晰地指向服务器上的特定资源，减少了资源冲突的可能。
2. 清晰的层次结构：不同的资源可以根据URI的不同路径进行区分，例如，当前天气、天气预报等。
3. 与现有标准兼容：MCP的设计理念与现代互联网的标准做法高度一致，特别是在API和RESTful服务的设计上。
4. 模块化设计：每个功能都被封装成独立的模块，提升了代码的可维护性和可扩展性。

尽管MCP协议在理论上非常优越，但在实际应用中仍然面临一些挑战。比如，当前支持MCP协议的客户端不多，Claude是目前支持得较好的客户端之一。此外，MCP协议目前只能在本地端调用，无法直接指向远程服务器。

如何使用MCP协议开发自己的天气服务？

如果你也想基于MCP协议开发自己的服务，比如一个简单的天气查询工具，下面是一些基本的步骤和配置。

1. 安装Python环境

首先，你需要确保安装了Python 3.10及以上版本。可以通过以下命令来验证：

python --version  # Should be 3.10 or higher

2. 安装必要的依赖

使用Homebrew安装 uv，然后通过 uvx 创建一个新的MCP项目。

brew install uv
uv --version  # Should be 0.4.18 or higher

接着，创建项目：

uvx create-mcp-server --path weather_service
cd weather_service

安装必要的Python依赖：

uv add httpx python-dotenv

3. 设置API密钥

在项目根目录下创建一个 .env 文件，加入你的API密钥：

WARNING:
本文将使用 OpenWeatherMap API API KEY

OPENWEATHER_API_KEY=your-api-key-here

4. 开发服务

在 weather_service/src/weather_service/server.py 文件中设置基本的导入和服务器配置。具体代码可以根据需求定制。

import os
import json
import logging
from datetime import datetime, timedelta
from collections.abc import Sequence
from functools import lru_cache
from typing import Anyimport httpx
import asyncio
from dotenv import load_dotenv
from mcp.server import Server
from mcp.types import (Resource,Tool,TextContent,ImageContent,EmbeddedResource,LoggingLevel
)
from pydantic import AnyUrl# Load environment variables
load_dotenv()# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("weather-server")# API configuration
API_KEY = os.getenv("OPENWEATHER_API_KEY")
if not API_KEY:raise ValueError("OPENWEATHER_API_KEY environment variable required")API_BASE_URL = "http://api.openweathermap.org/data/2.5"
DEFAULT_CITY = "London"
CURRENT_WEATHER_ENDPOINT = "weather"
FORECAST_ENDPOINT = "forecast"# The rest of our server implementation will go here
# Create reusable params
http_params = {"appid": API_KEY,"units": "metric"
}async def fetch_weather(city: str) -> dict[str, Any]:async with httpx.AsyncClient() as client:response = await client.get(f"{API_BASE_URL}/weather",params={"q": city, **http_params})response.raise_for_status()data = response.json()return {"temperature": data["main"]["temp"],"conditions": data["weather"][0]["description"],"humidity": data["main"]["humidity"],"wind_speed": data["wind"]["speed"],"timestamp": datetime.now().isoformat()}app = Server("weather-server")
app = Server("weather-server")@app.list_resources()
async def list_resources() -> list[Resource]:"""List available weather resources."""uri = AnyUrl(f"weather://{DEFAULT_CITY}/current")return [Resource(uri=uri,name=f"Current weather in {DEFAULT_CITY}",mimeType="application/json",description="Real-time weather data")]@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:"""Read current weather data for a city."""city = DEFAULT_CITYif str(uri).startswith("weather://") and str(uri).endswith("/current"):city = str(uri).split("/")[-2]else:raise ValueError(f"Unknown resource: {uri}")try:weather_data = await fetch_weather(city)return json.dumps(weather_data, indent=2)except httpx.HTTPError as e:raise RuntimeError(f"Weather API error: {str(e)}")app = Server("weather-server")# Resource implementation ...@app.list_tools()
async def list_tools() -> list[Tool]:"""List available weather tools."""return [Tool(name="get_forecast",description="Get weather forecast for a city",inputSchema={"type": "object","properties": {"city": {"type": "string","description": "City name"},"days": {"type": "number","description": "Number of days (1-5)","minimum": 1,"maximum": 5}},"required": ["city"]})]@app.call_tool()
async def call_tool(name: str, arguments: Any) -> Sequence[TextContent | ImageContent | EmbeddedResource]:"""Handle tool calls for weather forecasts."""if name != "get_forecast":raise ValueError(f"Unknown tool: {name}")if not isinstance(arguments, dict) or "city" not in arguments:raise ValueError("Invalid forecast arguments")city = arguments["city"]days = min(int(arguments.get("days", 3)), 5)try:async with httpx.AsyncClient() as client:response = await client.get(f"{API_BASE_URL}/{FORECAST_ENDPOINT}",params={"q": city,"cnt": days * 8,  # API returns 3-hour intervals**http_params,})response.raise_for_status()data = response.json()forecasts = []for i in range(0, len(data["list"]), 8):day_data = data["list"][i]forecasts.append({"date": day_data["dt_txt"].split()[0],"temperature": day_data["main"]["temp"],"conditions": day_data["weather"][0]["description"]})return [TextContent(type="text",text=json.dumps(forecasts, indent=2))]except httpx.HTTPError as e:logger.error(f"Weather API error: {str(e)}")raise RuntimeError(f"Weather API error: {str(e)}")
async def main():# Import here to avoid issues with event loopsfrom mcp.server.stdio import stdio_serverasync with stdio_server() as (read_stream, write_stream):await app.run(read_stream,write_stream,app.create_initialization_options())

在 weather_service/src/weather_service/__init__.py 文件中设置基本的导入和服务器配置。具体代码可以根据需求定制。

from . import server
import asynciodef main():"""Main entry point for the package."""asyncio.run(server.main())# Optionally expose other important items at package level
__all__ = ['main', 'server']

5. 配置Claude

将以下配置添加到 claude_desktop_config.json 文件中：

{"mcpServers": {"weather": {"command": "uv","args": ["--directory","path/to/your/project","run","weather-service"],"env": {"OPENWEATHER_API_KEY": "your-api-key"}}}
}

然后，重启Claude，确保你的天气服务已在应用中生效。

结语

MCP协议无疑为开发者提供了一个更高效、更简洁的方式来构建客户端与服务端之间的交互。虽然当前的支持还有待加强，但它在未来的发展潜力巨大，尤其是在AI与大数据应用的快速增长下，MCP协议无疑是一个值得关注的工具。

---

通过上述内容，你就已经基本理解了这个方法，基础用法我也都有展示。如果你能融会贯通，我相信你会很强

Best
Wenhao (楠博万)