Python으로 구현하는 실시간 날씨 정보 API 서버

ngrok http --url=quick-kit-enhanced.ngrok-free.app 80
라즈베리파이 : 192.168.0.136

linxu / mac

curl -LsSf https://astral.sh/uv/install.sh | sh

Create a new directory for our project

uv init weather
cd weather

Create virtual environment and activate it

uv venv
source .venv/bin/activate

Install dependencies

uv add mcp[cli] httpx

Create our server file

touch weather.py

windows

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

Create a new directory for our project

uv init weather
cd weather

Create virtual environment and activate it

uv venv
.venv\Scripts\activate

Install dependencies

uv add mcp[cli] httpx

Create our server file

new-item weather.py

# 필요한 타입 힌트를 위해 Any를 가져오고, 비동기 HTTP 요청을 위해 httpx를 사용합니다
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP  # FastMCP는 MCP(Micro Command Platform) 서버를 생성하기 위한 클래스입니다

# FastMCP 서버를 초기화합니다. 이 MCP 인스턴스는 "weather"라는 이름을 가집니다.
mcp = FastMCP("weather")

# 상수 설정
NWS_API_BASE = "https://api.weather.gov"  # 미국 국립기상청(NWS) API 기본 URL
USER_AGENT = "weather-app/1.0"  # 요청 시 사용할 사용자 에이전트 헤더

# NWS API에 요청을 보내는 비동기 함수
async def make_nws_request(url: str) -> dict[str, Any] | None:
    """
    NWS API에 요청을 보내고, 결과를 JSON으로 반환합니다.
    실패 시 None을 반환합니다.
    """
    headers = {
        "User-Agent": USER_AGENT,  # 요청을 보낼 때 자신의 앱을 식별하기 위한 헤더
        "Accept": "application/geo+json"  # 응답을 GeoJSON 형식으로 받기 위해 설정
    }
    async with httpx.AsyncClient() as client:
        try:
            # GET 요청을 보내고 응답을 기다립니다
            response = await client.get(url, headers=headers, timeout=30.0)
            # 응답 상태가 정상인지 확인하고, 그렇지 않으면 예외 발생
            response.raise_for_status()
            # JSON 응답을 반환
            return response.json()
        except Exception:
            # 예외 발생 시 None 반환
            return None

# 경고(Alert) 정보를 사람이 읽기 좋게 포맷하는 함수
def format_alert(feature: dict) -> str:
    """
    Alert 데이터를 보기 좋은 문자열 형태로 변환합니다.
    """
    props = feature["properties"]
    return f"""
Event: {props.get('event', 'Unknown')}
Area: {props.get('areaDesc', 'Unknown')}
Severity: {props.get('severity', 'Unknown')}
Description: {props.get('description', 'No description available')}
Instructions: {props.get('instruction', 'No specific instructions provided')}
"""

# 이 MCP 도구는 특정 주(state)에 대한 현재 기상 경고를 조회합니다
@mcp.tool()
async def get_alerts(state: str) -> str:
    """
    미국 주(state)의 기상 경고 정보를 가져옵니다.

    Args:
        state: 두 글자의 미국 주 코드 (예: CA, NY)
    """
    # API URL 생성
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    # API 요청
    data = await make_nws_request(url)

    # 응답이 없거나, features가 없다면 오류 메시지 반환
    if not data or "features" not in data:
        return "Unable to fetch alerts or no alerts found."

    # 알림이 없는 경우
    if not data["features"]:
        return "No active alerts for this state."

    # 각 알림을 보기 좋게 포맷
    alerts = [format_alert(feature) for feature in data["features"]]
    return "\n---\n".join(alerts)

# 이 MCP 도구는 위도/경도를 기반으로 해당 위치의 일기예보를 가져옵니다
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """
    위도, 경도를 기준으로 날씨 예보를 조회합니다.

    Args:
        latitude: 위도
        longitude: 경도
    """
    # 먼저, 해당 좌표에 대한 forecast URL을 얻기 위한 API 호출
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    # 포인트 정보를 못 받으면 오류 메시지 반환
    if not points_data:
        return "Unable to fetch forecast data for this location."

    # forecast URL 추출
    forecast_url = points_data["properties"]["forecast"]
    # 예보 데이터 요청
    forecast_data = await make_nws_request(forecast_url)

    # 예보 데이터를 못 받았을 경우
    if not forecast_data:
        return "Unable to fetch detailed forecast."

    # 예보 정보를 사람이 읽기 쉽게 정리 (최대 5개 시간대)
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:  # 다음 5개의 예보만 출력
        forecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
        forecasts.append(forecast)

    return "\n---\n".join(forecasts)

# 메인 실행 함수 - 실제 서버를 실행할 때 사용
if __name__ == "__main__":
    # SSE(서버 전송 이벤트) 대신 표준 입출력 방식으로 MCP 서버 실행
    # mcp.run(transport='stdio')
    mcp.run(transport='sse')