Anthropic의 모델 컨텍스트 프로토콜(MCP): AI 통합을 위한 "USB-C" 표준 (Part 2)

Part 1 확인하기: Anthropic의 모델 컨텍스트 프로토콜(MCP): AI 통합을 위한 ‘USB-C’ 표준 (Part 1)

Part 1에서는 Anthropic의 모델 컨텍스트 프로토콜(MCP)이 AI 시스템이 외부 데이터, 도구 및 리소스에 접근할 수 있도록 공통 언어를 제공하는 개방형 표준으로서 어떻게 작동하는지를 살펴보았습니다. 이는 마치 USB-C가 다양한 기기를 연결하는 범용 커넥터가 된 것과 같은 개념입니다.

이번 후속편인 Part 2에서는 개념에서 실제 구현으로 나아가며, MCP를 실질적으로 구현하는 간단하지만 동작하는 예제를 함께 살펴봅니다. Python을 사용하여 구조화된 문서를 제공하는 간단한 MCP 서버와, 해당 문서를 연결, 검색 및 처리하여 LLM 기반 애플리케이션의 컨텍스트를 구축하는 클라이언트를 구성합니다. 이 실습 기반의 예시는 MCP가 언어 모델과 기업 지식을 연결하는 다리 역할을 하며, 더 역동적이고 확장 가능한 AI 워크플로우를 가능하게 하는 방법을 보여줍니다.

python
import logging 
from mcp.server.fastmcp import FastMCP 
 
## 서버 기본 로깅 설정 (Configure basic logging for the server) 
logging.basicConfig(level=logging.INFO) 
logger = logging.getLogger(__name__) 
 
## MCP 서버 인스턴스 생성 (프로토콜 세부 처리를 담당) 
mcp = FastMCP("DocumentServer") 
 
## 제공할 예시 텍스트 기반 리소스(짧은 문서들) 
## 실제 환경에서는 파일 내용이나 DB 레코드가 될 수 있음 
documents = { 
    "doc1": ( 
        "Python은 Guido van Rossum이 만들고 1991년에 처음 출시된 고수준 인터프리터 프로그래밍 언어입니다. " 
        "가독성과 폭넓은 표준 라이브러리로 잘 알려져 있으며, Python은 절차적, 객체지향적, 기능적인 프로그래밍 등 다양한 패러다임을 지원합니다. " 
        "Django와 Flask와 같은 프레임워크를 사용하는 웹 개발, pandas, NumPy, scikit-learn 등의 라이브러리를 사용하는 데이터 과학, " 
        "TensorFlow, PyTorch를 이용한 인공지능 및 머신러닝, 자동화, 스크립팅, 교육 분야에서 널리 사용됩니다. " 
        "Python의 단순함과 커뮤니티 지원 덕분에 세계에서 가장 인기 있는 프로그래밍 언어 중 하나가 되었습니다." 

    ), 
    "doc2": ( 
        "OpenAI의 GPT-4는 자연어를 이해하고 생성하는 데 탁월한 최첨단 대형 언어 모델입니다. " 
        "라이선스가 부여된 데이터와 공개 데이터의 혼합으로 학습된 GPT-4는 뛰어난 추론, 요약, 번역, 질의응답 능력을 보여줍니다. " 
        "이 모델은 챗봇 구축, 프로그래밍 지원, 창의적인 콘텐츠 생성, 문서 분석 등에 활용될 수 있습니다. " 
        "GPT-4는 모든 LLMs처럼 강력한 기능을 갖추고 있지만, 그 한계도 존재합니다. 예를 들어, 그럴듯하지만 부정확한 정보를 생성하는 환각 현상, " 
        "컨텍스트 길이 제한, 실시간 인식 부족 등이 있습니다. OpenAI는 GPT-4에 대한 API 및 ChatGPT와 같은 도구를 통해 접근을 제공합니다." 

    ), 
    "doc3": ( 
        "Retrieval-Augmented Generation (RAG)은 실행 시 외부 지식을 통합하여 언어 모델의 성능을 향상시키는 기법입니다. " 
        "모델의 내부 가중치만을 사용하는 대신, RAG 시스템은 지식 베이스에서 관련 문서를 검색하고 이를 프롬프트에 전달합니다. " 
        "이 접근 방식은 사실 정확도를 높이고, 재학습 없이 모델의 지식을 동적으로 업데이트할 수 있으며, 도메인 특화 애플리케이션을 가능하게 합니다. " 
        "RAG는 고객 지원, 법률 기술, 검색 엔진, AI 어시스턴트 등 정확하고 최신 정보가 필요한 분야에서 널리 사용됩니다. " 
    ), 
    "doc4": ( 
        "모델 컨텍스트 프로토콜(MCP)은 Anthropic이 제안한 오픈 표준으로, AI 시스템이 외부 데이터 및 도구와 상호작용하는 방식을 정의합니다. " 
        "이는 JSON-RPC 기반 아키텍처를 사용하며, 호스트(LLM 기반 앱)는 stdio 또는 HTTP 전송 방식을 통해 서버(데이터 소스/도구)와 연결됩니다. " 
        "MCP는 문서 접근을 위한 'resources', 작업 수행을 위한 'tools', 사전 정의된 메시지 템플릿을 위한 'prompts'와 같은 기능을 지원합니다. " 
        "이를 통해 AI 어시스턴트는 데이터베이스, 문서, API와 같은 기업 데이터를 안전하게 접근할 수 있으며, 확장 가능하고 풍부한 컨텍스트의 AI 워크플로우를 가능하게 합니다. " 
        "MCP는 모델에 독립적이며, 재사용 가능한 통합의 오픈 생태계를 촉진하도록 설계되었습니다." 
    ) 
} 
 
