MCP로 완성하기 - Claude가 회사 지식을 쓰게 만들기

소규모 기업을 위한 제대로 된 RAG 시스템 [4/4]

3편까지 해서 꽤 쓸 만한 RAG 시스템을 만들었습니다. 문서를 전처리하고, Contextual Retrieval로 문맥을 붙여 인덱싱하고, 하이브리드 검색과 리랭킹으로 정확한 답변을 만들고, 가드레일과 RAGAS 평가까지 갖추었습니다. 그런데 이 시스템을 실제로 쓰려면 python search_pipeline.py를 터미널에서 실행해야 합니다.

여기서 현실적인 문제가 생깁니다. 이 시스템을 쓸 사람이 개발자만은 아니라는 겁니다. 인사팀에서 "우리 회사 연차 규정에서 3년차 이상은 며칠인가요?"라고 물어보고 싶고, 영업팀에서 "지난달 A고객한테 보낸 제안서에 어떤 가격을 제시했었지?"를 찾고 싶습니다. 이 사람들에게 터미널을 열고 Python 스크립트를 실행하라고 할 수는 없습니다.

MCP가 이 문제를 해결합니다. Model Context Protocol은 Anthropic이 만든 개방형 표준으로, AI 앱이 외부 데이터 소스나 도구에 표준 프로토콜로 연결해줍니다. 쉽게 말하면, 우리가 만든 RAG 시스템을 MCP 서버로 포장하면, Claude Desktop이나 Claude Code에서 바로 연결해서 쓸 수 있습니다. "우리 회사 휴가 규정이 뭐야?"라고 Claude에게 물어보면, Claude가 MCP 서버를 통해 사내 문서를 검색하고 답변합니다. 사용자 입장에서는 Claude에게 말을 거는 것뿐입니다.

이번 편에서는 MCP가 어떤 프로토콜이고 왜 RAG 시스템의 마지막 퍼즐인지 설명하고, FastMCP 3.0으로 3편까지 만든 검색 파이프라인을 MCP 서버로 변환합니다. Claude Desktop과 Claude Code에 연결하는 것은 물론, 인증 처리와 프로덕션 배포까지 다룹니다.

MCP가 필요한 이유

RAG 시스템을 만들고 나면 "이걸 어떻게 사용자에게 전달하느냐"라는 문제에 부딪힙니다. 웹 UI를 직접 만들거나, Slack 봇을 개발하거나, API 서버를 띄우고 프론트엔드를 붙이거나. 어떤 방식이든 추가 개발이 필요하고, 소규모 기업에서 RAG 파이프라인 구축에 이미 시간을 들인 뒤에 사용자 인터페이스까지 따로 만드는 건 꽤 부담스럽습니다.

MCP가 이 문제를 깔끔하게 풀어줍니다. Claude Desktop, Claude Code, Cursor 같은 AI 클라이언트가 이미 잘 만들어진 대화형 인터페이스를 갖고 있으니, 우리는 데이터와 도구만 MCP 표준에 맞게 내보내면 됩니다. 인터페이스는 AI 클라이언트가 알아서 처리하고요.

Anthropic이 2024년 11월에 MCP를 발표한 이후 채택이 빠르게 늘었습니다. 2025년 기준으로 MCP 채택은 340% 증가했고, 500개 이상의 공개 MCP 서버가 등록되어 있습니다. Cursor, Figma, Replit, Sourcegraph가 MCP 클라이언트를 지원하고, Claude Desktop과 Claude Code는 기본 지원합니다.

구조부터 보겠습니다. MCP는 클라이언트-서버 아키텍처입니다. Claude Desktop 같은 AI 앱이 클라이언트, 우리가 만드는 RAG 시스템이 서버입니다. 둘 사이에 JSON-RPC 2.0 기반의 메시지가 오가고요. 통신 방식은 두 가지인데, 로컬에서는 stdio(표준 입출력)를 쓰고 원격 서버와는 Server-Sent Events(SSE)를 씁니다.

MCP 서버가 내보낼 수 있는 것은 세 가지입니다.

Resources는 읽기 전용 데이터입니다. 파일 내용, 데이터베이스 레코드, 설정 정보처럼 AI가 참조할 수 있는 정보를 URI 형식으로 제공합니다. company://policies/vacation이면 휴가 정책 문서인 식이죠.

Tools는 AI가 호출할 수 있는 함수입니다. 검색, 계산, 외부 API 호출 같은 작업을 처리하는데, 우리의 RAG 검색 파이프라인이 여기에 해당합니다. Claude가 사용자의 질문을 받으면 search_documents라는 Tool을 호출해서 관련 문서를 찾아옵니다.

Prompts는 재사용 가능한 프롬프트 템플릿입니다. 자주 쓰는 질문 패턴이나 분석 틀을 미리 정의해둘 수 있는데, "문서 요약" 프롬프트를 등록해두면 사용자가 한 번의 클릭으로 호출할 수 있습니다.

우리의 RAG MCP 서버에서는 주로 Tools를 씁니다. 문서 검색, 유사 문서 탐색, 컬렉션 목록 조회 같은 기능을 Tool로 등록하고, Claude가 사용자 질문에 따라 적절한 Tool을 골라 호출하게 합니다.

다음 그림은 MCP를 통해 Claude가 사내 RAG 시스템에 접속하는 전체 흐름입니다.

사용자가 Claude Desktop에서 질문하면, Claude가 MCP를 통해 RAG 서버의 검색 도구를 호출합니다. RAG 서버는 하이브리드 검색과 리랭킹을 거쳐 관련 문서를 찾고, 그 결과를 Claude에게 돌려줍니다. Claude는 받은 문서를 바탕으로 자연어 답변을 만들어 사용자에게 전달합니다.

FastMCP 3.0으로 RAG MCP 서버 구현

FastMCP는 MCP 서버를 Python으로 간단하게 만들 수 있는 프레임워크입니다. 데코레이터로 Tool, Resource, Prompt를 등록하면 되고, MCP 프로토콜의 세부 사항은 프레임워크가 알아서 처리해줍니다. 2026년 1월에 나온 3.0 버전에서 컴포넌트 버전 관리, 세분화된 인증, OpenTelemetry 계측이 추가되었습니다.

먼저 필요한 패키지를 설치합니다.

