Apidog

올인원 협업 API 개발 플랫폼

API 설계

API 문서

API 디버깅

API 모킹

API 자동화 테스트

무료 회원가입
다운로드Mac 또는 Linux용
Home

/

MCP 서버와 함께 Langchain을 사용해 보았습니다, 절차는 다음과 같습니다:

MCP 서버와 함께 Langchain을 사용해 보았습니다, 절차는 다음과 같습니다:

Young-jae

Young-jae

Updated on April 10, 2025

모델 컨텍스트 프로토콜(MCP)은 Anthropic에서 개발한 오픈 소스 프로토콜로, 대형 언어 모델(LLM) 애플리케이션에서 외부 데이터 소스 및 도구와의 분리라는 근본적인 문제를 다룹니다. 이 포괄적인 튜토리얼은 LangChain으로 MCP를 구현하는 과정에서 여러분을 안내하며, 두 기술을 효과적으로 활용한 정교한 애플리케이션을 만드는 데 필요한 지식을 제공합니다.

💡
API 기반 애플리케이션 테스트 구현 시 개발자와 테스터는 점점 더 전문 도구인 Apidog와 같은 종합적인 Postman 대안을 사용하여 API 개발 라이프사이클을 간소화합니다. 
버튼

MCP의 이해 및 목적

모델 컨텍스트 프로토콜은 LLM 기반 애플리케이션이 다양한 외부 시스템에 연결되는 방식을 표준화하는 것을 목표로 합니다. MCP를 "AI를 위한 USB-C"로 생각해 보십시오. 이는 LLM/AI 에이전트와 외부 자원 간의 원활하고 안전하며 확장 가능한 데이터 교환을 가능하게 하는 범용 인터페이스입니다.

MCP는 클라이언트-서버 아키텍처를 사용합니다:

  • MCP 호스트: 외부 데이터에 접근해야 하는 AI 애플리케이션
  • MCP 서버: 호스트에게 정보를 제공하는 데이터 또는 도구 공급자

이 프로토콜은 명확한 관심사의 분리를 촉진하여 개발자가 모듈식이고 재사용 가능한 커넥터를 만들 때 강력한 보안을 유지할 수 있도록 세밀한 권한 제어를 제공합니다.

기술 아키텍처

MCP의 아키텍처는 세 가지 주요 구성 요소로 이루어집니다:

  1. 서버: MCP 서버는 표준화된 API를 통해 도구 및 데이터 소스를 공개합니다.
  2. 클라이언트: 클라이언트 애플리케이션은 도구 및 데이터에 접근하기 위해 서버와 통신합니다.
  3. 어댑터: LangChain은 MCP 서버와 LLM 애플리케이션 간의 통합을 단순화하는 어댑터를 제공합니다.

통신 흐름은 다음과 같은 패턴을 따릅니다:

  • LangChain 애플리케이션이 데이터/도구 실행을 요청합니다.
  • MCP 어댑터가 요청을 MCP 프로토콜 형식으로 변환합니다.
  • 서버가 요청을 처리하고 결과를 반환합니다.
  • 어댑터가 응답을 LangChain에서 사용할 수 있는 형식으로 변환합니다.

사전 요구 사항

시작하기 전에 다음 사항을 확인하십시오:

  • Python 3.8+ 설치됨
  • OpenAI API 키 (LangChain과 함께 GPT 모델을 사용하기 위해)
  • LangChain 개념에 대한 기본적인 이해
  • 터미널 접근 (예시는 macOS에 표시됨)

환경 설정

먼저 개발 환경을 만들고 구성해 보겠습니다:

# 가상 환경 생성
python3 -m venv MCP_Demo

# 가상 환경 활성화
source MCP_Demo/bin/activate

# 필수 패키지 설치
pip install langchain-mcp-adapters
pip install langchain-openai

# OpenAI API 키 설정
export OPENAI_API_KEY=your_api_key

간단한 MCP 서버 만들기

