mcp_server.py

import logging
from typing import Annotated

import httpx
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field


logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("mcp.server.lowlevel.server").setLevel(logging.WARNING)

mcp = FastMCP("weather")

WEATHER_CODES = {
    0: "clear sky",
    1: "mainly clear",
    2: "partly cloudy",
    3: "overcast",
    45: "fog",
    48: "depositing rime fog",
    51: "light drizzle",
    53: "moderate drizzle",
    55: "dense drizzle",
    61: "light rain",
    63: "moderate rain",
    65: "heavy rain",
    71: "light snow",
    73: "moderate snow",
    75: "heavy snow",
    80: "light rain showers",
    81: "moderate rain showers",
    82: "violent rain showers",
    95: "thunderstorm",
    96: "thunderstorm with light hail",
    99: "thunderstorm with heavy hail",
}


class WeatherResult(BaseModel):
    requested_city: str
    requested_country_code: str | None
    requested_region: str | None
    resolved_city: str
    resolved_region: str | None
    country: str
    country_code: str
    latitude: float
    longitude: float
    temperature_c: float
    apparent_temperature_c: float
    condition: str
    humidity_percent: int
    precipitation_mm: float
    wind_speed_kmh: float
    measured_at: str


@mcp.tool()
async def get_weather(
    city: Annotated[
        str,
        Field(description="City name to look up, for example Gdansk, Hel, or Warsaw."),
    ],
    country_code: Annotated[
        str | None,
        Field(
            description=(
                "Two-letter ISO country code. Always provide when known "
                "(for example PL, DE, US) to avoid wrong locations."
            ),
        ),
    ] = None,
    region: Annotated[
        str | None,
        Field(
            description=(
                "State, province, or administrative region used to disambiguate "
                "similarly named cities."
            ),
        ),
    ] = None,
) -> WeatherResult:
    """Fetch current weather for a city from Open-Meteo (no API key required).

    Use this tool before travel or weather recommendations. Returns structured
    JSON with resolved location, temperature, conditions, humidity, precipitation,
    wind, and measurement time. Always pass country_code when the country is known.
    """
    normalized_country_code = country_code.upper() if country_code else None
    geocoding_params = {
        "name": city,
        "count": 10,
        "language": "en",
        "format": "json",
    }
    if normalized_country_code:
        geocoding_params["countryCode"] = normalized_country_code

    async with httpx.AsyncClient(timeout=15.0) as client:
        geocoding_response = await client.get(
            "https://geocoding-api.open-meteo.com/v1/search",
            params=geocoding_params,
        )
        geocoding_response.raise_for_status()
        locations = geocoding_response.json().get("results", [])
        if not locations:
            raise ValueError(f"City not found: {city}")

        candidates = locations
        if normalized_country_code:
            candidates = [
                location
                for location in candidates
                if location.get("country_code", "").upper()
                == normalized_country_code
            ]

        if region:
            normalized_region = region.casefold().strip()
            region_matches = [
                location
                for location in candidates
                if any(
                    normalized_region in str(location.get(field, "")).casefold()
                    for field in ("admin1", "admin2", "admin3", "admin4")
                )
            ]
            if region_matches:
                candidates = region_matches
            elif len(candidates) > 1:
                details = ", ".join(
                    part
                    for part in (city, region, normalized_country_code)
                    if part
                )
                raise ValueError(f"Region did not match an unambiguous city: {details}")

        if not candidates:
            details = ", ".join(
                part
                for part in (city, region, normalized_country_code)
                if part
            )
            raise ValueError(f"No location matched: {details}")

        location = max(
            candidates,
            key=lambda candidate: candidate.get("population") or 0,
        )
        weather_response = await client.get(
            "https://api.open-meteo.com/v1/forecast",
            params={
                "latitude": location["latitude"],
                "longitude": location["longitude"],
                "current": (
                    "temperature_2m,apparent_temperature,"
                    "relative_humidity_2m,precipitation,"
                    "weather_code,wind_speed_10m"
                ),
                "timezone": "auto",
            },
        )
        weather_response.raise_for_status()

    current = weather_response.json()["current"]
    description = WEATHER_CODES.get(
        current["weather_code"],
        f"weather code {current['weather_code']}",
    )
    return WeatherResult(
        requested_city=city,
        requested_country_code=normalized_country_code,
        requested_region=region,
        resolved_city=location["name"],
        resolved_region=location.get("admin1"),
        country=location["country"],
        country_code=location["country_code"],
        latitude=location["latitude"],
        longitude=location["longitude"],
        temperature_c=current["temperature_2m"],
        apparent_temperature_c=current["apparent_temperature"],
        condition=description,
        humidity_percent=current["relative_humidity_2m"],
        precipitation_mm=current["precipitation"],
        wind_speed_kmh=current["wind_speed_10m"],
        measured_at=current["time"],
    )


if __name__ == "__main__":
    mcp.run(transport="stdio")