복사
pip install "mcp[cli]" fastmcp chromadb sentence-transformers

MCP 서버의 전체 코드입니다. 2편의 데이터 수집 파이프라인과 3편의 검색 파이프라인을 MCP Tool로 감싸는 구조입니다.

복사
# rag_mcp_server.py
"""사내 문서 검색 MCP 서버.

2-3편에서 구축한 RAG 파이프라인을 MCP 프로토콜로 노출합니다.
Claude Desktop, Claude Code, Cursor 등에서 연결하여 사용할 수 있습니다.
"""
import os
import json
import logging
from pathlib import Path
from dataclasses import dataclass, field

from mcp.server.fastmcp import FastMCP

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ──────────────────────────────────────────────
# 설정
# ──────────────────────────────────────────────

CHROMA_PERSIST_DIR = os.getenv("CHROMA_PERSIST_DIR", "./chroma_db")
OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434")
EMBED_MODEL = os.getenv("EMBED_MODEL", "nomic-embed-text")
LLM_MODEL = os.getenv("LLM_MODEL", "llama3.2")
RERANK_MODEL = os.getenv("RERANK_MODEL", "cross-encoder/ms-marco-MiniLM-L-6-v2")
TOP_K = int(os.getenv("TOP_K", "5"))

# ──────────────────────────────────────────────
# MCP 서버 인스턴스
# ──────────────────────────────────────────────

mcp = FastMCP(
    "Company Knowledge Base",
    description="사내 문서를 검색하고 질문에 답변하는 RAG 시스템",
)

# ──────────────────────────────────────────────
# RAG 파이프라인 컴포넌트 (2-3편 코드 재활용)
# ──────────────────────────────────────────────

# 전역 변수로 지연 초기화. MCP 서버가 뜰 때 한 번만 로드합니다.
_pipeline = None

def _get_pipeline():
    """검색 파이프라인을 지연 초기화합니다.
    
    ChromaDB 로드, BM25 인덱스 구축, 리랭킹 모델 로드를
    서버 시작 시 한 번만 수행합니다.
    """
    global _pipeline
    if _pipeline is not None:
        return _pipeline

    import chromadb
    from chromadb.utils.embedding_functions import OllamaEmbeddingFunction
    from sentence_transformers import CrossEncoder

    logger.info("RAG 파이프라인 초기화 시작...")

    # Ollama 연결 확인
    import urllib.request
    try:
        urllib.request.urlopen(f"{OLLAMA_BASE_URL}/api/tags", timeout=5)
        logger.info("  Ollama 연결 확인 완료")
    except Exception:
        raise RuntimeError(
            f"Ollama에 연결할 수 없습니다 ({OLLAMA_BASE_URL}). "
            "Ollama가 실행 중인지 확인하세요: ollama serve"
        )

    # ChromaDB 클라이언트
    chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR)
    embed_fn = OllamaEmbeddingFunction(
        url=f"{OLLAMA_BASE_URL}/api/embed",
        model_name=EMBED_MODEL,
    )

    # 컬렉션 목록
    collections = {}
    for col in chroma_client.list_collections():
        collections[col.name] = chroma_client.get_collection(
            name=col.name,
            embedding_function=embed_fn,
        )

    # 기본 컬렉션 (없으면 첫 번째 사용)
    default_collection_name = os.getenv(
        "DEFAULT_COLLECTION", 
        next(iter(collections), None)
    )

    # BM25 인덱스 구축
    from bm25_retriever import BM25Retriever, BM25Document
    bm25_indices = {}
    for name, col in collections.items():
        all_docs = col.get(include=["documents", "metadatas"])
        bm25 = BM25Retriever()
        docs = []
        for i, (doc_id, content) in enumerate(
            zip(all_docs["ids"], all_docs["documents"])
        ):
            metadata = all_docs["metadatas"][i] if all_docs["metadatas"] else {}
            docs.append(BM25Document(
                doc_id=doc_id, content=content, metadata=metadata
            ))
        bm25.index(docs)
        bm25_indices[name] = bm25
        logger.info(f"  BM25 인덱스 구축: {name} ({len(docs)}개 문서)")

    # 크로스인코더 리랭킹 모델
    reranker = CrossEncoder(RERANK_MODEL)
    logger.info(f"  리랭커 로드 완료: {RERANK_MODEL}")

    _pipeline = {
        "chroma_client": chroma_client,
        "collections": collections,
        "default_collection": default_collection_name,
        "bm25_indices": bm25_indices,
        "reranker": reranker,
        "embed_fn": embed_fn,
    }
    logger.info("RAG 파이프라인 초기화 완료")
    return _pipeline

def _hybrid_search(query: str, collection_name: str, top_k: int = 20):
    """하이브리드 검색 (벡터 + BM25 + RRF).
    
    3편에서 구현한 하이브리드 검색 로직을 그대로 사용합니다.
    """
    pipeline = _get_pipeline()
    collection = pipeline["collections"][collection_name]
    bm25 = pipeline["bm25_indices"][collection_name]

    # 1. 벡터 검색
    vector_results = collection.query(
        query_texts=[query],
        n_results=top_k,
        include=["documents", "metadatas", "distances"],
    )

    # 2. BM25 검색
    bm25_results = bm25.search(query, top_k=top_k)

    # 3. RRF 병합
    rrf_scores = {}
    k = 60  # RRF 상수

    # 벡터 검색 결과에 RRF 점수 부여
    for rank, doc_id in enumerate(vector_results["ids"][0]):
        rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + 1 / (k + rank + 1)

    # BM25 결과에 RRF 점수 부여
    for rank, (doc_id, _) in enumerate(bm25_results):
        rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + 1 / (k + rank + 1)

    # RRF 점수 기준 정렬
    sorted_ids = sorted(rrf_scores.keys(), key=lambda x: rrf_scores[x], reverse=True)

    # 상위 결과의 문서 내용 가져오기
    results = []
    seen = set()
    for doc_id in sorted_ids[:top_k]:
        if doc_id in seen:
            continue
        seen.add(doc_id)

        # ChromaDB에서 문서 내용 조회
        doc = collection.get(ids=[doc_id], include=["documents", "metadatas"])
        if doc["documents"]:
            results.append({
                "id": doc_id,
                "content": doc["documents"][0],
                "metadata": doc["metadatas"][0] if doc["metadatas"] else {},
                "rrf_score": rrf_scores[doc_id],
            })

    return results