수학 연산을 제공하는 기본 MCP 서버를 만드는 것부터 시작하겠습니다. math_server.py라는 이름의 파일을 만듭니다:

# math_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """두 숫자 더하기"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """두 숫자 곱하기"""
    return a * b

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

이 서버는 두 개의 수학 도구, 즉 add와 multiply를 공개합니다. FastMCP 클래스는 서버 생성을 단순화하며, 프로토콜 세부 사항을 자동으로 처리합니다. @mcp.tool()으로 장식된 각 함수는 클라이언트에게 사용 가능하며 문서는 docstring에서 파생됩니다.

LangChain 클라이언트 구현

다음으로 MCP 서버와 상호작용하는 LangChain 클라이언트를 생성합니다. 이를 client.py로 저장합니다:

# stdio 연결을 위한 서버 매개변수 생성
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import asyncio

model = ChatOpenAI(model="gpt-4o")

# 서버 매개변수 구성
server_params = StdioServerParameters(
    command="python",
    # 서버 파일 경로 지정
    args=["math_server.py"],
)

async def run_agent():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 연결 초기화
            await session.initialize()
            
            # LangChain 형식으로 MCP 도구 로드
            tools = await load_mcp_tools(session)
            
            # 에이전트를 생성하고 실행
            agent = create_react_agent(model, tools)
            agent_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
            return agent_response

# 비동기 함수 실행
if __name__ == "__main__":
    result = asyncio.run(run_agent())
    print(result)

이 클라이언트는 MCP 서버에 연결을 설정하고, 사용 가능한 도구를 로드하며, 이러한 도구를 사용하여 문제를 해결할 수 있는 LangChain 에이전트를 생성합니다.

예제 실행하기

이 예제를 실행하려면:

  1. 하나의 터미널 탭에서 MCP 서버를 시작합니다:
python3 math_server.py
  1. 다른 터미널 탭에서 클라이언트를 실행합니다:
python3 client.py

클라이언트는 LangChain 에이전트를 호출하며, 이 에이전트는:

  1. "(3 + 5) x 12"라는 질문을 분석합니다.
  2. 인수 3과 5로 add 도구를 호출합니다.
  3. 결과 8을 얻습니다.
  4. 인수 8과 12로 multiply 도구를 호출합니다.
  5. 최종 답변을 반환합니다: 96

고급 MCP 서버 구현

데이터베이스 접근을 제공하는 더 정교한 MCP 서버를 생성하기 위해 구현을 확장해 보겠습니다. 이 예제는 외부 시스템에 커넥터를 구축하는 방법을 보여줍니다:

# db_server.py
from mcp.server.fastmcp import FastMCP
import sqlite3
from typing import List, Dict, Any

class DatabaseConnector:
    def __init__(self, db_path):
        self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()
    
    def execute_query(self, query: str) -> List[Dict[str, Any]]:
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        results = []
        for row in self.cursor.fetchall():
            results.append({columns[i]: row[i] for i in range(len(columns))})
        return results

mcp = FastMCP("DatabaseTools")
db_connector = DatabaseConnector("example.db")

@mcp.tool()
def run_sql_query(query: str) -> List[Dict[str, Any]]:
    """데이터베이스에서 SQL 쿼리 실행 및 결과 반환"""
    try:
        return db_connector.execute_query(query)
    except Exception as e:
        return {"error": str(e)}

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

LangChain과 여러 MCP 서버 통합

더 복잡한 애플리케이션을 위해 여러 MCP 서버를 통합해야 할 수도 있습니다. 여러 서버에 연결하는 클라이언트를 만드는 방법은 다음과 같습니다:

# multi_server_client.py
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import asyncio
from typing import List, Dict

# 서버 구성 정의
servers = [
    {
        "name": "math",
        "params": StdioServerParameters(
            command="python", 
            args=["math_server.py"]
        )
    },
    {
        "name": "database",
        "params": StdioServerParameters(
            command="python", 
            args=["db_server.py"]
        )
    }
]

