데이터 에이전트: SQLite DB 연결하기
1. 실습용 더미 데이터베이스 생성 (setup_db.py)
먼저 조회할 데이터가 들어있는 DB 파일을 만들어야 합니다. 아래 코드를 실행해서 company.db를 생성하세요.
Python
import sqlite3
def init_db():
conn = sqlite3.connect('company.db')
cursor = conn.cursor()
# 사원 테이블 생성
cursor.execute('''
CREATE TABLE IF NOT EXISTS employees (
id INTEGER PRIMARY KEY,
name TEXT,
department TEXT,
position TEXT
)
''')
# 데이터 삽입
employees = [
(101, '이개발', '개발팀', '대리'),
(102, '김인사', '인사팀', '과장'),
(103, '박디자', 'UI팀', '사원')
]
cursor.executemany('INSERT OR REPLACE INTO employees VALUES (?,?,?,?)', employees)
conn.commit()
conn.close()
print("company.db 생성 완료!")
if __name__ == "__main__":
init_db()

Python
import sqlite3
import os
import sys
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("DatabaseAgent")
# DB 파일의 절대 경로 설정
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "company.db")
@mcp.tool()
def query_employee_info(name: str) -> str:
"""사원 이름을 입력받아 부서와 직급 정보를 DB에서 조회합니다."""
try:
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
# SQL 실행 (신입 교육 포인트: 파라미터 바인딩을 통해 SQL Injection 방지)
cursor.execute("SELECT department, position FROM employees WHERE name = ?", (name,))
result = cursor.fetchone()
conn.close()
if result:
return f"{name}님은 {result[0]} 소속이며 직급은 {result[1]}입니다."
else:
return f"'{name}' 사원을 찾을 수 없습니다."
except Exception as e:
return f"DB 조회 중 에러 발생: {str(e)}"
if __name__ == "__main__":
print("--- Database MCP Server Starting ---", file=sys.stderr)
mcp.run()
JSON
{
"mcpServers": {
"db-agent": {
"command": "C:/dev/llm/llm_env/Scripts/python.exe",
"args": ["C:/dev/llm/db_server.py"]
}
}
}
질문: "우리 회사에 김개발 대리님이 어느 부서 소속이야?"
에이전트의 내부 동작:
사용자의 질문에서 '김개발'이라는 이름을 추출.
query_employee_info(name="김개발") 도구를 호출해야겠다고 판단.
Python 코드가 SQL(SELECT...)을 실행하고 결과를 Claude에게 전달.
최종 답변: "김개발 대리님은 개발팀 소속입니다."

보안(Security): 에이전트에게 DB 전체 접속 권한을 주는 것이 아니라, 우리가 정의한 query_employee_info라는 특정 Skill만 허용해야 합니다. (이것이 MCP의 보안 철학입니다)
에러 핸들링: 만약 DB에 없는 이름을 물어봤을 때 "데이터가 없다"는 응답을 받은 Claude가 사용자에게 어떻게 대답하는지 확인 하세요.