def _rerank(query: str, documents: list[dict], top_k: int = 5):
    """크로스인코더 리랭킹.
    
    3편의 리랭킹 로직입니다. 쿼리-문서 쌍을 크로스인코더로 평가해서 재정렬합니다.
    """
    pipeline = _get_pipeline()
    reranker = pipeline["reranker"]

    if not documents:
        return []

    pairs = [(query, doc["content"]) for doc in documents]
    scores = reranker.predict(pairs)

    for doc, score in zip(documents, scores):
        doc["rerank_score"] = float(score)

    reranked = sorted(documents, key=lambda x: x["rerank_score"], reverse=True)
    return reranked[:top_k]

# ──────────────────────────────────────────────
# MCP Tools
# ──────────────────────────────────────────────

@mcp.tool()
def search_documents(
    query: str,
    collection: str = "",
    top_k: int = 5,
) -> str:
    """사내 문서를 검색합니다.

    하이브리드 검색(벡터 + BM25)과 크로스인코더 리랭킹을 거쳐
    가장 관련성 높은 문서를 반환합니다.

    Args:
        query: 검색할 질문 또는 키워드
        collection: 검색할 컬렉션 이름 (비우면 기본 컬렉션)
        top_k: 반환할 문서 수 (기본값 5)
    """
    pipeline = _get_pipeline()
    col_name = collection or pipeline["default_collection"]

    if col_name not in pipeline["collections"]:
        available = ", ".join(pipeline["collections"].keys())
        return json.dumps({
            "error": f"컬렉션 '{col_name}'을 찾을 수 없습니다. "
                     f"사용 가능한 컬렉션: {available}"
        }, ensure_ascii=False)

    # 하이브리드 검색
    hybrid_results = _hybrid_search(query, col_name, top_k=20)

    # 리랭킹
    reranked = _rerank(query, hybrid_results, top_k=top_k)

    # 결과 포맷팅
    output = []
    for i, doc in enumerate(reranked):
        source = doc["metadata"].get("source", "알 수 없음")
        output.append({
            "rank": i + 1,
            "content": doc["content"],
            "source": source,
            "relevance_score": round(doc["rerank_score"], 4),
        })

    return json.dumps({
        "query": query,
        "collection": col_name,
        "total_results": len(output),
        "results": output,
    }, ensure_ascii=False, indent=2)

@mcp.tool()
def list_collections() -> str:
    """사용 가능한 문서 컬렉션 목록을 반환합니다.

    각 컬렉션의 이름과 포함된 문서 수를 보여줍니다.
    """
    pipeline = _get_pipeline()
    collections_info = []
    for name, col in pipeline["collections"].items():
        count = col.count()
        collections_info.append({
            "name": name,
            "document_count": count,
            "is_default": name == pipeline["default_collection"],
        })

    return json.dumps({
        "collections": collections_info,
        "total": len(collections_info),
    }, ensure_ascii=False, indent=2)

@mcp.tool()
def find_similar_documents(
    document_id: str,
    collection: str = "",
    top_k: int = 5,
) -> str:
    """특정 문서와 유사한 문서를 찾습니다.

    문서 ID를 기반으로 해당 문서와 내용이 유사한 다른 문서를 검색합니다.
    관련 규정이나 연관 자료를 찾을 때 유용합니다.

    Args:
        document_id: 기준 문서의 ID
        collection: 검색할 컬렉션 이름 (비우면 기본 컬렉션)
        top_k: 반환할 유사 문서 수 (기본값 5)
    """
    pipeline = _get_pipeline()
    col_name = collection or pipeline["default_collection"]

    if col_name not in pipeline["collections"]:
        return json.dumps({"error": f"컬렉션 '{col_name}'을 찾을 수 없습니다."}, 
                         ensure_ascii=False)

    col = pipeline["collections"][col_name]

    # 기준 문서 조회
    source_doc = col.get(ids=[document_id], include=["documents", "embeddings"])
    if not source_doc["documents"]:
        return json.dumps({"error": f"문서 '{document_id}'를 찾을 수 없습니다."}, 
                         ensure_ascii=False)

    # 기준 문서의 임베딩으로 유사 문서 검색
    results = col.query(
        query_embeddings=source_doc["embeddings"],
        n_results=top_k + 1,  # 자기 자신 제외
        include=["documents", "metadatas", "distances"],
    )

    output = []
    for i, doc_id in enumerate(results["ids"][0]):
        if doc_id == document_id:
            continue
        output.append({
            "rank": len(output) + 1,
            "id": doc_id,
            "content": results["documents"][0][i][:500] + "...",
            "source": results["metadatas"][0][i].get("source", "알 수 없음"),
            "similarity": round(1 - results["distances"][0][i], 4),
        })
        if len(output) >= top_k:
            break

    return json.dumps({
        "source_document": document_id,
        "similar_documents": output,
    }, ensure_ascii=False, indent=2)

@mcp.tool()
def get_document(
    document_id: str,
    collection: str = "",
) -> str:
    """특정 문서의 전체 내용을 조회합니다.

    문서 ID로 해당 문서의 전체 텍스트와 메타데이터를 반환합니다.
    검색 결과에서 특정 문서를 더 자세히 보고 싶을 때 사용합니다.

    Args:
        document_id: 조회할 문서의 ID
        collection: 컬렉션 이름 (비우면 기본 컬렉션)
    """
    pipeline = _get_pipeline()
    col_name = collection or pipeline["default_collection"]

    if col_name not in pipeline["collections"]:
        return json.dumps({"error": f"컬렉션 '{col_name}'을 찾을 수 없습니다."}, 
                         ensure_ascii=False)

    col = pipeline["collections"][col_name]
    doc = col.get(ids=[document_id], include=["documents", "metadatas"])

    if not doc["documents"]:
        return json.dumps({"error": f"문서 '{document_id}'를 찾을 수 없습니다."}, 
                         ensure_ascii=False)

    return json.dumps({
        "id": document_id,
        "content": doc["documents"][0],
        "metadata": doc["metadatas"][0] if doc["metadatas"] else {},
    }, ensure_ascii=False, indent=2)

# ──────────────────────────────────────────────
# MCP Resources
# ──────────────────────────────────────────────

@mcp.resource("knowledge://collections")
def get_collections_resource() -> str:
    """전체 컬렉션 현황을 Resource로 노출합니다."""
    return list_collections()

@mcp.resource("knowledge://status")
def get_system_status() -> str:
    """RAG 시스템의 현재 상태를 반환합니다."""
    pipeline = _get_pipeline()
    total_docs = sum(
        col.count() for col in pipeline["collections"].values()
    )
    return json.dumps({
        "status": "running",
        "total_collections": len(pipeline["collections"]),
        "total_documents": total_docs,
        "embedding_model": EMBED_MODEL,
        "llm_model": LLM_MODEL,
        "rerank_model": RERANK_MODEL,
    }, ensure_ascii=False, indent=2)

# ──────────────────────────────────────────────
# MCP Prompts
# ──────────────────────────────────────────────

@mcp.prompt()
def summarize_search(query: str) -> str:
    """검색 결과를 요약하는 프롬프트 템플릿.

    사용자가 문서를 검색한 뒤, 결과를 간결하게 정리해달라고 할 때 사용합니다.
    """
    return f"""다음 질문에 대해 사내 문서를 검색한 결과를 분석하세요.

질문: {query}

search_documents 도구로 "{query}"를 검색한 뒤, 결과를 다음 형식으로 정리하세요.

1. 핵심 답변 (2-3문장)
2. 근거 문서 목록 (출처 포함)
3. 추가로 확인이 필요한 사항 (있는 경우)

답변은 검색된 문서의 내용만을 근거로 하세요. 문서에 없는 내용은 "해당 정보를 찾을 수 없습니다"라고 명시하세요."""

@mcp.prompt()
def compare_documents(doc_id_1: str, doc_id_2: str) -> str:
    """두 문서를 비교 분석하는 프롬프트 템플릿."""
    return f"""다음 두 문서를 비교 분석하세요.

get_document 도구로 두 문서를 각각 조회하세요.
- 문서 1: {doc_id_1}
- 문서 2: {doc_id_2}

비교 항목:
1. 두 문서의 주제와 범위
2. 공통점과 차이점
3. 상충되는 내용이 있는지 여부
4. 어떤 문서가 더 최신 정보를 담고 있는지"""

# ──────────────────────────────────────────────
# 서버 실행
# ──────────────────────────────────────────────

if __name__ == "__main__":
    mcp.run()

로컬 실행에서는 mcp.run()이 stdio 전송을 사용합니다. Claude Desktop이 이 스크립트를 직접 실행하고 표준 입출력으로 통신하는 방식입니다. 원격 서버에서 Docker로 배포할 때는 SSE 전송으로 전환해야 합니다. 이 부분은 Docker 배포 섹션에서 다룹니다.

코드 구조는 직관적입니다. FastMCP 인스턴스를 만들고, @mcp.tool(), @mcp.resource(), @mcp.prompt() 데코레이터로 각 기능을 등록하면 됩니다. 데코레이터를 붙이면 함수의 이름, 독스트링, 파라미터 타입 힌트가 MCP 스키마로 자동 변환되고, Claude가 이 스키마를 보고 어떤 도구를 언제 호출할지 판단합니다.

_get_pipeline() 함수가 핵심인데, 서버가 처음 요청을 받을 때 ChromaDB 클라이언트, BM25 인덱스, 크로스인코더 모델을 한 번만 로드합니다. 이후 요청에서는 이미 로드된 객체를 재사용하는 거죠. ChromaDB 로드에 1~2초, BM25 인덱싱에 문서 수에 비례하는 시간, 크로스인코더 모델 로드에 3~5초 정도 걸리니까, 첫 요청만 느리고 그 다음부터는 빠릅니다.

search_documents가 가장 중요한 Tool입니다. 사용자 질문을 받아서 하이브리드 검색(벡터 + BM25 + RRF) 후 크로스인코더 리랭킹을 거쳐 상위 결과를 반환하는데, 3편에서 구현한 로직을 그대로 쓰고 있습니다. 반환값은 JSON 문자열이고, Claude가 이 JSON을 읽고 사용자에게 자연어로 풀어서 답해줍니다.

여기서 중요한 설계 결정이 하나 있습니다. MCP Tool에서 RAG 파이프라인의 검색 단계만 내보내고, LLM 생성 단계는 빼놓았습니다. 답변 생성은 Claude가 직접 하기 때문입니다. MCP 서버가 할 일은 관련 문서를 정확하게 찾아서 Claude에게 건네주는 것까지고, 그걸 바탕으로 답변을 만드는 건 Claude의 몫입니다. 이렇게 하면 Ollama LLM을 따로 실행하지 않아도 되니 리소스를 아낄 수 있습니다. 임베딩 모델만 있으면 충분하거든요.

"가드레일은 어떻게 됐나?" 하는 질문이 나올 수 있습니다. 3편에서 구현한 2단계 가드레일(시맨틱 유사도 + LLM Judge)은 RAG 시스템이 직접 답변을 생성할 때 그 신뢰성을 검증하는 용도였습니다. MCP 서버에서는 답변 생성을 Claude에게 맡기니까, 가드레일의 역할이 달라집니다. 검색 결과의 관련성만 확인하면 되는 셈이죠. 리랭킹 점수가 이미 관련성 지표 역할을 하고 있어서, MCP 서버에서는 별도의 가드레일 없이 리랭킹 점수를 함께 반환하는 방식을 택했습니다. 점수가 낮은 결과는 Claude가 알아서 답변에 반영하지 않습니다.

MCP 서버 테스트

Claude Desktop에 연결하기 전에 서버가 제대로 작동하는지부터 확인해봐야 합니다. FastMCP는 mcp dev 명령으로 개발 모드를 제공하거든요.

복사
# MCP Inspector로 서버 테스트
mcp dev rag_mcp_server.py

이 명령을 실행하면 브라우저에서 MCP Inspector가 열립니다. 등록된 Tool, Resource, Prompt 목록을 확인하고, 직접 호출해서 결과를 볼 수 있는데요. search_documents에 테스트 쿼리를 넣어보고 JSON 응답이 제대로 나오는지 확인하면 됩니다. Ollama가 꺼져 있거나 ChromaDB 경로가 잘못되면 여기서 바로 에러가 나타납니다.

