MCP 데이터베이스 통합

Model Context Protocol (MCP) 입문

James Chapman

AI Curriculum Manager, DataCamp

타임존 서버에 데이터베이스를 추가하는 이유

쿼리 기능: 전체 데이터셋을 로드하지 않고 필터링, 검색, 정렬 가능
확장성: 데이터베이스 인덱스를 활용한 빠른 조회

데이터베이스 연결 아이콘

타임존 서버에 데이터베이스를 추가하는 이유

쿼리 기능: 전체 데이터셋을 로드하지 않고 필터링, 검색, 정렬 가능
확장성: 데이터베이스 인덱스를 활용한 빠른 조회
동시성: 여러 도구 호출 또는 클라이언트가 DB를 안전하게 사용
프로덕션 지원: 백업, 복제, 단일 정보 출처

데이터베이스 장점 다이어그램

데이터 접근: 도구 vs. 리소스

도구 아이콘

동적, 사용자 입력 의존도 높음, 쓰기 작업
예시: 데이터 분석 및 운영

리소스 아이콘

읽기 전용, 정적 데이터, 참조 정보
예시: 가이드라인, 문서, 정책

연결 수명 주기

데이터베이스 연결 다이어그램

데이터베이스 리소스: timezone_server.py

from mcp.server.fastmcp import FastMCP
import sqlite3

mcp = FastMCP("Timezone Converter")

# Create connection at startup; reuse in handlers
conn = sqlite3.connect("timezones.db")
conn.row_factory = sqlite3.Row

@mcp.resource("db://timezones")
def get_locations() -> str:
    try:
        cursor = conn.execute("SELECT timezone FROM locations")
        rows = cursor.fetchall()
        return "\n".join(r["timezone"] for r in rows)
    except sqlite3.Error as e: return f"Error: {e}"

데이터베이스 조회 도구: timezone_server.py

@mcp.tool()
def lookup_locations(prefix: str) -> str:
    """Find timezones that contain the given prefix."""
    try:
        cursor = conn.execute(
            "SELECT timezone FROM locations WHERE timezone LIKE ? LIMIT 50", (f"%{prefix}%",))
        rows = cursor.fetchall()
        return "\n".join(r["timezone"] for r in rows)
    except sqlite3.Error as e: return f"Error: {e}"


if __name__ == "__main__":
    try:
        mcp.run(transport="stdio")
    finally:
        conn.close()

안전성 및 프로덕션 권장 사항

매개변수화된 (?) 쿼리: 사용자 또는 LLM 입력을 SQL에 직접 삽입하지 말 것
- 프롬프트 인젝션 → 프롬프트를 통한 악성 코드 주입

SELECT timezone FROM locations WHERE timezone LIKE ? LIMIT 50

MCP 서버에는 읽기 전용 및 최소 권한 데이터베이스 접근 권장
제한 적용: 최대 행 수 및 타임아웃

연습해 봅시다!

Model Context Protocol (MCP) 입문