async def connect_to_server(server_config):
    """단일 MCP 서버에 연결하고 도구를 로드합니다."""
    name = server_config["name"]
    params = server_config["params"]
    
    read, write = await stdio_client(params).__aenter__()
    session = ClientSession(read, write)
    await session.__aenter__()
    await session.initialize()
    
    tools = await load_mcp_tools(session)
    return {
        "name": name,
        "session": session,
        "tools": tools,
        "cleanup": lambda: asyncio.gather(
            session.__aexit__(None, None, None),
            stdio_client(params).__aexit__(None, None, None)
        )
    }

async def run_multi_server_agent():
    # 모든 서버에 연결
    connections = await asyncio.gather(
        *[connect_to_server(server) for server in servers]
    )
    
    try:
        # 모든 서버에서 모든 도구 수집
        all_tools = []
        for connection in connections:
            all_tools.extend(connection["tools"])
        
        # 모든 도구로 에이전트 생성
        model = ChatOpenAI(model="gpt-4o")
        agent = create_react_agent(model, all_tools)
        
        # 여러 서버를 사용할 수 있는 복잡한 쿼리로 에이전트 실행
        response = await agent.ainvoke({
            "messages": "평균 주문 금액을 초과하여 지출한 고객을 찾아 그들의 총 지출을 계산합니다."
        })
        
        return response
    
    finally:
        # 모든 연결 정리
        for connection in connections:
            await connection["cleanup"]()

# 다중 서버 에이전트 실행
if __name__ == "__main__":
    result = asyncio.run(run_multi_server_agent())
    print(result)

오류 처리 및 대체 전략

견고한 MCP 구현은 오류 처리를 포함해야 합니다. 다음은 오류 처리 및 대체 전략을 보여주는 클라이언트의 향상된 버전입니다:

# robust_client.py
async def run_agent_with_fallbacks():
    try:
        # 기본 연결 시도
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                try:
                    await session.initialize()
                    tools = await load_mcp_tools(session)
                    agent = create_react_agent(model, tools)
                    return await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
                except Exception as e:
                    print(f"MCP 도구 사용 중 오류: {e}")
                    # 도구 없이 직접 모델 호출로 대체
                    return await model.ainvoke([
                        HumanMessage(content="what's (3 + 5) x 12?")
                    ])
    except Exception as connection_error:
        print(f"연결 오류: {connection_error}")
        # 궁극적인 대체
        return {"error": "MCP 서버에 연결할 수 없습니다."}

보안 고려 사항

LangChain과 함께 MCP를 구현할 때 다음 보안 모범 사례를 고려하십시오:

  1. 입력 검증: 주입 공격을 방지하기 위해 항상 MCP 도구에 대한 입력을 검증하십시오.
  2. 도구 권한: 각 도구에 대해 세밀한 권한을 구현하십시오.
  3. 요금 제한: 도구 남용을 방지하기 위해 요금 제한을 적용하십시오.
  4. 인증: 클라이언트와 서버 간의 올바른 인증을 구현하십시오.

도구 권한을 구현하는 예는 다음과 같습니다:

from mcp.server.fastmcp import FastMCP, Permission

mcp = FastMCP("SecureTools")

# 권한 수준 정의
READ_PERMISSION = Permission("read", "데이터를 읽을 수 있습니다.")
WRITE_PERMISSION = Permission("write", "데이터를 수정할 수 있습니다.")

@mcp.tool(permissions=[READ_PERMISSION])
def get_data(key: str) -> str:
    """키로 데이터 가져오기 (읽기 권한 필요)"""
    # 구현...
    return f"{key}에 대한 데이터"

@mcp.tool(permissions=[WRITE_PERMISSION])
def update_data(key: str, value: str) -> bool:
    """데이터 업데이트 (쓰기 권한 필요)"""
    # 구현...
    return True

성능 최적화