Python 코드에서 직접 테스트할 수도 있습니다.

복사
# test_mcp_server.py
"""MCP 서버 도구를 직접 호출하여 테스트합니다."""
import json

# 서버 모듈에서 Tool 함수를 직접 import
from rag_mcp_server import search_documents, list_collections

# 컬렉션 목록 확인
print("=== 컬렉션 목록 ===")
print(list_collections())

# 검색 테스트
print("\n=== 검색 테스트 ===")
result = search_documents("연차 휴가 규정")
parsed = json.loads(result)
print(f"검색 결과: {parsed['total_results']}건")
for r in parsed.get("results", []):
    print(f"  [{r['rank']}] score={r['relevance_score']} source={r['source']}")

테스트가 통과하면 Claude Desktop에 연결합니다.

Claude Desktop에 연결하기

MCP 서버 코드가 준비되었으니 Claude Desktop에 연결할 차례입니다. 설정 파일에 MCP 서버를 등록하면 자동으로 연결됩니다.

macOS 기준으로 설정 파일 위치는 ~/Library/Application Support/Claude/claude_desktop_config.json입니다. Windows는 %APPDATA%\Claude\claude_desktop_config.json입니다.

복사
{
  "mcpServers": {
    "company-knowledge": {
      "command": "python",
      "args": ["/path/to/rag_mcp_server.py"],
      "env": {
        "CHROMA_PERSIST_DIR": "/path/to/chroma_db",
        "OLLAMA_BASE_URL": "http://localhost:11434",
        "EMBED_MODEL": "nomic-embed-text",
        "RERANK_MODEL": "cross-encoder/ms-marco-MiniLM-L-6-v2",
        "DEFAULT_COLLECTION": "company_docs"
      }
    }
  }
}

command에 Python 경로, args에 서버 스크립트 경로를 지정하고, env에 환경 변수를 넣어서 ChromaDB 경로나 모델 이름을 설정합니다. Claude Desktop을 재시작하면 MCP 서버가 자동으로 뜹니다.

연결이 되면 Claude Desktop의 입력 창 옆에 도구 아이콘이 나타납니다. 클릭하면 search_documents, list_collections, find_similar_documents, get_document 네 개의 도구가 등록된 것을 확인할 수 있습니다.

Claude Code에서 연결하려면 프로젝트 디렉토리에 .claude/settings.json 파일을 만들면 됩니다.

복사
{
  "mcpServers": {
    "company-knowledge": {
      "command": "python",
      "args": ["rag_mcp_server.py"],
      "env": {
        "CHROMA_PERSIST_DIR": "./chroma_db"
      }
    }
  }
}

Claude Code를 실행하면 MCP 서버가 자동으로 시작되고, 대화 중에 사내 문서를 검색할 수 있습니다.

실제로 어떻게 되는지 보겠습니다. Claude Desktop에서 "우리 회사 연차 규정에서 3년차 이상은 연차가 며칠이야?"라고 질문하면, Claude는 질문을 분석해서 search_documents Tool을 호출합니다. query는 "연차 규정 3년차 이상 일수" 정도가 되겠죠. MCP 서버가 하이브리드 검색과 리랭킹을 돌려서 관련 문서 5개를 JSON으로 반환하고, Claude가 그 문서를 읽고 "인사규정 제15조에 따르면 3년차 이상 직원의 연차 유급휴가는 15일입니다" 같은 구체적인 답변을 만들어줍니다.

사용자 입장에서는 Claude에게 한국어로 질문한 것뿐입니다. RAG 시스템이 뒤에서 돌고 있다는 걸 의식할 필요가 없죠.

인증과 권한 관리

사내 문서에는 민감한 정보가 섞여 있기 마련입니다. 급여 정보, 인사 평가, 계약 조건 같은 것이 누구에게나 검색되면 곤란하죠. MCP 서버에 인증을 추가해야 합니다.

로컬 실행(stdio 방식)에서는 MCP 서버가 사용자의 로컬 머신에서 돌아가니까, 운영체제 수준의 접근 제어로 충분합니다. 문서 컬렉션을 팀별로 분리하고, 각 사용자의 Claude Desktop 설정에서 접근 가능한 컬렉션만 지정하면 됩니다.

복사
{
  "mcpServers": {
    "company-knowledge-hr": {
      "command": "python",
      "args": ["rag_mcp_server.py"],
      "env": {
        "CHROMA_PERSIST_DIR": "./chroma_db",
        "DEFAULT_COLLECTION": "hr_policies",
        "ALLOWED_COLLECTIONS": "hr_policies,company_general"
      }
    }
  }
}

서버 코드에 컬렉션 접근 제한을 추가합니다.

복사
# 컬렉션 접근 제한 추가
ALLOWED_COLLECTIONS = os.getenv("ALLOWED_COLLECTIONS", "")

def _get_allowed_collections():
    """허용된 컬렉션 목록을 반환합니다."""
    if not ALLOWED_COLLECTIONS:
        return None  # 제한 없음
    return set(ALLOWED_COLLECTIONS.split(","))

@mcp.tool()
def search_documents(
    query: str,
    collection: str = "",
    top_k: int = 5,
) -> str:
    """사내 문서를 검색합니다."""
    pipeline = _get_pipeline()
    col_name = collection or pipeline["default_collection"]

    # 접근 권한 확인
    allowed = _get_allowed_collections()
    if allowed and col_name not in allowed:
        return json.dumps({
            "error": f"컬렉션 '{col_name}'에 대한 접근 권한이 없습니다."
        }, ensure_ascii=False)

    # 이하 동일...

원격 배포(SSE 방식)에서는 API 키 인증이 필요합니다. 여러 사용자가 하나의 MCP 서버에 접속하는 구조이므로, 요청마다 사용자를 식별하고 권한을 확인해야 합니다.

FastMCP 3.0에서 SSE 전송과 인증 미들웨어를 추가하는 코드입니다.

복사
# rag_mcp_server_remote.py
"""원격 접속을 지원하는 MCP 서버.

SSE 전송과 API 키 인증을 추가합니다.
"""
import os
import json
import hashlib
import hmac
from functools import wraps

from mcp.server.fastmcp import FastMCP

