LangGraph 6 - MCP 도구 연계

1. 이번 단계에서 달라지는 것

지금까지 도구는 코드 안에 들어있었다. 이번에는 도구를 별도 프로세스(MCP 서버)로 분리하고, 표준 프로토콜로 통신하며 가져다 쓰는 구조를 만들었음.

2편에서 @tool 데코레이터로 만든 곱셈 함수는 에이전트와 같은 파일·같은 프로세스 안에 있었다. 이번에는 도구를 MCP(Model Context Protocol) 서버로 떼어내고, 에이전트는 그 서버에 접속해 도구 목록을 받아온 뒤 LangChain 도구로 변환해 LLM에 바인딩하였음.

• MCP 서버 — 도구의 구현체. 이름·인자·반환 스키마를 외부에 노출함. 에이전트와 분리된 별도 프로세스로 동작.
• MCP 클라이언트 — 서버에 접속해 도구 목록을 조회하고, 도구를 호출(JSON-RPC)하는 통로.
• 도구 어댑터 — MCP 도구를 LangChain StructuredTool로 변환. 이게 있어야 2~5편에서 써온 그래프에 그대로 끼울 수 있음.

2편이 도구를 "내 코드에 직접 넣은" 단계였다면, 이번엔 도구를 "외부 표준으로 빼낸" 단계다. 한 번 어댑터를 만들어 두면 어떤 MCP 서버든 같은 방식으로 붙는다.

---

2. MCP 서버 — 도구를 외부로 노출

FastMCP로 6개 도구를 구성함. 계산·시간·메모 CRUD·RAG 자리까지, LLM이 가져다 쓸 외부 기능들을 한 파일에 모았다.

from mcp.server.fastmcp import FastMCP

mcp = FastMCP('6ToolsMPCServer')

note_memory = dict()   # 메모 저장용 인메모리 저장소(데모용)

`FastMCP`로 서버 객체 하나를 만들고, 이후 `@mcp.tool()`을 붙인 함수들이 자동으로 도구로 등록된다. `note_memory`는 CRUD 도구가 공유하는 서버 측 임시 저장소.

2.1 계산·시간 도구

@mcp.tool()
def add(a: float, b: float) -> str:
    '''두 수를 더하는 계산기
    Args:
        a : 첫 번째 수치
        b : 두 번째 수치
    Returns
        계산 결과
    '''
    result = a + b
    logger.info(f'Tool 1 add 호출 : {a} + {b} = {result}')
    return f'계산결과 : {a}+{b} = {result}'

@mcp.tool()
def get_time() -> str:
    '''서버측 현재 시간을 조회
    Returns
        현재 시간 문자열
    '''
    cur_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return f'현재 시간: {cur_time}'

②편 111글의 `@tool`과 판박이임. 데코레이터로 함수를 도구로 등록하고, docstring·타입힌트가 곧 LLM이 읽는 설명서가 된다. 차이는 이 함수가 에이전트가 아니라 "서버 측"에 있다는 것뿐. `get_time`처럼 인자 없는 도구도 그대로 등록된다.

2.2 메모 CRUD 도구

상태를 가진 도구 묶음. 저장(Create/Update)·조회(Read)·삭제(Delete)를 note_memory 딕셔너리 위에서 처리함.

@mcp.tool()
def save_note(note_id: str, note_content: str) -> str:
    '''메모 저장
    Args:
        note_id: 메모의 고유 ID, 키값
        note_content: 메모 내용
    Returns
        저장 완료 메세지
    '''
    if not note_id or not note_content:        # 방어 코드
        return "Fail: 필수 파라미터 누락"
    note_memory[note_id] = {
        "content"    : note_content,
        "created_at" : datetime.now().isoformat()
    }
    return f"메모 저장 완료 {note_id}"

필수 인자 누락을 먼저 막고(방어 코드), 통과하면 딕셔너리에 저장한다. 같은 `note_id`로 다시 부르면 덮어쓰므로 Create와 Update를 겸함.

