파이썬으로 MCP 서버 구축하기: AI 에이전트의 실시간 데이터 연동 마스터 클래스

서론: MCP의 혁신적 역할과 가능성

MCP(Model Context Protocol)는 AI 모델과 외부 시스템 간의 표준화된 상호작용을 가능하게 하는 오픈 소스 프레임워크입니다^6. 이 기술은 Anthropic이 초기 개발을 주도했으며, 현재는 OpenAI와 Microsoft를 비롯한 주요 기업들이 적극 지원하고 있습니다^9. MCP 서버를 구축함으로써 개발자는 AI 어시스턴트가 데이터베이스, 웹사이트, 다양한 애플리케이션과 실시간으로 상호작용할 수 있는 인프라를 구축할 수 있습니다^6.

본 튜토리얼에서는 파이썬 기반 MCP 서버 구축을 통해 웹 스크래핑 기능을 구현하는 전 과정을 상세히 설명합니다. 클라우드 데스크톱 및 ChatGPT와의 연동을 통해 실제 비즈니스 시나리오에 적용 가능한 실용적인 솔루션 개발 방법을 제시합니다^6.

Set Up MCP Server In Python ❘ Step-By-Step Tutorial

Set Up MCP Server In Python | Step-By-Step Tutorial

 

---

1. 개발 환경 설정: 파이썬 및 필수 도구 설치

1.1 파이썬 3.10 이상 설치

1. 공식 파이썬 웹사이트(https://www.python.org) 접속
2. 운영체제에 맞는 최신 버전 다운로드
3. 설치 시 "Add Python to PATH" 옵션 반드시 체크^6
4. 관리자 권한 활성화(Windows: "Use admin privileges" 체크)

   # 설치 확인
   python --version
   # 출력: Python 3.12.1

1.2 UV 패키지 관리자 설치

pip install uv

1.3 프로젝트 디렉토리 생성

uv venv mcp-server
cd mcp-server
source .venv/bin/activate  # Windows: .venv\Scripts\activate

---

2. MCP 서버 핵심 의존성 설치

2.1 필수 패키지 설치

uv add "mcp[cli]" httpx beautifulsoup4 python-dotenv

2.2 MCP SDK 검증

from mcp.server.fastmcp import FastMCP
print(FastMCP.__doc__)  # MCP 서버 클래스 문서 확인[^9][^14]

---

3. 기본 MCP 서버 아키텍처 구축

3.1 서버 초기화 코드

# server.py
from mcp.server.fastmcp import FastMCP
import httpx
from bs4 import BeautifulSoup
import os
from dotenv import load_dotenv

load_dotenv()

mcp = FastMCP("WebScraper", debug=True)

3.2 웹 스크래핑 도구 구현

@mcp.tool()
async def scrape_web(url: str) -> str:
    """
    지정된 URL에서 텍스트 콘텐츠 추출
    Args:
        url: 스크랩할 웹사이트 주소
    Returns:
        정제된 텍스트 콘텐츠
    """
    headers = {"User-Agent": "MCP-WebScraper/1.0"}

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=10.0)
            response.raise_for_status()

            soup = BeautifulSoup(response.text, 'html.parser')

            # 불필요 요소 제거
            for element in soup(["script", "style", "nav", "footer", "header"]):
                element.decompose()

            # 텍스트 추출 및 정제
            text = soup.get_text(separator='\n')
            cleaned_text = '\n'.join([line.strip() for line in text.splitlines() if line.strip()])

            return cleaned_text[:5000]  # 5000자 제한

        except Exception as e:
            return f"Error: {str(e)}"

---

4. 서버 실행 및 테스트

4.1 서버 실행 명령어

uv run server.py --port 8000

4.2 테스트 클라이언트 구현

# test_client.py
from mcp.client import Client

async def test_scraping():
    async with Client("http://localhost:8000") as client:
        result = await client.tools.scrape_web(url="https://example.com")
        print(result)

if __name__ == "__main__":
    import asyncio
    asyncio.run(test_scraping())

---

5. Claude Desktop 연동 설정

5.1 구성 파일 편집

1. Claude Desktop 설정 → 개발자 모드 활성화
2. claude_config.json 파일 편집^6

   {
   "mcpServers": {
    "WebScraper": {
      "command": "uv",
      "args": ["run", "/path/to/server.py"],
      "env": {}
    }
   }
   }

5.2 기능 테스트

Claude 프롬프트에 다음 명령어 입력:

@scrape_web url="https://news.google.com"

---

6. 고급 기능: 동적 문서 검색 시스템 구현

6.1 문서 검색 도구 강화