# API 키 저장소 (프로덕션에서는 DB 또는 비밀 관리 서비스 사용)
API_KEYS = {
    # "키": {"user": "사용자명", "collections": ["허용_컬렉션"]}
}

# 환경 변수에서 API 키 로드
def _load_api_keys():
    """API_KEYS_FILE 환경 변수에서 키 파일을 로드합니다."""
    keys_file = os.getenv("API_KEYS_FILE", "./api_keys.json")
    if os.path.exists(keys_file):
        with open(keys_file) as f:
            return json.load(f)
    return {}

API_KEYS = _load_api_keys()

mcp = FastMCP(
    "Company Knowledge Base",
    description="사내 문서 검색 RAG 시스템 (인증 필요)",
)

def generate_api_key(user: str, collections: list[str]) -> str:
    """새 API 키를 생성합니다.
    
    관리자가 사용자별 API 키를 발급할 때 사용합니다.
    """
    import secrets
    key = secrets.token_urlsafe(32)
    key_hash = hashlib.sha256(key.encode()).hexdigest()

    # 키 저장
    API_KEYS[key_hash] = {
        "user": user,
        "collections": collections,
    }

    # 파일에 저장
    keys_file = os.getenv("API_KEYS_FILE", "./api_keys.json")
    with open(keys_file, "w") as f:
        json.dump(API_KEYS, f, indent=2, ensure_ascii=False)

    return key

def verify_api_key(key: str) -> dict | None:
    """API 키를 검증하고 사용자 정보를 반환합니다."""
    key_hash = hashlib.sha256(key.encode()).hexdigest()
    return API_KEYS.get(key_hash)

솔직히 말하면, MCP 프로토콜의 인증 방식은 아직 표준이 잡히는 중입니다. FastMCP 3.0이 세분화된 인증을 지원한다고는 하지만, Claude Desktop이 SSE 연결 시 인증 헤더를 전달하는 방식은 클라이언트마다 다를 수 있습니다. 소규모 기업에서 현실적인 접근은 앞서 보여드린 ALLOWED_COLLECTIONS 환경 변수 방식으로 사용자별 접근 범위를 제한하고, 원격 배포 시에는 VPN이나 네트워크 수준의 접근 제어를 병행하는 겁니다. API 키 기반 인증은 MCP 프로토콜의 인증 표준이 확정된 후에 도입해도 늦지 않습니다.

API 키 파일(api_keys.json)의 구조는 다음과 같습니다.

복사
{
  "a1b2c3...해시값": {
    "user": "김인사",
    "collections": ["hr_policies", "company_general"]
  },
  "d4e5f6...해시값": {
    "user": "이영업",
    "collections": ["sales_docs", "company_general"]
  }
}

관리자가 generate_api_key("김인사", ["hr_policies", "company_general"])를 호출하면 API 키가 생성되고, 이 키를 해당 사용자의 Claude Desktop 설정에 넣으면 됩니다. 소규모 기업에서는 이 정도면 충분합니다. 사용자가 수십 명 이상으로 늘어나면 그때 OAuth 2.0이나 SSO 통합을 고려하면 됩니다.

Docker로 프로덕션 배포

개발 환경에서는 python rag_mcp_server.py로 직접 실행하면 되지만, 프로덕션에서는 Docker로 패키징하는 편이 관리하기 수월합니다.

복사
# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# 시스템 패키지
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

# Python 패키지
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 애플리케이션 코드
COPY rag_mcp_server.py .
COPY bm25_retriever.py .

# ChromaDB 데이터 볼륨
VOLUME /data/chroma_db

# 환경 변수 기본값
ENV CHROMA_PERSIST_DIR=/data/chroma_db
ENV OLLAMA_BASE_URL=http://host.docker.internal:11434
ENV EMBED_MODEL=nomic-embed-text
ENV RERANK_MODEL=cross-encoder/ms-marco-MiniLM-L-6-v2

EXPOSE 8080

CMD ["python", "rag_mcp_server.py", "--transport", "sse"]

Docker 컨테이너에서는 SSE 전송을 써야 합니다. stdio는 로컬 프로세스 간 통신이라 컨테이너 외부에서 접근이 안 되거든요. 서버 코드의 __main__ 블록을 다음과 같이 수정합니다.

복사
if __name__ == "__main__":
    import sys
    if "--transport" in sys.argv and "sse" in sys.argv:
        mcp.run(transport="sse", host="0.0.0.0", port=8080)
    else:
        mcp.run()  # 기본 stdio

로컬에서는 python rag_mcp_server.py로 stdio를 쓰고, Docker에서는 --transport sse 인자로 SSE를 씁니다. Claude Desktop에서 원격 MCP 서버에 연결하려면 설정 파일의 형식이 달라집니다.

복사
{
  "mcpServers": {
    "company-knowledge": {
      "url": "http://your-server:8080/sse"
    }
  }
}

복사
# requirements.txt
mcp[cli]>=1.0.0
fastmcp>=3.0.0
chromadb>=0.5.0
sentence-transformers>=3.0.0

Docker Compose로 Ollama와 함께 실행하는 설정입니다.

복사
# docker-compose.yml
services:
  ollama:
    image: ollama/ollama:latest
    ports:
      - "11434:11434"
    volumes:
      - ollama_data:/root/.ollama
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: all
              capabilities: [gpu]

  rag-mcp:
    build: .
    ports:
      - "8080:8080"
    volumes:
      - ./chroma_db:/data/chroma_db
      - ./api_keys.json:/app/api_keys.json
    environment:
      - OLLAMA_BASE_URL=http://ollama:11434
      - CHROMA_PERSIST_DIR=/data/chroma_db
      - API_KEYS_FILE=/app/api_keys.json
    depends_on:
      - ollama

volumes:
  ollama_data:

복사
# Ollama 모델 다운로드 (최초 1회)
docker compose up ollama -d
docker compose exec ollama ollama pull nomic-embed-text

# 전체 스택 실행
docker compose up -d

GPU가 없는 서버에서는 deploy.resources 섹션을 제거하면 CPU 모드로 돌아갑니다. 소규모 기업의 쿼리 볼륨(하루 수백 건)이면 CPU 모드로도 충분히 쓸 만합니다. 임베딩 검색과 BM25 검색은 CPU에서도 빠르고, 가장 느린 구간인 크로스인코더 리랭킹도 10개 문서 기준 200~500ms 정도입니다.