@mcp.tool()
def list_note() -> str:
    '''저장된 모든 메모 목록 조회'''
    if not note_memory:
        return "저장된 메모 없음"
    notes = "\n".join([
        f'- id: {note_id},  content: {value["content"]}'
        for note_id, value in note_memory.items()
    ])
    return f'저장된 모든 메모:\n{notes}'

@mcp.tool()
def delete_note(note_id: str) -> str:
    '''특정 메모 삭제
    Args:
        note_id: 메모의 고유 ID, 키값
    '''
    if note_id in note_memory:
        del note_memory[note_id]
        return f'메모 삭제 완료! {note_id}'
    else:
        return f'메모 삭제 실패! {note_id}로 구분되는 메모가 없음'

`list_note`는 빈 저장소를 먼저 거르고, 존재하면 메모를 한 덩어리 문자열로 묶어 반환한다. `delete_note`는 키 존재 여부를 확인한 뒤 `del`로 지운다. 반환값이 전부 문자열인 점이 포인트 — LLM이 읽을 자연어 결과로 돌려준다.

2.3 RAG 자리 — 스텁

@mcp.tool()
def rag_search():
    pass   # 검색증강 -> ④편 RAG와 연결 시 구현 예정

빈 도구지만 자리를 잡아둠. 서버 측 도구로 검색증강을 노출하려는 다음 단계(④편 113글 RAG)의 연결점이다.

2.4 로깅은 반드시 stderr로 — STDIO 통신의 함정

STDIO 모드에서 stdout은 JSON-RPC 메시지 전용 채널임. print/로그를 stdout으로 흘리면 프로토콜이 깨짐.

import sys, logging

logging.basicConfig(
    level  = logging.INFO,
    format = '[MCP Server] %(levelname)s: %(message)s',
    stream = sys.stderr          # stdout 아님. 반드시 stderr
)

if __name__ == '__main__':
    mcp.run(transport="stdio")   # 표준 입출력으로 통신

서버와 클라이언트는 stdin/stdout으로 JSON-RPC를 주고받는다. 로그를 stdout에 찍으면 클라이언트가 그걸 RPC 응답으로 오해해 파싱이 깨진다. 그래서 사람이 볼 메시지는 전부 stderr로 분리함.

데이터엔지니어 감각으로 보면 stdout = 데이터 채널, stderr = 관측 채널의 분리다. 파이프라인에서 로그와 페이로드를 섞지 않는 것과 같은 원칙.

---

3. MCP 클라이언트 — 접속과 도구 호출

서버를 자식 프로세스로 띄우고, 세션을 열어 도구 목록을 받아 호출까지 해봄.

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

server_params = StdioServerParameters(
    command = sys.executable,   # 현재 파이썬으로
    args    = ['server.py'],    # 서버 스크립트를 띄움
    env     = None,
)

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()              # 1. 세션 초기화
        res = await session.list_tools()        # 2. 도구 목록 조회
        self.tools = res.tools

`stdio_client`가 서버 프로세스를 띄우고 입출력 스트림(read/write)을 돌려준다. 그 스트림으로 `ClientSession`을 만들고 초기화하면, `list_tools()`로 서버가 노출한 도구 6개를 받아온다.

각 도구는 name, description, inputSchema(JSON Schema)를 들고 옴. 인자가 있는 도구는 스키마에서 매개변수명을 뽑아 확인할 수 있음.

for tool in self.tools:
    if hasattr(tool, "inputSchema") and tool.inputSchema:
        props = tool.inputSchema.get('properties', {})
        if props:
            print(f'입력 : {", ".join(props.keys())}')   # 예: a, b

`inputSchema`가 비어 있을 수 있으니 `hasattr` + 기본값 `{}`로 방어한다. 이 스키마가 다음 단계 변환의 원재료다.

3.1 함정 — 결과는 문자열이 아니라 content 블록

`call_tool` 반환값을 그냥 출력하면 객체가 찍힘. 결과는 content 리스트 안에 들어 있어, text 필드를 꺼내야 함.

async def call_tool(self, session, tool_name: str, arguments: dict):
    result = await session.call_tool(tool_name, arguments)
    if hasattr(result, 'content') and result.content:
        for content in result.content:
            if hasattr(content, 'text'):
                print(f' 결과 : \n{content.text}\n')   # 여기서 text를 꺼냄
            else:
                print(f' 결과 : \n{content}\n')
    else:
        print(f' 결과 : \n{result}\n')
    return result

