langchain_mcp_adapters 를 이용해 Langchain 과 MCP 연동하기

langchain_mcp_adapters 를 이용해 Langchain 과 MCP 연동하기

 

LangChain은 다양한 LLM(대형 언어 모델)과 도구들을 조합하여 강력한 AI 애플리케이션을 개발할 수 있게 해주는 프레임워크입니다. 최근에는 Smithery에서 개발한 MCP (Model Context Protocol) 기반 도구와의 연동이 가능해지면서, LangChain 에이전트에 외부 도구를 유연하게 연결할 수 있게 되었습니다.

이번 글에서는 langchain_mcp_adapters 라이브러리를 활용해 LangChain과 MCP 서버를 연동하는 방법을 알아봅니다.

 

MCP란 무엇인가?

**MCP (Model Context Protocol)**는 다양한 도구(tool)를 LLM 기반 시스템에 플러그인처럼 연결할 수 있도록 표준화된 방식입니다. 이를 통해 날씨 조회, 이미지 생성, 파일 분석 등 다양한 기능을 LLM 에이전트에 붙일 수 있습니다.

 

langchain_mcp_adapters란?

langchain_mcp_adapters는 MCP 서버에 정의된 도구들을 LangChain의 Tool 객체로 변환하여 사용할 수 있도록 도와주는 라이브러리입니다. MCP OpenAPI 스펙을 기반으로 도구를 자동 등록하며, LangChain 에이전트에서 자연스럽게 호출할 수 있게 만들어줍니다.

 

Langchain 과 MCP 연동하기

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

pip install langchain-mcp-adapters langgraph "langchain[openai]" python-dotenv

.env 파일을 만들어 OPEN API KEY 를 입력합니다.

OPENAI_API_KEY=sk-***

 

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")

 

LangChain 에이전트 코드 (main.py)

# main.py
import os
from dotenv import load_dotenv
import asyncio

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langchain_mcp_adapters.client import MultiServerMCPClient

async def main():
    # 환경 변수 로드
    load_dotenv()
    openai_key = os.getenv("OPENAI_API_KEY")
    if not openai_key:
        raise RuntimeError("OPENAI_API_KEY 가 설정되지 않았습니다.")
    
    # MCP 서버 설정
    mcp_servers = {
        "math": {
            "command": "python",
            "args": ["math_server.py"],
            "transport": "stdio",
        }
    }

    client = MultiServerMCPClient(mcp_servers)
    tools = await client.get_tools()

    # OpenAI 모델 설정
    model = ChatOpenAI(openai_api_key=openai_key, model="gpt-4o", temperature=0)

    agent = create_react_agent(model=model, tools=tools)

    response = await agent.ainvoke({
        "messages": [{"role": "user", "content": "4 곱하기 7은 얼마야?"}]
    })

    # 응답 메시지 정리 출력 + MCP 도구 사용 여부 출력
    tool_used = None
    final_answer = None

    for message in response["messages"]:
        # MCP 도구 호출이 포함된 메시지 찾기
        if hasattr(message, "tool_calls") and message.tool_calls:
            for tool_call in message.tool_calls:
                tool_used = tool_call['name']
        # 최종 사용자 응답 추출
        if hasattr(message, "content") and message.content:
            final_answer = message.content

    # 결과 출력
    if tool_used:
        print(f"🔧 사용된 MCP 도구: {tool_used}")
    else:
        print("🔧 MCP 도구가 사용되지 않았습니다.")

    if final_answer:
        print(f"🧠 최종 응답: {final_answer}")
    else:
        print("❗응답을 읽을 수 없습니다.")


# 비동기 실행
if __name__ == "__main__":
    asyncio.run(main())

* 위 파이썬 파일들은 동일 폴더에 있어야합니다.

 

실행

python main.py

 

결과

 

Ollama 를 이용한 방법

패키지 설치

pip install langchain-ollama

코드

# main.py
import asyncio
from langchain_ollama import ChatOllama
from langgraph.prebuilt import create_react_agent
from langchain_mcp_adapters.client import MultiServerMCPClient

async def main():
    # MCP 서버 설정 (math_server.py 실행)
    mcp_servers = {
        "math": {
            "command": "python",
            "args": ["math_server.py"],
            "transport": "stdio",
        }
    }

    # MCP 클라이언트 생성 및 도구 불러오기
    client = MultiServerMCPClient(mcp_servers)
    tools = await client.get_tools()

    # Ollama 모델 설정 (설치된 모델 이름 사용: mistral, llama3 등)
    model = ChatOllama(model="llama3.3")  # 또는 llama3, phi, codellama 등

    # LangChain 에이전트 생성
    agent = create_react_agent(model=model, tools=tools)

    # 사용자 질문 수행
    response = await agent.ainvoke({
        "messages": [{"role": "user", "content": "4 곱하기 7은 얼마야?"}]
    })

    # 응답 메시지 정리 출력
    tool_used = None
    final_answer = None

    for message in response["messages"]:
        if hasattr(message, "tool_calls") and message.tool_calls:
            for tool_call in message.tool_calls:
                tool_used = tool_call['name']
        if hasattr(message, "content") and message.content:
            final_answer = message.content

    # 출력
    if tool_used:
        print(f"🔧 사용된 MCP 도구: {tool_used}")
    else:
        print("🔧 MCP 도구가 사용되지 않았습니다.")

    if final_answer:
        print(f"🧠 최종 응답: {final_answer}")
    else:
        print("❗응답을 읽을 수 없습니다.")

# 실행
if __name__ == "__main__":
    asyncio.run(main())

결과

 

참고

https://pypi.org/project/langchain-mcp-adapters/

langchain-mcp-adapters