들어가며
AI 에이전트가 외부 시스템과 연결되려면 각 시스템마다 커스텀 통합을 구현해야 했습니다. 이는 중복된 노력과 파편화를 초래했죠. Model Context Protocol (MCP)은 이 문제를 해결하기 위한 오픈 표준입니다. MCP를 한 번 구현하면 전체 통합 생태계가 열립니다.
이 글에서는 데이터베이스와 Redis에 접근할 수 있는 MCP 서버를 어떻게 구축했는지, 그리고 이를 에이전트와 어떻게 연동했는지 공유합니다.
왜 MCP 서버를 만들었나?
문제 상황
현재 운영 중인 플랫폼은 데이터 무결성을 위해 고도로 정규화된 데이터베이스 구조를 가지고 있습니다. 이로 인해 다음과 같은 비효율이 발생했습니다.
높은 조회 난이도: 운영팀에서 요청한 원인을 파악하려면 최소 6개 이상의 테이블을 조인하거나 서치를 해야 합니다.
운영 병목: DB에 직접적인 접근 권한 없는 운영팀은 개발자에게 데이터 조회를 의존해야 했고, 개발자는 반복적인 쿼리 작성 업무에 시달렸습니다.
MCP가 제공하는 해결책
MCP 서버를 구축하면 Claude 같은 AI 에이전트가:
- 자연어로 질문을 받아 적절한 SQL을 생성
- 여러 테이블을 자동으로 조인하여 데이터 수집
- 결과를 사람이 이해하기 쉽게 설명
이 모든 것이 가능해집니다. 더 중요한 것은, 한 번 구현하면 MCP를 지원하는 모든 클라이언트에서 사용할 수 있다는 점입니다.
또한, DB에 직접 접근이 어려운 운영자분들께서도 이 MCP를 사용하면 에이전트를 통해 쉽게 문제 파악을 할 수 있을 거라 생각했습니다. 이렇게 되면 개발자는 운영 이슈에 보다 자유롭고 운영자분들께서도 운영 업무에 효율성을 높일 수 있겠다고 생각했습니다.
구현 과정
1. Python과 FastMCP 선택
MCP 공식 문서를 보면 여러 언어의 SDK가 제공됩니다. Python을 선택한 이유는:
- FastMCP의 간결함: Python 타입 힌트와 docstring만으로 자동으로 도구 정의 생성
- 비동기 지원:
aiomysql과redis.asyncio로 효율적인 I/O 처리 - 빠른 프로토타이핑: 복잡한 보일러플레이트 없이 핵심 로직에 집중
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("mysql-db")
@mcp.tool()
async def query_db(query: str) -> str:
"""Execute SELECT query on the database."""
return await _execute_query(get_db, query)
FastMCP는 이 간단한 코드에서 자동으로:
- 도구 이름:
query_db - 설명: docstring에서 추출
- 파라미터 스키마: 타입 힌트에서 생성
2. 읽기 전용 쿼리 강제
AI가 생성한 SQL을 실행하는 것은 위험할 수 있습니다. 따라서 쿼리 검증 레이어를 구현했습니다:
def validate_readonly_query(query: str):
"""SELECT 쿼리만 허용하고 위험한 명령어 차단"""
query_upper = query.strip().upper()
if ";" in query and query.count(";") > 1:
raise ValueError("Multiple queries not allowed")
dangerous = ["INSERT", "UPDATE", "DELETE", "DROP", "CREATE", "ALTER"]
if any(cmd in query_upper for cmd in dangerous):
raise ValueError(f"Only SELECT queries allowed")
if not query_upper.startswith("SELECT"):
raise ValueError("Query must start with SELECT")
이 검증은 모든 쿼리 실행 전에 수행되며, 추가로 MySQL 세션 레벨에서도 읽기 전용을 강제합니다:
await cursor.execute("SET SESSION TRANSACTION READ ONLY")
3. AI가 올바른 쿼리를 작성하도록 스키마 정보 제공
AI가 정확한 SQL을 생성하려면 데이터베이스 구조를 알아야 합니다. get_db_schema 도구는 테이블과 컬럼 정보를 Markdown 형식으로 제공합니다:
@mcp.tool()
async def get_db_schema(database_name: str) -> str:
"""
지정한 데이터베이스의 전체 테이블 및 컬럼 구조(Schema)를 조회합니다.
테이블 관계를 파악하거나 쿼리를 작성하기 전에 반드시 먼저 실행하세요.
"""
sql = """
SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE,
IF(COLUMN_KEY = 'PRI', 'PK',
IF(COLUMN_KEY = 'MUL', 'FK/Index', '')) as KEY_TYPE,
IS_NULLABLE, COLUMN_COMMENT
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = %s
ORDER BY TABLE_NAME, ORDINAL_POSITION
"""
# ... 실행 및 Markdown 변환
이 도구의 docstring에 "반드시 먼저 실행하세요"라고 명시하여 AI가 쿼리 작성 전에 스키마를 확인하도록 유도했습니다.
4. 도메인 지식 리소스
MCP는 Tools 외에도 Resources를 제공할 수 있습니다. domain_knowledge.md 파일을 리소스로 노출하여 AI가 비즈니스 로직을 이해하도록 했습니다:
@mcp.resource("domain://knowledge")
async def get_domain_knowledge() -> str:
"""도메인 지식 및 데이터베이스 스키마 정보"""
with open("domain_knowledge.md", "r", encoding="utf-8") as f:
return f.read()
이러한 도메인 지식을 작성해주지 않는다면, 사용자가 에이전트를 통해 어떤 질문을 하게 된다면, 해당 질문에 대한 이해를 하지 못하고 요청과는 관련없는 쿼리를 작성할 수 있습니다.
클라이언트 설정
서버 코드를 작성했다면, 이제 에이전트가 앱이 이 서버를 인식할 수 있도록 설정해야 합니다.
설정 파일의 mcpServers 섹션에 실행 명령을 추가합니다.
저는 Python 패키지 매니저인 uv를 사용하여 의존성을 관리하고 실행했습니다.
{
"mcpServers": {
"mysql-db": {
"command": "uv",
"args": [
"run",
"/absolute/path/to/project/db_mcp_server.py"
]
}
}
}
mysql-db: 클라이언트에서 식별할 서버의 이름입니다.command: 실행할 명령어입니다.uv를 사용하면 가상환경 설정 없이도 스크립트를 즉시 실행할 수 있어 편리합니다.args: 실행 인자입니다. 이때 스크립트 파일의 경로는 절대 경로로 지정해야 Claude가 정확히 파일을 찾을 수 있습니다.
코드 실행 패턴의 장점
Anthropic의 Code execution with MCP 글도 함께 소개하려합니다.
이 mcp는 직접 도구 호출을 하지만, 대신 코드 실행을 사용하면:
1. 컨텍스트 효율성
도구 정의를 모두 로드하는 대신, 필요한 것만 탐색:
// 직접 호출: 모든 도구 정의 로드 (150,000 토큰)
query_users_db(query="...")
query_orders_db(query="...")
list_tables()
...
// 코드 실행: 필요한 것만 로드 (2,000 토큰)
import * as db from './servers/mysql-db';
const tables = await db.list_tables();
const result = await db.query_orders_db(`SELECT * FROM ${tables[0]}`);
2. 중간 결과 필터링
대용량 데이터를 AI 컨텍스트로 보내지 않고 코드에서 처리:
# 10,000개 행을 모두 컨텍스트로 보내지 않음
all_rows = await db.query_db("SELECT * FROM heavy_log_table")
pending = [r for r in all_rows if r['status'] == 'ERROR']
print(f"Found {len(pending)} error logs")
3. 복잡한 제어 흐름
루프와 조건문을 코드로 표현:
# 여러 테이블을 순회하며 데이터 수집
for table in ['orders_2024', 'payments_2024']:
result = await db.query_db(f"SELECT * FROM {table} WHERE user_id = 'user_123'")
# 결과 처리
이 프로젝트는 초기 단계이므로, 직접 호출 방식을 사용했습니다만, 추후에 자주 요청이 되는 쿼리나 CS 패턴을 수집한 후, 코드 실행 패턴으로 전환하면 더 효율적일 것입니다.
실제 사용 예시
에이전트로 자연어로 질문하면:
AI의 동작:
get_db_schema()호출 → 테이블 구조 파악query_db()호출 → 사용자의 질의 맞는 쿼리 작성 후 실행- 결과 종합하여 자연어로 설명
배운 점과 개선 방향
배운 점
- FastMCP의 생산성: 타입 힌트와 docstring만으로 도구 정의가 자동 생성되어 개발 속도가 빨랐습니다.
- 보안의 중요성: AI가 생성한 쿼리를 실행하므로 다층 방어(검증 + 읽기 전용 세션)가 필수입니다.
- 도메인 지식 제공: 스키마 정보와 비즈니스 로직을 함께 제공하니 AI의 답변 정확도가 크게 향상되었습니다.
개선 방향
- 코드 실행 패턴 도입: 파일 시스템 기반 도구 탐색으로 컨텍스트 효율성 개선
- 캐싱: 스키마 정보는 자주 변하지 않으므로 캐싱하여 반복 조회 방지
마치며
MCP는 AI 에이전트와 외부 시스템을 연결하는 표준 프로토콜입니다. 한 번 구현하면 Claude뿐만 아니라 MCP를 지원하는 모든 클라이언트에서 사용할 수 있습니다.
이 프로젝트를 통해:
- 복잡한 데이터베이스 쿼리를 자연어로 수행
- 운영 효율성 대폭 향상
- 재사용 가능한 도구 생태계 구축
을 달성했습니다.
참고 자료: