[2. 크롤링 & Markdown 출력 MCP 개발] 33m2 숙소 크롤링 MCP 서버 개발 일지

안녕하세요,

제가 원하는 지역의 숙소 데이터를 직접 뽑아보고 싶어서 시작하게 된 33m2 크롤링 MCP 서버 개발지입니다.

저는 33m2의 전국 단기 임대 숙소를 정보를 기반으로 숙소 입지, 가격 분석 등 데이터 기반 인사이트가 필요해서 직접 MCP 서버를 개발하였습니다.
(는 사실 핑계고 MCP 서버를 간단하게 입문해보고 싶었습니다)


지난 1편에선 POST 요청 / 기본 폼 / 쿠키 / 헤더 등을 맞춰 크롤링해주는 MCP 서버를 개발했었고,


이번 2편에서는 좀 더 디테일하게 데이터를 수집해오고 -> MCP 사용자에게 Markdown 형식으로 보여주는 MCP 서버를 개발하는 내용을 담게 됐습니다.
(중간 중간 요약으로 되어있는 부분만 보셔도 됩니다)


목차


1편 요약

  • 33m2 크롤링 MCP 서버.py

from typing import Any
import os
import json
from urllib.parse import quote
import httpx
from mcp.server.fastmcp import FastMCP
import asyncio
from datetime import datetime

mcp = FastMCP("33m2")

BASE_URL = 'https://33m2.co.kr/app/room/search'
DATA_DIR = 'data'
DEFAULT_COOKIE = {
    'SESSION': 'MmYwMjYxOTItYWE5ZC00ZTI2LWIxMzctNTk2MjUzZjU4MjUy',
}
HEADERS = {
    'accept': '*/*',
    'content-type': 'application/x-www-form-urlencoded; charset=UTF-8',
    'origin': 'https://33m2.co.kr',
    'referer': 'https://33m2.co.kr',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
    'x-requested-with': 'XMLHttpRequest',
}

async def crawl_listings(keyword: str) -> dict[str, Any]:
    """
    33m2에서 주어진 키워드로 숙소 정보를 크롤링하여 JSON으로 반환합니다.
    """
    encoded = quote(keyword, safe='')
    data = {
        'theme_type': '',
        'keyword': keyword,
        'room_cnt': '',
        'property_type': '',
        'animal': 'false',
        'subway': 'false',
        'longterm_discount': 'false',
        'early_discount': 'false',
        'parking_place': 'false',
        'start_date': '',
        'end_date': '',
        'week': '',
        'min_using_fee': '0',
        'max_using_fee': '1000000',
        'sort': 'popular',
        'now_page': '1',
        'map_level': '7',
        'by_location': 'true',
        'north_east_lng': '127.0027187813919',
        'north_east_lat': '37.62752588259554',
        'south_west_lng': '126.92651812946441',
        'south_west_lat': '37.51923950893502',
        'itemcount': '1000',
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(
            BASE_URL,
            cookies=DEFAULT_COOKIE,
            headers={**HEADERS, 'referer': f'https://33m2.co.kr/webpc/search/map?keyword={encoded}'},
            data=data,
            timeout=30.0
        )
        response.raise_for_status()
        return response.json()

@mcp.tool()
async def fetch_and_save(keyword: str) -> str:
    """
    지정한 키워드로 33m2 데이터를 크롤링하고 'data/33m2_{keyword}_{date}.json'에 저장합니다.
    """
    os.makedirs(DATA_DIR, exist_ok=True)
    result = await crawl_listings(keyword)
    today = datetime.now().strftime('%Y-%m-%d')
    file_path = os.path.join(DATA_DIR, f"33m2_{keyword}_{today}.json")
    with open(file_path, 'w', encoding='utf-8') as f:
        json.dump(result, f, ensure_ascii=False, indent=2)
    return f"Data saved to {file_path}."

def test(keyword: str) -> None:
    return asyncio.run(fetch_and_save(keyword))

if __name__ == '__main__':
    print(test("광화문"))

    mcp.run(transport='stdio')

문제점 발견!

위 코드의 POST 폼 용도로 만들어진 data 변수를 살펴보면, 위도와 경도가 특정 위치로 고정되어 있는 것을 알 수 있습니다.

# 위도 경도 고정
        'north_east_lng': '127.0027187813919',
        'north_east_lat': '37.62752588259554',
        'south_west_lng': '126.92651812946441',
        'south_west_lat': '37.51923950893502',

본 위도, 경도는 '경복궁'이라는 키워드로 검색했을 때 정해진 값입니다.

따라서, (그럴일은없지만) 제주도의 경복궁을 검색하면 서울의 경복궁 숙소만 출력된다는 문제가 있습니다. 따라서,

해결 방법 ①: POST 폼에서 위도·경도 변수 제거 -> 또다른 문제 발생!

가장 먼저 위도와 경도 값을 아예 삭제해서 크롤링을 해봤습니다.

async def crawl_listings(keyword: str) -> dict[str, Any]:
    """
    33m2에서 주어진 키워드로 숙소 정보를 크롤링하여 JSON으로 반환합니다.
    """
    encoded = quote(keyword, safe='')

    data = {
        'theme_type': '',
        'keyword': keyword,
        'room_cnt': '',
        'property_type': '',
        'animal': 'false',
        'subway': 'false',
        'longterm_discount': 'false',
        'early_discount': 'false',
        'parking_place': 'false',
        'start_date': '',
        'end_date': '',
        'week': '',
        'min_using_fee': '0',
        'max_using_fee': '1000000',
        'sort': 'popular',
        'now_page': '1',
        'map_level': '7',
        'by_location': 'true',
        # 'north_east_lng': '127.0027187813919',
        # 'north_east_lat': '37.62752588259554',
        # 'south_west_lng': '126.92651812946441',
        # 'south_west_lat': '37.51923950893502',
        'itemcount': '1000',
    }

위도, 경도 값을 제거하고 '북촌'이라는 키워드로 숙소 정보를 크롤링해봤더니 다음과 같이 데이터가 수집됐습니다.

{
  "airbridge": {
    "event_name": "airbridge.ecommerce.searchResults.viewed",
    "action": "북촌",
    "label": ""
  },
  "bucket_name": "samsamm2",
  "error_code": 0,
  "list": [
    {
      "rid": 38692,
      "room_name": "삼청동 유일 펜트하우스",
      ...(중략)
      "lat": 37.5820697,
      "lng": 126.983763
    },
    {
      "rid": 27564,
      "room_name": "경복궁,북촌 평지3룸",
      ...(중략)
      "lng": 126.9818217
    },
    {
      "rid": 49341,
      "room_name": "❤️조천 바다앞❤",
      "state": "제주특별자치도",
      "province": "제주시",
      "town": "조천읍",
      "pic_main": "room/2fccd79c-761e-4b45-ad09-550593fde1ee.jpg",
      "addr_lot": "제주특별자치도 제주시 조천읍 북촌리 360 대상다려마을연립",
      "addr_street": "제주특별자치도 제주시 조천읍 일주동로 1574",
      "using_fee": 219000,
      "pyeong_size": 12,
      "room_cnt": 1,
      "bathroom_cnt": 1,
      "cookroom_cnt": 1,
      "sittingroom_cnt": 0,
      "reco_type_1": false,
      "reco_type_2": false,
      "longterm_discount_per": 5,
      "early_discount_per": 0,
      "is_new": false,
      "is_super_host": false,
      "lat": 33.5517166,
      "lng": 126.7022009
    },
    {
      "rid": 24536,
      "room_name": "북촌 삼청3룸 주차가능",
      ...(중략)
    },
    ...(중략)
}

서울의 북촌에 있는 숙소를 검색하려 했지만, 제주도 조천읍 북촌리에 있는 숙소가 함께 검색되는 문제가 발생한 것입니다!!!

본 MCP 서버는 개발 취지상 새로 오픈할 숙소의 적절한 금액대 등을 추천받으려고 설계됐기 때문에, 다른 지역에 있는 숙소 정보는 크롤링해서는 안됩니다.

따라서, 숙소를 검색할 키워드의 위도와 경도를 직접 계산해서 form에 주입하는 방식으로 방법을 틀었습니다.

요약

'위치' 기반 데이터를 가져오는 크롤러인데, 데이터를 가져올 때 위도와 경도 정보가 들어감.
-> 위도와 경도를 아예 삭제하니, '북촌'이라는 키워드로 숙소를 검색했을 때 서울의 북촌 뿐만 아니라 제주도 북촌까지 검색됨
-> 키워드별 위도와 경도를 직접 계산해서 크롤링하기로 결정함

해결 방법 ②: 키워드별 위도/경도 직접 계산 & 폼에 주입

geopy의 Nominatim을 사용하면 위도, 경도를 매우 쉽게 계산할 수 있었습니다.

from typing import Any, Tuple
import os
import json
from urllib.parse import quote
import httpx
from mcp.server.fastmcp import FastMCP
import asyncio
from datetime import datetime
from geopy.geocoders import Nominatim
from geopy.exc import GeocoderTimedOut
import time
from geopy.distance import geodesic

mcp = FastMCP("33m2")

BASE_URL = 'https://33m2.co.kr/app/room/search'
DATA_DIR = 'data'
DEFAULT_COOKIE = {
    'SESSION': 'MmYwMjYxOTItYWE5ZC00ZTI2LWIxMzctNTk2MjUzZjU4MjUy',
}
HEADERS = {
    'accept': '*/*',
    'content-type': 'application/x-www-form-urlencoded; charset=UTF-8',
    'origin': 'https://33m2.co.kr',
    'referer': 'https://33m2.co.kr',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
    'x-requested-with': 'XMLHttpRequest',
}

def get_keyword_coordinates(keyword: str) -> Tuple[float, float]:
    """
    키워드의 위도와 경도를 반환합니다.
    """
    geolocator = Nominatim(user_agent="33m2_crawler")
    try:
        location = geolocator.geocode(keyword)
        if location:
            return location.latitude, location.longitude
        raise ValueError(f"Could not find coordinates for keyword: {keyword}")
    except Exception as e:
        raise ValueError(f"Error getting coordinates for keyword {keyword}: {str(e)}")

def calculate_bounding_box(lat: float, lng: float, radius_km: float = 5.0) -> dict:
    """
    주어진 위도, 경도를 중심으로 radius_km 반경의 사각형 영역의 좌표를 계산합니다.
    """
    # 북동쪽 좌표 계산
    north_east = geodesic(kilometers=radius_km).destination((lat, lng), 45)
    # 남서쪽 좌표 계산
    south_west = geodesic(kilometers=radius_km).destination((lat, lng), 225)

    return {
        'north_east_lat': north_east.latitude,
        'north_east_lng': north_east.longitude,
        'south_west_lat': south_west.latitude,
        'south_west_lng': south_west.longitude
    }

async def crawl_listings(keyword: str) -> dict[str, Any]:
    """
    33m2에서 주어진 키워드로 숙소 정보를 크롤링하여 JSON으로 반환합니다.
    """
    # 키워드의 위도, 경도 구하기
    lat, lng = get_keyword_coordinates(keyword)

    # 5km 반경의 사각형 영역 계산
    bounding_box = calculate_bounding_box(lat, lng)

    encoded = quote(keyword, safe='')

    data = {
        'theme_type': '',
        'keyword': keyword,
        'room_cnt': '',
        'property_type': '',
        'animal': 'false',
        'subway': 'false',
        'longterm_discount': 'false',
        'early_discount': 'false',
        'parking_place': 'false',
        'start_date': '',
        'end_date': '',
        'week': '',
        'min_using_fee': '0',
        'max_using_fee': '1000000',
        'sort': 'popular',
        'now_page': '1',
        'map_level': '7',
        'by_location': 'true',
        'north_east_lng': str(bounding_box['north_east_lng']),
        'north_east_lat': str(bounding_box['north_east_lat']),
        'south_west_lng': str(bounding_box['south_west_lng']),
        'south_west_lat': str(bounding_box['south_west_lat']),
        'itemcount': '1000',
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(
            BASE_URL,
            cookies=DEFAULT_COOKIE,
            headers={**HEADERS, 'referer': f'https://33m2.co.kr/webpc/search/map?keyword={encoded}'},
            data=data,
            timeout=30.0
        )
        response.raise_for_status()
        return response.json()

@mcp.tool()
async def fetch_and_save(keyword: str) -> str:
    """
    지정한 키워드로 33m2 데이터를 크롤링하고 'data/33m2_{keyword}_{date}.json'에 저장합니다.
    """
    os.makedirs(DATA_DIR, exist_ok=True)
    result = await crawl_listings(keyword)
    today = datetime.now().strftime('%Y-%m-%d')
    file_path = os.path.join(DATA_DIR, f"33m2_{keyword}_{today}.json")

    with open(file_path, 'w', encoding='utf-8') as f:
        json.dump(result, f, ensure_ascii=False, indent=2)

    return f"Data saved to {file_path}."

def test(keyword: str) -> None:
    return asyncio.run(fetch_and_save(keyword))

if __name__ == '__main__':
    print(test("북촌"))

    mcp.run(transport='stdio')

이번에도 마찬가지로 '북촌'이라는 키워드로 숙소 정보를 크롤링해봤고, 마침내 원하는 데이터가 수집됐습니다.

{
  "airbridge": {
    "event_name": "airbridge.ecommerce.searchResults.viewed",
    "action": "북촌",
    "label": ""
  },
  "bucket_name": "samsamm2",
  "error_code": 0,
  "list": [
    {
      "rid": 38692,
      "room_name": "삼청동 유일 펜트하우스",
    ...(중략)
    },
    {
      "rid": 27564,
      "room_name": "경복궁,북촌 평지3룸",
    ...(중략)
      "lat": 37.58078,
      "lng": 126.9818
    },
    {
      "rid": 24536,
      "room_name": "북촌 삼청3룸 주차가능",
    ...(중략)
      "lat": 37.58549,
      "lng": 126.98407
    },
    ...(중략)
}

다만, Nominatim이 한국의 특정 지역에 대한 위도, 경도 검색 기능이 엉망이어서 그냥 Google Geocoding API를 사용해 위도와 경도를 검색했습니다.

이 코드는 아래에서 다루도록 하겠습니다.

요약

geopy의 Nominatim을 사용해 숙소를 검색할 키워드의 위도와 경도를 검색함
-> return된 위도와 경도를 기반으로 5km 반경의 영역을 계산한 뒤
-> POST 요청 값에 집어 넣음
-> 다만, Nominatim 자체가 한국 특정 지역의 위도와 경도를 이해하지 못하기 때문에 최종적으로는 Google Geocoding API를 사용함

최신 숙소 정보 Markdown 표로 출력하기

# Google Geocoding API 사용한 버전
from typing import Any, Tuple
import os
import json
from urllib.parse import quote
import httpx
from mcp.server.fastmcp import FastMCP
import asyncio
from datetime import datetime
import time
from geopy.distance import geodesic

mcp = FastMCP("33m2")

BASE_URL = 'https://33m2.co.kr/app/room/search'
DATA_DIR = 'data'
DEFAULT_COOKIE = {
    'SESSION': 'MmYwMjYxOTItYWE5ZC00ZTI2LWIxMzctNTk2MjUzZjU4MjUy',
}
HEADERS = {
    'accept': '*/*',
    'content-type': 'application/x-www-form-urlencoded; charset=UTF-8',
    'origin': 'https://33m2.co.kr',
    'referer': 'https://33m2.co.kr',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
    'x-requested-with': 'XMLHttpRequest',
}

# Google Geocoding API 설정
GOOGLE_GEOCODING_API_KEY = os.getenv('GOOGLE_GEOCODING_API_KEY', 'None')
GOOGLE_GEOCODING_BASE_URL = 'https://maps.googleapis.com/maps/api/geocode/json'

async def get_keyword_coordinates(keyword: str) -> Tuple[float, float]:
    """
    Google Geocoding API를 사용하여 키워드의 위도와 경도를 반환합니다.
    """
    if not GOOGLE_GEOCODING_API_KEY:
        raise ValueError("Google Geocoding API 키가 설정되지 않았습니다. 환경 변수 GOOGLE_GEOCODING_API_KEY를 설정해주세요.")

    params = {
        'address': keyword,
        'key': GOOGLE_GEOCODING_API_KEY,
        'language': 'ko'  # 한국어 결과를 받기 위해 설정
    }

    async with httpx.AsyncClient() as client:
        response = await client.get(GOOGLE_GEOCODING_BASE_URL, params=params)
        response.raise_for_status()
        data = response.json()

        if data['status'] != 'OK':
            raise ValueError(f"Geocoding API 오류: {data['status']}")

        location = data['results'][0]['geometry']['location']
        return location['lat'], location['lng']

def calculate_bounding_box(lat: float, lng: float, radius_km: float = 5.0) -> dict:
    """
    주어진 위도, 경도를 중심으로 radius_km 반경의 사각형 영역의 좌표를 계산합니다.
    """
    north_east = geodesic(kilometers=radius_km).destination((lat, lng), 45)
    south_west = geodesic(kilometers=radius_km).destination((lat, lng), 225)
    return {
        'north_east_lat': north_east.latitude,
        'north_east_lng': north_east.longitude,
        'south_west_lat': south_west.latitude,
        'south_west_lng': south_west.longitude
    }

async def crawl_listings(keyword: str) -> dict[str, Any]:
    """
    33m2에서 주어진 키워드로 숙소 정보를 크롤링하여 JSON으로 반환합니다.
    """
    lat, lng = await get_keyword_coordinates(keyword)
    # print(lat, lng)
    bounding_box = calculate_bounding_box(lat, lng)
    encoded = quote(keyword, safe='')
    data = {
        'theme_type': '',
        'keyword': keyword,
        'room_cnt': '',
        'property_type': '',
        'animal': 'false',
        'subway': 'false',
        'longterm_discount': 'false',
        'early_discount': 'false',
        'parking_place': 'false',
        'start_date': '',
        'end_date': '',
        'week': '',
        'min_using_fee': '0',
        'max_using_fee': '1000000',
        'sort': 'popular',
        'now_page': '1',
        'map_level': '7',
        'by_location': 'true',
        'north_east_lng': str(bounding_box['north_east_lng']),
        'north_east_lat': str(bounding_box['north_east_lat']),
        'south_west_lng': str(bounding_box['south_west_lng']),
        'south_west_lat': str(bounding_box['south_west_lat']),
        'itemcount': '1000',
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(
            BASE_URL,
            cookies=DEFAULT_COOKIE,
            headers={**HEADERS, 'referer': f'https://33m2.co.kr/webpc/search/map?keyword={encoded}'},
            data=data,
            timeout=30.0
        )
        response.raise_for_status()
        return response.json()

@mcp.tool()
async def fetch_and_save(keyword: str) -> str:
    """
    지정한 키워드로 33m2 데이터를 크롤링하고 'data/33m2_{keyword}_{date}.json'에 저장합니다.
    """
    os.makedirs(DATA_DIR, exist_ok=True)
    result = await crawl_listings(keyword)
    today = datetime.now().strftime('%Y-%m-%d')
    file_path = os.path.join(DATA_DIR, f"33m2_{keyword}_{today}.json")

    with open(file_path, 'w', encoding='utf-8') as f:
        json.dump(result, f, ensure_ascii=False, indent=2)

    return f"Data saved to {file_path}."

@mcp.tool()
async def display_listings(keyword: str) -> str:
    """
    지정한 키워드의 오늘자 33m2 데이터를 마크다운 표 형식으로 출력합니다.
    오늘자 파일이 없으면 fetch_and_save를 호출한 후 표를 출력합니다.
    """
    # 오늘 날짜를 YYYY-MM-DD 형식으로 가져오기
    today = datetime.now().strftime('%Y-%m-%d')
    file_path = os.path.join(DATA_DIR, f"33m2_{keyword}_{today}.json")

    # 오늘자 파일이 없으면 fetch_and_save 호출
    if not os.path.exists(file_path):
        await fetch_and_save(keyword)

    # JSON 파일 읽기
    with open(file_path, 'r', encoding='utf-8') as f:
        data = json.load(f)

    # 표 헤더 생성
    headers = [
        "숙소 이름", "도로명 주소", "금액", "평 사이즈", "방 수",
        "화장실 수", "거실 수", "주방 수", "33m2 추천",
        "인기 호스트", "신규 숙소", "장기 할인", "빠른 입주 할인"
    ]

    # 표 헤더 행 생성
    markdown = "| " + " | ".join(headers) + " |\n"
    markdown += "| " + " | ".join(["---"] * len(headers)) + " |\n"

    # 데이터 행 생성
    for item in data.get('list', []):
        row = [
            item.get('room_name', ''),
            item.get('addr_street', ''),
            f"{item.get('using_fee', 0):,}원",
            f"{item.get('pyeong_size', 0)}평",
            str(item.get('room_cnt', 0)),
            str(item.get('bathroom_cnt', 0)),
            str(item.get('sittingroom_cnt', 0)),
            str(item.get('cookroom_cnt', 0)),
            "✓" if item.get('reco_type_1') else "X",
            "✓" if item.get('is_super_host') else "X",
            "✓" if item.get('is_new') else "X",
            f"{item.get('longterm_discount_per', 0)}%" if item.get('longterm_discount_per') else "-",
            f"{item.get('early_discount_per', 0)}%" if item.get('early_discount_per') else "-"
        ]
        markdown += "| " + " | ".join(row) + " |\n"

    # 마크다운 파일로 저장
    md_file_path = os.path.join(DATA_DIR, f"33m2_{keyword}_{today}.md")
    with open(md_file_path, 'w', encoding='utf-8') as f:
        f.write(markdown)

    return markdown

def test(keyword: str) -> None:
    FS = asyncio.run(fetch_and_save(keyword))
    print(FS)
    MD = asyncio.run(display_listings(keyword))
    print(MD)

if __name__ == '__main__':
    test("북촌")

    mcp.run(transport='stdio')

본 코드를 사용하시기 위해선 GOOGLE_GEOCODING_API_KEY를 환경 변수로 설정하셔야 합니다.

구글 지오코딩 API 키를 할당받으시고, 시스템 환경 변수에 'GOOGLE_GEOCODING_API_KEY' 이름으로 키 값을 저장해주시기 바랍니다!

실전 사용 예시: 데이터 기반 신규 숙소 가격 추천

개발이 끝난 MCP 서버는 클로드를 이용해서 테스트해봤습니다.

검은 색 컴퓨터 화면의 스크린 샷

추가로, 한옥이 아니라서 한옥 프리미엄 빼서 계산해달라고 하니까 추천 금액이 10만원 정도 떨어졌습니다. 🙂

마무리

직접 데이터를 수집하고 제가 원하는 방식대로 가공할 수 있다는 점에서 확실히 많은 것을 배웠고,

특히 커서를 이용해서 바이브 코딩 형식으로 개발했는데 이젠 정말 개발이 쉬워졌다는 것을 많이 느꼈습니다.

또한, MCP 서버를 개발하다보니 '어디까지 내가 하고, 어디서부터 AI가 분석하게 하면 될까'를 구별하고 고민하는게 꽤나 어렵고 흥미로운 과정이었습니다.

한편, 이번 프로젝트를 진행하며 다음과 같은 한계점도 느꼈습니다.

  1. 여러 사이트에서 데이터를 수집해 오려면 먼저 수집해올 데이터의 공통 폼을 결정하는게 낫겠다.

  2. 데이터 활용도: 데이터 수집 이후 진짜 필요한 건 '어떻게 해석해서 쓸 것인가?' 인데, 저의 경우는 AI가 데이터를 바탕으로 분석한 데이터가 신뢰성이 떨어지다보니 데이터 활용도가 매우 떨어졌습니다. 즉, 인사이트 추출/분석 설계가 더 중요하다는 걸 느꼈습니다.

  3. 실사용 동기: 개인적으로 실제 '단기 임대 숙소 시세 분석' 필요성이 크지 않아 프로젝트 완성 동기부여가 약해졌습니다. (그래도 쓸 상황이 생기면 바로 업데이트해서 쓸 수 있을 듯 합니다!)

1편에서 기획했던 것처럼 "모든 단기 임대 플랫폼 통합 MCP"까지는 가지 못했지만, 작은 프로젝트라도 실전 문제를 하나하나 부딪혀가면서 개발했기에, 배운 점이 많았던 경험입니다.

혹시 이후에라도 더 개선하거나, 다른 플랫폼에 적용해 본다면 그때 다시 글을 남겨보겠습니다.

읽어주셔서 감사합니다!

👉 이 게시글도 읽어보세요