Model Context Protocol(MCP). AI 모델을 위한 새로운 표준 (3)

복잡한 AI 앱을 구축하는 데 어려움을 겪고 계신가요? MCP는 AI 모델에 구조화된 컨텍스트를 제공하는 새로운 패러다임을 제시합니다. Anthropic에서 개발한 이 프로토콜이 어떻게 AI 개발 방식을 변화시키고 있는지 알아봅시다.

안녕하세요. 요새 개인 프로젝트로 MCP들을 몇개 만들어 활용하는 놀이(?)를 하고 있습니다. 그래서 오늘 이 MCP에 대해서 다뤄보는 포스팅도 한번 작성해보고 싶어서 이렇게 왔습니다...ㅎ

 

최근 AI 애플리케이션 개발이 급속도로 발전하면서, 대규모 언어 모델(LLM)에 효과적으로 컨텍스트를 제공하는 방법이 중요한 과제로 떠올랐습니다. Model Context Protocol(MCP)은 Anthropic에서 개발한 새로운 표준으로, AI 모델과 컨텍스트 제공자 간의 표준화된 인터페이스를 정의합니다. 이 글에서는 MCP가 무엇인지, 어떻게 작동하는지, 그리고 왜 AI 애플리케이션 개발자에게 중요한지 자세히 알아보겠습니다.

2025.04.14 - [Developer] - Model Context Protocol(MCP). AI 모델을 위한 새로운 표준 (1)

Model Context Protocol(MCP). AI 모델을 위한 새로운 표준 (1)

2025.04.14 - [Developer] - Model Context Protocol(MCP). AI 모델을 위한 새로운 표준 (2)

Model Context Protocol(MCP). AI 모델을 위한 새로운 표준 (2)

 

---

MCP는 어떻게 활용할 수 있을까?

MCP는 다양한 AI 애플리케이션에서 활용될 수 있으며, 몇 가지 실제 사용 사례를 살펴보겠습니다.

1. 지식 기반 챗봇

지식 기반 챗봇은 MCP를 활용하여 사용자 질문에 더 정확하고 관련성 높은 응답을 제공할 수 있습니다.

작동 과정:

1. 사용자가 질문을 입력합니다. (예: "우리 회사의 휴가 정책은 어떻게 되나요?")
2. 모델이 질문을 분석하고 필요한 컨텍스트를 요청합니다. (예: "회사 정책 문서에서 휴가 관련 섹션 검색")
3. MCP 서버가 내부 문서 데이터베이스에서 관련 정보를 찾아 제공합니다.
4. 모델이 제공된 정보를 검토하고 추가 정보가 필요하면 다시 요청합니다. (예: "병가 정책에 대한 세부 정보 요청")
5. 충분한 정보를 수집한 후, 모델이 종합적인 응답을 생성합니다.

이 접근 방식은 모든 회사 정책을 모델에 미리 제공하는 대신, 필요한 정보만 선별적으로 검색하여 효율성을 높입니다.

2. 개인화된 추천 시스템

MCP를 활용한 추천 시스템은 사용자 프로필과 선호도에 기반한 맞춤형 추천을 제공할 수 있습니다.

단계 MCP 없이 MCP 활용 시

단계	MCP 없을 경우	MCP 활용할 경우
초기 데이터	사용자 프로필 전체를 모델에 제공	기본 사용자 정보만 제공
컨텍스트 획득	모든 제품 데이터를 한 번에 로드	모델이 필요한 카테고리나 특성별로 제품 데이터 요청
사용자 이력	전체 구매 이력을 모델에 제공	모델이 관련 구매 이력만 선별적으로 요청
상품 분석	제한된 데이터로 분석	필요시 상품 상세 정보 추가 요청
최종 추천	제한된 컨텍스트 내에서 추론	단계적 추론을 통해 더 정확한 추천 제공

3. 다중 소스 연구 어시스턴트

연구 어시스턴트는 다양한 소스에서 정보를 수집하여 종합적인 분석을 제공해야 합니다.

 

MCP 활용 과정:

1. 사용자가 연구 주제를 제시합니다. (예: "재생 에너지의 경제적 영향 분석")
2. 모델이 필요한 다양한 정보 유형을 식별합니다:
   • 학술 논문에서의 최신 연구 결과
   • 산업 보고서의 시장 데이터
   • 정부 정책 문서의 규제 정보
   • 뉴스 기사의 최근 개발 동향