MCP 응답은 평범한 문자열이 아니라 content 블록의 리스트다. 텍스트·이미지 등 여러 타입을 담을 수 있는 구조라, `content.text`로 명시적으로 꺼내야 사람이 읽을 결과가 나온다. 이 패턴은 클라이언트와 다음 장 어댑터 양쪽에 똑같이 등장함.

도구 호출 시나리오는 다음과 같이 굴렸음.

await self.call_tool(session, "add", {"a": 100, "b": 5})
await self.call_tool(session, "get_time", {})
await self.call_tool(session, "save_note", {"note_id": "de-001", "note_content": "MCP 1"})
await self.call_tool(session, "save_note", {"note_id": "de-002", "note_content": "MCP 2"})
await self.call_tool(session, "list_note", {})
await self.call_tool(session, "delete_note", {"note_id": "de-002"})
await self.call_tool(session, "list_note", {})

저장 2건 → 목록 → 1건 삭제 → 목록으로, CRUD가 서버 측 note_memory에 실제로 반영되는지 확인. 인자 없는 도구는 빈 딕셔너리 `{}`를 넘긴다.

여기까지는 사람이 직접 도구 이름과 인자를 적어 호출한 단계다. 다음 장부터는 이 호출을 LLM이 스스로 판단해 날리게 만든다. 그러려면 MCP 도구를 LangChain 도구로 바꿔야 한다.

---

4. 도구 어댑터 — MCP → LangChain 변환 (핵심)

MCP 도구를 LangChain `StructuredTool`로 자동 변환하는 어댑터. 이번 편에서 가장 손이 많이 간 부분임.

4.1 함정 1 — async with를 못 쓴다

세션·스트림을 여러 메서드(initialize / call_tool / cleanup)에서 계속 써야 하므로, with 블록으로 자동 종료시킬 수 없음. 컨텍스트 매니저를 수동으로 연다.

async def initialize(self):
    server_params = StdioServerParameters(
        command=sys.executable, args=[self.server_script], env=None
    )
    # with문은 블록을 벗어나면 세션을 닫아버림 -> 못 씀
    self._stdio_context = stdio_client(server_params)
    stdio_tuple = await self._stdio_context.__aenter__()   # 직접 진입
    self.read_stream, self.write_stream = stdio_tuple

    self.session = ClientSession(self.read_stream, self.write_stream)
    await self.session.__aenter__()      # 세션도 수동 활성화
    await self.session.initialize()

    res = await self.session.list_tools()
    self.mcp_tools = res.tools
    return self

3장 client.py처럼 `async with` 한 블록에서 다 끝내면 편하지만, 그건 "접속→호출→종료"가 한 흐름일 때 얘기다. 어댑터는 호출 시점이 다 다르므로, `__aenter__`로 열고 나중에 cleanup에서 `__aexit__`로 닫는다.

async def cleanup(self):
    try:
        if self.session:
            await self.session.__aexit__(None, None, None)
    except Exception as e:
        print('세션 종료 에러', e)
    try:
        if self._stdio_context:
            await self._stdio_context.__aexit__(None, None, None)
    except Exception as e:
        print('입출력 스트림 종료 에러', e)

수동으로 연 자원은 수동으로 닫아야 한다. 세션 → 스트림 순서로 `__aexit__`을 호출하고, 각각 try로 감싸 한쪽 종료 실패가 다른 쪽을 막지 않게 한다.

4.2 함정 2 — 클로저 늦은 바인딩

도구 호출 함수를 반복문 안에서 그냥 만들면, 모든 함수가 마지막 도구 이름만 참조하게 됨. 한 겹 감싸 독립시킨다.

def create_tool_func(name: str):          # name을 인자로 고정(캡처)
    async def async_tool_func(**kwargs) -> str:
        return await self.call_tool(name, kwargs)
    return async_tool_func

tool_func = create_tool_func(tool_name)   # 도구마다 독립 함수 생성

