Python으로 MCP 서버 구축: 웹 검색 및 스크레이핑 가이드

AI 어시스턴트를 위한 MCP 서버를 Python 예제와 함께 구축하세요.

Page content

모델 컨텍스트 프로토콜(MCP)은 AI 어시스턴트가 외부 데이터 소스 및 도구와 상호 작용하는 방식을 혁신하고 있습니다. 이 가이드에서는 웹 검색 및 스크래핑 기능에 초점을 맞춘 MCP 서버를 Python으로 구축 방법을 살펴보겠습니다.

MCP 로봇

모델 컨텍스트 프로토콜이란?

모델 컨텍스트 프로토콜(MCP)은 Anthropic에서 도입한 오픈 프로토콜로, AI 어시스턴트가 외부 시스템에 연결하는 방식을 표준화합니다. 각 데이터 소스에 맞춤형 통합을 구축하는 대신, MCP는 다음을 가능하게 하는 통합된 인터페이스를 제공합니다:

  • AI 어시스턴트(Claude, ChatGPT 또는 커스텀 LLM 애플리케이션 등)가 도구를 발견하고 사용할 수 있도록
  • 개발자가 표준화된 프로토콜을 통해 데이터 소스, 도구 및 프롬프트를 노출할 수 있도록
  • 각 사용 사례에 맞춤형 솔루션을 다시 개발하지 않고도 원활한 통합이 가능하도록

이 프로토콜은 다음과 같은 클라이언트-서버 아키텍처에서 작동합니다:

  • MCP 클라이언트(AI 어시스턴트)는 기능을 발견하고 사용합니다.
  • MCP 서버는 자원, 도구 및 프롬프트를 노출합니다.
  • 통신은 stdio 또는 HTTP/SSE를 통해 JSON-RPC를 사용합니다.

다른 언어로 MCP 서버를 구현하려는 경우, Go로 MCP 서버 구현 가이드를 참조하세요. 이 가이드는 프로토콜 사양 및 메시지 구조에 대해 자세히 설명합니다.

왜 Python으로 MCP 서버를 구축해야 하나요?

Python은 MCP 서버 개발에 있어 다음과 같은 이유로 탁월한 선택입니다:

  1. 풍부한 생태계: requests, beautifulsoup4, selenium, playwright와 같은 라이브러리가 웹 스크래핑을 간단하게 만들어 줍니다.
  2. MCP SDK: 공식 Python SDK(mcp)는 견고한 서버 구현 지원을 제공합니다.
  3. 빠른 개발: Python의 간결함은 빠른 프로토타이핑 및 반복을 가능하게 합니다.
  4. AI/ML 통합: langchain, openai와 같은 AI 라이브러리 및 데이터 처리 도구와의 통합이 용이합니다.
  5. 커뮤니티 지원: 대규모 커뮤니티와 함께 광범위한 문서 및 예제가 제공됩니다.

개발 환경 설정

먼저 가상 환경을 생성하고 필요한 의존성을 설치하세요. 가상 환경은 Python 프로젝트의 격리에 필수적이며, 갱신이 필요하다면 venv Cheatsheet를 참조하세요.

# 가상 환경 생성 및 활성화
python -m venv mcp-env
source mcp-env/bin/activate  # Windows: mcp-env\Scripts\activate

# MCP SDK 및 웹 스크래핑 라이브러리 설치
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # Playwright 브라우저 드라이버 설치

현대적인 대안: 더 빠른 의존성 해결 및 설치를 원한다면, uv - 현대적인 Python 패키지 및 환경 관리자를 고려하세요. 이는 pip보다 대규모 프로젝트에서 훨씬 빠를 수 있습니다.

기본 MCP 서버 구축

먼저 최소한의 MCP 서버 구조부터 시작해 보겠습니다. Python에 익숙하지 않거나 구문 및 일반 패턴에 대한 빠른 참조가 필요하다면, Python Cheatsheet를 참조하세요. 이 문서는 Python의 기본 사항에 대한 종합적인 개요를 제공합니다.

import asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio

