Claude Desktop에서 Database 서버에 접근하는 MCP 서버를 만들자.

데이터베이스 서버에 접근하는 local MCP 서버 코딩

Claude Desktop에서 다음 프롬프트를 이용하여 MariaDB에 접근하여 자료를 읽는 local MCP 서버 코딩을 한다.

mariadb 서버에 연결해서 레코드를 읽어보는 local mcp 서버 코딩해줘. 
query는 select item_no, item_name, owner, time_create from item where item_name='010'. 
fastmcp 프레임워크 사용해줘

 

Calude는 결과로 다음 3개 파일을 생성한다.

• server.py - MCP 서버를 구현한 파이썬 파일 -> dbserver.py로 이름을 변경
• readme.md - Cladue Desktop에서 설명한 생성 코드 설명과 필요 모듈에 대한 설명 정리 파일
• requirements.txt - 설치해야할 파이썬 모듈 이름과 버전 조건

uv로 mcp 서버 프로젝트를 생성한다.

uv init dbserver

 

프로젝트 디렉터리에서 uv를 이용하여 필요한 파이썬 모듈을 설치한다.

cd dbserver
pip install requirements.txt

 

requirements.txt 에서 설치해야 할 파이썬 모듈이 기록되어 있다. 다음 deserver.py파일과 함께 Claude Desktop에서 코등 생성 시 자동으로 제공된다. requirements.txt 파일 내용은 다음과 같다.

mcp>=1.2.0
pymysql>=1.1.0

 

프로젝트 디렉터리에 mcp 서버를 Python으로 구현한 dbserver.py를 생성한다(생성된 server.py 파일을 dbserver.py로 변경).

"""
MariaDB에 연결하여 item테이블 레코드를 조회하는 로컬 MCP 서버.

사용 프레임워크: FastMCP (mcp 패키지의 fastmcp)
전송 방식: stdio (로컬 MCP 서버용 표준 방식)

실행 전 준비:
    pip install -r requirements.txt

환경변수로 DB 접속 정보를 설정합니다 (하드코딩 방지):
    MARIADB_HOST     (기본값: 127.0.0.1)
    MARIADB_PORT     (기본값: 3306)
    MARIADB_USER     (필수)
    MARIADB_PASSWORD (필수)
    MARIADB_DATABASE (필수)
"""

import os
import logging
from typing import Any

import pymysql
import pymysql.cursors
from fastmcp import FastMCP

# ---------------------------------------------------------------------------
# 로깅 설정 (stdout은 MCP 프로토콜 전용이므로 로그는 stderr로)
# ---------------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("mariadb-mcp")

# ---------------------------------------------------------------------------
# DB 접속 설정
# ---------------------------------------------------------------------------
DB_CONFIG = {
    "host": os.environ.get("MARIADB_HOST", "127.0.0.1"),
    "port": int(os.environ.get("MARIADB_PORT", "3306")),
    "user": os.environ.get("MARIADB_USER"),
    "password": os.environ.get("MARIADB_PASSWORD"),
    "database": os.environ.get("MARIADB_DATABASE"),
    "cursorclass": pymysql.cursors.DictCursor,
    "connect_timeout": 5,
    "read_timeout": 10,
}

REQUIRED_ENV_VARS = ["MARIADB_USER", "MARIADB_PASSWORD", "MARIADB_DATABASE"]


def _check_config() -> None:
    missing = [name for name in REQUIRED_ENV_VARS if not os.environ.get(name)]
    if missing:
        raise RuntimeError(
            f"다음 환경변수가 설정되지 않았습니다: {', '.join(missing)}. "
            "MCP 클라이언트 설정(예: claude_desktop_config.json)의 env 항목을 확인하세요."
        )


def get_connection() -> pymysql.connections.Connection:
    """매 호출마다 새 커넥션을 열고 사용 후 닫는다 (로컬/저빈도 조회용 단순 구현)."""
    _check_config()
    return pymysql.connect(**DB_CONFIG)


# ---------------------------------------------------------------------------
# MCP 서버 정의
# ---------------------------------------------------------------------------
mcp = FastMCP("mariadb-student")