## 문서들을 노출할 MCP 리소스 정의 
## 각 리소스는 URI로 식별됨. 여기서는 "docs://" 스킴 사용 
@mcp.resource("docs://doc1") 
def get_doc1() -> str: 
    """문서 1의 내용을 반환""" 
    logger.info("Serving resource docs://doc1") 
    return documents["doc1"] 
 
@mcp.resource("docs://doc2") 
def get_doc2() -> str: 
    """문서 2의 내용을 반환""" 
    logger.info("Serving resource docs://doc2") 
    return documents["doc2"] 
 
## 선택적으로 docs://{name} 같은 동적 리소스를 정의할 수도 있으나, 
## 여기서는 단순성과 목록화를 위해 고정 URI 사용 
 
## 이 스크립트를 실행할 때 MCP 서버 시작 
## STDIO를 통해 들어오는 MCP 클라이언트 연결을 수신 
if __name__ == "__main__": 
    ## STDIO 전송으로 서버 실행. 클라이언트(예: MCP 클라이언트)가 
    ## STDIN/STDOUT을 통해 연결할 때까지 대기 
    mcp.run() 

설명:

FastMCP를 사용하여 이름이 지정된 MCP 서버를 생성하고, 두 개의 텍스트 리소스인 docs://doc1과 docs://doc2를 등록합니다. 각 리소스는 URI를 통해 노출되며 문자열(문서 텍스트)을 반환합니다. @mcp.resource 데코레이터는 이러한 함수들을 MCP 서버에 연결하며, MCP 클라이언트는 이들을 나열하고 가져올 수 있습니다.

스크립트가 실행되면(예: STDIO를 통해 클라이언트가 실행했을 때), mcp.run()이 서버의 이벤트 루프를 시작하고 표준 입력/출력을 통해 클라이언트 요청을 기다립니다. 우리는 Python의 로깅 기능을 사용하여 리소스가 제공될 때마다 기록하며, 이는 디버깅 및 감사에 도움이 됩니다.

python
iimport asyncio 
import logging 
import os 
from mcp import ClientSession, StdioServerParameters 
from mcp.client.stdio import stdio_client 
from openai import OpenAI 
 
## 클라이언트 로깅 설정 (Configure logging for the client) 
logging.basicConfig(level=logging.INFO) 
logger = logging.getLogger(__name__) 
 
async def main(): 
    ## 환경 변수에서 OpenAI API 키 설정 (실행 전 반드시 설정)     

