Model Context Protocol(MCP)

Model Context Protocol

최근 MCP가 AI의 확장을 도움을 주는 사례가 많아지면서 크게 주목받고 있습니다. MCP는 Anthropic에서 처음 제안되어 LLM에 컨텍스트를 제공하는 방법을 표준화하는 개방형 프로토콜입니다.

표준화된 프로토콜을 사용하는 점은 사용자 입장에서 Tool을 추가하고 삭제할 때 편리하다는 장점이 있습니다. MCP 이전 Langchain 애플리케이션에서 Tool을 정의했을 때는 Tool의 추가, 삭제 때마다 새로 애플리케이션을 배포했어야 했습니다. 이런 점에서 사용자에게 MCP는 LLM의 확장을 도우면서 편의성을 제공하는 역할을 맡았습니다.
Langchain도 mcp_adapters를 지원하게 되면서 langchain 애플리케이션에서도 mcp를 사용할 수 있게 지원하고 있어 개발자에게도 tool을 모듈화할 수 있는 장점이 생겼습니다.

그렇다면 Gemini-2.0-flash 모델을 이용해 Langchain의 ReAct Agent를 사용하면서 MCP 서버를 붙이는 예시를 들어보겠습니다.

Tools

get_weather

가장 간단하게 테스트해볼 수 있는 예시입니다. 위치를 받게되면 날씨와 기온을 반환하는 임의의 함수를 작성하고 mcp로 연결해보겠습니다.

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("TEST")

@mcp.tool()
def get_weather(location: str) -> str:
    """ """
    import random, json
    weather, celsius = [random.choice(['맑음','흐림','비']), random.randint(10,25)]
    return f"""{{"지역": {location}, "날씨": {weather}, "기온": {celsius}}}"""

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

mcp의 실행 방식은 mcp.run(transport='')에서 정의할 수 있으며 stdio와 sse 두 가지 방식이 있습니다. stdio는 로컬에서 실행하는 방식, sse는 서버로 실행하는 방식을 말합니다. sse는 mcp로 추가하기 전에 서버로 실행되어 있어야 LLM이 정상적으로 접근할 수 있고 stdio 방식은 LLM이 접근할 때 실행되었다가 종료되는 방식입니다.

위 코드는 위치를 받아 랜덤하게 날씨와 기온을 생성하고 값을 리턴하는 함수입니다. langchain에서 이 mcp를 실행할 때 아래와 같은 동작을 수행합니다.

from dotenv import load_dotenv
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_google_genai import ChatGoogleGenerativeAI

load_dotenv()

model = ChatGoogleGenerativeAI(model="gemini-2.0-flash")

server_params = StdioServerParameters(command="python", args=["mcp_server.py"])

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        tools = await load_mcp_tools(session)
        agent = create_react_agent(model, tools)
        result = await agent.ainvoke({"messages": "오늘 서울 날씨는 어때?"})

for message in result['messages']:
    print(f"{message.type}:", message.content, message.additional_kwargs)

human: 오늘 서울 날씨는 어때? {}
ai:  {'function_call': {'name': 'get_weather', 'arguments': '{"location": "\\uc11c\\uc6b8"}'}}
tool: {"지역": 서울, "날씨": 흐림, "기온": 21} {}
ai: 오늘 서울 날씨는 흐리고, 기온은 21도입니다. {}

LLM에게 서울이라는 위치정보와 날씨를 물어보고 있고 LLM이 알아서 판단해서 function_call에 get_weather라는 tool을 호출하고 있습니다. tool의 반환된 결과로 서울, 흐림, 21도라는 정보를 이용해 LLM이 사용자에게 최종 답변을 해주고 있습니다.

Execute Code

또 다른 예제로는 코드를 실행하는 mcp입니다. ChatGPT처럼 코드를 실행해달라고 요청하면 내부에서 실행하고 결과를 리턴해주는 기능이 있는데 mcp로도 충분히 구현가능하다는 것을 보여줍니다.

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("TEST")