문서 업데이트 자동화

RAG 시스템에서 빠지기 쉬운 함정이 문서 업데이트입니다. 1편에서 다뤘던 "Knowledge Drift" 문제인데, 문서가 변경되었는데 벡터DB는 이전 버전을 가리키고 있으면 시스템이 오래된 정보를 자신있게 답하는 상황이 벌어집니다.

MCP 서버에 문서 수집을 트리거하는 Tool을 추가합니다.

복사
@mcp.tool()
def refresh_collection(
    collection: str,
    source_dir: str = "",
) -> str:
    """문서 컬렉션을 새로 인덱싱합니다.

    지정된 디렉토리의 문서를 다시 읽어서 벡터DB를 갱신합니다.
    문서가 추가, 수정, 삭제된 후에 실행하세요.

    Args:
        collection: 갱신할 컬렉션 이름
        source_dir: 문서가 있는 디렉토리 경로 (비우면 기존 경로 사용)
    """
    pipeline = _get_pipeline()
    
    if collection not in pipeline["collections"]:
        return json.dumps({
            "error": f"컬렉션 '{collection}'을 찾을 수 없습니다."
        }, ensure_ascii=False)

    # 2편의 수집 파이프라인 호출
    try:
        from document_loader import load_documents
        from chunker import chunk_documents
        from contextual_retrieval import add_context
        
        # 문서 로드
        docs = load_documents(source_dir)
        
        # 청킹
        chunks = chunk_documents(docs)
        
        # 컨텍스트 추가
        enriched = add_context(chunks, docs)
        
        # ChromaDB 컬렉션 갱신
        col = pipeline["collections"][collection]
        
        # 기존 문서 삭제 후 새로 추가
        existing = col.get()
        if existing["ids"]:
            col.delete(ids=existing["ids"])
        
        col.add(
            ids=[c["id"] for c in enriched],
            documents=[c["content"] for c in enriched],
            metadatas=[c["metadata"] for c in enriched],
        )
        
        # BM25 인덱스 재구축
        from bm25_retriever import BM25Retriever, BM25Document
        bm25 = BM25Retriever()
        bm25_docs = [
            BM25Document(doc_id=c["id"], content=c["content"], metadata=c["metadata"])
            for c in enriched
        ]
        bm25.index(bm25_docs)
        pipeline["bm25_indices"][collection] = bm25
        
        return json.dumps({
            "status": "success",
            "collection": collection,
            "documents_indexed": len(enriched),
        }, ensure_ascii=False)
        
    except Exception as e:
        logger.error(f"컬렉션 갱신 실패: {e}")
        return json.dumps({
            "error": f"갱신 실패: {str(e)}"
        }, ensure_ascii=False)

이 Tool을 등록하면 Claude에게 "영업팀 문서 컬렉션을 새로 인덱싱해줘"라고 말하는 것으로 문서 업데이트가 가능합니다.

주의할 점이 하나 있습니다. 위 코드는 기존 문서를 전부 삭제한 뒤 새로 추가하는 방식입니다. 재인덱싱이 진행되는 동안(문서 수에 따라 수초~수분) 검색 요청이 들어오면 빈 컬렉션에서 검색하게 되어 결과가 안 나옵니다. 동시 접속이 적은 소규모 기업이라면 큰 문제가 아닐 수 있지만, 서비스 중단을 피하려면 새 컬렉션에 인덱싱을 완료한 뒤 교체하는 블루-그린 방식이 낫습니다. 새 컬렉션 이름에 타임스탬프를 붙여서(예: company_docs_20260301) 만들고, 인덱싱이 끝나면 DEFAULT_COLLECTION 환경 변수를 교체하고 서버를 재시작하는 식이죠.

정기적인 자동 업데이트를 원한다면 cron이나 systemd timer를 사용합니다.

복사
# crontab -e
# 매일 새벽 2시에 문서 재인덱싱
0 2 * * * cd /path/to/rag && python pipeline.py --input-dir ./documents --persist-dir ./chroma_db >> /var/log/rag-reindex.log 2>&1

시리즈 전체 아키텍처 회고

4편에 걸쳐 구축한 시스템의 전체 모습을 돌아봅니다.

다음 그림은 1편부터 4편까지 구축한 전체 RAG 아키텍처를 한눈에 보여줍니다.

1편에서 기본 RAG의 실패 원인을 분석하고 프로덕션급 아키텍처의 방향을 잡았습니다. 2편에서 문서 전처리, 시맨틱 청킹, Contextual Retrieval, Ollama 임베딩으로 데이터 수집 파이프라인을 만들었습니다. 3편에서 하이브리드 검색, 크로스인코더 리랭킹, 가드레일, RAGAS 평가, Langfuse 모니터링으로 검색 파이프라인을 완성했습니다. 4편에서 이 모든 것을 MCP 서버로 포장해서 Claude와 연결했습니다.

각 편에서 다룬 컴포넌트와 운영 비용을 한 눈에 보겠습니다.

Ollama를 로컬에서 실행하고, ChromaDB와 Langfuse를 셀프 호스팅하면 추가 비용은 사실상 없습니다. 서버 하드웨어 비용은 별도이지만요. GPU 없이 CPU만으로도 전부 돌아가긴 하는데, 리랭킹과 임베딩 생성 속도가 느려집니다. M1 이상의 MacBook이나 16GB RAM 이상의 리눅스 서버면 소규모 기업의 쿼리 볼륨은 무난히 처리합니다.

Claude 쪽 비용도 짚어보면, MCP 서버가 검색만 하고 답변 생성은 Claude Desktop이 처리하는 구조이니 Claude Pro 구독($20/월)으로 상당한 양의 질의를 소화할 수 있습니다. 팀 전체가 쓰려면 Claude Team($30/사용자/월) 또는 Claude Enterprise를 검토해야 하고요. 이건 RAG 시스템 자체의 비용이 아니라 Claude 구독 비용이니까, RAG 인프라 운영비와는 별개입니다.

시리즈를 처음부터 따라하려는 독자를 위해 전체 프로젝트 구조와 빠른 시작 가이드를 정리합니다.

