133 lines
4.7 KiB
Python
133 lines
4.7 KiB
Python
"""Weather MCP server — exposes a get_weather tool backed by wttr.in.
|
|
|
|
Runs over stdio inside the voice container. The agent attaches it via
|
|
MCPServerStdio, so the LLM can call get_weather during a conversation to
|
|
report current conditions and today's forecast for any location.
|
|
|
|
Backed by wttr.in (no API key needed):
|
|
- GET https://wttr.in/{location}?format=j1 -> full JSON weather report
|
|
|
|
The tool returns a short, conversational summary suitable for speech
|
|
(1-2 sentences), not raw data.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
WTTR_BASE = os.environ.get("WTTR_BASE", "https://wttr.in").rstrip("/")
|
|
|
|
mcp = FastMCP("weather")
|
|
|
|
|
|
def _c_to_f(c: str | int) -> int:
|
|
"""Convert Celsius to Fahrenheit (rounds to whole degrees)."""
|
|
return round(int(c) * 9 / 5 + 32)
|
|
|
|
|
|
def _summarize(data: dict, location: str) -> str:
|
|
"""Build a short spoken-style weather summary from wttr.in j1 JSON."""
|
|
current = data.get("current_condition", [{}])[0] or {}
|
|
area = data.get("nearest_area", [{}])[0] or {}
|
|
|
|
city = (area.get("areaName", [{}])[0].get("value") or location).strip()
|
|
temp_f = _c_to_f(current["temp_C"]) if current.get("temp_C") is not None else None
|
|
feels_f = _c_to_f(current["FeelsLikeC"]) if current.get("FeelsLikeC") is not None else None
|
|
desc = (current.get("weatherDesc", [{}])[0].get("value") or "").strip().lower()
|
|
|
|
astronomy = data.get("astronomy", [{}])[0] or {}
|
|
sunrise = astronomy.get("sunrise", "")
|
|
sunset = astronomy.get("sunset", "")
|
|
|
|
today = (data.get("weather") or [None])[0] or {}
|
|
high_f = _c_to_f(today["maxtempC"]) if today.get("maxtempC") is not None else None
|
|
low_f = _c_to_f(today["mintempC"]) if today.get("mintempC") is not None else None
|
|
|
|
parts: list[str] = []
|
|
|
|
if temp_f is not None:
|
|
lead = f"It's {temp_f} degrees and {desc or 'clear'} in {city}"
|
|
if feels_f is not None and feels_f != temp_f:
|
|
lead += f", feels like {feels_f}"
|
|
parts.append(lead + ".")
|
|
|
|
if high_f is not None and low_f is not None:
|
|
parts.append(f"Today's high is {high_f} and the low is {low_f}.")
|
|
|
|
if sunrise and sunset:
|
|
parts.append(f"Sunrise was at {sunrise} and sunset is at {sunset}.")
|
|
|
|
return " ".join(parts) or f"I have no detailed weather for {city} right now."
|
|
|
|
|
|
@mcp.tool()
|
|
async def get_weather(location: str) -> str:
|
|
"""Get the current weather and today's forecast for a location.
|
|
|
|
Use this when the user asks about current conditions, temperature,
|
|
or the forecast for a place. Returns a short conversational summary
|
|
in Fahrenheit suitable for reading aloud (1-2 sentences).
|
|
"""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
resp = await client.get(
|
|
f"{WTTR_BASE}/{location}?format=j1",
|
|
headers={"User-Agent": "curl/8.0"},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
except Exception: # noqa: BLE001
|
|
return "I couldn't get the weather for that location"
|
|
|
|
try:
|
|
summary = _summarize(data, location)
|
|
except Exception: # noqa: BLE001
|
|
return "I couldn't get the weather for that location"
|
|
return summary
|
|
|
|
|
|
@mcp.tool()
|
|
def get_time(location: str = "") -> str:
|
|
"""Get the current date and time.
|
|
|
|
If a location is provided, returns the local time there (approximated
|
|
via wttr.in's timezone data). Otherwise returns UTC time.
|
|
Use this when the user asks what time or day it is.
|
|
"""
|
|
if not location:
|
|
now = datetime.now(timezone.utc)
|
|
return f"It's {now.strftime('%I:%M %p')} on {now.strftime('%A, %B %d')}, UTC."
|
|
|
|
try:
|
|
async def _fetch():
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.get(
|
|
f"{WTTR_BASE}/{location}?format=j1",
|
|
headers={"User-Agent": "curl/8.0"},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
import asyncio
|
|
data = asyncio.get_event_loop().run_until_complete(_fetch())
|
|
current = data.get("current_condition", [{}])[0] or {}
|
|
tz_offset = current.get("localObsDateTime", "")
|
|
area = data.get("nearest_area", [{}])[0] or {}
|
|
city = (area.get("areaName", [{}])[0].get("value") or location).strip()
|
|
|
|
if tz_offset:
|
|
return f"It's {tz_offset.split(' ')[1][:5]} on {tz_offset.split(' ')[0]} in {city}."
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
now = datetime.now(timezone.utc)
|
|
return f"It's {now.strftime('%I:%M %p')} on {now.strftime('%A, %B %d')}, UTC."
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run(transport="stdio")
|