Python MCP Korea Weather Service

Instructions

Implementation Reference

• src/server.py:55-58 (registration)

  MCP tool registration for 'get_forecast' with name and description.
  @mcp.tool( name="get_forecast", description="한국 기상청의 초단기예보 API를 호출하여 특정 지역의 날씨 예보 정보를 제공합니다. 사용자가 입력한 지역 정보와 격자 좌표를 바탕으로 현재 시점에서의 기상 정보를 조회합니다. 이 도구는 온도, 강수량, 하늘상태, 습도, 풍향, 풍속 등 상세한 기상 정보를 포함하며, 6시간 이내의 단기 예보를 제공합니다." )
• src/server.py:59-70 (handler)

  Thin MCP handler for get_forecast tool that delegates to get_forecast_api.
  async def get_forecast(city: str, gu: str, dong: str, nx: int, ny: int) -> str: """Get weather forecast for a location. Args: city: City Name (e.g. 서울특별시) gu: Gu Name (e.g. 서초구) dong: Dong Name (e.g. 양재1동) nx: Grid X coordinate ny: Grid Y coordinate """ return await get_forecast_api(city, gu, dong, nx, ny)
• src/api.py:67-202 (handler)

  Core implementation of get_forecast: fetches data from Korean weather API, processes categories like sky, rain, temp, etc., and formats the forecast.
  async def get_forecast_api(city: str, gu: str, dong: str, nx: float, ny: float) -> str: try: serviceKey = os.environ.get("KO_WEATHER_API_KEY") if not serviceKey: raise ValueError("KO_WEATHER_API_KEY 환경변수가 설정되지 않았습니다.") base_date = datetime.now().strftime("%Y%m%d") # 발표 일자 base_time = datetime.now().strftime("%H%M") # 발표 시간 # nx = '62' # 예보 지점 x좌표 # ny = '123' # 예보 지점 y좌표 # 알고 싶은 시간 input_d = datetime.strptime(base_date + base_time, "%Y%m%d%H%M" ) # 실제 입력 시간 input_d = datetime.strptime(base_date + base_time, "%Y%m%d%H%M" ) - timedelta(hours=1) input_datetime = datetime.strftime(input_d, "%Y%m%d%H%M") input_date = input_datetime[:-4] input_time = input_datetime[-4:] # url url = f"http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getUltraSrtFcst?serviceKey={serviceKey}&numOfRows=60&pageNo=1&dataType=json&base_date={input_date}&base_time={input_time}&nx={nx}&ny={ny}" data = await make_api_request(url) if not data: raise ValueError("API 요청 결과가 비어있습니다.") if 'response' not in data: raise KeyError("API 응답에 'response' 항목이 없습니다.") if 'body' not in data['response']: raise KeyError("API 응답에 'body' 항목이 없습니다.") if 'items' not in data['response']['body']: raise KeyError("API 응답에 'items' 항목이 없습니다.") if 'item' not in data['response']['body']['items']: raise KeyError("API 응답에 'item' 항목이 없습니다.") res = data['response']['body']['items']['item'] if not res: return [f"{city} {gu} {dong} 지역의 날씨 정보를 찾을 수 없습니다."] informations = dict() for items in res: try: cate = items['category'] fcstTime = items['fcstTime'] fcstValue = items['fcstValue'] if fcstTime not in informations.keys(): informations[fcstTime] = dict() informations[fcstTime][cate] = fcstValue except KeyError as e: print(f"날씨 데이터 항목 누락: {e}") continue if not informations: return [f"{city} {gu} {dong} 지역의 날씨 정보를 처리할 수 없습니다."] forecasts = [] for key, val in zip(informations.keys(), informations.values()): features = dict() try: template = f"""{base_date[:4]}년 {base_date[4:6]}월 {base_date[-2:]}일 {key[:2]}시 {key[2:]}분 {city} {gu} {dong} 지역의 날씨는 """ # 하늘 상태 if 'SKY' in val and val['SKY']: try: sky_temp = sky_code[int(val['SKY'])] features['sky'] = sky_temp except (ValueError, KeyError): print(f"하늘 상태 코드 처리 오류: {val['SKY']}") # 강수 형태 if 'PTY' in val and val['PTY']: try: pty_temp = pyt_code[int(val['PTY'])] features['rain'] = pty_temp template += pty_temp # 강수 있는 경우 if 'RN1' in val and val['RN1'] != '강수없음': rn1_temp = val['RN1'] # template += f"시간당 {rn1_temp}mm " features['rain'] = rn1_temp except (ValueError, KeyError): print(f"강수 형태 코드 처리 오류: {val['PTY']}") # 기온 if 'T1H' in val and val['T1H']: try: t1h_temp = float(val['T1H']) # template += f" 기온 {t1h_temp}℃ " features['temp'] = t1h_temp except ValueError: print(f"기온 값 처리 오류: {val['T1H']}") # 습도 if 'REH' in val and val['REH']: try: reh_temp = float(val['REH']) # template += f"습도 {reh_temp}% " features['humidity'] = reh_temp except ValueError: print(f"습도 값 처리 오류: {val['REH']}") # 풍향/ 풍속 if 'VEC' in val and val['VEC'] and 'WSD' in val and val['WSD']: try: vec_temp = deg_to_dir(float(val['VEC'])) wsd_temp = val['WSD'] # template += f"풍속 {vec_temp} 방향 {wsd_temp}m/s" features['wind_direction'] = vec_temp features['wind_speed'] = wsd_temp except ValueError: print(f"풍향/풍속 값 처리 오류: VEC={val.get('VEC')}, WSD={val.get('WSD')}") forecasts.append(template + format_weather_features(features)) except Exception as e: print(f"날씨 정보 처리 중 오류 발생: {e}") continue if not forecasts: return f"{city} {gu} {dong} 지역의 날씨 정보를 생성할 수 없습니다." print(forecasts) return "\n---\n".join(forecasts) except Exception as e: print(f"날씨 API 요청 중 오류 발생: {e}") return [f"날씨 정보를 가져오는 중 오류가 발생했습니다: {str(e)}"]
• src/api.py:35-52 (helper)

  Helper function to format weather features into readable strings.
  def format_weather_features(features: dict) -> str: formatted_features = [] for key, value in features.items(): if key == 'sky': formatted_features.append(f"하늘 상태: {value}") elif key == 'rain': formatted_features.append(f"강수 형태: {value}") elif key == 'rain_amount': formatted_features.append(f"강수량: {value}mm") elif key == 'temp': formatted_features.append(f"기온: {value}℃") elif key == 'humidity': formatted_features.append(f"습도: {value}%") elif key == 'wind_direction': formatted_features.append(f"풍향: {value}") elif key == 'wind_speed': formatted_features.append(f"풍속: {value}m/s") return "\n".join(formatted_features)
• src/utils.py:6-18 (helper)

  Utility to make async HTTP requests to the weather API.
  async def make_api_request(url: str) -> dict[str, Any] | None: """Make a request to the API with proper error handling.""" headers = { "User-Agent": USER_AGENT, "Accept": "application/json" } async with httpx.AsyncClient() as client: try: response = await client.get(url, headers=headers, timeout=30.0) response.raise_for_status() return response.json() except Exception: return None