이렇게 안 감싸고 루프 변수 `tool_name`을 직접 참조하면, 파이썬 클로저 특성상 모든 함수가 루프 종료 후의 마지막 값을 본다. add를 부르려 했는데 delete_note가 불리는 식. `create_tool_func(name)`으로 값을 인자에 박아 끊어준다.

어댑터 측 call_tool도 결과에서 text를 꺼내는 건 동일함.

async def call_tool(self, tool_name: str, arguments: dict) -> str:
    result = await self.session.call_tool(tool_name, arguments)
    if hasattr(result, 'content') and result.content:
        for content in result.content:
            if hasattr(content, 'text'):
                return content.text
    return str(result)

3.1장의 content 블록 패턴을 LangChain 도구가 기대하는 "문자열 반환" 형태로 정리한 것. LLM에는 결국 이 문자열이 도구 실행 결과로 전달된다.

4.3 스키마 → pydantic 동적 모델

MCP의 JSON Schema를 읽어 pydantic 모델을 런타임에 생성함. 타입별로 매핑하고, required면 기본값을 강제(...)로 둔다.

from pydantic import BaseModel, Field, create_model

tool_params = {}
props    = mcp_tool.inputSchema.get('properties', {})
required = mcp_tool.inputSchema.get('required',   [])

for param_name, param_info in props.items():
    param_type = param_info.get('type', 'string')
    param_desc = param_info.get('description', '')
    default = ... if param_name in required else None   # ... = 필수

    if   param_type == 'number':  py_type = float
    elif param_type == 'integer': py_type = int
    elif param_type == 'boolean': py_type = bool
    else:                         py_type = str

    tool_params[param_name] = (py_type, Field(default=default, description=param_desc))

if tool_params:
    args_schema = create_model(f'{tool_name}_args', **tool_params)
else:
    class EmptyArgs(BaseModel): pass     # get_time처럼 인자 없는 도구
    args_schema = EmptyArgs

MCP는 JSON으로 느슨하게 스키마를 주고, LangChain은 pydantic으로 엄격한 타입 검증을 건다. 그 사이를 메우는 게 이 변환. `create_model`로 도구 개수만큼 인자 스키마 클래스를 찍어내고, 인자 없는 도구는 빈 `EmptyArgs`로 처리한다.

from langchain_core.tools import StructuredTool

try:
    langchain_tool = StructuredTool.from_function(
        func        = tool_func,
        name        = tool_name,
        description = tool_description,
        args_schema = args_schema,
        coroutine   = tool_func,      # 비동기 함수임을 명시
    )
except Exception:
    langchain_tool = StructuredTool.from_function(   # coroutine 미지원 버전 폴백
        func        = tool_func,
        name        = tool_name,
        description = tool_description,
        args_schema = args_schema,
    )

실행 함수 + 이름 + 설명 + 인자 스키마. 이 네 가지가 곧 ②편에서 LLM이 도구를 고를 때 보던 "판단 재료"다. `coroutine` 인자를 못 받는 버전을 대비해 try/except로 폴백을 둠. 이제 어떤 MCP 서버를 붙여도 이 어댑터 한 벌이면 도구 6개가 전부 LangChain 도구로 변환된다.

---

5. 에이전트에 결합 — 그래프는 그대로

⑤편 114글에서 만든 그래프 구조(agent 노드 + tools 노드 + 조건부 엣지)를 그대로 재사용함. 도구 출처만 로컬 `@tool`에서 MCP로 바뀐다.

from langgraph.graph import StateGraph, MessagesState, END
from langgraph.prebuilt import ToolNode

# MCP 도구 로드 -> LangChain 도구로 변환
self.mcp_adapter = MCPToolAdapter('server.py')
await self.mcp_adapter.initialize()
self.tools = self.mcp_adapter.create_langchain_tools()

llm_with_tools = self.llm.bind_tools(self.tools)   # ②편과 동일

초기화 순서가 곧 의존 순서다. MCP 서버에 붙어 도구를 받아오고(initialize) → LangChain 도구로 변환하고(create_langchain_tools) → LLM에 바인딩(bind_tools). 이 셋이 어긋나면 LLM이 도구를 못 본다.

