Python MCP Korea Weather Service

Instructions

Implementation Reference

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

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

  The tool handler function decorated by @mcp.tool. It receives city, gu, dong, nx, ny parameters and delegates execution to the get_forecast_api helper.
  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 handler implementation in get_forecast_api: authenticates with API key, constructs URL for ultra-short forecast API using grid coordinates, fetches data asynchronously, parses items, maps weather codes (SKY, PTY, T1H, REH, VEC, WSD) to Korean descriptions, formats per time slot, handles errors.
  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)

  format_weather_features: utility to convert raw weather dict into formatted Korean strings for sky, rain, temp, humidity, wind.
  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)

  make_api_request: async HTTP client utility using httpx to GET the weather API URL with user-agent and timeout, returns JSON or None on error.
  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