# 서버 인스턴스 생성
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """사용 가능한 도구 정의"""
    return [
        Tool(
            name="search_web",
            description="정보를 검색합니다.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "검색 쿼리"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "최대 결과 수",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """도구 실행 처리"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # 검색 로직 구현
        results = await perform_web_search(query, max_results)
        
        return [TextContent(
            type="text",
            text=f"'{query}'에 대한 검색 결과:\n\n{results}"
        )]
    
    raise ValueError(f"알 수 없는 도구: {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """실제 검색 구현을 위한 플레이스홀더"""
    return f"{max_results}개의 결과가 발견되었습니다: {query}"

async def main():
    """서버 실행"""
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

if __name__ == "__main__":
    asyncio.run(main())

웹 검색 기능 구현

이제 DuckDuckGo를 사용하여 실제 웹 검색 도구를 구현해 보겠습니다(이 도구는 API 키가 필요하지 않습니다):

import asyncio
import requests
from bs4 import BeautifulSoup
from urllib.parse import quote_plus

async def search_duckduckgo(query: str, max_results: int = 5) -> list[dict]:
    """DuckDuckGo 검색 및 결과 파싱"""
    
    url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        
        soup = BeautifulSoup(response.text, 'html.parser')
        results = []
        
        for result in soup.select('.result')[:max_results]:
            title_elem = result.select_one('.result__title')
            snippet_elem = result.select_one('.result__snippet')
            url_elem = result.select_one('.result__url')
            
            if title_elem and snippet_elem:
                results.append({
                    "title": title_elem.get_text(strip=True),
                    "snippet": snippet_elem.get_text(strip=True),
                    "url": url_elem.get_text(strip=True) if url_elem else "N/A"
                })
        
        return results
        
    except Exception as e:
        raise Exception(f"검색 실패: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """검색 결과 표시를 위한 포맷팅"""
    if not results:
        return "결과가 없습니다."
    
    formatted = []
    for i, result in enumerate(results, 1):
        formatted.append(
            f"{i}. {result['title']}\n"
            f"   {result['snippet']}\n"
            f"   URL: {result['url']}\n"
        )
    
    return "\n".join(formatted)

웹 스크래핑 기능 추가

이제 웹 페이지에서 콘텐츠를 스크래핑하고 추출하는 도구를 추가해 보겠습니다. LLM과 함께 HTML 콘텐츠를 사용할 때는 Markdown 형식으로 변환하여 처리가 더 용이할 수 있습니다. 이 목적을 위해, Python으로 HTML을 Markdown으로 변환을 참조하세요. 이 가이드는 6개의 다른 라이브러리를 벤치마크와 실용적인 추천과 함께 비교합니다.

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """Playwright를 사용하여 웹페이지 콘텐츠 스크래핑"""
    
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        try:
            await page.goto(url, timeout=30000)
            
            # 콘텐츠 로드 대기
            await page.wait_for_load_state('networkidle')
            
            if selector:
                # 특정 요소 추출
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "선택자 발견되지 않음"
            else:
                # 메인 콘텐츠 추출
                content = await page.inner_text('body')
            
            title = await page.title()
            
            return {
                "title": title,
                "content": content[:5000],  # 콘텐츠 길이 제한
                "url": url,
                "success": True
            }
            
        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# MCP 서버에 스크래퍼 도구 추가
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="DuckDuckGo를 사용하여 웹 검색",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "검색 쿼리"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="웹페이지 콘텐츠 스크래핑",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "스크래핑할 URL"},
                    "selector": {
                        "type": "string",
                        "description": "특정 콘텐츠를 위한 선택적 CSS 선택자"
                    }
                },
                "required": ["url"]
            }
        )
    ]

완전한 MCP 서버 구현

검색 및 스크래핑 기능을 모두 갖춘 완전한, 프로덕션 준비가 된 MCP 서버입니다:

#!/usr/bin/env python3
"""
웹 검색 및 스크래핑을 위한 MCP 서버
웹 검색 및 페이지 콘텐츠 추출을 위한 도구 제공
"""

import asyncio
import logging
from typing import Any
import requests
from bs4 import BeautifulSoup
from urllib.parse import quote_plus
from playwright.async_api import async_playwright

from mcp.server import Server
from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource
import mcp.server.stdio

# 로깅 구성
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websearch-scraper")

# 서버 생성
app = Server("websearch-scraper")

# 검색 구현
async def search_web(query: str, max_results: int = 5) -> str:
    """DuckDuckGo 검색 및 포맷팅된 결과 반환"""
    url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
    headers = {"User-Agent": "Mozilla/5.0"}
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        soup = BeautifulSoup(response.text, 'html.parser')
        results = []
        
        for result in soup.select('.result')[:max_results]:
            title = result.select_one('.result__title')
            snippet = result.select_one('.result__snippet')
            link = result.select_one('.result__url')
            
            if title and snippet:
                results.append({
                    "title": title.get_text(strip=True),
                    "snippet": snippet.get_text(strip=True),
                    "url": link.get_text(strip=True) if link else ""
                })
        
        # 결과 포맷팅
        if not results:
            return "결과가 없습니다."
        
        formatted = [f"총 {len(results)}개의 결과가 '{query}'에 대해 발견되었습니다:\n"]
        for i, r in enumerate(results, 1):
            formatted.append(f"\n{i}. **{r['title']}**")
            formatted.append(f"   {r['snippet']}")
            formatted.append(f"   {r['url']}")
        
        return "\n".join(formatted)
        
    except Exception as e:
        logger.error(f"검색 실패: {e}")
        return f"검색 오류: {str(e)}"

# 스크래퍼 구현
async def scrape_page(url: str, selector: str = None) -> str:
    """Playwright를 사용하여 웹페이지 콘텐츠 스크래핑"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        try:
            await page.goto(url, timeout=30000)
            await page.wait_for_load_state('networkidle')
            
            title = await page.title()
            
            if selector:
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "선택자 발견되지 않음"
            else:
                content = await page.inner_text('body')
            
            # 콘텐츠 길이 제한
            content = content[:8000] + "..." if len(content) > 8000 else content
            
            result = f"**{title}**\n\nURL: {url}\n\n{content}"
            return result
            
        except Exception as e:
            logger.error(f"스크래핑 실패: {e}")
            return f"스크래핑 오류: {str(e)}"
        finally:
            await browser.close()

# MCP 도구 정의
@app.list_tools()
async def list_tools() -> list[Tool]:
    """사용 가능한 MCP 도구 목록"""
    return [
        Tool(
            name="search_web",
            description="DuckDuckGo를 사용하여 웹 검색. 제목, 스니펫, URL을 반환합니다.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "검색 쿼리"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "최대 결과 수 (기본값: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="웹페이지 콘텐츠 추출. CSS 선택자를 사용하여 특정 요소를 대상으로 할 수 있습니다.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "스크래핑할 URL"
                    },
                    "selector": {
                        "type": "string",
                        "description": "특정 콘텐츠 추출을 위한 선택적 CSS 선택자"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """도구 실행 처리"""
    try:
        if name == "search_web":
            query = arguments["query"]
            max_results = arguments.get("max_results", 5)
            result = await search_web(query, max_results)
            return [TextContent(type="text", text=result)]
        
        elif name == "scrape_webpage":
            url = arguments["url"]
            selector = arguments.get("selector")
            result = await scrape_page(url, selector)
            return [TextContent(type="text", text=result)]
        
        else:
            raise ValueError(f"알 수 없는 도구: {name}")
    
    except Exception as e:
        logger.error(f"도구 실행 실패: {e}")
        return [TextContent(
            type="text",
            text=f"{name} 실행 오류: {str(e)}"
        )]

async def main():
    """MCP 서버 실행"""
    logger.info("웹검색-스크래퍼 MCP 서버 시작")
    
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

if __name__ == "__main__":
    asyncio.run(main())

MCP 서버 구성

Claude Desktop 또는 다른 MCP 클라이언트와 함께 MCP 서버를 사용하려면 구성 파일을 생성하세요:

Claude Desktop용 (claude_desktop_config.json):

{
  "mcpServers": {
    "websearch-scraper": {
      "command": "python",
      "args": [
        "/path/to/your/mcp_server.py"
      ],
      "env": {}
    }
  }
}

위치:

  • macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows: %APPDATA%\Claude\claude_desktop_config.json
  • Linux: ~/.config/Claude/claude_desktop_config.json

MCP 서버 테스트

기능을 확인하기 위해 테스트 스크립트를 생성하세요:

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """로컬에서 MCP 서버 테스트"""
    
    # 검색 테스트
    print("웹 검색 테스트...")
    results = await search_web("Python MCP 튜토리얼", 3)
    print(results)
    
    # 스크래퍼 테스트
    print("\n\n웹페이지 스크래퍼 테스트...")
    content = await scrape_page("https://example.com")
    print(content[:500])

if __name__ == "__main__":
    asyncio.run(test_mcp_server())

고급 기능 및 최선의 실천

1. 요청 제한

타겟 서버를 과도하게 사용하지 않도록 요청 제한을 구현하세요:

import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_requests: int = 10, time_window: int = 60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = defaultdict(list)
    
    async def acquire(self, key: str):
        now = time.time()
        self.requests[key] = [
            t for t in self.requests[key] 
            if now - t < self.time_window
        ]
        
        if len(self.requests[key]) >= self.max_requests:
            raise Exception("요청 제한 초과")
        
        self.requests[key].append(now)

limiter = RateLimiter(max_requests=10, time_window=60)

2. 캐싱

성능 향상을 위해 캐싱을 추가하세요:

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
async def cached_search(query: str, max_results: int):
    return await search_web(query, max_results)

3. 오류 처리

강력한 오류 처리를 구현하세요:

from enum import Enum

class ErrorType(Enum):
    NETWORK_ERROR = "network_error"
    PARSE_ERROR = "parse_error"
    RATE_LIMIT = "rate_limit_exceeded"
    INVALID_INPUT = "invalid_input"

def handle_error(error: Exception, error_type: ErrorType) -> str:
    logger.error(f"{error_type.value}: {str(error)}")
    return f"오류 ({error_type.value}): {str(error)}"

4. 입력 검증

처리 전에 사용자 입력을 검증하세요:

from urllib.parse import urlparse

def validate_url(url: str) -> bool:
    try:
        result = urlparse(url)
        return all([result.scheme, result.netloc])
    except:
        return False

def validate_query(query: str) -> bool:
    return len(query.strip()) > 0 and len(query) < 500

배포 고려사항

웹 배포를 위한 SSE 전송 사용

웹 기반 배포를 위해 SSE(Server-Sent Events) 전송을 사용하세요. 서버리스 배포를 고려 중이라면, AWS Lambda에서 JavaScript, Python, Golang 성능 비교을 참조하여 런타임에 대한 정보 있는 결정을 내릴 수 있습니다:

import mcp.server.sse

async def main_sse():
    """SSE 전송으로 서버 실행"""
    from starlette.applications import Starlette
    from starlette.routing import Mount
    
    sse = mcp.server.sse.SseServerTransport("/messages")
    
    starlette_app = Starlette(
        routes=[
            Mount("/mcp", app=sse.get_server())
        ]
    )
    
    import uvicorn
    await uvicorn.Server(
        config=uvicorn.Config(starlette_app, host="0.0.0.0", port=8000)
    ).serve()

AWS Lambda 배포

MCP 서버는 특히 SSE 전송을 사용할 때 AWS Lambda 함수로도 배포할 수 있습니다. Lambda 배포에 대한 종합 가이드:

Docker 배포

컨테이너화된 배포를 위해 Dockerfile을 생성하세요:

FROM python:3.11-slim

WORKDIR /app

# 시스템 의존성 설치
RUN apt-get update && apt-get install -y \
    wget \
    && rm -rf /var/lib/apt/lists/*

# Python 의존성 설치
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# 애플리케이션 복사
COPY mcp_server.py .

CMD ["python", "mcp_server.py"]

성능 최적화

비동기 작업

동시 작업을 위해 asyncio를 사용하세요:

async def search_multiple_queries(queries: list[str]) -> list[str]:
    """여러 쿼리에 대해 동시 검색"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

연결 풀링

더 나은 성능을 위해 연결을 재사용하세요:

import aiohttp

session = None

async def get_session():
    global session
    if session is None:
        session = aiohttp.ClientSession()
    return session

async def fetch_url(url: str) -> str:
    session = await get_session()
    async with session.get(url) as response:
        return await response.text()

보안 최선의 실천 방법

  1. 입력 정화: 사용자 입력을 항상 검증하고 정화하세요
  2. URL 허용 목록: 스크래핑을 위해 URL 허용 목록을 고려해 구현하세요
  3. 타임아웃 제어: 자원 고갈을 방지하기 위해 적절한 타임아웃을 설정하세요
  4. 콘텐츠 제한: 스크래핑된 콘텐츠의 크기를 제한하세요
  5. 인증: 프로덕션 배포에 인증을 구현하세요
  6. HTTPS: 프로덕션에서 SSE 전송에 HTTPS를 사용하세요

다양한 LLM 제공업체와의 협업

MCP는 Anthropic이 Claude를 위해 개발했지만, 프로토콜은 모든 LLM과 호환되도록 설계되었습니다. 여러 AI 제공업체와 상호작용하는 MCP 서버를 구축하고 구조화된 출력이 필요한 경우, 우리의 인기 있는 LLM 제공업체의 구조화된 출력 비교를 참고하세요. 이 비교에는 OpenAI, Gemini, Anthropic, Mistral, AWS Bedrock이 포함됩니다.

유용한 링크 및 자료

관련 자료

MCP 및 프로토콜 구현

Python 개발

웹 스크래핑 및 콘텐츠 처리

서버리스 배포 자료

LLM 통합

결론

Python으로 MCP 서버를 구축하면 AI 어시스턴트에 맞춤형 도구와 데이터 소스를 추가하는 강력한 가능성을 열 수 있습니다. 여기서 보여주는 웹 검색 및 스크래핑 기능은 시작일 뿐이며, 이 기반을 확장해 데이터베이스, API, 파일 시스템, 거의 모든 외부 시스템을 통합할 수 있습니다.

Model Context Protocol은 여전히 발전 중이지만, AI 도구 통합에 대한 표준화된 접근 방식은 다음 세대의 AI 기반 애플리케이션을 개발하는 개발자들에게 흥미로운 기술이 됩니다. 조직 내부 도구를 만들거나 커뮤니티를 위한 공개 MCP 서버를 구축하든간에, Python은 빠른 개발 및 배포를 위한 탁월한 기반이 됩니다.