프로덕션 배포에 대한 성능 최적화 고려 사항:

  1. 연결 풀링: 각 요청을 위해 새로운 연결을 생성하는 대신 MCP 연결을 재사용합니다.
  2. 배치 처리: 가능하면 여러 도구 호출을 그룹화합니다.
  3. 비동기 처리: asyncio를 사용하여 여러 요청을 동시에 처리합니다.

연결 풀링의 예:

class MCPConnectionPool:
    def __init__(self, max_connections=10):
        self.available_connections = asyncio.Queue(max_connections)
        self.max_connections = max_connections
        self.current_connections = 0
    
    async def initialize(self):
        # 일부 연결을 미리 생성합니다.
        for _ in range(3):  # 3개 연결로 시작
            await self._create_connection()
    
    async def _create_connection(self):
        if self.current_connections >= self.max_connections:
            raise Exception("최대 연결 수에 도달했습니다.")
        
        read, write = await stdio_client(server_params).__aenter__()
        session = await ClientSession(read, write).__aenter__()
        await session.initialize()
        
        self.current_connections += 1
        await self.available_connections.put(session)
    
    async def get_connection(self):
        if self.available_connections.empty() and self.current_connections < self.max_connections:
            await self._create_connection()
        
        return await self.available_connections.get()
    
    async def release_connection(self, connection):
        await self.available_connections.put(connection)

MCP 구현 테스트

철저한 테스트는 신뢰할 수 있는 MCP 구현에 필수적입니다. 다음은 pytest를 사용한 테스트 접근 방식입니다:

# test_mcp.py
import pytest
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools

@pytest.fixture
async def mcp_session():
    server_params = StdioServerParameters(
        command="python",
        args=["math_server.py"],
    )
    
    read, write = await stdio_client(server_params).__aenter__()
    session = ClientSession(read, write)
    await session.__aenter__()
    await session.initialize()
    
    yield session
    
    await session.__aexit__(None, None, None)
    await stdio_client(server_params).__aexit__(None, None, None)

@pytest.mark.asyncio
async def test_add_tool(mcp_session):
    tools = await load_mcp_tools(mcp_session)
    add_tool = next(tool for tool in tools if tool.name == "add")
    
    result = await add_tool.invoke({"a": 5, "b": 7})
    assert result == 12

@pytest.mark.asyncio
async def test_multiply_tool(mcp_session):
    tools = await load_mcp_tools(mcp_session)
    multiply_tool = next(tool for tool in tools if tool.name == "multiply")
    
    result = await multiply_tool.invoke({"a": 6, "b": 8})
    assert result == 48

결론

모델 컨텍스트 프로토콜은 LangChain 애플리케이션과 외부 도구 및 데이터 소스를 연결하기 위한 강력한 프레임워크를 제공합니다. 이러한 연결을 표준화함으로써 MCP는 개발자가 환경과 원활하게 상호작용할 수 있는 정교한 AI 에이전트를 만들 수 있게 해 줍니다.

LangChain의 에이전트 기능과 MCP의 연결성을 결합하면 진정으로 강력하고 컨텍스트 인식 애플리케이션을 구축하기 위한 기반이 마련됩니다. MCP 생태계가 계속 확장됨에 따라 더 많은 사전 구축된 서버와 도구가 등장할 것으로 기대되며, 개발 프로세스가 더욱 간소화될 것입니다.

이 튜토리얼에서는 기본 설정에서 연결 풀링 및 오류 처리와 같은 고급 패턴에 이르기까지 LangChain과 함께 MCP를 사용하는 기본 개념 및 구현 세부 사항을 다루었습니다. 이러한 관행을 따르면 두 기술의 장점을 활용한 견고하고 프로덕션 준비가 완료된 애플리케이션을 만들 수 있습니다.

더욱 더 탐색하려면 GitHub에 있는 MCP 서버의 성장하는 생태계를 살펴보거나 커뮤니티에 자신의 서버를 기여해 보십시오. AI 에이전트의 미래는 외부 도구와 데이터를 효과적으로 활용하는 능력에 달려 있으며, MCP는 이러한 비전을 현실로 만드는 중요한 단계입니다.