在本教程中，我们将构建一个简单的MCP天气服务器，并将其连接到主机Claude for Desktop。我们将从一个基本的设置开始，然后发展到更复杂的用例。
我们将建造什么
许多llm目前无法获取天气预报和恶劣天气警报。让我们使用MCP来解决这个问题!
我们将构建一个公开两个工具的服务器:get-alerts和get-forecast。然后我们将服务器连接到MCP主机 (在国内推荐使用 Continue https://www.claudemcp.com/zh)
核心MCP概念
MCP服务器可以提供三种主要类型的功能:
资源: 客户端可以读取的类似文件的数据 (如API响应或文件内容)
工具: 可由LLM调用的函数 (经用户批准)
提示: 帮助用户完成特定任务的预写模板
Python 方法 （点击获得完整代码）
前提
本快速入门假设您熟悉:
Python
LLMs像Qwen，deepseek
系统要求
已安装Python 3.10或更高版本。
您必须使用Python MCP SDK 1.2.0或更高版本。
设置您的环境
首先，让我们安装uv并设置我们的Python项目和环境:
curl -LsSf https://astral.sh/uv/install.sh | sh
确保之后重新启动您的终端，以确保uv命令被拿起。
现在，让我们创建并设置我们的项目:
# Create a new directory for our project
uv init weather
cd weather
# Create virtual environment and activate it
uv venv
source .venv/bin/activate
# Install dependencies
uv add "mcp[cli]" httpx
# Create our server file
touch weather.py现在让我们开始构建您的服务器。
构建您的服务器
导入包并设置实例
将这些添加到您的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"FastMCP类使用Python类型提示和文档字符串自动生成工具定义，使创建和维护MCP工具变得容易。
辅助函数
接下来，让我们添加用于查询和格式化来自National Weather Service API的数据的辅助函数:
（作为demo，也没必要去接入，模拟几个地区的天气数据就好）
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')您的服务器已完成!运行uv run weather.py确认一切正常。
现在让我们从现有的MCP主机Continue测试您的服务器。
你一个prompt下发生了什么
当你问一个问题:
客户将您的问题发送给LLM
LLM分析可用的工具，并决定使用哪一个 (MCP工具)
客户端通过MCP服务器执行所选工具
结果被送回给LLM
LLM制定了自然语言反应
响应显示给你!
评论区