WeatherFlow Tempest MCP服务器
一种模型上下文协议(MCP)服务器,提供对WeatherFlow Tempest气象站数据的无缝访问。 该服务器使人工智能助手和应用程序能够检索实时天气观测、预报和台站元数据。
🌤️ 特性
- 实时天气数据:从个人气象站获取当前状况
- 天气预报:使用专业气象模型获取每小时和每天的预报
- 车站管理:发现和管理多个气象站
- 设备信息:有关连接的气象设备的详细元数据
- 智能高速缓存:具有可配置TTL的自动缓存,可实现最佳性能
- 用于交互式查询和结构化天气数据的工具
- 综合数据:温度、湿度、压力、风、降水、太阳辐射、紫外线指数和雷电探测
🚀 快速开始
先决条件
- Python 3.13或更高版本
- WeatherFlow API代币(在 stormwx.com/settings/tokens)
安装
虽然每个客户端都有自己的指定方式,但通常会使用以下值:
| 字段 | 值 |
|---|---|
| 命令 | uvx |
| 参数 | mcp-server-tempest |
| 环境 | WEATHERFLOW_API_TOKEN = `` |
开发版本
如果你想使用最新和最好的,服务器可以直接从GitHub中提取。 只需添加一个额外的 --from 论点:
| 字段 | 值 |
|---|---|
| 命令 | uvx |
| 参数 | --from, git+https://github.com/briandconnelly/mcp-server-tempest, mcp-server-tempest |
| 环境 | WEATHERFLOW_API_TOKEN = `` |
📋 配置
环境变量
| 变量 | 描述 | 默认值 | 必填 |
|---|---|---|---|
WEATHERFLOW_API_TOKEN | 您的WeatherFlow API代币 | - | ✅ 是的 |
WEATHERFLOW_CACHE_TTL | 内存缓存TTL(秒) | 300 | 否 |
WEATHERFLOW_CACHE_SIZE | 最大内存缓存条目数 | 100 | 否 |
WEATHERFLOW_DISK_CACHE_TTL | 磁盘缓存TTL(秒) | 86400 | 否 |
缓存和数据新鲜度
服务器将响应缓存在两层中:
- 内存中 (
WEATHERFLOW_CACHE_TTL/WEATHERFLOW_CACHE_SIZE):全部四个
工具。
- 在磁盘上 (
WEATHERFLOW_DISK_CACHE_TTL,默认24小时):get_stations和
get_station_details 只有。存储在 platformdirs.user_cache_dir("mcp-server-tempest") 在每个令牌中 (哈希键)子目录。
要清除:重新启动服务器(在内存中)或删除缓存目录 (磁盘)。
运输
stdio(的默认值 uvx mcp-server-tempest 以及README配置)。
🛠️ 用法
可用工具
get_stations()
获取所有气象站和连接设备的列表。
# Get all available stations
stations = await client.call_tool("get_stations")
for station in stations["stations"]:
print(f"Station: {station['name']} (ID: {station['station_id']})")
print(f"Location: {station['latitude']}, {station['longitude']}")get_observation(station_id)
获取特定站点的当前天气状况。
# Get current conditions
obs = await client.call_tool("get_observation", {"station_id": 12345})
current = obs["obs"][0]
print(f"Temperature: {current['air_temperature']}°")
print(f"Humidity: {current['relative_humidity']}%")
print(f"Wind: {current['wind_avg']} {obs['station_units']['units_wind']}")get_forecast(station_id)
获取天气预报和当前状况。
# Get forecast
forecast = await client.call_tool("get_forecast", {"station_id": 12345})
# Current conditions
current = forecast["current_conditions"]
print(f"Current: {current['air_temperature']}°")
print(f"Conditions: {current['conditions']}")
# Today's forecast
today = forecast["forecast"]["daily"][0]
print(f"High/Low: {today['air_temp_high']}°/{today['air_temp_low']}°")
print(f"Rain chance: {today['precip_probability']}%")get_station_details(station_id)
获取特定电台的详细信息。
# Get station details
station = await client.call_tool("get_station_details", {"station_id": 12345})
print(f"Station: {station['name']}")
print(f"Elevation: {station['station_meta']['elevation']}m")
print(f"Devices: {len(station['devices'])}")🌟 示例
基本天气检查
# Get your stations
stations = await client.call_tool("get_stations")
station_id = stations["stations"][0]["station_id"]
# Get current conditions
obs = await client.call_tool("get_observation", {"station_id": station_id})
current = obs["obs"][0]
units = obs["station_units"]
print(f"🌡️ Temperature: {current['air_temperature']}°{units['units_temp']}")
print(f"💧 Humidity: {current['relative_humidity']}%")
print(f"💨 Wind: {current['wind_avg']} {units['units_wind']}")
print(f"🌧️ Precipitation: {current['precip_accum_local_day']} {units['units_precip']}")天气预报
from datetime import datetime
# Get forecast
forecast = await client.call_tool("get_forecast", {"station_id": station_id})
# Today's weather
today = forecast["forecast"]["daily"][0]
print(f"📅 Today: {today['conditions']}")
print(f"🌡️ High: {today['air_temp_high']}° / Low: {today['air_temp_low']}°")
print(f"🌧️ Rain chance: {today['precip_probability']}%")
# Next few hours
for hour in forecast["forecast"]["hourly"][:6]:
time = datetime.fromtimestamp(hour["time"])
print(f"🕐 {time.strftime('%H:%M')}: {hour['air_temperature']}° - {hour['conditions']}")车站信息
# Get station details
station = await client.call_tool("get_station_details", {"station_id": station_id})
print(f"🏠 Station: {station['name']}")
print(f"📍 Location: {station['latitude']}°, {station['longitude']}°")
print(f"⛰️ Elevation: {station['station_meta']['elevation']}m")
print(f"🕐 Timezone: {station['timezone']}")
# Check device status
for device in station["devices"]:
if device.get("serial_number"):
status = "🟢 Online" if device.get("device_meta") else "🔴 Offline"
print(f"📡 {device['device_type']}: {status}")🤝 贡献
- 分叉存储库
- 创建要素分支(
git checkout -b feature/amazing-feature) - 提交您的更改(
git commit -m 'Add amazing feature') - 推到分支(
git push origin feature/amazing-feature) - 打开拉取请求
📄 许可证
此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。
🙏 致谢
📞 支持
- 问题:
- 文档: WeatherFlow API文件
- 社区: WeatherFlow社区
