[gemini cli + 커스텀 mcp 만들기] - 3. 기상청 mcp 서버 만들고 적용하기

이전시간에 초단기실황을 알려주는 파이썬 코드를 작성했는데, 이를 mcp 서버에 활용할 수 있도록 하자. 그러기 위해선 몇 가지 추가적인 작업이 또(...) 필요하다.

 

1. API_KEY 코드 수정

일반적으로 API 키는 노출되지 않는다. 그래서 API_KEY를 환경변수로 등록하고 사용할 것이다. cmd를 이용하면 영구적으로 환경변수로 등록할 수 있는데, 본인은 그런 것을 별로 좋아하지 않으므로 vs 코드 터미널에서 임시로만 등록할 것이다. 즉, 현재 터미널 세션에만 임시로 설정하는 방법을 진행할 것이다.

 

파이썬에서는 일반적으로 os 라이브러리를 이용하여 os.environ.get 메서드를 이용하는데 아래와 같이 코드를 작성하고 간단하게 테스트 해 보자.

import os

API_KEY = os.environ.get('KOREA_WEATHER_API_KEY')

if API_KEY:
    print("KOREA_WEATHER_API_KEY 환경 변수가 정상적으로 설정되었습니다.")

if not API_KEY:
    print("오류: KOREA_WEATHER_API_KEY 환경 변수가 설정되지 않았습니다.")

 

 

'당연히' 아무것도 등록하지 않았으므로 되지 않는다. 

 

그래서 임시로 환경변수를 등록하기 위해 아래를 터미널에 입력한다.

$env:KOREA_WEATHER_API_KEY = "<your_api_key>"

 

그리고 다시 테스트 코드를 실행하면 아래와 같이 정상 등록된다. 이 방법은 vs code를 끄지 않는 한 유지된다.

 

 

2. st_forecast 함수를 비동기 함수로 설정

st_forecast는 기상청으로부터 날씨 정보를 응답받는 함수인데, requests.get()은 '동기(Synchronous)' 함수여서 호출되면, API 서버로부터 응답이 올 때 까지 프로그램 전체가 그 자리에서 '멈춰 기다린다.' 이를 블로킹 문제라고 하는데 한 번에 하나의 요청을 처리하면 큰 문제가 되지 않지만, 여러 요청을 동시에 처리해야 하는 상황이라면 심각한 성능 저하를 유발할 수 있다.

 

따라서 async def로 함수를 정의하고, 비동기 I/O 라이브러리를 사용하면 네트워크 요청을 보내놓고 응답을 기다리는 동안 다른 작업을 처리할 수있다. 따라서 서버의 자원을 훨씬 효율적으로 사용하게 되어 더 많은 동시 요청을 빠르게 처리할 수 있다.

 

여기서는 requests 대신 이전에 설치했던 httpx라는 비동기 HTTP 클라이언트 라이브러리를 이용할 것이다.

 

코드를 아래와 같이 수정한다.

async def st_forecast(api_key, url, nx, ny):
    """
    지정된 격자 좌표(nx, ny)의 초단기 실황을 비동기 방식으로 요청하고
    응답을 딕셔너리로 반환합니다.
    """
    date, time = get_datetime()
    parameters = {
        'serviceKey': api_key,
        'numOfRows': 30,
        'pageNo': 1,
        'dataType': 'XML',
        'base_date': date,
        'base_time': time,
        'nx': nx,
        'ny': ny
    }

    # httpx의 비동기 클라이언트를 사용합니다.
    async with httpx.AsyncClient() as client:
        try:
            # client.get은 await 키워드가 필요한 코루틴(coroutine)입니다.
            response = await client.get(url, params=parameters, timeout=10)
            
            # 응답 상태 코드가 200(OK)이 아닐 경우 예외를 발생시킵니다.
            response.raise_for_status()
            
            # xmltodict는 동기 함수이므로 await 없이 그대로 사용합니다.
            return xmltodict.parse(response.text)
        
        except httpx.HTTPStatusError as e:
            # HTTP 상태 코드 오류 (4xx, 5xx 등)
            return {"error": f"API 서버 오류: 상태 코드 {e.response.status_code}"}
        except httpx.RequestError as e:
            # 네트워크 연결 오류, 타임아웃 등
            return {"error": f"API 요청 실패: {e}"}

 

 

 

 

이 함수를 호출하는 get_current_weather 함수 내에서도 await 키워드를 사용해야 한다.

# get_current_weather 함수 내부 (호출 부분 예시)

# ... (grid 좌표 계산 후) ...
my_response = await st_forecast(API_KEY, URL, nx, ny) # 'await' 추가!
parsed_weather = parse_ultra_short_term_weather(my_response)

# ...

 

 

 

 

 

3. get_current_weather 함수 설정

get_current_weather 함수는 @mcp.tool 이라는 데코레이터를 설정하여 이 함수를 mcp tool에 등록한다.

이를 위에 첫 줄에 FastMCP라는 인스턴스를 설정해야 한다.

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my_korea_weather")

 

 

그리고 mcp tool에 등록되는 get_current_weather 함수를 아래와 같이 정의한다. 마찬가지로 비동기로 정의한다.