@mcp.tool()
def execute_python_code(code: str) -> str:
    import tempfile, subprocess
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as tmp:
        tmp.write(code)
        tmp_path = tmp.name
    
    result = subprocess.run(
        ['python', tmp_path],
        capture_output=True,
        text=True
    )

    return f"""{{"stdout": {result.stdout}, "result.stderr": {result.stderr}, "result.returncode": {result.returncode}}}"""

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

위 코드는 subprocess를 이용해 python 코드를 실행하는 함수입니다. code를 입력받으면 임시폴더에 작성하고 실행한 결과를 리턴하는 함수입니다. 제가 실행한 코드는 아래와 같습니다.

# test.py
import numpy as np
import sqlite3

db = sqlite3.connect("test.db")

def create_table():
    cur = db.cursor()
    cur.execute("CREATE TABLE IF NOT EXISTS phone (name text, phoneNumber text)")
    db.commit()
    cur.close()

def insert_data(name: str, phone_number: str):
    cur = db.cursor()
    cur.execute("INSERT INTO phone VALUES ('{name}', '{phone_number}')".format(name=name, phone_number=phone_number))
    db.commit()
    cur.close()

def get_data():
    cur = db.cursor()
    data = cur.execute("SELECT * FROM phone").fetchall()
    cur.close()
    return data

if __name__ == "__main__":
    create_table()
    insert_data("A", "00000")
    insert_data("B", "00001")
    print(get_data())

sqlite3를 이용해 테이블을 만들고 데이터 삽입 및 조회하는 간단한 코드입니다. 이 코드를 LLM에게 전달해서 실행한 결과는 다음과 같습니다.

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        tools = await load_mcp_tools(session)
        agent = create_react_agent(model, tools)

        with open('test.py', 'r') as f:
            code = f.read()
         
        result = await agent.ainvoke({"messages": f"파이썬 파일을 실행해. <code>{code}</code>"})

human: 파이썬 파일을 실행해. <code># test.py\nimport numpy as np\nimport sqlite3...insert_data("B", "00001")\nprint(get_data())</code> {}
ai:  {'function_call': {'name': 'execute_python_code', 'arguments': '{"code": "\\n#...print(get_data())\\n"}'}}
tool: {"stdout": [('A', '00000'), ('B', '00001')]\n, "result.stderr": , "result.returncode": 0} {}
ai: 파이썬 파일이 성공적으로 실행되었습니다. 결과는 `[('A', '00000'), ('B', '00001')]`입니다. {}

이미 phone이라는 테이블이 존재하고 있는 상황에서 IF NOT EXISTS를 빼버리고 일부러 에러를 일으킨다면 다음의 결과를 받아볼 수 있습니다.

human: 파이썬 파일을 실행해. <code># test.py...</code> {}
ai:  {'function_call': {'name': 'execute_python_code', 'arguments': '{"code": "\\n# test.py\\n...print(get_data())\\n"}'}}
tool: {"stdout": , "result.stderr": Traceback (most recent call last):
  File "/tmp/tmpklro9fgx.py", line 27, in <module>
    create_table()
  File "/tmp/tmpklro9fgx.py", line 10, in create_table
    cur.execute("CREATE TABLE phone (name text, phoneNumber text)")
sqlite3.OperationalError: table phone already exists, "result.returncode": 1} {}
ai: 파이썬 코드를 실행하는 동안 오류가 발생했습니다. 특히 'phone'이라는 테이블이 이미 존재하기 때문에 테이블을 만들 수 없습니다. {}

이처럼 stderr를 같이 리턴하고 있으므로 에러 메시지를 LLM이 파악할 수 있어 사용자에게 어떠한 에러로 실행되지 않았는지 설명하는 모습을 볼 수 있습니다.

따라서, 개발자는 MCP를 이용해 Agent 애플리케이션 개발시 Tool을 같이 개발할 필요가 없고 애플리케이션과 독립적으로 Tool을 모듈로서 개발할 수 있게 되었습니다. 사용자는 원하는 Tool을 쉽게 추가, 삭제할 수 있어 편의성을 제공받을 수 있었습니다. 물론 보안의 문제가 당연히 존재하지만 결국 해결될 문제라고 생각하고 있어 다음이 기대되는 기능입니다.

Reference

• 모델 컨텍스트 프로토콜 (MCP)