3. 모델이 각 정보 유형에 맞는 컨텍스트 제공자에게 요청을 보냅니다.
4. 초기 정보를 분석한 후, 모델은 추가 정보가 필요한 영역을 식별하고 후속 요청을 생성합니다.
5. 수집된 정보를 바탕으로 종합적인 분석 보고서를 작성합니다.

이 접근 방식은 모델이 다양한 소스에서 정보를 능동적으로 수집하고 통합할 수 있게 합니다.

4. 코드 분석 및 디버깅 어시스턴트

MCP는 프로그래밍 지원 도구에서도 유용하게 활용될 수 있습니다.

 

구현 방식:

1. 사용자가 코드 문제나 오류를 제시합니다.
2. 모델이 코드를 분석하고 다음과 같은 컨텍스트를 요청합니다:
   • 관련 API 문서
   • 유사한 오류 패턴에 대한 스택오버플로우 답변
   • 프로젝트의 다른 관련 파일 내용
   • 환경 설정 정보
3. 모델이 문제를 진단하고 해결책을 제안합니다.

이 방식은 개발자가 직접 여러 소스를 검색하는 시간을 절약하며, 모델이 필요한 기술 문서를 정확히 요청할 수 있습니다.

💡 TIP: 효과적인 MCP 활용을 위한 모범 사례

1. 명확한 컨텍스트 제공자 설계: 각 제공자의 기능과 범위를 명확히 정의하여 모델이 적절한 제공자를 선택할 수 있게 합니다.
2. 컨텍스트 요청 최적화: 너무 빈번하거나 중복된 요청이 발생하지 않도록 모델 프롬프트를 최적화합니다.
3. 단계적 접근: 복잡한 작업은 여러 단계로 나누어 각 단계에서 필요한 컨텍스트만 요청하도록 설계합니다.
4. 캐싱 전략: 자주 요청되는 컨텍스트는 캐싱하여 응답 시간을 단축하고 비용을 절감합니다.
5. 요청 파라미터 검증: 모델이 제공하는 파라미터가 유효한지 검증하여 오류를 방지합니다.
6. 지속적인 모니터링: 모델의 컨텍스트 요청 패턴을 모니터링하여 최적화 기회를 식별합니다.

실제 MCP 구현 해보자! 검색 기반 질의응답 시스템

다음은 MCP를 활용한 검색 기반 질의응답 시스템의 구현 예시입니다. 이 예제에서는 사용자 질문에 답변하기 위해 웹 검색과 데이터베이스 쿼리를 수행하는 컨텍스트 제공자를 구현합니다.

from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel, Field
from typing import Dict, List, Any, Optional, Union
import httpx
import json
import os
import asyncio
import logging
from datetime import datetime

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp_server")

app = FastAPI(title="MCP Server", description="Model Context Protocol Server Implementation")

# 모델 클래스 정의
class PromptRequest(BaseModel):
    prompt: str
    model_params: Optional[Dict[str, Any]] = Field(default=None, description="모델 파라미터 (온도, 최대 토큰 등)")
    session_id: Optional[str] = Field(default=None, description="세션 식별자")

class ContextRequest(BaseModel):
    provider: str = Field(..., description="컨텍스트 제공자 이름")
    params: Dict[str, Any] = Field(..., description="컨텍스트 요청 파라미터")
    request_id: Optional[str] = Field(default=None, description="요청 식별자")

class ContextResponse(BaseModel):
    content: Any = Field(..., description="제공자의 응답 내용")
    metadata: Optional[Dict[str, Any]] = Field(default=None, description="응답 관련 메타데이터")

class ErrorResponse(BaseModel):
    error: str = Field(..., description="오류 메시지")
    details: Optional[Dict[str, Any]] = Field(default=None, description="추가 오류 세부 정보")

class PromptResponse(BaseModel):
    response: str = Field(..., description="모델의 최종 응답")
    context_requests: List[Dict[str, Any]] = Field(default=[], description="처리된 컨텍스트 요청 목록")
    processing_time: float = Field(..., description="처리 시간(초)")

# 캐시 저장소
context_cache = {}