@mcp.tool()
def get_items(limit: int = 100) -> list[dict[str, Any]]:
    """
    MariaDB의 student 테이블에서 address, name 컬럼을 조회합니다.

    실행 쿼리: SELECT item_no,item_name, owner, time_create FROM student where item_no = '010' LIMIT %s

    Args:
        limit: 반환할 최대 행 수 (기본 100, 최대 1000). 대량 조회로 인한
               응답 폭주를 막기 위한 안전장치입니다.

    Returns:
        각 item을 {"item_no": ..., "item_name": ...,"owner": ..., "time_create": ...} 형태의 dict로 담은 리스트.
    """
    safe_limit = max(1, min(limit, 1000))

    conn = None
    try:
        conn = get_connection()
        with conn.cursor() as cursor:
            cursor.execute("SELECT item_no, item_name, owner, time_create FROM item where item_no='010' LIMIT %s", (safe_limit,))
            rows = cursor.fetchall()
            logger.info("item 테이블 조회 완료: %d행", len(rows))
            return rows
    except pymysql.err.OperationalError as e:
        logger.error("DB 연결 오류: %s", e)
        raise RuntimeError(
            f"MariaDB 연결에 실패했습니다: {e}. host/port/자격증명을 확인하세요."
        ) from e
    except pymysql.err.ProgrammingError as e:
        logger.error("쿼리 오류: %s", e)
        raise RuntimeError(
            f"쿼리 실행 중 오류가 발생했습니다: {e}. "
            "student 테이블과 add, name 컬럼이 실제로 존재하는지 확인하세요."
        ) from e
    finally:
        if conn is not None:
            conn.close()


if __name__ == "__main__":
    # stdio 전송 방식으로 실행 (Claude Desktop 등 로컬 MCP 클라이언트용)
    mcp.run(transport="stdio")

 

Claude Desktop에 MCP 서버 등록

readme.txt에서 Claude Desktop 등록 json을 읽어 등록한다.

{
  "mcpServers": {
    "dbserver": {
      "command": "python",
      "args": ["c:/do/systems/trip/mcp/dbserver/dbserver.py"],
      "env": {
        "MARIADB_HOST": "localhost",
        "MARIADB_PORT": "3306",
        "MARIADB_USER": "your_user", #자신의 login으로 변경
        "MARIADB_PASSWORD": "your_password", #자신의 패스워드로 변경
        "MARIADB_DATABASE": "your_db" #자신의 database로 변경
      }
    }
  }
}

 

Claude Desktop에서 데이터베이스 검색 결과를 분석한다.

local MCP 서버가 실행되는 것을 확인한다. "dbserver"가 실행되고 있다.

 

Claude Desktop에게 item 정보를 데이터베이스에서 읽어오도록 명령한다.

item을 읽어줘

 

다음과 같은 결과를 얻는다.

 

총 165개의 item 정보를 읽어와서 다양한 질문이 가능하다.

 

> 아이템 정보를 출력해줘

165개 아이템 목록을 모두 출력한다.

 

> 아이템 번호가 중복된 행을 하나로 묶고 갯수 행을 추가해줘

중본 아이템 번호와 중복 갯수를 포함한 테이블로 바꿔 출력해준다.

 

> item no가 'table'인 것만 다시 정리해줘

item name이 'table'인 것을 착각하고 명령했지만 스스로 적절한 검색으로 바꿔 item name이 정확히 'table'인 것으로 다시 정리해줬다. 내용에 대한 간단한 분석도 대답에 추가하였다.

 

> 아이템 이름이 Table인 것들만 다시 정리해줘

> 대소문자 구분없이 그리고 한국말도 포함해서 정리해줘

> 같은 것은 하나로 묶고 갯수 열을 포함시켜 정리해줘

 

위 질문으로 영문 Table과 의미가 같은 "탁자"를 포함한 아이템 이름을 가진 아이템의 갯수를 확인할 수 있었다.

 

Claude Desktop에 MCP 서버를 이용하여 데이터베이스를 연결할 경우 읽은 자료에 대한 자연어를 이용한 검색과 분석이 가능하다. 간단하고 빠른 처리를 위해서는 SQL 사용이 옳을 것 같고, 자주 쓰는 질의는 SQL을 내재한 응용 프로그램이 적합하다. 그러므로 Generative AI 채팅을 이용한 데이터베이스 질의는 Ad hoc(때때로) 분석에 쓰일 수 있다.

 

2026 NDoStudio

2026/8/21 처음 2026/8/21

c:\do\systems\trip\mcp\dbserver\dbserver.py