Claude가 내 DB를 못 읽어서 답답하다면 — MCP로 직접 연결하기

"이 데이터 분석해줘" 할 때마다 복붙하는 당신에게

Claude한테 데이터 분석을 부탁할 때마다 CSV를 복사해서 붙여넣고 있나요? 또는 쿼리 결과를 스크린샷으로 찍어서 보여주고 있나요?

MCP(Model Context Protocol)를 쓰면 Claude가 당신의 DB에 직접 접근해서 쿼리를 실행합니다. "지난달 매출 상위 10개 상품 보여줘"라고 말하면, Claude가 알아서 SQL을 짜고 실행해서 결과를 보여줍니다.

이 글에서는 Python 50줄로 MCP 서버를 만들고, Claude Desktop에 연결하는 전체 과정을 다룹니다.

MCP가 뭔가요? — 5분 개념 정리

MCP(Model Context Protocol)는 Anthropic이 만든 AI와 외부 시스템을 연결하는 표준 프로토콜입니다.

핵심 구성요소 3가지

┌─────────────────┐     JSON-RPC      ┌─────────────────┐
│   Claude        │ ←───────────────→ │   MCP Server    │
│   (Client)      │    stdio/SSE      │   (Your Code)   │
└─────────────────┘                   └─────────────────┘
                                              │
                                              ▼
                                      ┌───────────────┐
                                      │  Your DB/API  │
                                      └───────────────┘

Server: 당신이 Python/TypeScript로 만드는 프로그램. DB 연결, 쿼리 실행 등의 기능을 제공합니다.

Client: Claude Desktop이 클라이언트 역할. 서버에 연결해서 기능을 사용합니다.

Protocol: 서버와 클라이언트가 JSON-RPC로 통신하는 규약입니다.

MCP가 제공하는 3가지 기능

Python 50줄로 MCP 서버 만들기

기본 구조

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

# 1. 서버 생성
server = Server("my-analytics")

# 2. 도구 정의 — AI가 호출할 수 있는 함수들
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="run_query",
            description="SQL SELECT 쿼리 실행",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "SQL 쿼리"}
                },
                "required": ["query"]
            }
        )
    ]

# 3. 도구 호출 처리
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "run_query":
        result = execute_sql(arguments["query"])
        return [TextContent(type="text", text=result)]

# 4. 서버 실행
async def main():
    async with mcp.server.stdio.stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

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

핵심 포인트:

• @server.list_tools(): AI에게 "이런 함수들 쓸 수 있어"라고 알려줌
• inputSchema: JSON Schema로 파라미터 정의 → AI가 올바른 형식으로 호출
• @server.call_tool(): 실제 함수 로직 구현

실전 DB 분석 서버

import sqlite3
import pandas as pd
import json

db_path = "analytics.db"

def get_connection():
    return sqlite3.connect(db_path)

def validate_query(query: str) -> tuple[bool, str]:
    """보안을 위한 쿼리 검증"""
    query_upper = query.strip().upper()

    if not query_upper.startswith("SELECT"):
        return False, "SELECT 쿼리만 허용됩니다"

    dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER"]
    for keyword in dangerous:
        if keyword in query_upper:
            return False, f"허용되지 않는 키워드: {keyword}"

    return True, "OK"

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="execute_query",
            description="SQL SELECT 쿼리 실행, 결과를 JSON으로 반환",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "limit": {"type": "integer", "default": 100}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="get_table_info",
            description="테이블 스키마와 샘플 데이터 조회",
            inputSchema={
                "type": "object",
                "properties": {
                    "table_name": {"type": "string"}
                },
                "required": ["table_name"]
            }
        ),
        Tool(
            name="list_tables",
            description="데이터베이스의 모든 테이블 목록",
            inputSchema={"type": "object", "properties": {}}
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    conn = get_connection()

    if name == "execute_query":
        query = arguments["query"]
        is_valid, msg = validate_query(query)
        if not is_valid:
            return [TextContent(type="text", text=f"오류: {msg}")]

        limit = arguments.get("limit", 100)
        df = pd.read_sql_query(f"{query} LIMIT {limit}", conn)
        return [TextContent(type="text", text=df.to_json(orient="records", indent=2))]

    elif name == "get_table_info":
        table = arguments["table_name"]

        # 컬럼 정보
        cursor = conn.execute(f"PRAGMA table_info({table})")
        columns = [{"name": c[1], "type": c[2]} for c in cursor.fetchall()]

        # 행 수
        count = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]

        # 샘플 데이터
        sample = pd.read_sql_query(f"SELECT * FROM {table} LIMIT 5", conn)

        result = {
            "table": table,
            "row_count": count,
            "columns": columns,
            "sample": sample.to_dict(orient="records")
        }
        return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))]

    elif name == "list_tables":
        cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
        tables = [row[0] for row in cursor.fetchall()]
        return [TextContent(type="text", text=json.dumps(tables, indent=2))]

