1. 背景
在介绍 Model Context Protocol (MCP) 的概念、架构及解决的问题后，为了进一步理解整套 MCP 框架如何落地，本文基于官方示例——获取天气预报，演示 MCP 落地的整条链路。
2. MCP 示例
该案例是构建一个简单的 MCP 天气预报服务器，并将其连接到主机（如 Claude for Desktop）。大模型虽然能力强大，但其内容往往是非实时的。例如，它无法直接获取最新的天气预报和严重天气警报。使用 MCP 可以解决这一问题。
构建一个服务器，提供两个工具：get-alerts（获取警报）和 get-forecast（获取预报）。然后将该服务器连接到 MCP 主机。
环境配置
（1）安装 uv
curl -LsSf https://astral.sh/uv/install.sh | sh
安装完成后会提示安装成功。
（2）安装所需的依赖包
需安装 mcp 及相关 HTTP 库。
（3）构建服务端代码
在 server.py 中构建相应的 get-alerts 和 get-forecast 工具。
from typing import Any
import asyncio
import httpx
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"
server = Server("weather")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""List available tools."""
return [
types.Tool(
name="get-alerts",
description="Get weather alerts for a state",
inputSchema={
"type": "object",
"properties": {
"state": {"type": "string", "description": "Two-letter state code (e.g. CA, NY)",},
},
"required": ["state"],
},
),
types.Tool(
name="get-forecast",
description="Get weather forecast for a location",
inputSchema={
"type": "object",
"properties": {
"latitude": {"type": "number", "description": "Latitude of the location",},
"longitude": {"type": "number", "description": "Longitude of the location",},
},
"required": ["latitude", "longitude"],
},
),
]
async def make_nws_request(client: httpx.AsyncClient, 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"}
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 concise string."""
props = feature["properties"]
return (
f"Event: {props.get('event', 'Unknown')}\n"
f"Area: {props.get('areaDesc', 'Unknown')}\n"
f"Severity: {props.get('severity', 'Unknown')}\n"
f"Status: {props.get('status', 'Unknown')}\n"
f"Headline: {props.get('headline', 'No headline')}\n"
"---"
)
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""Handle tool execution requests."""
if not arguments:
raise ValueError("Missing arguments")
if name == "get-alerts":
state = arguments.get("state")
if not state:
raise ValueError("Missing state parameter")
state = state.upper()
if len(state) != 2:
raise ValueError("State must be a two-letter code (e.g. CA, NY)")
async with httpx.AsyncClient() as client:
alerts_url = f"{NWS_API_BASE}/alerts?area={state}"
alerts_data = await make_nws_request(client, alerts_url)
if not alerts_data:
return [types.TextContent(type="text", text="Failed to retrieve alerts data")]
features = alerts_data.get("features", [])
if not features:
return [types.TextContent(type="text", text=f"No active alerts for {state}")]
formatted_alerts = [format_alert(feature) for feature in features[:20]]
alerts_text = f"Active alerts for {state}:\n\n" + "\n".join(formatted_alerts)
return [types.TextContent(type="text", text=alerts_text)]
elif name == "get-forecast":
try:
latitude = float(arguments.get("latitude"))
longitude = float(arguments.get("longitude"))
except (TypeError, ValueError):
return [types.TextContent(type="text", text="Invalid coordinates. Please provide valid numbers.")]
if not (-90 <= latitude <= 90) or not (-180 <= longitude <= 180):
return [types.TextContent(type="text", text="Invalid coordinates.")]
async with httpx.AsyncClient() as client:
points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
points_data = await make_nws_request(client, points_url)
if not points_data:
return [types.TextContent(type="text", text=f"Failed to retrieve grid point data.")]
properties = points_data.get("properties", {})
forecast_url = properties.get("forecast")
if not forecast_url:
return [types.TextContent(type="text", text="Failed to get forecast URL.")]
forecast_data = await make_nws_request(client, forecast_url)
if not forecast_data:
return [types.TextContent(type="text", text="Failed to retrieve forecast data")]
periods = forecast_data.get("properties", {}).get("periods", [])
if not periods:
return [types.TextContent(type="text", text="No forecast periods available")]
formatted_forecast = []
for period in periods:
forecast_text = (
f"{period.get('name', 'Unknown')}:\n"
f"Temperature: {period.get('temperature', 'Unknown')}°{period.get('temperatureUnit', 'F')}\n"
f"Wind: {period.get('windSpeed', 'Unknown')} {period.get('windDirection', '')}\n"
f"{period.get('shortForecast', 'No forecast available')}\n"
"---"
)
formatted_forecast.append(forecast_text)
forecast_text = f"Forecast for {latitude}, {longitude}:\n\n" + "\n".join(formatted_forecast)
return [types.TextContent(type="text", text=forecast_text)]
else:
raise ValueError(f"Unknown tool: {name}")
async def main():
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="weather",
server_version="0.1.0",
capabilities=server.get_capabilities(notification_options=NotificationOptions(), experimental_capabilities={},),
),
)
if __name__ == "__main__":
asyncio.run(main())
核心逻辑在于 @server.list_tools() 注册可用工具处理器，以及 @server.call_tool() 注册执行工具调用的处理器。调用函数匹配工具名称，抽取输入参数，发起 API 请求并处理结果。
服务端与客户端交互
测试服务器与 Claude for Desktop。构建 MCP 客户端的核心逻辑如下：
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 = [{"name": tool.name, "description": tool.description, "input_schema": tool.inputSchema} for tool in response.tools]
response = self.anthropic.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1000,
messages=messages,
tools=available_tools
)
final_text = []
assistant_message_content = []
for content in response.content:
if content.type == 'text':
final_text.append(content.text)
assistant_message_content.append(content)
elif content.type == 'tool_use':
tool_name = content.name
tool_args = content.input
result = await self.session.call_tool(tool_name, tool_args)
final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")
assistant_message_content.append(content)
messages.append({"role": "assistant", "content": assistant_message_content})
messages.append({"role": "user", "content": [{"type": "tool_result", "tool_use_id": content.id, "content": result.content}]})
response = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022", max_tokens=1000, messages=messages, tools=available_tools)
final_text.append(response.content[0].text)
return "\n".join(final_text)
启动客户端需要打开配置文件：~/Library/Application Support/Claude/claude_desktop_config.json。如果文件不存在则创建，并配置以下信息（以示例说明，假设 uv init 的是 weather）：
{
"mcpServers": {
"weather": {
"command": "uv",
"args": [
"--directory",
"/ABSOLUTE/PATH/TO/PARENT/FOLDER/weather",
"run",
"server.py"
]
}
}
}
保存文件并重新启动 Claude for Desktop，即可识别天气服务器暴露的工具。
3. MCP 到底解决了什么问题
工具是智能体框架的重要组成部分，允许大模型与外界互动并扩展其能力。即使没有 MCP 协议，也可以实现 LLM 智能体，但存在弊端：当有许多不同的 API 时，启用工具使用变得很麻烦，因为任何工具都需要手动构建 prompt，且每当其 API 发生变化时需手动更新。
MCP 解决了当存在大量工具时，能够自动发现并自动构建 prompt 的问题。
整体流程示例：
- 工具发现：MCP 主机（如 Claude Desktop）首先调用 MCP 服务器，询问有哪些工具可用。
- 工具调用：MCP 客户端接收到列出的可用工具后发给 LLM，LLM 选择使用某个工具，通过主机向 MCP 服务器发送请求，接收结果。
- 输出答案：LLM 收到工具处理结果后，向用户输出最终答案。
总结来说，MCP 协议让智能体更容易管理、发现和使用工具。