복사
rag-system/
├── document_loader.py       # 2편: 문서 로딩
├── chunker.py               # 2편: 시맨틱 청킹
├── contextual_retrieval.py  # 2편: Contextual Retrieval
├── pipeline.py              # 2편: 데이터 수집 파이프라인
├── bm25_retriever.py        # 3편: BM25 검색기
├── search_pipeline.py       # 3편: 검색 파이프라인 (CLI)
├── evaluator.py             # 3편: RAGAS 평가
├── rag_mcp_server.py        # 4편: MCP 서버
├── test_mcp_server.py       # 4편: MCP 서버 테스트
├── requirements.txt
├── Dockerfile
├── docker-compose.yml
├── documents/               # 사내 문서 디렉토리
│   ├── hr_policies/
│   └── general/
└── chroma_db/               # ChromaDB 저장소 (자동 생성)

복사
# 빠른 시작 (5분)
# 1. 의존성 설치
pip install -r requirements.txt

# 2. Ollama 모델 다운로드
ollama pull nomic-embed-text

# 3. 사내 문서 인덱싱 (2편)
python pipeline.py --input-dir ./documents --persist-dir ./chroma_db

# 4. MCP 서버 테스트 (4편)
python test_mcp_server.py

# 5. Claude Desktop 설정 후 사용
# ~/Library/Application Support/Claude/claude_desktop_config.json 편집

솔직한 평가를 덧붙이자면, 이 시리즈에서 구축한 시스템은 "소규모 기업이 현실적으로 구축하고 운영할 수 있는 프로덕션급 RAG"를 목표로 했습니다. 하지만 분명 한계가 있습니다.

지식 그래프를 통합하지 않았습니다. 1편에서 GraphRAG의 가능성을 소개했지만, 구현에서는 하이브리드 검색 + 리랭킹에 집중했습니다. 지식 그래프는 다중 홉 추론이 필요한 복잡한 질문에서 큰 차이를 만드는데, Neo4j 운영과 엔티티 추출 파이프라인까지 추가하면 소규모 기업에게는 부담이 됩니다. 문서 검색 수준에서는 하이브리드 검색 + 리랭킹으로 충분하고, 지식 그래프는 시스템이 안정적으로 돌아가고 더 복잡한 질문 처리가 필요해진 뒤에 도입해도 됩니다.

멀티모달도 빠져 있습니다. 이미지가 포함된 문서, 표와 차트에서 데이터를 추출하려면 별도의 파이프라인이 필요한데, LlamaParse가 테이블 추출은 지원하지만 차트나 다이어그램의 의미를 이해하는 건 아직 성숙하지 않은 영역입니다.

실시간 데이터 연동도 없습니다. Kafka나 Flink를 이용한 실시간 스트리밍 수집이 엔터프라이즈 RAG에서는 중요하지만, 소규모 기업에서는 일일 배치 갱신으로 충분한 경우가 대부분이거든요.

이 한계는 시스템이 성장하면서 필요에 따라 붙여나갈 수 있는 것이지, 처음부터 다 갖춰야 하는 건 아닙니다.

마무리

이 시리즈를 쓰게 된 계기는 개발자 커뮤니티에서 자주 올라오는 질문 때문이었습니다. "RAG를 만들었는데 답변 품질이 안 좋습니다. 어떻게 해야 하나요?" 댓글에 달리는 답은 대부분 비슷합니다. 청킹을 바꿔보세요, 하이브리드 검색을 써보세요, 리랭킹을 추가해보세요. 각각 맞는 말인데, 전체 아키텍처를 처음부터 끝까지 보여주는 글은 드물었습니다.

4편에 걸쳐 데이터 수집부터 사용자 서빙까지 전체 파이프라인을 코드와 함께 다뤘습니다. 핵심만 다시 짚으면, RAG 실패의 80%는 데이터 준비 단계에서 시작됩니다. 시맨틱 청킹과 Contextual Retrieval로 문맥을 보존하고, 하이브리드 검색과 리랭킹으로 검색 정확도를 올리고, 가드레일과 평가 체계로 신뢰성을 확보합니다. MCP로 이 모든 것을 Claude에 연결하면, 비개발자도 자연어로 회사 지식에 접근할 수 있게 됩니다.

개발자 커뮤니티에서 "RAG는 죽었다"는 말이 종종 나옵니다. CAG나 롱컨텍스트 LLM이 RAG를 대체한다는 건데, 1편에서 다뤘듯이 100만 토큰 이하의 정적 데이터에는 CAG가 맞습니다. 하지만 문서가 계속 업데이트되고, 여러 소스에서 들어오고, 수백만 토큰을 넘어가는 엔터프라이즈 환경에서는 여전히 RAG가 현실적인 유일한 선택입니다.

이 시리즈의 모든 코드는 Ollama 로컬 실행 기준이니까, 서버 한 대와 약간의 시간만 있으면 구축할 수 있습니다. 소규모 기업이라서 못 하는 게 아니라, 소규모 기업이기 때문에 오히려 빠르게 도입할 수 있습니다. 의사결정 과정이 짧고, 기존 시스템과의 통합 포인트가 적고, 데이터 규모가 관리 가능한 수준이니까요.

---

참고 자료

• Model Context Protocol - Anthropic 공식 소개
• RAG MCP Server Tutorial - Medium
• FastMCP 3.0 Tutorial - Firecrawl
• Integrating Agentic RAG with MCP Servers - Technical Implementation Guide
• MCP Server for RAG (Qdrant 기반) - LobeHub
• $5/month Production RAG System - DEV Community
• Ollama Open-Source Embeddings for RAG - Ollama 공식 문서
• Enterprise RAG Architecture: A Practitioner's Guide - Applied AI
• RAGAS Documentation - RAGAS 공식 문서
• RAG Observability and Evals - Langfuse 블로그

---

시리즈 전체 목차

• [1편] 왜 당신의 RAG는 실패하는가 - 온톨로지, 지식 그래프, 그리고 제대로 된 아키텍처의 조건
• [2편] 데이터가 전부입니다 - 문서를 지식으로 바꾸는 파이프라인
• [3편] 검색, 평가, 그리고 신뢰할 수 있는 시스템
• [4편] MCP로 완성하기 - Claude가 회사 지식을 쓰게 만들기 (현재 글)