workflow = StateGraph(MessagesState)
workflow.add_node('agent', call_model)
workflow.add_node('tools', ToolNode(self.tools))

workflow.set_entry_point('agent')
workflow.add_edge('tools', 'agent')           # 도구 실행 후 다시 판단
workflow.add_conditional_edges('agent', should_continue, {
    'tools': 'tools',
    'end'  : END
})
self.graph = workflow.compile()

`bind_tools` → `ToolNode` → 조건부 엣지 패턴은 ②편 111글, 분기 함수가 `tool_calls`를 보는 건 ④편 113글 그대로다.

def should_continue(state: MessagesState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        return "tools"      # 도구 호출이 있으면 -> tools 노드
    return "end"            # 없으면 -> 종료

바뀐 건 `self.tools`를 만드는 출처뿐. 그래프 입장에선 도구가 로컬 함수인지 MCP 원격 도구인지 구분하지 않는다.

어댑터가 변환을 책임지므로, 앞서 4편에 걸쳐 쌓은 그래프 자산을 한 줄도 안 고치고 재사용했다. 추상화가 잘 되면 교체 비용이 0에 수렴한다.

5.1 troubleshooting — 첫 작성본이 안 돌던 이유

초안에서 막혔다가 정리하며 잡은 실수들.

• import명 불일치 — 어댑터가 정의한 클래스는 MCPToolAdapter인데 MCPClient로 import해 ImportError. 클래스명을 맞춰 해결.
• await 누락 — initialize()가 비동기인데 await 없이 호출해 코루틴 객체만 생기고 실제 연결이 안 됨. await agent.initialize()로 수정.
• 빈 도구로 bind — MCP 로드를 건너뛴 채 bind_tools([])를 호출해 LLM이 도구를 못 봄. 초기화 순서를 MCP 로드 → 변환 → 바인딩으로 고정.

---

6. 실행 흐름

"10과 5를 더해줘" 같은 자연어를 넣으면, LLM이 add 도구가 필요하다 판단 → tools 노드에서 MCP 서버 호출 → 결과를 받아 다시 판단 → 최종 답변.

async def process_query(self, user_input: str) -> str:
    messages = [HumanMessage(content=user_input)]
    result = self.graph.invoke({"messages": messages})
    final_message = result["messages"][-1]
    return final_message.content if hasattr(final_message, 'content') else str(final_message)

사용자 입력을 HumanMessage로 감싸 그래프에 넣고, 누적된 messages의 마지막 응답을 꺼낸다. 그 사이 agent ↔ tools 순환은 LangGraph가 알아서 돌린다.

사용자 입력	LLM 판단	실행 경로
"10과 5 더해줘"	add 필요	agent → tools(add) → agent → 응답
"지금 몇 시야?"	get_time 필요	agent → tools(get_time) → agent → 응답
"안녕?"	도구 불필요	agent → 응답 (tools 건너뜀)

---

요약

도구를 MCP 서버로 외부화하고, 어댑터로 LangChain 도구 6개로 변환해 ⑤편 그래프에 그대로 결합했음. 자연어 한 줄이 MCP 도구 호출로 이어지는 데까지 묶었다.

항목	2~5편 (로컬 도구)	6편 (MCP 연계)
도구 위치	에이전트와 같은 프로세스	별도 MCP 서버 프로세스
도구 등록	@tool 데코레이터	서버에서 list_tools로 수신
그래프 결합	함수를 직접 바인딩	어댑터가 StructuredTool로 변환
통신	함수 호출	JSON-RPC (STDIO)
결과 형태	반환값 그대로	content 블록에서 text 추출

1~5편이 "한 그래프 안에서 능력을 키운" 과정이었다면, ⑥편은 도구를 그래프 밖 표준으로 빼냈다. 핵심은 어댑터다 — MCP의 느슨한 JSON 스키마와 LangChain의 엄격한 pydantic 검증 사이를 메우는 한 벌만 갖추면, 세상의 어떤 MCP 도구든 기존 에이전트에 그대로 꽂힌다. 에이전트가 복잡해진 만큼 "어디서 느리고 얼마를 쓰는지" 관측이 필요해지는데, 그건 다음 편 LangSmith에서 다룬다.