# 웹 검색 컨텍스트 제공자
async def web_search_provider(query: str, result_count: int = 3) -> ContextResponse:
    # 캐시 확인
    cache_key = f"web_search:{query}:{result_count}"
    if cache_key in context_cache:
        logger.info(f"Cache hit for {cache_key}")
        return context_cache[cache_key]
    
    logger.info(f"Performing web search for query: {query}")
    
    # 실제 구현에서는 검색 API 호출
    # 이 예제에서는 모의 응답 반환
    await asyncio.sleep(0.8)  # API 호출 시뮬레이션
    
    search_results = [
        {
            "title": "Model Context Protocol(MCP) 소개", 
            "url": "https://modelcontextprotocol.io/introduction", 
            "snippet": "Model Context Protocol은 AI 모델에 컨텍스트를 제공하는 표준화된 방법입니다."
        },
        {
            "title": "Anthropic, MCP 공식 발표", 
            "url": "https://www.anthropic.com/news/mcp-announcement", 
            "snippet": "Anthropic은 오늘 Model Context Protocol을 공식 발표했습니다."
        },
        {
            "title": "MCP vs RAG: 차이점과 장단점", 
            "url": "https://example.com/mcp-vs-rag", 
            "snippet": "MCP와 RAG의 주요 차이점과 각 접근 방식의 장단점을 비교합니다."
        },
        {
            "title": "MCP 구현 가이드", 
            "url": "https://modelcontextprotocol.io/quickstart", 
            "snippet": "MCP를 애플리케이션에 구현하는 단계별 가이드입니다."
        },
        {
            "title": "Claude와 MCP 통합하기", 
            "url": "https://docs.anthropic.com/claude/docs/mcp-integration", 
            "snippet": "Claude 모델과 MCP를 통합하는 방법에 대한 공식 문서입니다."
        },
    ]
    
    # 결과 수 제한
    limited_results = search_results[:result_count]
    
    response = ContextResponse(
        content=limited_results,
        metadata={
            "query": query, 
            "result_count": len(limited_results),
            "timestamp": datetime.now().isoformat()
        }
    )
    
    # 캐시에 저장
    context_cache[cache_key] = response
    
    return response

# 데이터베이스 컨텍스트 제공자
async def database_provider(table: str, query: Dict[str, Any]) -> ContextResponse:
    logger.info(f"Database query on table {table}: {query}")
    
    # 실제 구현에서는 데이터베이스 쿼리 실행
    await asyncio.sleep(0.5)  # DB 쿼리 시뮬레이션
    
    if table == "products":
        results = [
            {"id": 1, "name": "AI Assistant Premium", "price": 29.99, "description": "고급 AI 비서 서비스"},
            {"id": 2, "name": "MCP Toolkit", "price": 149.99, "description": "MCP 구현을 위한 개발자 도구"},
            {"id": 3, "name": "Context Provider Bundle", "price": 99.99, "description": "다양한 컨텍스트 제공자 패키지"},
        ]
    elif table == "users":
        results = [
            {"id": 101, "name": "홍길동", "membership": "premium"},
            {"id": 102, "name": "김철수", "membership": "basic"},
        ]
    else:
        results = []
    
    # 필터링 적용 (간단한 예시)
    if "filter" in query and results:
        for key, value in query["filter"].items():
            if key in results[0]:
                results = [item for item in results if item[key] == value]
    
    return ContextResponse(
        content=results,
        metadata={"table": table, "query": query, "timestamp": datetime.now().isoformat()}
    )

# 컨텍스트 제공자 레지스트리
context_providers = {
    "web_search": web_search_provider,
    "database": database_provider,
}

# 컨텍스트 요청 처리 API
@app.post("/context", response_model=Union[ContextResponse, ErrorResponse])
async def get_context(request: ContextRequest):
    start_time = datetime.now()
    
    logger.info(f"Context request received: {request.provider}")
    
    if request.provider not in context_providers:
        logger.warning(f"Unknown provider requested: {request.provider}")
        return ErrorResponse(error=f"Provider {request.provider} not found")
    
    try:
        provider_func = context_providers[request.provider]
        response = await provider_func(**request.params)
        
        # 처리 시간 기록
        processing_time = (datetime.now() - start_time).total_seconds()
        if response.metadata is None:
            response.metadata = {}
        response.metadata["processing_time"] = processing_time
        
        logger.info(f"Context request processed in {processing_time:.2f}s")
        return response
    
    except Exception as e:
        logger.error(f"Error processing context request: {str(e)}", exc_info=True)
        return ErrorResponse(
            error="Context provider error",
            details={"message": str(e), "provider": request.provider}
        )