@mcp.tool()
async def get_current_weather(lat: float, lon: float) -> str:
    """지정된 위도와 경도를 기반으로 현재 날씨 정보를 조회하여 정리된 문자열로 반환합니다."""
    if not API_KEY or API_KEY == '<your_api_key>':
        return "오류: 서버에 API 키가 설정되지 않았습니다. 관리자에게 문의하세요."

    grid = convert_to_grid(lat, lon)
    nx, ny = grid['x'], grid['y']
    
    my_response = await st_forecast(API_KEY, URL, nx, ny) # 'await' 추가!
    parsed_weather = parse_ultra_short_term_weather(my_response)

    if 'error' in parsed_weather:
        return f"날씨 정보를 가져오는 데 실패했습니다: {parsed_weather['error']}"
    
    date_str = parsed_weather.get('발표일자', '00000000')
    time_str = parsed_weather.get('발표시각', '0000')
    
    result = f"""# 현재 날씨 정보 (위도: {lat}, 경도: {lon})
- 기준 시각: {date_str[:4]}년 {date_str[4:6]}월 {date_str[6:]}일 {time_str[:2]}시 {time_str[2:]}분
- 격자 좌표: X={nx}, Y={ny}

## 기상 상태
- 기온: {parsed_weather.get('기온(℃)', 'N/A')}℃
- 습도: {parsed_weather.get('습도(%)', 'N/A')}%
- 강수 형태: {parsed_weather.get('강수형태', 'N/A')}
- 1시간 강수량: {parsed_weather.get('1시간 강수량', 'N/A')}

## 바람 정보
- 풍향: {parsed_weather.get('풍향', 'N/A')}
- 풍속: {parsed_weather.get('풍속', 'N/A')}
- 동서성분: {parsed_weather.get('동서성분', 'N/A')}
- 남북성분: {parsed_weather.get('남북성분', 'N/A')}
"""
    return result

 

 

마지막으로 (테스트) 서버를 실행하기 위해 아래 코드를 작성한다.

if __name__ == "__main__":
    print("MCP 날씨 정보 서버 시작...")
    print("테스트 메시지를 JSON-RPC 형식으로 입력하세요.")
    mcp.run(transport='stdio')

 

 

4. 테스트 실행하기

mcp 서버로 구현하기 전, 터미널에서 테스트를 해 볼 수 있다. 이를 위해 FastMCP 서버와 클라이언트가 대화하는 정해진 절차와 규칙을 따라야 하는데 이 3단계 통신 절차를 언어 서버 프로토콜(Language Server Protocol, LSP)라는 표준 규약을 따른다. 

마치 악수하는 과정과 비유할 수 있다.

 

1단계. initialize 요청

"안녕하세요. 저는 test_client 입니다. 이제부터 통신을 시작하고 싶습니다"

클라이언트의 자기소개 및 기능 협상으로 클라이언트가 서버에 보내는 첫 공식적인 메세지이다. "이제부터 대화를 시작하자"라는 뜻이며 아래와 같은 정보를 서버에 전달한다.

{"jsonrpc": "2.0", "method": "initialize", "params": {"protocolVersion": "1.0", "capabilities": {}, "clientInfo": {"name": "test_client", "version": "0.1.0"}}, "id": 0}

• "method": "initialize": "초기화 절차를 시작하겠습니다."
• "params": {"clientInfo": ...}: "제 이름은 'test_client'이고 버전은 '0.1.0'입니다." (클라이언트의 신원 정보)
• "params": {"protocolVersion": "1.0"}: "저는 '1.0' 버전의 통신 규칙으로 대화하고 싶습니다." (언어 버전 맞추기)
• " params": {"capabilities": {}}: "제가 특별히 지원하는 고급 기능은 이런 것들이 있습니다." (우리의 경우 비어 있음)

 

이러한 요청애 대해, 서버는 아래와 같은 응답을 한다.

{"jsonrpc":"2.0","id":0,"result":{"protocolVersion":"2025-06-18","capabilities":{"experimental":{},"prompts":{"listChanged":false},"resources":{"subscribe":false,"listChanged":false},"tools":{"listChanged":false}},"serverInfo":{"name":"my_korea_weather","version":"1.12.0"}}}

"네, 반갑습니다. 저는 'my_korea_weather' 서버입니다. 제가 지원하는 기능은 다음과 같습니다."

 

result에 서버의 기능 목록(capabilities)과 우리가 등록한 도구(tools) 목록이 담겨서 돌아오며. 이 응답을 통해 클라이언트는 이 서버가 get_current_weather라는 도구를 가지고 있다는 사실을 알게 된다.

 

 

 

2단계. notifications/initialized 알림

"네 당신의 기능들을 잘 확인했습니다. 이제 저도 모든 준비가 끝났습니다."

