weather_server.py•3.9 kB
# weather_server.py
from __future__ import annotations
from typing import Any, Dict, Optional
import httpx
from fastmcp import FastMCP
mcp = FastMCP("Weather Server")
GEOCODE_URL = "https://geocoding-api.open-meteo.com/v1/search"
FORECAST_URL = "https://api.open-meteo.com/v1/forecast"
async def geocode(client: httpx.AsyncClient, location: str) -> Optional[Dict[str, Any]]:
"""Geocode with Open‑Meteo first; fall back to Nominatim if no hits."""
# 1) Try Open‑Meteo Geocoding
try:
r = await client.get(
GEOCODE_URL,
params={
"name": location,
"count": 1,
"language": "en",
"format": "json",
},
)
r.raise_for_status()
data = r.json()
if data.get("results"):
return data["results"][0]
except Exception:
pass
# 2) Fallback: OpenStreetMap Nominatim (no key).
try:
nominatim = "https://nominatim.openstreetmap.org/search"
r2 = await client.get(
nominatim,
params={
"q": location,
"format": "jsonv2",
"limit": 1,
"addressdetails": 1,
},
headers={"User-Agent": "fastmcp-weather/0.1"},
)
r2.raise_for_status()
js = r2.json()
if js:
hit = js[0]
return {
"name": hit.get("display_name"),
"country": (hit.get("address") or {}).get("country", ""),
"latitude": float(hit["lat"]),
"longitude": float(hit["lon"]),
}
except Exception:
pass
return None
async def _get_forecast_impl(location: str, days: int = 1) -> Dict[str, Any]:
"""Return a simple forecast for a location."""
days = max(1, min(7, int(days)))
async with httpx.AsyncClient(timeout=15) as client:
place = await geocode(client, location)
if not place:
return {"ok": False, "error": f"Location not found: {location}"}
lat, lon = place["latitude"], place["longitude"]
params = {
"latitude": lat,
"longitude": lon,
"current_weather": True,
"daily": [
"temperature_2m_max",
"temperature_2m_min",
"precipitation_probability_max",
"precipitation_sum",
"windspeed_10m_max",
],
"timezone": "auto",
}
r = await client.get(FORECAST_URL, params=params)
r.raise_for_status()
data = r.json()
# Trim daily arrays to requested days (Open‑Meteo returns several days by default)
daily = data.get("daily", {})
trimmed_daily = {
k: (v[:days] if isinstance(v, list) else v) for k, v in daily.items()
}
return {
"ok": True,
"query": {
"requested_location": location,
"resolved": {
"name": place.get("name"),
"country": place.get("country"),
"latitude": lat,
"longitude": lon,
},
"days": days,
},
"current": data.get("current_weather", {}),
"daily": trimmed_daily,
}
@mcp.tool(name="get_forecast")
async def get_forecast_tool(location: str, days: int = 1) -> Dict[str, Any]:
"""MCP-exposed tool that calls the underlying implementation."""
return await _get_forecast_impl(location, days)
@mcp.tool
async def ping() -> str:
"""Health check for the server."""
return "pong"
try:
http_app = mcp.asgi_app()
except AttributeError:
# For older fastmcp versions without asgi helper, raise a clear error
http_app = None
if __name__ == "__main__":
mcp.run()