api_key = os.getenv("OPENAI_API_KEY") 
    if not api_key: 

        logger.error("OpenAI API 키가 설정되지 않았습니다. OPENAI_API_KEY 환경 변수를 설정하세요.") 
        return 
 
    ## OpenAI 클라이언트 초기화 (API 키 사용) 
    client = OpenAI(api_key=api_key) 
 
    ## stdio 전송을 통해 MCP 서버를 실행할 파라미터 정의 
    ## 이 호출은 server.py 스크립트를 서브프로세스로 실행함 
    server_params = StdioServerParameters( 
        command="python3", 
        args=["server.py"] 
    ) 
    logger.info("STDIO를 통해 MCP 서버 시작...") 
 
    ## MCP 서버를 실행하고 stdio 연결을 수립 
    async with stdio_client(server_params) as (read_stream, write_stream): 
        async with ClientSession(read_stream, write_stream) as session: 
            ## MCP 세션 초기화(클라이언트-서버 핸드셰이크) 
            try: 
                await session.initialize() 
                logger.info("MCP 세션 초기화 완료 (핸드셰이크 성공)") 
            except Exception as e: 
                logger.error(f"MCP 세션 초기화 실패: {e}") 
                return 
 
            ## 서버에서 사용 가능한 리소스 목록 조회 
            try: 
                resources_result = await session.list_resources() 
                logger.info(f"사용 가능한 리소스: {resources_result}") 
            except Exception as e: 
                logger.error(f"리소스 목록 조회 오류: {e}") 
                return 
 
            ## ListResourcesResult 객체에서 실제 리소스 리스트 추출 
            resources_list = resources_result.resources if hasattr(resources_result, 'resources') else [] 
 
            if not resources_list: 
                logger.error("MCP 서버에서 리소스를 찾을 수 없음.") 
                return 
 
            ## LLM 컨텍스트 제공을 위해 모든 리소스를 읽기 
            all_docs = {} 
            for resource in resources_list: 
                resource_uri = resource.uri if hasattr(resource, "uri") else str(resource) 
                resource_name = resource.name if hasattr(resource, "name") else resource_uri.split('/')[-1] 
 
                logger.info(f"리소스 읽는 중: {resource_uri}") 
                try: 
                    meta, content_tuple = await session.read_resource(resource_uri) 
 
                    ## 텍스트 콘텐츠 추출 
                    text_content = "" 
                    if (isinstance(content_tuple, tuple) 
                        and len(content_tuple) == 2 
                        and content_tuple[0] == 'contents'): 
                        contents_list = content_tuple[1] 
                        if isinstance(contents_list, list) and contents_list: 
                            for item in contents_list: 
                                if hasattr(item, 'text'): 
                                    text_content += item.text 
 
                    ## URI와 함께 문서 저장 
                    all_docs[resource_uri] = { 
                        "name": resource_name, 
                        "content": text_content 
                    } 
                    logger.info(f"문서 {resource_uri} 컨텍스트에 추가 완료") 
                except Exception as e: 
                    logger.error(f"리소스 {resource_uri} 읽기 실패: {e}") 
 
            if not all_docs: 
                logger.error("문서 콘텐츠를 가져오지 못함.") 
                return 
 
            ## 컨텍스트용으로 모든 문서 포맷팅 
            context = "" 
            for uri, doc_info in all_docs.items(): 
                context += f"Document: {uri}\n" 
                context += f"Content: {doc_info['content']}\n\n" 
 
            ## 특정 문서에서 정보를 찾도록 LLM에 질문 
            user_question = "What does MCP stand for, and what is its purpose according to the documents?" 
 
            prompt = ( 
                "You have access to several documents. Review them carefully and answer the question " 
                "based only on the information contained in these documents.\n\n" 
                f"{context}\n\nQuestion: {user_question}" 
            ) 
 
            messages = [ 
                {"role": "user", "content": prompt} 
            ] 
 
            ## 새 v1.0+ 포맷으로 OpenAI API 호출 
            try: 
                response = client.chat.completions.create( 
                    model="gpt-4o-mini", 
                    messages=messages, 
                    temperature=0.3  ## 더 집중된 답변을 위한 낮은 온도 
                ) 
            except Exception as e: 
                logger.error(f"OpenAI API 호출 실패: {e}") 
                return 
 
            ## 응답에서 어시스턴트의 답변 추출 
            answer = response.choices[0].message.content.strip() 
            print("OpenAI 응답:", answer) 
 
