어... 생각보다 일이 커지게 되었다.
지금까지 방식은 server.py와 같은 폴더에 2025_student_record_request.db와 interviews_master.db라는 데이터베이스가 있고, 사용자컴퓨터에 각자 설치하는 방식이었는데, 개인정보가 각자 컴퓨터에 설치되다보니 관리가 힘들어지게 되었다.
개발한 MCP 서버의 이용 후기와 개선할 점을 선생님분들과 논의하던 중, 이러한 정보를 한 곳에서 일괄적으로 관리 할 수 있으면 좋겠다는 의견을 주셨다.
생각보다 많은 선생님분들께서 이용에 적극적이셨고 엄청 편하다는 의견을 주셨다. 그런데 개인정보가 각자 컴퓨터에 설치되니 함부로 공유할 수도 없는 노릇이고...
그러다가 문득 '대교협 프로그램처럼 개인정보 DB를 서버용 컴퓨터에 설치하고, 호출하면 되지 않을까? 어차피 로컬내에서 작동하는 방식이라 학교 랜선 꼽지 않는 이상 접속이 불가하니까 보안도 괜찮을 것 같고...'
1. 방법 구상하기
방법은 알아보니 PostgreSQL과 MySQL 와 같은 전문 데이터베이스 서버 프로그램을 설치하는 방법이 있었다. 이걸 해 보니 생각보다 다루기가 너무 어려웠고 서버컴퓨터는 내가 하지만 생각보다 내가 갖고 있는 권한이 많이 없어 접근이 너무 어려웠다. 그래서 포기
두 번째로 단순히 '읽기'만 할 것이므로 FastAPI를 이용하여 API 통신으로 호출하는 방식이었다. 이 방법이 학교와 같은 소규모 프로젝트에서는 쉬웠다. 간단하게 말해 "학생 정보 조회", "면접 기록 조회"와 같은 엔드포인트(Endpoint)를 URL로 정의하고 작동하는 방식으로 유지보수가 간단하다는 장점이 있다.
이에 학교 서버용 컴퓨터에 DB_SERVER라는 uv 가상환경을 준비하고 여기에 DB를 옳긴 다음, API 서버를 켜고 클라이언트에서 API 서버에 접속하여 정보를 가져오는 방식으로 변경하게 되었다.
2. 구현하기
uv로 DB_SERVER라는 가상환경을 생성한 다음, uv add fastapi "uvicorn[standard]" 로 설치하자. uv 구성방법에 대해서는 아래 포스트 참고해주세요.
[gemini cli + 커스텀 mcp 만들기] - 1. uv 환경구성
[gemini cli + 커스텀 mcp 만들기] - 1. uv 환경구성
1. uv 설치최근 대세는 uv라고 한다. 지금까지 miniconda를 사용했었는데 과연 이게 어떨지 궁금하긴 하다. 참고로 uv를 설치하는 과정은 3. gemini cli + mcp(SQLite) 3. gemini cli + mcp(SQLite)1. SQLite란SQLite는 데
toyourlight.tistory.com
같은 가상환경 폴더에 2025_student_record_request.db(생활기록부DB)와 interviews_master.db(면접 후기 DB)를 놓고 아래와 같이 api_server.py 파일을 작성한다.
import sqlite3
import json
import importlib.resources
import os
import logging
import contextlib
from fastapi import FastAPI, HTTPException, Query
from typing import List, Optional
# --- 1. 기본 설정 ---
# 서버 터미널에 상세한 로그를 출력하도록 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# FastAPI 애플리케이션 생성
app = FastAPI(
title="MCP API Server",
description="면접 후기 및 학생 활동 기록 조회를 위한 중앙 데이터베이스 API 서버",
version="1.1.0"
)
# --- 2. 상수 및 데이터베이스 경로 탐색 ---
# 사용할 데이터베이스 테이블 이름을 상수로 지정
INTERVIEWS_TABLE_NAME = 'interviews'
STUDENT_RECORDS_TABLE_NAME = 'student_records_request'
# 패키지 모드와 로컬 개발 모드를 자동으로 감지
try:
# 패키지로 설치되었을 때 ('pip install -e .' 등)
pkg_files = importlib.resources.files('univ_interviews')
INTERVIEWS_DB_REF = pkg_files.joinpath('interviews_master.db')
STUDENT_DB_REF = pkg_files.joinpath('2025_student_record_request.db')
IS_PKG_MODE = True
print("정보: '패키지 모드'로 DB 파일을 로드합니다.")
except ModuleNotFoundError:
# 로컬에서 'python api_server.py'로 직접 실행했을 때
print("경고: '로컬 개발 모드'로 실행합니다. DB 파일을 현재 스크립트 위치에서 찾습니다.")
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
INTERVIEWS_DB_REF = os.path.join(BASE_DIR, 'interviews_master.db')
STUDENT_DB_REF = os.path.join(BASE_DIR, '2025_student_record_request.db')
IS_PKG_MODE = False
# --- 3. 경로 처리를 위한 헬퍼 함수 ---
@contextlib.contextmanager
def get_db_path(db_ref):
"""
패키지 모드와 로컬 모드 양쪽에서 모두 안전하게 DB 파일 경로를 얻는 컨텍스트 매니저.
'with' 구문과 함께 사용합니다.
"""
if IS_PKG_MODE:
# 패키지 모드: importlib.resources를 통해 임시 파일 경로를 얻음
with importlib.resources.as_file(db_ref) as path:
yield path
else:
# 로컬 모드: db_ref는 이미 문자열 경로이므로 그대로 반환
yield db_ref
# --- 4. API 엔드포인트(Endpoint) 정의 ---
@app.get("/", summary="서버 상태 확인")
def read_root():
"""서버가 정상적으로 실행 중인지 확인하는 기본 엔드포인트입니다."""
return {"status": "MCP API Server is running"}
@app.get("/interviews/search", summary="면접 후기 검색")
def search_interview_reviews(
university: Optional[str] = None,
department: Optional[str] = None,
question_keywords: Optional[str] = None
):
"""대학교, 학과, 키워드를 조합하여 면접 후기를 검색합니다."""
try:
with get_db_path(INTERVIEWS_DB_REF) as db_path:
with sqlite3.connect(db_path) as conn:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
base_query = f"SELECT id, university, department, qa_list_json, tips_json, question_keywords FROM {INTERVIEWS_TABLE_NAME}"
conditions, params = [], []
if university:
conditions.append("university LIKE ?")
params.append(f"%{university}%")
if department:
conditions.append("department LIKE ?")
params.append(f"%{department}%")
if question_keywords:
conditions.append("question_keywords LIKE ?")
params.append(f"%{question_keywords}%")
if conditions:
base_query += " WHERE " + " AND ".join(conditions)
cursor.execute(base_query, tuple(params))
rows = cursor.fetchall()
if not rows:
return {"message": "검색 조건에 해당하는 면접 후기를 찾을 수 없습니다."}
return [dict(row) for row in rows]
except Exception as e:
logging.error(f"'/interviews/search' 처리 중 예외 발생: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"서버 내부 오류 발생: {e}")
@app.get("/students/activities", summary="학생 활동 기록 검색")
def get_student_activities(
student_id: int,
name: str,
domain: Optional[str] = None,
keywords: Optional[List[str]] = Query(None)
):
"""학번(필수)과 추가 조건을 조합하여 특정 학생의 활동 기록을 검색합니다."""
try:
with get_db_path(STUDENT_DB_REF) as db_path:
with sqlite3.connect(db_path) as conn:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
base_query = f"SELECT student_id, name, grade, domain, content, keywords FROM {STUDENT_RECORDS_TABLE_NAME}"
conditions, params = ["student_id = ?"], [student_id]
if name:
conditions.append("name = ?")
params.append(name)
if domain:
conditions.append("domain LIKE ?")
params.append(f"%{domain}%")
if keywords:
keyword_conditions = []
for kw in keywords:
keyword_conditions.append("content LIKE ? OR keywords LIKE ?")
params.extend([f"%{kw}%", f"%{kw}%"])
conditions.append(f"({ ' OR '.join(keyword_conditions) })")
base_query += " WHERE " + " AND ".join(conditions)
cursor.execute(base_query, tuple(params))
rows = cursor.fetchall()
if not rows:
return {"message": "검색 조건에 해당하는 학생 활동 기록을 찾을 수 없습니다."}
return [dict(row) for row in rows]
except Exception as e:
logging.error(f"'/students/activities' 처리 중 예외 발생: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"서버 내부 오류 발생: {e}")
# --- 5. 서버 실행 (로컬 테스트용) ---
if __name__ == "__main__":
import uvicorn
print("--- MCP API 서버 (로컬 디버그 모드) 시작 ---")
print(f"면접 후기 DB 경로 참조: {INTERVIEWS_DB_REF}")
print(f"학생 기록 DB 경로 참조: {STUDENT_DB_REF}")
print("\n서버 주소: http://127.0.0.1:8000")
print("자동 API 문서: http://127.0.0.1:8000/docs")
print("-----------------------------------------")
uvicorn.run(app, host="127.0.0.1", port=8000)
이에 맞게 MCP 서버 역할을 하는 server.py도 수정해주어야 한다.
import json
import requests # API 호출을 위한 라이브러리
from typing import List, Optional # FastAPI 서버와 타입 힌트를 맞추기 위해 추가
from mcp.server.fastmcp import FastMCP
# --- API 서버 설정 ---
# 공용 컴퓨터에서 실행 중인 API 서버의 주소를 입력합니다. (실제 공용컴퓨터 아이피 입력)
API_BASE_URL = "http://0.0.0.0:8000"
# --- MCP 서버 설정 ---
mcp = FastMCP("univ_interviews_v2.5")
# --- Tool 1: 면접 후기 검색 (API 호출 방식) ---
@mcp.tool()
def search_interview_reviews(
university: str | None = None,
department: str | None = None,
question_keywords: str | None = None
) -> str:
"""
대학교, 학과, 키워드를 조합하여 일반적인 면접 후기를 API 서버에서 검색합니다.
(기능 설명은 기존과 동일)
"""
# API 서버의 '/interviews/search' 엔드포인트에 전달할 파라미터를 구성합니다.
params = {
"university": university,
"department": department,
"question_keywords": question_keywords
}
# 값이 None인 파라미터는 API 요청에 포함하지 않도록 필터링합니다.
params = {k: v for k, v in params.items() if v is not None}
try:
# API 서버에 GET 요청을 보냅니다.
response = requests.get(f"{API_BASE_URL}/interviews/search", params=params)
# API 서버가 에러(4xx, 5xx)를 반환하면 예외를 발생시킵니다.
response.raise_for_status()
# 성공적으로 받은 결과를 JSON 텍스트로 변환하여 반환합니다.
# API 서버가 이미 JSON 형식으로 응답하지만, mcp tool의 반환 형식에 맞게 문자열로 다시 직렬화합니다.
results = response.json()
return json.dumps(results, ensure_ascii=False, indent=2)
except requests.exceptions.RequestException as e:
return json.dumps({"error": f"API 서버 연결 실패: {e}"}, ensure_ascii=False)
except Exception as e:
return json.dumps({"error": f"응답 데이터 처리 중 오류 발생: {e}"}, ensure_ascii=False)
# --- Tool 2: 학생 활동 기록 검색 (API 호출 방식) ---
@mcp.tool()
def get_student_activities(
student_id: int,
name: str | None = None,
domain: str | None = None,
keywords: List[str] | None = None
) -> str:
"""
학번(student_id)을 기반으로 특정 학생의 생활기록부 활동 내용을 API 서버에서 검색합니다.
(기능 설명은 기존과 동일)
"""
# API 서버의 '/students/activities' 엔드포인트에 전달할 파라미터를 구성합니다.
params = {
"student_id": student_id,
"name": name,
"domain": domain,
"keywords": keywords # requests는 리스트를 여러 개의 쿼리 파라미터로 자동 변환해줍니다.
}
params = {k: v for k, v in params.items() if v is not None}
try:
# API 서버에 GET 요청을 보냅니다.
response = requests.get(f"{API_BASE_URL}/students/activities", params=params)
response.raise_for_status()
results = response.json()
return json.dumps(results, ensure_ascii=False, indent=2)
except requests.exceptions.RequestException as e:
return json.dumps({"error": f"API 서버 연결 실패: {e}"}, ensure_ascii=False)
except Exception as e:
return json.dumps({"error": f"응답 데이터 처리 중 오류 발생: {e}"}, ensure_ascii=False)
# --- 서버 실행 (로컬 클라이언트 모드) ---
if __name__ == "__main__":
print("--- 개인화 면접 정보 MCP 클라이언트 서버 (univ_interviews_v2.5) 시작 ---")
print(f" [연결 대상 API 서버: {API_BASE_URL}]")
# API 서버가 켜져 있는지 간단히 확인
try:
health_check = requests.get(API_BASE_URL, timeout=3)
if health_check.status_code == 200:
print(f" [성공] API 서버에 연결되었습니다: {health_check.json()}")
else:
print(f" [경고] API 서버가 응답하지만 상태가 정상이 아닙니다 (상태 코드: {health_check.status_code})")
except requests.exceptions.RequestException:
print(f" [오류] API 서버({API_BASE_URL})에 연결할 수 없습니다. 서버가 실행 중인지 확인하세요.")
print("\n 클라이언트의 요청을 기다립니다. 종료하려면 Ctrl+C를 누르세요.")
mcp.run(transport='stdio')
마지막으로 uv 가상환경을 켠 상태에서 'uv run uvicorn api_server:app --host 0.0.0.0 --port 8000'을 입력하면 서버가 켜진다.
(DB_SERVER)D:\\uv_project\DB_SERVER uv run uvicorn api_server:app --host 0.0.0.0 --port 8000
3. 구동 결과
잘 출력된다.
'🗜 MCP' 카테고리의 다른 글
| [국가법령 MCP 서버 만들기] - 3. 판례, 헌재결정례, 법령해석례, 행정심판례 조회하기 (0) | 2025.11.16 |
|---|---|
| 대입 면접 상담 MCP 만들기 (0) | 2025.11.12 |
| [국가법령 MCP 서버 만들기] - 2. 판례 목록, 판례 본문 조회하기 (8) | 2025.08.29 |
| [국가법령 MCP 서버 만들기] - 1. API 발급, 환경 구성 (0) | 2025.08.28 |
| [gemini cli + 커스텀 mcp 만들기] - 5. 기상청 mcp 서버 배포하기 (6) | 2025.08.10 |
