AI导哈

天气数据服务

高性能天气数据服务，为农业应用提供实时天气和预测数据，优化缓存性能，适用于农业决策支持系统。

  • by srishhhhx
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 srishhhhx/weather-mcp

使用说明

气象MCP服务器

一个高性能的模型上下文协议（MCP）服务器，为农业应用提供天气数据，针对 克里希米特拉 项目。

快速开始

1.安装

# Clone and setup
cd /Users/srishtikn/Docs/weather-mcp
source .venv/bin/activate

# Set API key
cp .env.example .env
# Edit .env and add your OPENWEATHER_API_KEY

2.与KrishiMitra整合

# In your KrishiMitra .env
echo "MCP_WEATHER_PATH=/Users/srishtikn/Docs/weather-mcp/weather_server_free.py" >> .env

# Install FastMCP
pip install fastmcp

3.在代码中使用

import os
import json
from fastmcp.client import Client

async def get_weather(lat, lon):
    async with Client(os.getenv("MCP_WEATHER_PATH")) as client:
        result = await client.call_tool("get_weather", {"lat": lat, "lon": lon})
        return json.loads(result.content[0].text)

# Usage
weather = await get_weather(12.9716, 77.5946)
print(f"Temperature: {weather['temperature']}°C")

无需手动运行服务器，FastMCP客户端会自动处理！

业绩亮点

度量 值 影响
缓存加速 速度提高5000倍 0.01毫秒vs 40毫秒
API节省 减少91-99% 保持免费等级
响应时间 平均26ms 实时用户体验
容量 10000+用户/天 高度可扩展
可靠性 100%正常运行时间 过时的缓存回退

特性

核心功能

  • 当前天气 -实时天气数据
  • 5天预测 -天气预报
  • 健康检查 -服务器状态监控

演出

  • 5000倍缓存加速 -闪电般快速的响应
  • 90%+缓存命中率 -最少的API调用
  • 100%架构一致性 -可靠的数据结构
  • 优雅的错误处理 -过时的缓存回退

API合规性

  • 免费版兼容 -开放天气免费API
  • 速率限制安全 -智能缓存
  • 可扩展的 -处理10000+用户/天

可用工具

1. get_weather

获取坐标的当前天气。

参数：

  • lat （浮动）：纬度（-90到90）
  • lon （浮动）：经度（-180到180）

退货：

{
  "temperature": 20.33,
  "humidity": 93,
  "rainfall": 0.0,
  "wind_speed": 7.6,
  "description": "mist",
  "provider": "openweather",
  "cached": false,
  "timestamp": "2025-11-14T02:30:00"
}

2. get_forecast

获取坐标的天气预报。

参数：

  • lat （浮动）：纬度（-90到90）
  • lon （浮动）：经度（-180到180）
  • days （int）：天数（1-5，默认值5）

退货：

{
  "forecast": [
    {
      "date": "2025-11-14",
      "temp_min": 13.17,
      "temp_max": 27.5,
      "humidity": 27,
      "rainfall": 0.0,
      "description": "clear sky"
    }
  ],
  "provider": "openweather",
  "cached": false,
  "timestamp": "2025-11-14T02:30:00"
}

3. health_check

检查服务器运行状况。

参数： 无

退货：

{
  "status": "healthy",
  "service": "weather-mcp-server-free-tier",
  "timestamp": "2025-11-14T02:30:00",
  "api": "OpenWeather Free Tier (2.5)"
}

测试

运行所有测试

# Mock tests (31 tests)
pytest tests/test_weather_server.py -v

# Free tier real API tests (7 tests)
pytest tests/test_free_tier.py -v -m real_api

# Performance tests (5 tests)
pytest tests/test_performance_simple.py -v -s

# MCP client integration test
python test_mcp_client.py

测试结果

Mock Tests:        31/31 passed (100%)
Real API Tests:    7/7 passed (100%)
Performance Tests: 5/5 passed (100%)
Integration Test:  Passed

Total: 43/43 tests passing

建筑

Weather MCP Server
├── FastMCP Framework
│   ├── MCP Protocol (stdio)
│   └── Tool Registration
├── OpenWeather Free Tier API
│   ├── Current Weather (2.5)
│   └── 5-Day Forecast
├── In-Memory Cache
│   ├── TTL: 10 minutes
│   └── 5000x speedup
└── Error Handling
    ├── Stale cache fallback
    └── Graceful degradation

KrishiMitra的使用案例

1.灌溉调度

async def should_irrigate(lat, lon):
    weather = await get_weather(lat, lon)
    forecast = await get_forecast(lat, lon, days=2)
    
    # Don't irrigate if rain expected
    upcoming_rain = sum(day['rainfall'] for day in forecast['forecast'])
    return upcoming_rain < 10  # mm

2.作物建议

async def recommend_crops(lat, lon):
    forecast = await get_forecast(lat, lon, days=5)
    avg_temp = sum(day['temp_max'] for day in forecast['forecast']) / 5
    
    if avg_temp > 30:
        return ["Heat-resistant crops"]
    else:
        return ["Standard crops"]

3.虫害风险评估

async def check_pest_risk(lat, lon):
    weather = await get_weather(lat, lon)
    
    if weather['humidity'] > 80 and weather['temperature'] > 25:
        return {"risk": "high", "alert": "Monitor for fungal diseases"}
    return {"risk": "low"}

配置

环境变量

# Required
OPENWEATHER_API_KEY=your_api_key_here

# Optional
CACHE_TTL=600          # Cache duration in seconds (default: 10 min)
PORT=10000             # MCP server port (default: 10000)

缓存设置

  • 默认TTL： 10分钟（600秒）
  • 缓存类型： 内存中（每个服务器实例）
  • 缓存密钥格式： {lat}|{lon}|{endpoint}
  • 过期缓存： 用作API错误的回退

性能基准

高速缓存性能

First Request (API):     39.22 ms
Cached Request:          0.01 ms
Speedup:                 4984.7x

可扩展性

100 users/day:           300 API calls (50% cache hit)
1,000 users/day:         500 API calls (91.7% cache hit)
10,000 users/day:        1,000 API calls (98.75% cache hit)

自由层合规

Free Tier Limit:         1,000 calls/day
With Cache:              ✅ Safe for 10,000+ users
Without Cache:           ❌ Exceeds limit at 167 users

生产部署

清单

  • 已配置API密钥
  • 缓存TTL优化（10分钟）
  • 已测试错误处理
  • 性能验证（5000倍加速）
  • 免费套餐合规性已确认
  • 已创建集成指南
  • 通过测试（43/43）

监控

# Track cache hit rate
cache_hits = sum(1 for r in responses if r.get('cached'))
hit_rate = cache_hits / len(responses)

# Alert if < 80%
if hit_rate < 0.8:
    logger.warning(f"Low cache hit rate: {hit_rate:.1%}")

集成方法

方法1:FastMCP客户端（推荐）

from fastmcp.client import Client

async with Client(MCP_WEATHER_PATH) as client:
    weather = await client.call_tool("get_weather", {"lat": 12.9, "lon": 77.6})

优势：

  • 本地MCP协议
  • 自动启动/停止服务器
  • 类型安全
  • 无HTTP开销
  • 更简单的代码

项目统计

Lines of Code:           ~800
Test Coverage:           95%+
Tests:                   43 (all passing)
Performance:             5000x cache speedup

状态

生产就绪

  • 演出 5000倍缓存加速
  • 可靠性： 100%测试通过率
  • 一致性： 100%模式匹配

相关 MCP