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": {: , : ,},
},
: [],
},
),
types.Tool(
name=,
description=,
inputSchema={
: ,
: {
: {: , : ,},
: {: , : ,},
},
: [, ],
},
),
]
() -> [, ] | :
headers = {: USER_AGENT, : }
:
response = client.get(url, headers=headers, timeout=)
response.raise_for_status()
response.json()
Exception:
() -> :
props = feature[]
(
)
() -> [types.TextContent | types.ImageContent | types.EmbeddedResource]:
arguments:
ValueError()
name == :
state = arguments.get()
state:
ValueError()
state = state.upper()
(state) != :
ValueError()
httpx.AsyncClient() client:
alerts_url =
alerts_data = make_nws_request(client, alerts_url)
alerts_data:
[types.TextContent(=, text=)]
features = alerts_data.get(, [])
features:
[types.TextContent(=, text=)]
formatted_alerts = [format_alert(feature) feature features[:]]
alerts_text = + .join(formatted_alerts)
[types.TextContent(=, text=alerts_text)]
name == :
:
latitude = (arguments.get())
longitude = (arguments.get())
(TypeError, ValueError):
[types.TextContent(=, text=)]
(- <= latitude <= ) (- <= longitude <= ):
[types.TextContent(=, text=)]
httpx.AsyncClient() client:
points_url =
points_data = make_nws_request(client, points_url)
points_data:
[types.TextContent(=, text=)]
properties = points_data.get(, {})
forecast_url = properties.get()
forecast_url:
[types.TextContent(=, text=)]
forecast_data = make_nws_request(client, forecast_url)
forecast_data:
[types.TextContent(=, text=)]
periods = forecast_data.get(, {}).get(, [])
periods:
[types.TextContent(=, text=)]
formatted_forecast = []
period periods:
forecast_text = (
)
formatted_forecast.append(forecast_text)
forecast_text = + .join(formatted_forecast)
[types.TextContent(=, text=forecast_text)]
:
ValueError()
():
mcp.server.stdio.stdio_server() (read_stream, write_stream):
server.run(
read_stream,
write_stream,
InitializationOptions(
server_name=,
server_version=,
capabilities=server.get_capabilities(notification_options=NotificationOptions(), experimental_capabilities={},),
),
)
__name__ == :
asyncio.run(main())