## 비동기 main 함수 실행 
if __name__ == "__main__": 
    asyncio.run(main()) 

설명:

이 스크립트는 StdioServerParameters를 통해 server.py라는 MCP 서버를 실행하고, 표준 입력/출력을 통해 읽기/쓰기 스트림을 제공하는 stdio_client를 사용하여 서버에 연결합니다. 이 스트림은 ClientSession을 생성하는 데 사용되며, session.initialize()는 MCP 핸드셰이크를 수행합니다.

초기화가 완료된 후, 클라이언트는 session.list_resources()를 통해 사용 가능한 리소스를 나열하고, session.read_resource(uri)로 각 리소스의 내용을 읽습니다. 이 콘텐츠는 프롬프트 컨텍스트로 집계됩니다. 다음으로, 문서를 문맥으로 제공한 상태에서 OpenAI ChatCompletion.create(gpt-4o-mini 사용)를 통해 질문이 제시됩니다. 모델의 응답이 출력됩니다. 이 스크립트는 API 키 문제, 연결 실패, 읽기 오류 등을 처리하는 예외 처리 로직도 포함하고 있습니다.

이 예제를 로컬 환경에서 실행하려면, 필요한 패키지를 설치하고 OpenAI API 키를 설정해야 합니다.

• 의존성 설치: pip install mcp openai (MCP SDK와 OpenAI SDK를 설치합니다).
• API 키 설정: 환경 변수 OPENAI API_KEY를 OpenAI API 키로 설정하거나, 코드에서 직접 openai.api_key에 할당해도 됩니다.
• 서버 및 클라이언트 실행: 클라이언트 스크립트를 실행하면 서버가 자동으로 실행됩니다. 예를 들어 python client.py 입니다. 클라이언트는 서버에서 제공하는 리소스를 나열하고, 선택한 리소스의 내용을 읽은 후 해당 내용를 기반으로 OpenAI 모델의 응답을 출력합니다.

이 예제는 기본적인 검색 증강 생성 흐름을 보여줍니다. MCP 서버는 외부 지식(문서)을 표준화된 리소스로 제공하고, 클라이언트는 해당 데이터를 검색하여 LLM(OpenAI GPT-4o-mini)에 입력함으로써 정보에 기반한 대답을 생성합니다. MCP의 공식 SDK를 사용함으로써 2025년 MCP 표준과의 호환성을 보장하고, 데이터 소스를 LLM에 연결하는 데 있어 견고하고 표준화된 인터페이스를 활용할 수 있습니다. 실행 중 시스템의 유지 보수성과 투명성을 높이기 위해 로깅 및 오류 처리 기능도 포함되어 있습니다. 이 패턴은 MCP를 통해 다양한 데이터 소스와 도구를 통합하는 데 확장 가능하며, OpenAI 모델을 활용한 더욱 발전된 RAG 워크플로우를 구현할 수 있게 해줍니다.

이번 실용적인 데모는 모델 컨텍스트 프로토콜(MCP)이 얼마나 강력하면서도 접근하기 쉬운지를 잘 보여줍니다. 표준 입출력 기반의 가벼운 서버-클라이언트 아키텍처를 통해, LLM이 모델 자체를 복잡하게 만들거나 별도의 통합 작업 없이도 외부 지식 소스에 안전하게 접근할 수 있게 되었습니다.

AI 에이전트, 내부 코파일럿, 또는 지식 기반의 어시스턴트를 구축하고 있다면, MCP는 모델에 구애받지 않는 확장 가능한 프레임워크로서 리소스 탐색, 문맥 기반 연결, 모듈형 상호 운용성을 지원합니다.

더 많은 AI 시스템이 MCP를 채택함에 따라, 우리는 모델이 USB-C처럼 보편적이고, 신뢰할 수 있으며, 지능적으로 지식에 연결될 수 있는 생태계에 한 걸음 더 가까워지고 있습니다.