Claude Desktop에 연결하기 — 설정 파일 하나로 끝

설정 파일 위치

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

설정 예시

{
  "mcpServers": {
    "my-analytics": {
      "command": "python",
      "args": ["/Users/yourname/mcp/analytics_server.py"],
      "env": {
        "DB_PATH": "/Users/yourname/data/analytics.db"
      }
    }
  }
}

설정 후 Claude Desktop 재시작하면 연결 완료입니다.

연결 확인 방법

Claude Desktop에서 망치 아이콘(🔨)을 클릭하면 사용 가능한 도구 목록이 보입니다. execute_query, get_table_info 등 당신이 정의한 도구들이 나타나면 성공입니다.

실전 활용 시나리오

1. 자연어 → SQL 자동 변환

당신: "지난 달 매출 상위 10개 상품 보여줘"

Claude: (내부적으로 execute_query 호출)

SELECT product_name, SUM(revenue) as total_revenue
FROM sales
WHERE sale_date >= date('now', '-1 month')
GROUP BY product_name
ORDER BY total_revenue DESC
LIMIT 10

2. 데이터 탐색

당신: "customers 테이블 구조 알려줘"

Claude: (get_table_info 호출)

{
  "table": "customers",
  "row_count": 15420,
  "columns": [
    {"name": "id", "type": "INTEGER"},
    {"name": "name", "type": "TEXT"},
    {"name": "email", "type": "TEXT"},
    {"name": "created_at", "type": "DATETIME"}
  ]
}

3. 복잡한 분석도 대화로

당신: "연령대별 평균 구매 금액을 분석하고, 어떤 연령대를 타겟팅해야 할지 추천해줘"

Claude가 쿼리를 실행하고 결과를 분석해서 인사이트까지 제공합니다.

리소스로 컨텍스트 제공하기

도구(Tools) 외에도 리소스(Resources)로 AI에게 배경 지식을 제공할 수 있습니다.

@server.list_resources()
async def list_resources():
    conn = get_connection()
    cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
    tables = [row[0] for row in cursor.fetchall()]

    return [
        Resource(
            uri=f"db://schema/{table}",
            name=f"테이블: {table}",
            description=f"{table}의 스키마 정보",
            mimeType="application/json"
        )
        for table in tables
    ]

@server.read_resource()
async def read_resource(uri: str):
    if uri.startswith("db://schema/"):
        table = uri.replace("db://schema/", "")
        conn = get_connection()
        cursor = conn.execute(f"PRAGMA table_info({table})")
        columns = [{"name": c[1], "type": c[2]} for c in cursor.fetchall()]
        return json.dumps({"table": table, "columns": columns}, indent=2)

리소스를 정의하면 Claude가 쿼리 전에 테이블 구조를 미리 파악해서 더 정확한 SQL을 생성합니다.

보안 베스트 프랙티스

1. 쿼리 화이트리스트

ALLOWED_TABLES = ["sales", "customers", "products"]

def validate_table(table: str) -> bool:
    return table.lower() in ALLOWED_TABLES

2. 결과 크기 제한

MAX_ROWS = 1000

df = pd.read_sql_query(f"{query} LIMIT {min(limit, MAX_ROWS)}", conn)

3. 로깅

import logging

logging.info(f"Query executed: {query[:100]}...")

마치며

MCP를 사용하면 Claude가 진짜 데이터 분석 파트너가 됩니다:

• 복붙 없이 자연어로 쿼리
• 실시간 데이터 분석
• 컨텍스트 유지 — 테이블 구조를 매번 설명할 필요 없음

Python 50줄로 시작해서, 점점 기능을 확장해 나가면 됩니다. PostgreSQL, BigQuery, Snowflake 등 어떤 데이터 소스든 연결 가능합니다.

GitHub에서 전체 예제 코드를 확인하세요: 노트북 예제도 포함되어 있습니다.