# 프롬프트 처리 API (시뮬레이션)
@app.post("/prompt", response_model=Union[PromptResponse, ErrorResponse])
async def process_prompt(request: PromptRequest):
    start_time = datetime.now()
    logger.info(f"Prompt request received: {request.prompt[:50]}...")
    
    try:
        # 이 예제에서는 모델 호출을 시뮬레이션하고 MCP 패턴을 수동으로 구현
        context_requests = []
        
        # 1단계: 초기 컨텍스트 요청 (웹 검색)
        search_query = request.prompt
        web_search_request = ContextRequest(
            provider="web_search",
            params={"query": search_query, "result_count": 3}
        )
        web_search_response = await get_context(web_search_request)
        
        if isinstance(web_search_response, ErrorResponse):
            return web_search_response
        
        context_requests.append({
            "provider": "web_search",
            "params": web_search_request.params,
            "response": web_search_response
        })
        
        # 2단계: 추가 컨텍스트 요청 (데이터베이스)
        if "제품" in request.prompt or "상품" in request.prompt:
            db_request = ContextRequest(
                provider="database",
                params={"table": "products", "query": {"filter": {}}}
            )
            db_response = await get_context(db_request)
            
            if not isinstance(db_response, ErrorResponse):
                context_requests.append({
                    "provider": "database",
                    "params": db_request.params,
                    "response": db_response
                })
        
        # 최종 응답 생성 (실제 구현에서는 모델이 생성)
        search_results = web_search_response.content
        response_text = f"'{request.prompt}'에 대한 정보를 찾아보았습니다:\n\n"
        
        for idx, result in enumerate(search_results, 1):
            response_text += f"{idx}. {result['title']}\n{result['snippet']}\n[{result['url']}]\n\n"
        
        # 제품 정보가 있으면 추가
        for req in context_requests:
            if req["provider"] == "database" and req["params"]["table"] == "products":
                products = req["response"].content
                if products:
                    response_text += "\n관련 제품 정보:\n"
                    for product in products:
                        response_text += f"- {product['name']}: {product['price']}원 - {product['description']}\n"
        
        processing_time = (datetime.now() - start_time).total_seconds()
        logger.info(f"Prompt processed in {processing_time:.2f}s")
        
        return PromptResponse(
            response=response_text,
            context_requests=context_requests,
            processing_time=processing_time
        )
        
    except Exception as e:
        logger.error(f"Error processing prompt: {str(e)}", exc_info=True)
        return ErrorResponse(
            error="Prompt processing error",
            details={"message": str(e)}
        )

# 서버 상태 확인 API
@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "providers": list(context_providers.keys()),
        "cache_entries": len(context_cache),
        "timestamp": datetime.now().isoformat()
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

 

이 예제는 웹 검색과 데이터베이스 조회를 수행하는 기본적인 MCP 서버를 구현합니다.

실제 프로덕션 환경에서는 오류 처리, 인증, 로깅, 그리고 실제 외부 서비스 연동이 필요합니다.

결론

Model Context Protocol(MCP)은 AI 모델과 컨텍스트 제공자 간의 상호작용을 표준화하여 더 효율적이고 확장 가능한 AI 애플리케이션 개발을 가능하게 합니다. 기존의 컨텍스트 제공 방식과 달리, MCP는 모델이 필요한 정보를 능동적으로 요청할 수 있게 함으로써 토큰 사용을 최적화하고 더 관련성 높은 응답을 생성할 수 있습니다.

 

• 필요한 컨텍스트만 선별적으로 요청하여 토큰 사용 효율화
• 다양한 데이터 소스와 도구를 표준화된 방식으로 통합
• 복잡한 다단계 작업에서 단계별로 필요한 정보만 요청 가능
• 확장 가능한 아키텍처로 새로운 컨텍스트 제공자 쉽게 추가
• 동적 컨텍스트 수집을 통한 더 정확하고 관련성 높은 응답 생성

MCP는 이러한 이점이 있으며, MCP는 아직 초기 단계이지만, Anthropic의 Claude와 같은 강력한 LLM과 결합하여 AI 시스템의 상호작용 방식을 근본적으로 변화시킬 잠재력을 가지고 있습니다. 개발자들은 이러한 새로운 패러다임을 채택함으로써 더 지능적이고 효율적인 AI 애플리케이션을 구축할 수 있을 것입니다.