@mcp.tool()
async def search_docs(query: str, library: str = "python") -> str:
    """
    주요 라이브러리 문서에서 관련 내용 검색
    Args:
        query: 검색 키워드
        library: 대상 라이브러리 (python, pandas, numpy)
    """
    DOC_URLS = {
        "python": "https://docs.python.org/3/search.html?q={query}",
        "pandas": "https://pandas.pydata.org/docs/search.html?q={query}",
        "numpy": "https://numpy.org/doc/stable/search.html?q={query}"
    }

    target_url = DOC_URLS.get(library.lower(), DOC_URLS["python"]).format(query=query)
    return await scrape_web(target_url)

6.2 사용 예시

# Python 공식 문서에서 "async" 키워드 검색
result = await search_docs("async", library="python")

---

7. 성능 최적화 기법

7.1 캐싱 메커니즘 구현

from functools import lru_cache

@lru_cache(maxsize=100)
async def cached_scrape(url: str) -> str:
    return await scrape_web(url)

@mcp.tool()
async def scrape_with_cache(url: str) -> str:
    return await cached_scrape(url)

7.2 병렬 처리 구현

import asyncio

@mcp.tool()
async def bulk_scrape(urls: list[str]) -> dict:
    tasks = [scrape_web(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return {url: result for url, result in zip(urls, results)}

---

8. 보안 강화 방안

8.1 URL 화이트리스트 시스템

ALLOWED_DOMAINS = {"example.com", "python.org", "pandas.pydata.org"}

def is_allowed(url: str) -> bool:
    from urllib.parse import urlparse
    domain = urlparse(url).netloc
    return any(domain.endswith(d) for d in ALLOWED_DOMAINS)

@mcp.tool()
async def safe_scrape(url: str) -> str:
    if not is_allowed(url):
        return "Error: Domain not allowed"
    return await scrape_web(url)

8.2 요청 제한 시스템

from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self, max_requests=100, period=60):
        self.max_requests = max_requests
        self.period = timedelta(seconds=period)
        self.requests = []

    def check(self):
        now = datetime.now()
        self.requests = [t for t in self.requests if now - t < self.period]
        return len(self.requests) < self.max_requests

limiter = RateLimiter()

@mcp.tool()
async def rate_limited_scrape(url: str) -> str:
    if not limiter.check():
        return "Error: Rate limit exceeded"
    limiter.requests.append(datetime.now())
    return await scrape_web(url)

---

9. 모니터링 및 로깅 시스템

9.1 통합 모니터링 구현

import logging
from prometheus_client import start_http_server, Counter

SCRAPE_REQUESTS = Counter('scrape_requests', 'Total scrape requests')
SCRAPE_ERRORS = Counter('scrape_errors', 'Total scrape errors')

@mcp.tool()
async def monitored_scrape(url: str) -> str:
    SCRAPE_REQUESTS.inc()
    try:
        result = await scrape_web(url)
        return result
    except Exception as e:
        SCRAPE_ERRORS.inc()
        logging.error(f"Scrape failed: {str(e)}")
        raise

# Prometheus 메트릭 서버 시작
start_http_server(8001)

---

10. 실전 적용 사례 연구

10.1 금융 데이터 분석 파이프라인

@mcp.tool()
async def get_stock_data(symbol: str) -> dict:
    url = f"https://finance.yahoo.com/quote/{symbol}"
    content = await scrape_web(url)
    # 추출 로직 구현
    return parse_stock_data(content)

@mcp.tool()
async def compare_stocks(symbols: list[str]) -> dict:
    tasks = [get_stock_data(sym) for sym in symbols]
    results = await asyncio.gather(*tasks)
    return analyze_comparison(results)

10.2 AI 기반 마켓 리서치

@mcp.tool()
async def market_analysis(keyword: str) -> dict:
    news = await search_docs(keyword, library="news")
    trends = await scrape_web(f"https://trends.google.com/trends/explore?q={keyword}")
    return generate_report(news, trends)

---

결론: MCP 생태계의 진화 방향

본 튜토리얼을 통해 구현한 MCP 서버는 AI 어시스턴트의 기능을 혁신적으로 확장합니다^6. 실시간 데이터 연동, 사용자 정의 도구 개발, 보안 강화 기법 등을 통해 기업의 디지털 인프라를 혁신할 수 있습니다.

향후 발전 방향:

1. 자동 확장 기능: Kubernetes 기반 오케스트레이션 연동
2. 플러그인 시스템: 사용자 정의 모듈 동적 로딩
3. 분산 처리 아키텍처: 다중 MCP 서버 클러스터링
4. AI 모델 임베딩: 온디바이스 추론 기능 통합

MCP는 단순한 프로토콜을 넘어 AI와 인간의 협업을 재정의하는 핵심 인프라로 진화하고 있습니다^9. 본 가이드를 출발점으로 독자 여러분만의 혁신적인 AI 솔루션을 구축하기를 기대합니다.

---

#MCP #파이썬개발 #AI에이전트 #웹스크래핑 #실시간데이터 #Claude통합 #자동화프레임워크 #AI인프라 #PythonSDK #개발자도구

⁂