{"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}

 

서버에 알려주는 일종의 '알림(Notification)'으로 응답을 받기 위한 id 필드가 없다. 서버는 이 알림을 받아야만 '아 클라이언트가 내 말을 알아들었고, 이제 정말로 대화를 나눌 준비가 되었구나' 라고 판단하여 자신의 모든 기능을 활성화한다.

이 알림을 보내기 전 까지 서버는 계속 "아직 초기화 중..." 상태로 대기하고 있다.

 

 

 

3단계. tools/call 요청

"좋습니다. 그럼 이제 본론으로 들어가서, 'get_current_weather' 도구를 사용해 주세요."

{"jsonrpc": "2.0", "method": "tools/call", "params": {"name": "get_current_weather", "arguments": {"lat": 37.5665, "lon": 126.9780}}, "id": 1}

 

모든 절차가 끝나고, 드디어 실제 용건을 말하는 단계이다. 이제 서버는 모든 준비가 끝난 상태이므로 이 요청을 받아 이름에 맞는 함수(get _current_weather)를 찾아 실행하고, 결과를 담아 최종적으로 응답해 준다.

• " method": "tools/call": "도구를 사용하겠습니다." (서버는 이 요청을 받고 도구 관리자를 호출합니다.)
• " params": {"name": "get_current_weather", ...}: "사용할 도구의 이름은 get_current_weather입니다."
• " params": {"arguments": ...}: "이 도구에 전달할 인자는 위도와 경도입니다."

요약하자면 아래와 같다.

단계	역할	클라이언트가 하는 말
1	소개 및 협상	initialize: "안녕하세요, 통신 시작할까요?"
2	준비 완료 확인	notifications/initialized: "네, 당신 말 잘 들었어요. 저도 준비 끝!"
3	실제 업무 요청	tools/call: "그럼 이 일 좀 해주세요."

 

 

<출력결과>

{"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"# 현재 날씨 정보 (위도: 37.5665, 경도: 126.978)\n- 기준 시각: 2025년 08월 07일 00시 00분\n- 격자 좌표:zL0LuSKAwnSJZG3ziyKOfKJHqVls6zCcvqG%2FoVNnmmUs68%2F2A%3D%3D&numOfRows=30&pageNo=1&dataType=XML&base_date=20250807&base_time=0053&nx=60&ny=127   _client.py:1740 X=60, Y=127\n\n## 기상 상태\n- 기온: 26.0℃\n- 습도: 90.0%\n- 강수 형태: 없음\n- 1시간 강수량: 강수없음\n\n## 바람 정보\n- 풍향: 남남서\n- 풍속: 2.6m/s, 바람이 약하게 느 
껴집니다.\n- 동서성분: 동풍 1.4m/s\n- 남북성분: 북풍 2.2m/s\n"}],"structuredContent":{"result":"# 현재 날씨 정보 (위도: 37.5665, 경도: 126.978)\n- 기준 시각: 2025년 08월  X=60, Y=127\n\n## 기상 상태\n- 기온: 26.0℃\n- 습도: 90.0%\n- 강수 형태: 없음\n- 1시간 강수량: 강수없음\n\n## 바람 정보\n- 풍향: 남남서\n- 풍속: 2.6m/s, 바람이
07일 00시 00분\n- 격자 좌표: X=60, Y=127\n\n## 기상 상태\n- 기온: 26.0℃\n- 습도: 90.0%\n- 강수 형태: 없음\n- 1시간 강수량: 강수없음\n\n## 바람 정보\n- 풍향: 남남서\n- 풍 25년 08월 07일 00시 00분\n- 격자 좌표: X=60, Y=127\n\n## 기상 상태\n- 기온: 26.0℃\n- 습도: 90.0%\n- 강수 형태: 없음\n- 1시간 강수량: 강수없음\n\n## 바람 정보\n
속: 2.6m/s, 바람이 약하게 느껴집니다.\n- 동서성분: 동풍 1.4m/s\n- 남북성분: 북풍 2.2m/s\n"},"isError":false}}

 

 

5. gemini cli에 적용하기

이제 만든 mcp 서버를 gemini cli에 적용해 보자. 일단 "오늘 대전광역시 중구 현재 날씨를 알려줘" 이렇게 물어보자. 

예상대로 'GoogleSearch'를 사용했다. 

 

 

이번엔 우리가 만든 mcp 커스텀 서버를 적용해 보자.

{
  "selectedAuthType": "oauth-personal",
  "mcpServers": {
    "my_korea_weather": {
      "command": "uv",
      "args": [
        "--directory",
        "E:\\my_mcp\\my_weather",
        "run",
        "weather_server.py"
      ],
      "env": {
           "KOREA_WEATHER_API_KEY": "<your_api_key>"
      }
    }
  }
}

 

 

my_korea_weather mcp 서버를 사용해 달라는 요청을 추가하여 질문해 보았다.

먼저 직접 접근이 불가능하다. 이 mcp 서버를 사용하려면 위도와 경도를 입력해야 하는데, 대전광역시 중구에 대한 위도와 경도를 웹 검색으로 찾고 이를 커스텀 mcp 서버에 적용하여 출력하는 모습을 나타낸다.

 

 

잘된다! 다음 시간에는 mcp 서버의 기능을 좀 더 끌어올려줄, 리소스와 프롬프트를 추가로 만들어 보자!